feat: 添加风险规则及 agent assets 功能增强

server/tests/test_agent_foundation_endpoints.py

@@ -1,98 +1,98 @@
-from __future__ import annotations
-
-from collections.abc import Generator
-
-from fastapi.testclient import TestClient
-from sqlalchemy import create_engine
-from sqlalchemy.orm import Session, sessionmaker
-from sqlalchemy.pool import StaticPool
-
-from app.api.deps import get_db
-from app.core.agent_enums import AgentAssetStatus
-from app.db.base import Base
-from app.main import create_app
-from app.services.agent_assets import AgentAssetService
-
-
-def build_client() -> tuple[TestClient, sessionmaker[Session]]:
-    engine = create_engine(
-        "sqlite+pysqlite:///:memory:",
-        connect_args={"check_same_thread": False},
-        poolclass=StaticPool,
-    )
-    Base.metadata.create_all(bind=engine)
-    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
-
-    app = create_app()
-
-    def override_db() -> Generator[Session, None, None]:
-        db = session_factory()
-        try:
-            yield db
-        finally:
-            db.close()
-
-    app.dependency_overrides[get_db] = override_db
-    return TestClient(app), session_factory
-
-
-def test_list_agent_assets_endpoint_returns_seeded_items() -> None:
-    client, _ = build_client()
-
-    response = client.get("/api/v1/agent-assets", params={"asset_type": "rule"})
-
-    assert response.status_code == 200
-    payload = response.json()
-    assert payload
-    assert all(item["asset_type"] == "rule" for item in payload)
-    assert any(item["code"] == "rule.expense.travel_risk_control_standard" for item in payload)
-
-
-def test_get_agent_asset_detail_endpoint_returns_version_history() -> None:
-    client, _ = build_client()
-
-    list_response = client.get("/api/v1/agent-assets", params={"asset_type": "rule"})
-    asset_id = next(
-        item["id"]
-        for item in list_response.json()
-        if item["code"] == "rule.expense.travel_risk_control_standard"
-    )
-
-    response = client.get(f"/api/v1/agent-assets/{asset_id}")
-
-    assert response.status_code == 200
-    payload = response.json()
-    assert payload["recent_versions"]
-    assert payload["current_version_content_type"] == "markdown"
-    assert payload["current_version"] == "v1.1.0"
-    assert "行程闭环" in payload["current_version_content"]
-
-
-def test_activate_pending_rule_endpoint_is_blocked() -> None:
-    client, session_factory = build_client()
-
-    with session_factory() as db:
-        pending_rule = next(
-            item
-            for item in AgentAssetService(db).list_assets(asset_type="rule")
-            if item.status == AgentAssetStatus.REVIEW.value
-        )
-
-    response = client.post(
-        f"/api/v1/agent-assets/{pending_rule.id}/activate",
-        headers={"x-actor": "pytest"},
-    )
-
-    assert response.status_code == 400
-    assert "审核" in response.json()["detail"]
-
-
-def test_list_audit_logs_endpoint_returns_seeded_logs() -> None:
-    client, _ = build_client()
-
-    response = client.get("/api/v1/audit-logs")
-
-    assert response.status_code == 200
-    payload = response.json()
-    assert payload
-    assert any(item["action"] == "review_rule" for item in payload)
+from __future__ import annotations
+
+from collections.abc import Generator
+
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.api.deps import get_db
+from app.core.agent_enums import AgentAssetStatus
+from app.db.base import Base
+from app.main import create_app
+from app.services.agent_assets import AgentAssetService
+
+
+def build_client() -> tuple[TestClient, sessionmaker[Session]]:
+    engine = create_engine(
+        "sqlite+pysqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(bind=engine)
+    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+    app = create_app()
+
+    def override_db() -> Generator[Session, None, None]:
+        db = session_factory()
+        try:
+            yield db
+        finally:
+            db.close()
+
+    app.dependency_overrides[get_db] = override_db
+    return TestClient(app), session_factory
+
+
+def test_list_agent_assets_endpoint_returns_seeded_items() -> None:
+    client, _ = build_client()
+
+    response = client.get("/api/v1/agent-assets", params={"asset_type": "rule"})
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload
+    assert all(item["asset_type"] == "rule" for item in payload)
+    assert any(item["code"] == "rule.expense.travel_risk_control_standard" for item in payload)
+
+
+def test_get_agent_asset_detail_endpoint_returns_version_history() -> None:
+    client, _ = build_client()
+
+    list_response = client.get("/api/v1/agent-assets", params={"asset_type": "rule"})
+    asset_id = next(
+        item["id"]
+        for item in list_response.json()
+        if item["code"] == "rule.expense.travel_risk_control_standard"
+    )
+
+    response = client.get(f"/api/v1/agent-assets/{asset_id}")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["recent_versions"]
+    assert payload["current_version_content_type"] == "markdown"
+    assert payload["current_version"] == "v1.1.0"
+    assert "行程闭环" in payload["current_version_content"]
+
+
+def test_activate_pending_rule_endpoint_is_blocked() -> None:
+    client, session_factory = build_client()
+
+    with session_factory() as db:
+        pending_rule = next(
+            item
+            for item in AgentAssetService(db).list_assets(asset_type="rule")
+            if item.status == AgentAssetStatus.REVIEW.value
+        )
+
+    response = client.post(
+        f"/api/v1/agent-assets/{pending_rule.id}/activate",
+        headers={"x-actor": "pytest"},
+    )
+
+    assert response.status_code == 400
+    assert "审核" in response.json()["detail"]
+
+
+def test_list_audit_logs_endpoint_returns_seeded_logs() -> None:
+    client, _ = build_client()
+
+    response = client.get("/api/v1/audit-logs")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload
+    assert any(item["action"] == "review_rule" for item in payload)