mi
331 lines
12 KiB
Python
331 lines
12 KiB
Python
import asyncio
|
|
import base64
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yaml
|
|
from alembic.config import Config
|
|
from alembic.script import ScriptDirectory
|
|
from pydantic import SecretStr
|
|
|
|
from app.main import (
|
|
EXPECTED_API_DB_REVISION,
|
|
app,
|
|
app_config,
|
|
otp_settings,
|
|
refresh_jwks_cache,
|
|
websocket_token,
|
|
)
|
|
from app.services import SettingsSnapshot
|
|
|
|
EXPECTED_PATHS = {
|
|
"/health/live",
|
|
"/health/ready",
|
|
"/api/v1/public/app-config",
|
|
"/api/v1/public/content",
|
|
"/api/v1/public/notifications",
|
|
"/api/v1/public/notification-types",
|
|
"/api/v1/auth/bootstrap",
|
|
"/api/v1/consents",
|
|
"/api/v1/analytics/session-start",
|
|
"/api/v1/me",
|
|
"/api/v1/me/documents",
|
|
"/api/v1/notifications",
|
|
"/api/v1/notifications/counter",
|
|
"/api/v1/notifications/{notification_id}",
|
|
"/api/v1/notifications/{notification_id}/read",
|
|
"/api/v1/notifications/{notification_id}/hide",
|
|
"/api/v1/notifications/{notification_id}/buttons/{button_code}",
|
|
"/api/v1/notifications/{notification_id}/cta",
|
|
"/api/v1/notifications/{notification_id}/documents/{document_id}/download-url",
|
|
"/api/v1/uploads/init",
|
|
"/api/v1/uploads/{draft_id}/complete",
|
|
"/api/v1/uploads",
|
|
"/api/v1/uploads/{draft_id}",
|
|
"/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/notifications/v1/notifications",
|
|
"/internal/notifications/v1/notifications/cancel",
|
|
"/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_readiness_expected_revision_matches_alembic_head() -> None:
|
|
scripts = ScriptDirectory.from_config(Config("alembic.ini"))
|
|
assert EXPECTED_API_DB_REVISION == scripts.get_current_head()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jwks_refresh_loop_recovers_after_startup_race(monkeypatch) -> None:
|
|
class FakeJWKS:
|
|
has_keys = False
|
|
refresh_calls = 0
|
|
|
|
async def refresh(self) -> None:
|
|
self.refresh_calls += 1
|
|
self.has_keys = True
|
|
|
|
jwks = FakeJWKS()
|
|
test_app = SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
jwks=jwks,
|
|
settings=SimpleNamespace(jwks_cache_ttl_seconds=300),
|
|
)
|
|
)
|
|
delays: list[int] = []
|
|
|
|
async def fake_sleep(delay: int) -> None:
|
|
delays.append(delay)
|
|
if len(delays) > 1:
|
|
raise asyncio.CancelledError
|
|
|
|
monkeypatch.setattr("app.main.asyncio.sleep", fake_sleep)
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await refresh_jwks_cache(test_app)
|
|
|
|
assert jwks.refresh_calls == 1
|
|
assert delays == [5, 300]
|
|
|
|
|
|
def test_websocket_route_is_registered() -> None:
|
|
assert any(getattr(route, "path", None) == "/api/v1/realtime" for route in app.routes)
|
|
|
|
|
|
def test_notification_http_methods_match_contract() -> None:
|
|
paths = app.openapi()["paths"]
|
|
expected = {
|
|
"/api/v1/public/notifications": {"get"},
|
|
"/api/v1/public/notification-types": {"get"},
|
|
"/api/v1/notifications": {"get"},
|
|
"/api/v1/notifications/counter": {"get"},
|
|
"/api/v1/notifications/{notification_id}": {"get"},
|
|
"/api/v1/notifications/{notification_id}/read": {"post"},
|
|
"/api/v1/notifications/{notification_id}/hide": {"post"},
|
|
"/api/v1/notifications/{notification_id}/buttons/{button_code}": {"post"},
|
|
"/api/v1/notifications/{notification_id}/cta": {"post"},
|
|
(
|
|
"/api/v1/notifications/{notification_id}/documents/"
|
|
"{document_id}/download-url"
|
|
): {"get"},
|
|
"/api/v1/uploads/init": {"post"},
|
|
"/api/v1/uploads/{draft_id}/complete": {"post"},
|
|
"/api/v1/uploads": {"get"},
|
|
"/api/v1/uploads/{draft_id}": {"delete"},
|
|
"/internal/notifications/v1/notifications": {"post"},
|
|
"/internal/notifications/v1/notifications/cancel": {"post"},
|
|
}
|
|
for path, methods in expected.items():
|
|
assert methods <= paths[path].keys()
|
|
|
|
|
|
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_public_config_contract_is_strict_and_exposes_mobile_update() -> None:
|
|
generated = app.openapi()
|
|
response = generated["paths"]["/api/v1/public/app-config"]["get"]["responses"]["200"]
|
|
schema_ref = response["content"]["application/json"]["schema"]["$ref"]
|
|
schema = generated["components"]["schemas"][schema_ref.rsplit("/", 1)[-1]]
|
|
|
|
assert schema["additionalProperties"] is False
|
|
assert set(schema["required"]) == {
|
|
"auth",
|
|
"operator",
|
|
"messages",
|
|
"consents",
|
|
"attachments",
|
|
"notification",
|
|
"ux",
|
|
"mobile_update",
|
|
}
|
|
mobile_ref = schema["properties"]["mobile_update"]["$ref"]
|
|
mobile = generated["components"]["schemas"][mobile_ref.rsplit("/", 1)[-1]]
|
|
assert mobile["additionalProperties"] is False
|
|
assert set(mobile["required"]) == {"google_play", "rustore", "app_store"}
|
|
store_ref = mobile["properties"]["rustore"]["$ref"]
|
|
store = generated["components"]["schemas"][store_ref.rsplit("/", 1)[-1]]
|
|
assert "release_notes" in store["required"]
|
|
release_notes = store["properties"]["release_notes"]
|
|
assert {"type": "string", "maxLength": 4000} in release_notes["anyOf"]
|
|
assert {"type": "null"} in release_notes["anyOf"]
|
|
assert "304" in generated["paths"]["/api/v1/public/app-config"]["get"]["responses"]
|
|
|
|
|
|
async def test_public_config_returns_mobile_policy_and_supports_etag(monkeypatch) -> None:
|
|
values = {
|
|
"rate_limit.public_endpoints.per_ip": "60/minute",
|
|
"security.public_cache.max_age_seconds": "60",
|
|
"auth.phone.enabled": "true",
|
|
"auth.password.enabled": "false",
|
|
"operator.call.phone": "+74999591007",
|
|
"chat.message.max_length": "4000",
|
|
"consent.personal_data.required": "true",
|
|
"consent.personal_data.document_url": "https://example.ru/personal",
|
|
"consent.privacy_policy.document_url": "https://example.ru/privacy",
|
|
"consent.personal_data.version": "2026-06-10",
|
|
"consent.user_agreement.required": "true",
|
|
"consent.user_agreement.document_url": "https://example.ru/agreement",
|
|
"consent.user_agreement.version": "2026-06-10",
|
|
"consent.marketing.required": "false",
|
|
"consent.marketing.document_url": "https://example.ru/marketing",
|
|
"consent.marketing.version": "2026-06-10",
|
|
"chat.attachments.allowed_extensions": "jpg,pdf",
|
|
"chat.attachments.allowed_mime_types": "image/jpeg,application/pdf",
|
|
"chat.attachments.max_size_mb": "5",
|
|
"notification.carousel.autoplay_enabled": "false",
|
|
"notification.carousel.autoplay_interval_ms": "5000",
|
|
"ux.session.idle_timeout_minutes": "30",
|
|
"mobile_update.google_play.enabled": "true",
|
|
"mobile_update.google_play.latest_build": "2",
|
|
"mobile_update.google_play.minimum_build": "1",
|
|
"mobile_update.google_play.latest_version": "1.0.1",
|
|
"mobile_update.google_play.store_url": (
|
|
"https://play.google.com/store/apps/details?id=ru.han.chat"
|
|
),
|
|
"mobile_update.google_play.release_notes": "",
|
|
"mobile_update.rustore.enabled": "true",
|
|
"mobile_update.rustore.latest_build": "2",
|
|
"mobile_update.rustore.minimum_build": "1",
|
|
"mobile_update.rustore.latest_version": "1.0.1",
|
|
"mobile_update.rustore.store_url": (
|
|
"https://www.rustore.ru/catalog/app/ru.han.chat"
|
|
),
|
|
"mobile_update.rustore.release_notes": "",
|
|
"mobile_update.app_store.enabled": "false",
|
|
"mobile_update.app_store.latest_build": "",
|
|
"mobile_update.app_store.minimum_build": "",
|
|
"mobile_update.app_store.latest_version": "",
|
|
"mobile_update.app_store.store_url": "",
|
|
"mobile_update.app_store.release_notes": "",
|
|
}
|
|
settings = SettingsSnapshot(values, "settings-version")
|
|
request = SimpleNamespace(
|
|
headers={},
|
|
client=None,
|
|
app=SimpleNamespace(state=SimpleNamespace()),
|
|
)
|
|
|
|
async def no_limit(*args, **kwargs) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr("app.main.enforce_limit", no_limit)
|
|
|
|
response = await app_config(request, settings)
|
|
body = json.loads(response.body)
|
|
assert response.headers["etag"] == '"settings-version"'
|
|
assert response.headers["cache-control"] == "public, max-age=60"
|
|
assert body["mobile_update"]["google_play"]["latest_build"] == 2
|
|
assert body["mobile_update"]["google_play"]["release_notes"] is None
|
|
assert body["mobile_update"]["rustore"]["store_url"] == (
|
|
"https://www.rustore.ru/catalog/app/ru.han.chat"
|
|
)
|
|
assert body["mobile_update"]["app_store"] == {
|
|
"enabled": False,
|
|
"latest_build": None,
|
|
"minimum_build": None,
|
|
"latest_version": None,
|
|
"store_url": None,
|
|
"release_notes": None,
|
|
}
|
|
|
|
cached = await app_config(request, settings, 'W/"settings-version", "old"')
|
|
assert cached.status_code == 304
|
|
assert cached.headers["etag"] == '"settings-version"'
|
|
|
|
|
|
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
|