File size: 2,033 Bytes
ea6f215
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import requests
import time
import subprocess
import os
import signal
import sys

def test_api():
    print("๐Ÿš€ Starting FastAPI server for deployment testing...")
    
    # Start the server in a subprocess
    server_process = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "fastapi_app.main:app", "--host", "127.0.0.1", "--port", "8000"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        preexec_fn=os.setsid
    )
    
    time.sleep(5) # Give it time to load the model
    
    try:
        # 1. Health Check
        print("๐Ÿ” Checking API health...")
        health_resp = requests.get("http://127.0.0.1:8000/health")
        print(f"Health Response: {health_resp.json()}")
        assert health_resp.status_code == 200
        
        # 2. Prediction Test
        print("๐Ÿ”ฎ Testing Sales Prediction endpoint...")
        payload = {
            "Store": 1,
            "Date": "2015-09-17",
            "Promo": 1,
            "StateHoliday": "0",
            "SchoolHoliday": 0
        }
        pred_resp = requests.post("http://127.0.0.1:8000/predict", json=payload)
        
        if pred_resp.status_code == 200:
            result = pred_resp.json()
            print(f"โœ… Prediction Successful!")
            print(f"   Store: {result['Store']}")
            print(f"   Date: {result['Date']}")
            print(f"   Predicted Sales: โ‚ฌ{result['PredictedSales']:.2f}")
        else:
            print(f"โŒ Prediction Failed: {pred_resp.text}")
            
        assert pred_resp.status_code == 200
        
    except Exception as e:
        print(f"๐Ÿ’ฅ Error during API test: {str(e)}")
        # Print server logs if it failed
        out, err = server_process.communicate(timeout=1)
        print(f"Server STDOUT: {out.decode()}")
        print(f"Server STDERR: {err.decode()}")
        raise e
    finally:
        print("๐Ÿ›‘ Shutting down server...")
        os.killpg(os.getpgid(server_process.pid), signal.SIGTERM)

if __name__ == "__main__":
    test_api()