|
|
import sys |
|
|
import os |
|
|
from fastapi.testclient import TestClient |
|
|
|
|
|
|
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) |
|
|
|
|
|
from app.main import app, load_models |
|
|
|
|
|
|
|
|
|
|
|
print("βοΈ Forcing Model Load for Testing...") |
|
|
load_models() |
|
|
|
|
|
client = TestClient(app) |
|
|
|
|
|
def test_health_check(): |
|
|
"""Test if the API is alive and models are loaded.""" |
|
|
print("\nπ Testing Health Endpoint...") |
|
|
response = client.get("/health") |
|
|
|
|
|
|
|
|
print(f" Response: {response.json()}") |
|
|
|
|
|
assert response.status_code == 200 |
|
|
data = response.json() |
|
|
|
|
|
|
|
|
assert data["status"] == "healthy" |
|
|
print("β
Health Check Passed!") |
|
|
|
|
|
def test_prediction_endpoint(): |
|
|
"""Test the End-to-End Prediction pipeline with valid data.""" |
|
|
print("\nπ Testing Prediction Endpoint...") |
|
|
|
|
|
payload = { |
|
|
"tmmn": 290.0, "tmmx": 305.0, "rmin": 15.0, "rmax": 45.0, |
|
|
"vs": 6.5, "pr": 0.0, "erc": 50.0, |
|
|
"latitude": 34.0, "longitude": -118.0 |
|
|
} |
|
|
|
|
|
response = client.post("/predict", json=payload) |
|
|
|
|
|
|
|
|
if response.status_code != 200: |
|
|
print(f"β API Error: {response.json()}") |
|
|
|
|
|
assert response.status_code == 200 |
|
|
data = response.json() |
|
|
|
|
|
|
|
|
assert "burning_index_prediction" in data |
|
|
assert "risk_level_prediction" in data |
|
|
assert "cluster_zone" in data |
|
|
|
|
|
print("β
Prediction Logic Passed!") |
|
|
print(f" π₯ Predicted BI: {data['burning_index_prediction']}") |
|
|
print(f" β οΈ Risk Level: {data['risk_level_prediction']}") |
|
|
|
|
|
if __name__ == "__main__": |
|
|
try: |
|
|
test_health_check() |
|
|
test_prediction_endpoint() |
|
|
print("\nπ ALL API TESTS PASSED SUCCESSFULLY.") |
|
|
except AssertionError as e: |
|
|
print(f"\nβ TEST FAILED: Assertion Error") |
|
|
except Exception as e: |
|
|
print(f"\nβ CRITICAL ERROR: {e}") |