rebuild(llm): add audited model connectivity tests

next/tests/test_model_pool.py

@@ -6,6 +6,7 @@ import httpx
 
 from backend.bootstrap.application import create_application
 from backend.bootstrap.settings import Settings
+from backend.llm.provider import ProviderFailure
 from tests.support import run_scenario
 from tests.test_accounts import (
     ADMIN_PASSWORD,
@@ -25,6 +26,17 @@ def model_payload(index: int, api_key: str | None = None) -> dict[str, str]:
     }
 
 
+class ScriptedProvider:
+    def __init__(self, scripts: list[object]) -> None:
+        self.scripts = scripts
+
+    def stream(self, _profile, _messages):
+        script = self.scripts.pop(0)
+        if isinstance(script, Exception):
+            raise script
+        yield from script
+
+
 def test_model_pool_is_admin_only_and_never_exposes_keys(tmp_path) -> None:
     application = create_application(Settings.for_test(tmp_path))
 
@@ -62,6 +74,45 @@ def test_model_pool_is_admin_only_and_never_exposes_keys(tmp_path) -> None:
         assert encrypted_before != "private-alpha-key"
         assert "private-alpha-key" not in encrypted_before
 
+        application.state.container.llm._provider = ScriptedProvider(
+            [["连接", "成功"], ProviderFailure("authentication")]
+        )
+        tested = await client.post(
+            "/api/admin/models/1/test", headers=csrf_headers(admin_session)
+        )
+        assert tested.status_code == 200
+        assert tested.json()["connected"] is True
+        assert tested.json()["message"] == "连接成功"
+        assert "private-alpha-key" not in tested.text
+        failed_test = await client.post(
+            "/api/admin/models/1/test", headers=csrf_headers(admin_session)
+        )
+        assert failed_test.status_code == 503
+        assert failed_test.json()["error"]["code"] == "model_authentication"
+        with sqlite3.connect(application.state.settings.database_path) as connection:
+            requests = connection.execute(
+                """
+                SELECT feature, business_id, status, error_type
+                FROM llm_requests ORDER BY started_at, rowid
+                """
+            ).fetchall()
+            attempts = connection.execute(
+                """
+                SELECT model_id, role, status, error_type
+                FROM llm_attempts ORDER BY id
+                """
+            ).fetchall()
+            usage = connection.execute("SELECT COUNT(*) FROM llm_usage_daily").fetchone()[0]
+        assert requests == [
+            ("model_connectivity", "model:1", "success", ""),
+            ("model_connectivity", "model:1", "failed", "authentication"),
+        ]
+        assert attempts == [
+            (1, "primary", "success", ""),
+            (1, "primary", "failed", "authentication"),
+        ]
+        assert usage == 0
+
         updated_payload = model_payload(1)
         updated_payload.pop("api_key")
         updated_payload["display_name"] = "主模型"