test: 补齐上线关键链路自动验证

ai_knowledge_base_v2/apps/backend/tests/test_production_readiness.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+from datetime import date
+import asyncio
+import json
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.api import chat
+from app.models import Base
+from app.models.ai_config import SystemConfig
+from app.models.user import User
+from app.services.chat_queue_runtime import ChatQueueLease
+from app.services.chat_queue_service import ChatQueueConfig, QueueRequest
+from app.services.chat_service import ChatService
+from app.services.secret_service import ENCRYPTED_PREFIX, MASKED_SECRET, SecretService
+from app.services.security_state_service import SecurityStateService
+
+
+@pytest.fixture(autouse=True)
+def reset_security_state(monkeypatch):
+    monkeypatch.setattr("app.services.security_state_service.get_sync_redis_client", lambda: None)
+    SecurityStateService._counters.clear()
+
+
+def test_secret_is_encrypted_and_masked():
+    encrypted = SecretService.encrypt("production-secret")
+    assert encrypted.startswith(ENCRYPTED_PREFIX)
+    assert "production-secret" not in encrypted
+    assert SecretService.decrypt(encrypted) == "production-secret"
+    assert SecretService.masked(encrypted) == MASKED_SECRET
+
+
+def test_rate_limit_blocks_after_limit():
+    SecurityStateService.enforce_limit("test:rate", limit=2, window_seconds=60, message="too many")
+    SecurityStateService.enforce_limit("test:rate", limit=2, window_seconds=60, message="too many")
+    with pytest.raises(HTTPException) as exc:
+        SecurityStateService.enforce_limit("test:rate", limit=2, window_seconds=60, message="too many")
+    assert exc.value.status_code == 429
+
+
+def test_revoked_token_is_shared_security_state():
+    SecurityStateService.revoke_token("token-id", None)
+    assert SecurityStateService.is_token_revoked("token-id") is True
+
+
+def test_daily_quota_resets_for_new_day():
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(engine)
+    with Session(engine) as db:
+        user = User(
+            id=1,
+            phone="13800000001",
+            name="测试用户",
+            daily_chat_limit=10,
+            daily_chat_used=10,
+            daily_chat_reset_date=date(2020, 1, 1),
+        )
+        db.add(user)
+        db.commit()
+        prepared = ChatService.prepare_daily_quota(db, user)
+        assert prepared.daily_chat_used == 0
+        assert prepared.daily_chat_reset_date > date(2020, 1, 1)
+
+
+def test_queued_chat_reports_position_and_completes(monkeypatch):
+    config = ChatQueueConfig(1, 10, 10, 60)
+    queued = ChatQueueLease("memory", QueueRequest("queued", token=object(), position=2, active_count=1, waiting_count=2))
+    acquired = ChatQueueLease("memory", QueueRequest("acquired", token=queued.request.token, active_count=1))
+    released = False
+
+    async def request_slot(_config):
+        return queued
+
+    async def wait_slot(_lease, _config):
+        return acquired
+
+    async def release_slot(_lease):
+        nonlocal released
+        released = True
+
+    async def stream_answer(*_args, **_kwargs):
+        yield "回答"
+
+    monkeypatch.setattr(chat, "load_chat_queue_config", lambda _db: config)
+    monkeypatch.setattr(chat, "request_chat_slot", request_slot)
+    monkeypatch.setattr(chat, "wait_for_chat_slot", wait_slot)
+    monkeypatch.setattr(chat, "release_chat_slot", release_slot)
+    monkeypatch.setattr(chat.ChatStreamService, "stream_answer_async", stream_answer)
+
+    payload = type("Payload", (), {"sessionId": 1, "message": "问题"})()
+    async def collect_events():
+        return [item async for item in chat._chat_stream(payload, object(), object())]
+
+    events = asyncio.run(collect_events())
+    decoded = [json.loads(item.removeprefix("data: ")) for item in events if item != "data: [DONE]\n\n"]
+    assert decoded[0]["type"] == "queued"
+    assert any(item["type"] == "content" and item["content"] == "回答" for item in decoded)
+    assert released is True
+
+
+def test_sensitive_system_config_never_returns_plaintext():
+    from app.api.admin_settings import _config_dict
+
+    config = SystemConfig(config_key="feishu_app_secret", config_value="plain-secret")
+    result = _config_dict(config)
+    assert result["configValue"] == MASKED_SECRET