| import uuid |
| import pytest |
| from app.core.config import get_settings |
|
|
| def test_health_check_improved(client) -> None: |
| resp = client.get("/health") |
| assert resp.status_code == 200 |
| data = resp.json() |
| assert data["status"] == "ok" |
| assert "timestamp" in data |
| assert "environment" in data |
| assert "app_version" in data |
| assert data["app_version"] == "0.1.0" |
|
|
| def test_deep_health_check_success(client) -> None: |
| from app.core.admin_auth import require_admin |
|
|
| client.app.dependency_overrides[require_admin] = lambda: None |
| try: |
| resp = client.get("/health/deep") |
| finally: |
| client.app.dependency_overrides.pop(require_admin, None) |
|
|
| assert resp.status_code == 200 |
| data = resp.json() |
| assert data["status"] == "ok" |
| assert "checks" in data |
| checks = data["checks"] |
| assert checks["database"]["status"] == "ok" |
| assert checks["storage"]["status"] == "ok" |
| assert checks["ai_provider"]["status"] == "ok" |
| assert checks["tts_provider"]["status"] == "ok" |
| assert checks["render_output"]["status"] == "ok" |
| assert checks["env_presence"]["status"] == "ok" |
|
|
| def test_request_id_middleware_generated_if_missing(client) -> None: |
| resp = client.get("/health") |
| assert resp.status_code == 200 |
| assert "X-Request-ID" in resp.headers |
| request_id = resp.headers["X-Request-ID"] |
| uuid.UUID(request_id) |
|
|
| def test_request_id_middleware_propagated_if_provided(client) -> None: |
| custom_id = "test-request-id-12345" |
| resp = client.get("/health", headers={"X-Request-ID": custom_id}) |
| assert resp.status_code == 200 |
| assert resp.headers["X-Request-ID"] == custom_id |
|
|
| def test_request_id_middleware_too_long_replaced(client) -> None: |
| unsafe_id = "a" * 101 |
| resp = client.get("/health", headers={"X-Request-ID": unsafe_id}) |
| assert resp.status_code == 200 |
| res_id = resp.headers["X-Request-ID"] |
| assert res_id != unsafe_id |
| uuid.UUID(res_id) |
|
|
| def test_request_id_middleware_unsafe_characters_replaced(client) -> None: |
| unsafe_id = "test-id-with-newlines\nand<script>alert(1)</script>" |
| resp = client.get("/health", headers={"X-Request-ID": unsafe_id}) |
| assert resp.status_code == 200 |
| res_id = resp.headers["X-Request-ID"] |
| assert res_id != unsafe_id |
| uuid.UUID(res_id) |
|
|
| def test_deep_health_never_exposes_secrets(client) -> None: |
| from app.core.admin_auth import require_admin |
|
|
| client.app.dependency_overrides[require_admin] = lambda: None |
| try: |
| resp = client.get("/health/deep") |
| finally: |
| client.app.dependency_overrides.pop(require_admin, None) |
|
|
| assert resp.status_code == 200 |
| data = resp.json() |
| |
| import json |
| body_str = json.dumps(data).lower() |
| |
| |
| settings = get_settings() |
| |
| secrets_to_check = [ |
| settings.database_url, |
| settings.jwt_secret_key, |
| settings.sarvam_api_key, |
| settings.openrouter_api_key, |
| settings.beta_invite_code, |
| ] |
| |
| for secret in secrets_to_check: |
| if secret and len(secret) > 5: |
| assert secret.lower() not in body_str, f"Leaked secret value in /health/deep: {secret}" |
|
|
| def test_setup_request_id_logging_is_idempotent() -> None: |
| from app.utils.request_id import setup_request_id_logging |
| import logging |
| |
| setup_request_id_logging() |
| |
| root_logger = logging.getLogger("") |
| filters_count_1 = len(root_logger.filters) |
| |
| setup_request_id_logging() |
| filters_count_2 = len(root_logger.filters) |
| |
| assert filters_count_1 == filters_count_2 |
|
|
| def test_standard_error_shape_not_found(client) -> None: |
| resp = client.get("/non-existent-route-for-testing") |
| assert resp.status_code == 404 |
| data = resp.json() |
| assert data["success"] is False |
| assert "error" in data |
| assert data["error"]["code"] == "NOT_FOUND" |
| assert "message" in data["error"] |
| assert isinstance(data["error"]["details"], dict) |
|
|
| def test_standard_error_shape_validation_error(client) -> None: |
| resp = client.post("/video/generate-audio", json={}) |
| assert resp.status_code == 422 |
| data = resp.json() |
| assert data["success"] is False |
| assert "error" in data |
| assert data["error"]["code"] == "VALIDATION_ERROR" |
| assert "message" in data["error"] |
| assert "fields" in data["error"]["details"] |
|
|