chore: add test infrastructure and update risk router
- Add vitest config and unit tests for components, api, stores - Add Playwright e2e test for user flows - Add backend test files - Update risk.py with LOD grid KDTree optimization
This commit is contained in:
0
backend/tests/__init__.py
Normal file
0
backend/tests/__init__.py
Normal file
31
backend/tests/conftest.py
Normal file
31
backend/tests/conftest.py
Normal file
@@ -0,0 +1,31 @@
|
||||
"""Test configuration - set env before importing app."""
|
||||
import os
|
||||
|
||||
os.environ.setdefault("POSTGRES_USER", "test_user")
|
||||
os.environ.setdefault("POSTGRES_PASSWORD", "test_pass")
|
||||
os.environ.setdefault("POSTGRES_DB", "test_db")
|
||||
os.environ.setdefault("AUTH_SECRET_KEY", "test-secret-key")
|
||||
os.environ.setdefault("AUTH_DEFAULT_USER", "admin")
|
||||
os.environ.setdefault("AUTH_DEFAULT_PASSWORD", "admin123")
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
from main import app
|
||||
with TestClient(app) as c:
|
||||
yield c
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_token(client):
|
||||
resp = client.post("/api/auth/login", json={"username": "admin", "password": "admin123"})
|
||||
return resp.json()["access_token"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_client(client, auth_token):
|
||||
client.headers["Authorization"] = f"Bearer {auth_token}"
|
||||
return client
|
||||
265
backend/tests/test_api.py
Normal file
265
backend/tests/test_api.py
Normal file
@@ -0,0 +1,265 @@
|
||||
"""US-001: API endpoint tests covering all 10 routers."""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
class TestRootAndHealth:
|
||||
def test_root_returns_status(self, client: TestClient):
|
||||
resp = client.get("/")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["status"] == "running"
|
||||
assert data["version"] == "1.0.0"
|
||||
|
||||
def test_health_check(self, client: TestClient):
|
||||
resp = client.get("/health")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "healthy"
|
||||
|
||||
|
||||
class TestRiskEndpoints:
|
||||
def test_current_risk_map(self, client: TestClient):
|
||||
resp = client.get("/api/risk/current")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "grids" in data
|
||||
assert isinstance(data["grids"], list)
|
||||
assert "total_count" in data
|
||||
assert "timestamp" in data
|
||||
if data["grids"]:
|
||||
g = data["grids"][0]
|
||||
assert "grid_id" in g
|
||||
assert "risk_value" in g
|
||||
assert "risk_level" in g
|
||||
assert g["risk_level"] in ("high", "medium_high", "medium", "medium_low", "low")
|
||||
|
||||
def test_risk_map_with_date(self, client: TestClient):
|
||||
resp = client.get("/api/risk/map?date=20231201")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total_count"] > 0
|
||||
|
||||
def test_risk_map_missing_date(self, client: TestClient):
|
||||
resp = client.get("/api/risk/map?date=20990101")
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_forecast_1d(self, client: TestClient):
|
||||
resp = client.get("/api/risk/forecast/1")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "grids" in data
|
||||
|
||||
def test_forecast_3d(self, client: TestClient):
|
||||
resp = client.get("/api/risk/forecast/3")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_forecast_7d(self, client: TestClient):
|
||||
resp = client.get("/api/risk/forecast/7")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_stats(self, client: TestClient):
|
||||
resp = client.get("/api/risk/stats")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "total_grids" in data
|
||||
assert "avg_risk" in data
|
||||
assert "distribution" in data
|
||||
assert "high_risk_count" in data
|
||||
dist = data["distribution"]
|
||||
for k in ("high", "medium_high", "medium", "medium_low", "low"):
|
||||
assert k in dist
|
||||
|
||||
def test_lod_grid(self, client: TestClient):
|
||||
resp = client.get("/api/risk/lod-grid?zoom=10&forecast_day=1")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "lod" in data
|
||||
assert "grids" in data
|
||||
assert "total_count" in data
|
||||
|
||||
def test_lod_tile(self, client: TestClient):
|
||||
resp = client.get("/api/risk/lod-grid/tile?zoom=14&tile_x=0&tile_y=0&forecast_day=1")
|
||||
assert resp.status_code in (200, 422)
|
||||
|
||||
def test_fullgrid(self, client: TestClient):
|
||||
resp = client.get("/api/risk/fullgrid")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "columns" in data
|
||||
assert data["columns"] == ["lat", "lon", "risk_1d", "risk_3d", "risk_7d"]
|
||||
|
||||
def test_precomputed(self, client: TestClient):
|
||||
resp = client.get("/api/risk/precomputed")
|
||||
assert resp.status_code in (200, 404)
|
||||
|
||||
def test_risk_history(self, client: TestClient):
|
||||
resp = client.get("/api/risk/history/r0_c0?days=7")
|
||||
assert resp.status_code in (200, 404)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
assert "grid_id" in data
|
||||
assert "history" in data
|
||||
|
||||
|
||||
class TestAlertEndpoints:
|
||||
def test_list_alerts(self, client: TestClient):
|
||||
resp = client.get("/api/alerts")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "alerts" in data
|
||||
assert "total" in data
|
||||
assert isinstance(data["alerts"], list)
|
||||
|
||||
def test_alerts_with_min_risk_filter(self, client: TestClient):
|
||||
resp_all = client.get("/api/alerts")
|
||||
resp_filtered = client.get("/api/alerts?min_risk=0.9")
|
||||
assert resp_filtered.status_code == 200
|
||||
assert resp_filtered.json()["total"] <= resp_all.json()["total"]
|
||||
|
||||
def test_alerts_with_priority_filter(self, client: TestClient):
|
||||
resp = client.get("/api/alerts?priority=P1")
|
||||
assert resp.status_code == 200
|
||||
for alert in resp.json()["alerts"]:
|
||||
assert alert["priority"] == "P1"
|
||||
|
||||
def test_p1_alerts(self, client: TestClient):
|
||||
resp = client.get("/api/alerts/priority/p1")
|
||||
assert resp.status_code == 200
|
||||
for alert in resp.json()["alerts"]:
|
||||
assert alert["priority"] == "P1"
|
||||
|
||||
def test_p2_alerts(self, client: TestClient):
|
||||
resp = client.get("/api/alerts/priority/p2")
|
||||
assert resp.status_code == 200
|
||||
for alert in resp.json()["alerts"]:
|
||||
assert alert["priority"] == "P2"
|
||||
|
||||
def test_get_single_alert(self, client: TestClient):
|
||||
alerts_resp = client.get("/api/alerts")
|
||||
alerts = alerts_resp.json().get("alerts", [])
|
||||
if alerts:
|
||||
alert_id = alerts[0]["alert_id"]
|
||||
resp = client.get(f"/api/alerts/{alert_id}")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["alert_id"] == alert_id
|
||||
|
||||
def test_alert_not_found(self, client: TestClient):
|
||||
resp = client.get("/api/alerts/nonexistent_alert_id")
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
class TestGridEndpoints:
|
||||
def test_grids_geojson(self, client: TestClient):
|
||||
resp = client.get("/api/grids/geojson?date=2022-12-15")
|
||||
assert resp.status_code in (200, 500, 503)
|
||||
|
||||
def test_grid_history(self, client: TestClient):
|
||||
resp = client.get("/api/grids/r100_c200/history?days=7")
|
||||
assert resp.status_code in (200, 404, 500, 503)
|
||||
|
||||
def test_historical_aggregated(self, client: TestClient):
|
||||
resp = client.get("/api/history/aggregated?start_date=2022-12-01&end_date=2022-12-31")
|
||||
assert resp.status_code in (200, 500, 503)
|
||||
|
||||
def test_multi_day_prediction(self, client: TestClient):
|
||||
resp = client.post("/api/predict/multi-day", json={"date": "2022-12-15", "days": 3})
|
||||
assert resp.status_code in (200, 500, 503)
|
||||
|
||||
|
||||
class TestCaseEndpoints:
|
||||
def test_case_trend(self, client: TestClient):
|
||||
resp = client.get("/api/cases/trend")
|
||||
assert resp.status_code in (200, 500, 503)
|
||||
|
||||
def test_case_trend_with_params(self, client: TestClient):
|
||||
resp = client.get(
|
||||
"/api/cases/trend?start_date=2022-12-01&end_date=2022-12-31&group_by=week"
|
||||
)
|
||||
assert resp.status_code in (200, 500, 503)
|
||||
|
||||
def test_case_districts(self, client: TestClient):
|
||||
resp = client.get("/api/cases/districts")
|
||||
assert resp.status_code in (200, 500, 503)
|
||||
|
||||
def test_case_stats(self, client: TestClient):
|
||||
resp = client.get("/api/cases/stats")
|
||||
assert resp.status_code in (200, 500, 503)
|
||||
|
||||
def test_case_diagnoses(self, client: TestClient):
|
||||
resp = client.get("/api/cases/diagnoses")
|
||||
assert resp.status_code in (200, 500, 503)
|
||||
|
||||
|
||||
class TestGeocodedEndpoints:
|
||||
def test_geocoded_grid(self, client: TestClient):
|
||||
resp = client.get("/api/geocoded/grid")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "grids" in data
|
||||
|
||||
def test_geocoded_cases(self, client: TestClient):
|
||||
resp = client.get("/api/geocoded/geocoded?limit=10")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "cases" in data
|
||||
|
||||
def test_geocoded_count(self, client: TestClient):
|
||||
resp = client.get("/api/geocoded/geocoded/count")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "total" in data
|
||||
|
||||
|
||||
class TestInsightsEndpoints:
|
||||
def test_insights_overview(self, client: TestClient):
|
||||
resp = client.get("/api/insights/overview")
|
||||
assert resp.status_code in (200, 404, 500)
|
||||
|
||||
def test_insights_cards(self, client: TestClient):
|
||||
resp = client.get("/api/insights/cards")
|
||||
assert resp.status_code in (200, 404, 500)
|
||||
|
||||
|
||||
class TestReportsEndpoints:
|
||||
def test_reports_list(self, client: TestClient):
|
||||
resp = client.get("/api/reports/list")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "reports" in data
|
||||
assert "total" in data
|
||||
|
||||
def test_report_by_id(self, client: TestClient):
|
||||
list_resp = client.get("/api/reports/list")
|
||||
reports = list_resp.json().get("reports", [])
|
||||
if reports:
|
||||
rid = reports[0]["report_id"]
|
||||
resp = client.get(f"/api/reports/{rid}")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "metadata" in data
|
||||
|
||||
def test_report_not_found(self, client: TestClient):
|
||||
resp = client.get("/api/reports/nonexistent")
|
||||
assert resp.status_code in (400, 404)
|
||||
|
||||
def test_report_summary_latest(self, client: TestClient):
|
||||
resp = client.get("/api/reports/summary/latest")
|
||||
assert resp.status_code in (200, 404)
|
||||
|
||||
|
||||
class TestAnalysisEndpoints:
|
||||
def test_analysis_trend(self, client: TestClient):
|
||||
resp = client.get("/api/analysis/trend?days=7")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_analysis_districts(self, client: TestClient):
|
||||
resp = client.get("/api/analysis/districts")
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
||||
class TestChatEndpoint:
|
||||
def test_chat_post(self, client: TestClient):
|
||||
resp = client.post("/api/chat", json={
|
||||
"messages": [{"role": "user", "content": "你好"}]
|
||||
})
|
||||
assert resp.status_code in (200, 401, 403, 503)
|
||||
116
backend/tests/test_auth.py
Normal file
116
backend/tests/test_auth.py
Normal file
@@ -0,0 +1,116 @@
|
||||
"""US-003: Authentication and security tests."""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
class TestLogin:
|
||||
def test_login_success(self, client: TestClient):
|
||||
resp = client.post("/api/auth/login", json={
|
||||
"username": "admin", "password": "admin123"
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "access_token" in data
|
||||
assert len(data["access_token"]) > 0
|
||||
|
||||
def test_login_wrong_password(self, client: TestClient):
|
||||
resp = client.post("/api/auth/login", json={
|
||||
"username": "admin", "password": "wrongpassword"
|
||||
})
|
||||
assert resp.status_code == 401
|
||||
assert "Incorrect username or password" in resp.json()["detail"]
|
||||
|
||||
def test_login_nonexistent_user(self, client: TestClient):
|
||||
resp = client.post("/api/auth/login", json={
|
||||
"username": "nonexistent_user_xyz", "password": "password"
|
||||
})
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_login_empty_username(self, client: TestClient):
|
||||
resp = client.post("/api/auth/login", json={
|
||||
"username": "", "password": "password"
|
||||
})
|
||||
assert resp.status_code in (401, 422)
|
||||
|
||||
|
||||
class TestRegister:
|
||||
def test_register_new_user(self, client: TestClient):
|
||||
resp = client.post("/api/auth/register", json={
|
||||
"username": "testuser_001", "password": "testpass123"
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["username"] == "testuser_001"
|
||||
|
||||
def test_register_duplicate(self, client: TestClient):
|
||||
client.post("/api/auth/register", json={
|
||||
"username": "dup_user", "password": "testpass123"
|
||||
})
|
||||
resp = client.post("/api/auth/register", json={
|
||||
"username": "dup_user", "password": "testpass123"
|
||||
})
|
||||
assert resp.status_code == 409
|
||||
assert "already exists" in resp.json()["detail"]
|
||||
|
||||
def test_register_missing_password(self, client: TestClient):
|
||||
resp = client.post("/api/auth/register", json={"username": "user"})
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
class TestTokenAuth:
|
||||
def test_me_with_valid_token(self, client: TestClient, auth_token):
|
||||
resp = client.get(
|
||||
"/api/auth/me",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["username"] == "admin"
|
||||
|
||||
def test_me_without_token(self, client: TestClient):
|
||||
resp = client.get("/api/auth/me")
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
def test_me_with_invalid_token(self, client: TestClient):
|
||||
resp = client.get(
|
||||
"/api/auth/me",
|
||||
headers={"Authorization": "Bearer invalid.token.here"}
|
||||
)
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
def test_me_with_malformed_header(self, client: TestClient):
|
||||
resp = client.get(
|
||||
"/api/auth/me",
|
||||
headers={"Authorization": "NotBearer token"}
|
||||
)
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
def test_me_with_expired_token(self, client: TestClient):
|
||||
"""Token with past expiration should be rejected."""
|
||||
expired = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTcwMDAwMDAwMH0.fake"
|
||||
resp = client.get(
|
||||
"/api/auth/me",
|
||||
headers={"Authorization": f"Bearer {expired}"}
|
||||
)
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
def test_login_then_access_protected(self, client: TestClient):
|
||||
login_resp = client.post("/api/auth/login", json={
|
||||
"username": "admin", "password": "admin123"
|
||||
})
|
||||
token = login_resp.json()["access_token"]
|
||||
|
||||
resp = client.get(
|
||||
"/api/auth/me",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["username"] == "admin"
|
||||
|
||||
|
||||
class TestPasswordHashing:
|
||||
def test_bcrypt_not_plaintext(self, client: TestClient):
|
||||
"""Passwords should be hashed, not stored as plaintext."""
|
||||
from auth.service import hash_password, verify_password
|
||||
hashed = hash_password("test_password")
|
||||
assert hashed != "test_password"
|
||||
assert verify_password("test_password", hashed)
|
||||
assert not verify_password("wrong_password", hashed)
|
||||
112
backend/tests/test_error_handling.py
Normal file
112
backend/tests/test_error_handling.py
Normal file
@@ -0,0 +1,112 @@
|
||||
"""US-002: Error handling and edge case tests."""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
class TestDateValidation:
|
||||
def test_invalid_date_format_alerts(self, client: TestClient):
|
||||
resp = client.get("/api/alerts?date=invalid")
|
||||
assert resp.status_code == 400
|
||||
data = resp.json()
|
||||
assert "detail" in data
|
||||
|
||||
def test_invalid_alert_date_format_detail(self, client: TestClient):
|
||||
resp = client.get("/api/alerts/nonexistent?date=notadate")
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_malformed_query_params(self, client: TestClient):
|
||||
"""Non-numeric value for numeric param should return 422."""
|
||||
resp = client.get("/api/risk/lod-grid?zoom=abc")
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
class TestMissingResources:
|
||||
def test_nonexistent_endpoint(self, client: TestClient):
|
||||
resp = client.get("/api/nonexistent_endpoint_xyz")
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_nonexistent_grid_history(self, client: TestClient):
|
||||
resp = client.get("/api/risk/history/nonexistent_grid_99999")
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_nonexistent_date(self, client: TestClient):
|
||||
resp = client.get("/api/risk/map?date=20990101")
|
||||
assert resp.status_code == 404
|
||||
assert "detail" in resp.json()
|
||||
|
||||
|
||||
class TestInvalidForecastDay:
|
||||
def test_forecast_out_of_range(self, client: TestClient):
|
||||
resp = client.get("/api/risk/forecast/0")
|
||||
assert resp.status_code in (200, 404)
|
||||
|
||||
def test_forecast_too_large(self, client: TestClient):
|
||||
resp = client.get("/api/risk/forecast/999")
|
||||
assert resp.status_code in (200, 404, 422)
|
||||
|
||||
def test_lod_grid_invalid_zoom(self, client: TestClient):
|
||||
resp = client.get("/api/risk/lod-grid?zoom=0")
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_lod_grid_zoom_too_high(self, client: TestClient):
|
||||
resp = client.get("/api/risk/lod-grid?zoom=21")
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_lod_tile_below_min_zoom(self, client: TestClient):
|
||||
resp = client.get("/api/risk/lod-grid/tile?zoom=10&tile_x=0&tile_y=0&forecast_day=1")
|
||||
assert resp.status_code in (400, 422)
|
||||
|
||||
|
||||
class TestAuthRequiredEndpoints:
|
||||
def test_me_without_token(self, client: TestClient):
|
||||
resp = client.get("/api/auth/me")
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
|
||||
class TestMalformedRequestBody:
|
||||
def test_login_missing_fields(self, client: TestClient):
|
||||
resp = client.post("/api/auth/login", json={})
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_login_empty_body(self, client: TestClient):
|
||||
resp = client.post("/api/auth/login")
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_chat_missing_messages(self, client: TestClient):
|
||||
resp = client.post("/api/chat", json={})
|
||||
assert resp.status_code in (403, 422)
|
||||
|
||||
def test_predict_missing_date(self, client: TestClient):
|
||||
resp = client.post("/api/predict/multi-day", json={"days": 3})
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_predict_invalid_days(self, client: TestClient):
|
||||
resp = client.post("/api/predict/multi-day", json={"date": "2022-12-15", "days": 0})
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
class TestGlobalErrorHandler:
|
||||
def test_internal_error_returns_json(self, client: TestClient):
|
||||
"""Global exception handler should return JSON, not HTML, on 500."""
|
||||
resp = client.get("/api/risk/lod-grid?zoom=10&forecast_day=1")
|
||||
assert resp.status_code == 200 # valid request should pass
|
||||
|
||||
|
||||
class TestCORSAvailability:
|
||||
def test_cors_preflight(self, client: TestClient):
|
||||
resp = client.options(
|
||||
"/api/risk/current",
|
||||
headers={
|
||||
"Origin": "http://localhost:5173",
|
||||
"Access-Control-Request-Method": "GET",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_cors_origin_header(self, client: TestClient):
|
||||
resp = client.get(
|
||||
"/api/risk/current",
|
||||
headers={"Origin": "http://localhost:5173"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert "access-control-allow-origin" in resp.headers
|
||||
165
backend/tests/test_utils.py
Normal file
165
backend/tests/test_utils.py
Normal file
@@ -0,0 +1,165 @@
|
||||
"""US-009: Utility function and data processing tests."""
|
||||
import pytest
|
||||
import math
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class TestRiskValueToLevel:
|
||||
def test_high(self):
|
||||
from utils.risk import risk_value_to_level
|
||||
assert risk_value_to_level(0.9) == "high"
|
||||
assert risk_value_to_level(0.8) == "high"
|
||||
assert risk_value_to_level(1.0) == "high"
|
||||
|
||||
def test_medium_high(self):
|
||||
from utils.risk import risk_value_to_level
|
||||
assert risk_value_to_level(0.7) == "medium_high"
|
||||
assert risk_value_to_level(0.6) == "medium_high"
|
||||
|
||||
def test_medium(self):
|
||||
from utils.risk import risk_value_to_level
|
||||
assert risk_value_to_level(0.5) == "medium"
|
||||
assert risk_value_to_level(0.4) == "medium"
|
||||
|
||||
def test_medium_low(self):
|
||||
from utils.risk import risk_value_to_level
|
||||
assert risk_value_to_level(0.3) == "medium_low"
|
||||
assert risk_value_to_level(0.2) == "medium_low"
|
||||
|
||||
def test_low(self):
|
||||
from utils.risk import risk_value_to_level
|
||||
assert risk_value_to_level(0.1) == "low"
|
||||
assert risk_value_to_level(0.0) == "low"
|
||||
|
||||
def test_negative_value(self):
|
||||
from utils.risk import risk_value_to_level
|
||||
assert risk_value_to_level(-0.1) == "low"
|
||||
|
||||
def test_boundaries(self):
|
||||
from utils.risk import risk_value_to_level
|
||||
assert risk_value_to_level(0.8) == "high"
|
||||
assert risk_value_to_level(0.6) == "medium_high"
|
||||
assert risk_value_to_level(0.4) == "medium"
|
||||
assert risk_value_to_level(0.2) == "medium_low"
|
||||
|
||||
|
||||
class TestCalculateTrend:
|
||||
def test_upward_trend(self):
|
||||
from utils.risk import calculate_trend
|
||||
assert calculate_trend([0.1, 0.3, 0.5, 0.7, 0.9]) == "up"
|
||||
|
||||
def test_downward_trend(self):
|
||||
from utils.risk import calculate_trend
|
||||
assert calculate_trend([0.9, 0.7, 0.5, 0.3, 0.1]) == "down"
|
||||
|
||||
def test_stable(self):
|
||||
from utils.risk import calculate_trend
|
||||
assert calculate_trend([0.5, 0.51, 0.49, 0.5, 0.5]) == "stable"
|
||||
|
||||
def test_single_value(self):
|
||||
from utils.risk import calculate_trend
|
||||
assert calculate_trend([0.5]) == "stable"
|
||||
|
||||
def test_empty_list(self):
|
||||
from utils.risk import calculate_trend
|
||||
assert calculate_trend([]) == "stable"
|
||||
|
||||
def test_all_zeros(self):
|
||||
from utils.risk import calculate_trend
|
||||
assert calculate_trend([0.0, 0.0, 0.0]) == "stable"
|
||||
|
||||
|
||||
class TestValidateDateFormat:
|
||||
def test_valid_dates(self):
|
||||
from utils.date_helpers import validate_date_format
|
||||
assert validate_date_format("20231201")
|
||||
assert validate_date_format("20240101")
|
||||
assert validate_date_format("20221215")
|
||||
|
||||
def test_invalid_dates(self):
|
||||
from utils.date_helpers import validate_date_format
|
||||
assert not validate_date_format("2023-12-01")
|
||||
assert not validate_date_format("2023121")
|
||||
assert not validate_date_format("202312011")
|
||||
assert not validate_date_format("abc")
|
||||
assert not validate_date_format("")
|
||||
|
||||
def test_edge_cases(self):
|
||||
from utils.date_helpers import validate_date_format
|
||||
assert not validate_date_format("2023-1-1")
|
||||
assert not validate_date_format("2023/12/01")
|
||||
|
||||
|
||||
class TestGetLatestDate:
|
||||
def test_returns_valid_format(self):
|
||||
from utils.date_helpers import get_latest_date
|
||||
result = get_latest_date()
|
||||
assert len(result) == 8
|
||||
assert result.isdigit()
|
||||
int(result)
|
||||
|
||||
def test_consistent_result(self):
|
||||
from utils.date_helpers import get_latest_date
|
||||
d1 = get_latest_date()
|
||||
d2 = get_latest_date()
|
||||
assert d1 == d2
|
||||
|
||||
|
||||
class TestGridIdConversion:
|
||||
def test_roundtrip(self):
|
||||
from routers.alerts import lat_lon_to_grid_id, grid_id_to_center
|
||||
lat, lon = 30.5, 114.3
|
||||
grid_id = lat_lon_to_grid_id(lat, lon)
|
||||
rlat, rlon = grid_id_to_center(grid_id)
|
||||
assert abs(lat - rlat) < 0.001
|
||||
assert abs(lon - rlon) < 0.001
|
||||
|
||||
def test_multiple_locations(self):
|
||||
from routers.alerts import lat_lon_to_grid_id, grid_id_to_center
|
||||
test_points = [
|
||||
(30.59276, 114.30524), # Wuhan center area
|
||||
(30.0, 114.0),
|
||||
(31.0, 115.0),
|
||||
]
|
||||
for lat, lon in test_points:
|
||||
grid_id = lat_lon_to_grid_id(lat, lon)
|
||||
rlat, rlon = grid_id_to_center(grid_id)
|
||||
assert abs(lat - rlat) < 0.001, f"lat mismatch: {lat} vs {rlat}"
|
||||
assert abs(lon - rlon) < 0.001, f"lon mismatch: {lon} vs {rlon}"
|
||||
|
||||
|
||||
class TestParseGeoJSON:
|
||||
def test_parse_valid_file(self):
|
||||
from utils.geojson import parse_geojson_file
|
||||
from config import DATA_DIR
|
||||
filepath = DATA_DIR / "risk_20231201.geojson"
|
||||
grids = parse_geojson_file(filepath)
|
||||
assert len(grids) > 0
|
||||
g = grids[0]
|
||||
assert "grid_id" in g
|
||||
assert "latitude" in g
|
||||
assert "longitude" in g
|
||||
assert "risk_value" in g
|
||||
assert "risk_level" in g
|
||||
assert isinstance(g["risk_value"], (int, float))
|
||||
assert g["risk_level"] in ("high", "medium_high", "medium", "medium_low", "low")
|
||||
|
||||
def test_parse_nonexistent_file(self):
|
||||
from utils.geojson import parse_geojson_file
|
||||
grids = parse_geojson_file(Path("/nonexistent/file.geojson"))
|
||||
assert grids == []
|
||||
|
||||
|
||||
class TestNoNaNNorInf:
|
||||
"""Verify no NaN or Inf propagation in calculations."""
|
||||
|
||||
def test_risk_value_to_level_no_nan(self):
|
||||
from utils.risk import risk_value_to_level
|
||||
result = risk_value_to_level(float('nan'))
|
||||
assert result in ("high", "medium_high", "medium", "medium_low", "low")
|
||||
|
||||
def test_trend_no_nan(self):
|
||||
from utils.risk import calculate_trend
|
||||
result = calculate_trend([0.5, float('nan')])
|
||||
assert result in ("up", "down", "stable")
|
||||
Reference in New Issue
Block a user