| """Production security contract tests.""" |
| import os |
| import sys |
| from unittest.mock import MagicMock |
|
|
| os.environ.setdefault("CEPHEUS_CLOUD", "1") |
| os.environ.setdefault("CEPHEUS_API_KEY", "test-key") |
| os.environ.pop("CEPHEUS_PRODUCTION", None) |
|
|
| for mod in ("google.adk", "google.adk.agents", "google.adk.runners", "google.adk.sessions"): |
| sys.modules.setdefault(mod, MagicMock()) |
|
|
| BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| if BACKEND_DIR not in sys.path: |
| sys.path.insert(0, BACKEND_DIR) |
|
|
| from fastapi.testclient import TestClient |
|
|
| import main |
|
|
| API_HEADERS = {"X-API-Key": "test-key"} |
| client = TestClient(main.app) |
|
|
|
|
| def test_cors_allows_delete(): |
| r = client.options( |
| "/site/signage-placements/s1", |
| headers={ |
| "Origin": "http://localhost:5173", |
| "Access-Control-Request-Method": "DELETE", |
| }, |
| ) |
| assert r.status_code == 200 |
| allow = r.headers.get("access-control-allow-methods", "") |
| assert "DELETE" in allow |
|
|
|
|
| def test_face_files_not_publicly_mounted(): |
| r = client.get("/face_database/test/person.jpg") |
| assert r.status_code == 404 |
|
|
|
|
| def test_secure_face_proxy_missing_returns_404(): |
| r = client.get("/files/face/test/person.jpg", headers=API_HEADERS) |
| assert r.status_code == 404 |
|
|
|
|
| def test_ws_ticket_requires_auth_when_enabled(): |
| os.environ["CEPHEUS_AUTH_DEV_MODE"] = "1" |
| try: |
| r = client.post("/auth/ws-ticket", headers=API_HEADERS) |
| assert r.status_code in (200, 401) |
| finally: |
| os.environ.pop("CEPHEUS_AUTH_DEV_MODE", None) |
|
|
|
|
| def test_fight_model_toggle_cloud_returns_bad_request(): |
| """Cloud stub engine does not support YOLO-based fight heuristic.""" |
| r = client.post("/ai/toggle/fight", headers=API_HEADERS) |
| assert r.status_code == 400 |
| detail = r.json().get("detail", "").lower() |
| assert "not available" in detail or "gpu" in detail |
|
|
|
|
| def test_ai_status_includes_capabilities(): |
| r = client.get("/ai/status", headers=API_HEADERS) |
| assert r.status_code == 200 |
| body = r.json() |
| assert "models" in body |
| assert "capabilities" in body |
| assert body["capabilities"]["fire"]["supported"] is False |
|
|
|
|
| def test_demo_login_rejected_without_demo_mode(): |
| os.environ.pop("CEPHEUS_AUTH_DEV_MODE", None) |
| os.environ.pop("CEPHEUS_JWT_SECRET", None) |
| try: |
| r = client.post("/auth/login", json={"username": "anyone", "password": "anything"}) |
| assert r.status_code == 503 |
| finally: |
| os.environ["CEPHEUS_AUTH_DEV_MODE"] = "1" |
|
|
|
|
| def test_strict_startup_requires_jwt_secret(): |
| import security_config |
|
|
| env = { |
| "CEPHEUS_CLOUD": "1", |
| "CEPHEUS_PRODUCTION": "1", |
| "CEPHEUS_API_KEY": "prod-key-not-default", |
| "CORS_ORIGINS": "https://example.com", |
| } |
| saved = {k: os.environ.get(k) for k in env} |
| os.environ.update(env) |
| os.environ.pop("CEPHEUS_AUTH_DEV_MODE", None) |
| os.environ.pop("CEPHEUS_JWT_SECRET", None) |
| try: |
| try: |
| security_config.validate_startup() |
| assert False, "expected SystemExit" |
| except SystemExit: |
| pass |
| finally: |
| for k, v in saved.items(): |
| if v is None: |
| os.environ.pop(k, None) |
| else: |
| os.environ[k] = v |
|
|
|
|
| def test_readonly_api_key_blocks_writes(): |
| os.environ["CEPHEUS_READONLY_API_KEY"] = "readonly-test-key" |
| try: |
| import importlib |
| import main as main_module |
|
|
| importlib.reload(main_module) |
| ro_client = TestClient(main_module.app) |
| r = ro_client.post( |
| "/alert", |
| headers={"X-API-Key": "readonly-test-key"}, |
| json={"type": "test", "message": "x", "location": "y", "severity": "low"}, |
| ) |
| assert r.status_code == 403 |
| g = ro_client.get("/alerts", headers={"X-API-Key": "readonly-test-key"}) |
| assert g.status_code == 200 |
| finally: |
| os.environ.pop("CEPHEUS_READONLY_API_KEY", None) |
|
|
|
|
| def test_query_string_auth_blocked_in_production(): |
| os.environ["CEPHEUS_PRODUCTION"] = "1" |
| os.environ["CEPHEUS_JWT_SECRET"] = "prod-jwt-secret-min-32-characters-long" |
| os.environ["CEPHEUS_AUTH_DEV_MODE"] = "0" |
| try: |
| import importlib |
| import auth_service |
| import main as main_module |
|
|
| importlib.reload(auth_service) |
| importlib.reload(main_module) |
| |
| os.environ["CEPHEUS_PRODUCTION"] = "1" |
| os.environ["CEPHEUS_JWT_SECRET"] = "prod-jwt-secret-min-32-characters-long" |
| os.environ["CEPHEUS_AUTH_DEV_MODE"] = "0" |
| prod_client = TestClient(main_module.app) |
| r = prod_client.get("/files/face/test/person.jpg?api_key=test-key", headers=API_HEADERS) |
| assert r.status_code == 400 |
| finally: |
| os.environ.pop("CEPHEUS_PRODUCTION", None) |
| os.environ.pop("CEPHEUS_JWT_SECRET", None) |
|
|
