import os import tempfile from pathlib import Path _db = tempfile.NamedTemporaryFile(delete=False, suffix=".db") _db.close() os.environ["DATABASE_URL"] = "sqlite:///" + _db.name.replace("\\", "/") os.environ["API_KEY"] = "test-key" os.environ["MODEL_PROVIDER"] = "noop" os.environ["SCHEDULER_ENABLED"] = "false" from fastapi.testclient import TestClient from app.core.database import Base, engine from app.main import app Base.metadata.create_all(bind=engine) client = TestClient(app) headers = {"X-API-Key": "test-key"} def teardown_module() -> None: engine.dispose() path = Path(_db.name) if path.exists(): path.unlink() def test_project_report_and_feishu_command_preview() -> None: response = client.post( "/api/v1/business/projects", headers=headers, json={ "actor": "pytest", "data": { "code": "P-SMOKE-001", "name": "Smoke Project", "owner": "tester", "status": "执行中", "budget_amount": 1000, "actual_amount": 200, }, }, ) assert response.status_code == 200 assert response.json()["data"]["code"] == "P-SMOKE-001" response = client.get("/api/v1/reports/daily-brief", headers=headers) assert response.status_code == 200 assert response.json()["title"] == "每日经营晨报" response = client.post( "/api/v1/integrations/feishu/commands/preview", headers=headers, json={"text": "日报", "auto_reply": False}, ) assert response.status_code == 200 assert response.json()["command"] == "daily_brief" def test_approval_gate_for_high_risk_update() -> None: create_response = client.post( "/api/v1/business/fund-accounts", headers=headers, json={ "actor": "pytest", "data": { "code": "FUND-SMOKE-001", "name": "Main Account", "current_balance": 1000, "safety_line": 500, }, }, ) assert create_response.status_code == 200 record_id = create_response.json()["data"]["id"] blocked_response = client.patch( f"/api/v1/business/fund-accounts/{record_id}", headers=headers, json={"actor": "pytest", "data": {"current_balance": 100}}, ) assert blocked_response.status_code == 409 approval_response = client.post( "/api/v1/approvals", headers=headers, json={ "domain": "fund-accounts", "record_id": str(record_id), "action": "update:fund-accounts", "applicant": "pytest", "reason": "Smoke test balance adjustment", "payload": {"current_balance": 100}, }, ) assert approval_response.status_code == 200 ticket_id = approval_response.json()["ticket_id"] pending_response = client.patch( f"/api/v1/business/fund-accounts/{record_id}", headers=headers, json={ "actor": "pytest", "approval_ticket_id": ticket_id, "data": {"current_balance": 100}, }, ) assert pending_response.status_code == 403 approve_response = client.post( f"/api/v1/approvals/{ticket_id}/approve", headers=headers, json={"approver": "manager", "comment": "ok"}, ) assert approve_response.status_code == 200 assert approve_response.json()["status"] == "approved" update_response = client.patch( f"/api/v1/business/fund-accounts/{record_id}", headers=headers, json={ "actor": "pytest", "approval_ticket_id": ticket_id, "data": {"current_balance": 100}, }, ) assert update_response.status_code == 200 assert update_response.json()["data"]["current_balance"] == 100.0 def test_ai_noop_provider() -> None: response = client.post( "/api/v1/ai/ask", headers=headers, json={"prompt": "生成项目摘要", "actor": "pytest", "context": {"project": "P-SMOKE-001"}}, ) assert response.status_code == 200 assert response.json()["provider"] == "noop"