mi
127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
import base64
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import yaml
|
|
from pydantic import SecretStr
|
|
|
|
from app.main import app, otp_settings, websocket_token
|
|
from app.services import SettingsSnapshot
|
|
|
|
EXPECTED_PATHS = {
|
|
"/health/live",
|
|
"/health/ready",
|
|
"/api/v1/public/app-config",
|
|
"/api/v1/public/content",
|
|
"/api/v1/auth/bootstrap",
|
|
"/api/v1/consents",
|
|
"/api/v1/analytics/session-start",
|
|
"/api/v1/me",
|
|
"/api/v1/me/documents",
|
|
"/api/v1/documents/{document_id}",
|
|
"/api/v1/documents/{document_id}/download-url",
|
|
"/api/v1/dialogs",
|
|
"/api/v1/dialogs/{dialog_id}",
|
|
"/api/v1/dialogs/{dialog_id}/messages",
|
|
"/api/v1/dialogs/{dialog_id}/attachments/init",
|
|
"/api/v1/dialogs/{dialog_id}/attachments/{attachment_id}/complete",
|
|
"/api/v1/dialogs/{dialog_id}/attachments/{attachment_id}/download-url",
|
|
"/internal/openlines/v1/inbox",
|
|
"/internal/settings/v1/otp",
|
|
}
|
|
|
|
|
|
def test_openapi_31_contains_all_http_contracts() -> None:
|
|
schema = app.openapi()
|
|
assert schema["openapi"].startswith("3.1.")
|
|
assert EXPECTED_PATHS <= schema["paths"].keys()
|
|
assert all(not path.startswith("/internal/safety") for path in schema["paths"])
|
|
committed = yaml.safe_load(Path("openapi.yaml").read_text(encoding="utf-8"))
|
|
assert committed["openapi"] == "3.1.0"
|
|
assert committed["paths"].keys() == schema["paths"].keys()
|
|
|
|
|
|
def test_websocket_route_is_registered() -> None:
|
|
assert any(getattr(route, "path", None) == "/api/v1/realtime" for route in app.routes)
|
|
|
|
|
|
def test_websocket_accepts_canonical_base64url_jwt_protocol() -> None:
|
|
jwt = "header.payload.signature"
|
|
encoded = base64.urlsafe_b64encode(jwt.encode()).decode().rstrip("=")
|
|
websocket = SimpleNamespace(
|
|
headers={"sec-websocket-protocol": f"han-chat-v1, han.jwt.{encoded}"},
|
|
query_params={},
|
|
)
|
|
assert websocket_token(websocket) == (jwt, f"han.jwt.{encoded}")
|
|
|
|
|
|
def test_committed_openapi_server_does_not_double_api_prefix() -> None:
|
|
committed = yaml.safe_load(Path("openapi.yaml").read_text(encoding="utf-8"))
|
|
assert committed["servers"] == [{"url": "/"}]
|
|
|
|
|
|
def test_otp_settings_contract_is_strict_and_complete() -> None:
|
|
generated = app.openapi()
|
|
response = generated["paths"]["/internal/settings/v1/otp"]["get"]["responses"]["200"]
|
|
schema_ref = response["content"]["application/json"]["schema"]["$ref"]
|
|
schema = generated["components"]["schemas"][schema_ref.rsplit("/", 1)[-1]]
|
|
|
|
assert set(schema["required"]) == {
|
|
"max_send_attempts_per_24h",
|
|
"min_seconds_between_attempts",
|
|
"max_verify_attempts",
|
|
"code_length",
|
|
"ttl_seconds",
|
|
"sms_order_timeout_ms",
|
|
"version",
|
|
"cache_ttl_seconds",
|
|
}
|
|
assert schema["additionalProperties"] is False
|
|
assert schema["properties"]["code_length"] == {
|
|
"type": "integer",
|
|
"maximum": 10.0,
|
|
"minimum": 4.0,
|
|
"title": "Code Length",
|
|
}
|
|
assert schema["properties"]["ttl_seconds"]["multipleOf"] == 60
|
|
|
|
|
|
async def test_otp_settings_returns_runtime_values_and_supports_etag() -> None:
|
|
request = SimpleNamespace(
|
|
headers={"Authorization": "Bearer bridge-token"},
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
settings=SimpleNamespace(
|
|
keycloak_settings_bridge_token=SecretStr("bridge-token")
|
|
)
|
|
)
|
|
),
|
|
)
|
|
settings = SettingsSnapshot(
|
|
{
|
|
"otp.phone.max_send_attempts_per_24h": "3",
|
|
"otp.phone.min_seconds_between_attempts": "30",
|
|
"otp.phone.max_verify_attempts": "5",
|
|
"otp.phone.code_length": "6",
|
|
"otp.phone.ttl_seconds": "60",
|
|
"otp.phone.sms_order_timeout_ms": "3000",
|
|
},
|
|
"settings-version",
|
|
)
|
|
|
|
response = await otp_settings(request, settings)
|
|
assert json.loads(response.body) == {
|
|
"max_send_attempts_per_24h": 3,
|
|
"min_seconds_between_attempts": 30,
|
|
"max_verify_attempts": 5,
|
|
"code_length": 6,
|
|
"ttl_seconds": 60,
|
|
"sms_order_timeout_ms": 3000,
|
|
"version": "settings-version",
|
|
"cache_ttl_seconds": 60,
|
|
}
|
|
|
|
cached = await otp_settings(request, settings, response.headers["etag"])
|
|
assert cached.status_code == 304
|