File size: 4,360 Bytes
7c6ffa6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
f415181
 
 
 
 
 
 
 
7c6ffa6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
f415181
 
 
 
 
 
 
 
7c6ffa6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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()
    
    # Assert actual secret values are not in the response
    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"]