company-ai-platform/tests/test_smoke.py

import json
import os
import tempfile
from datetime import date, timedelta
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["FEISHU_APP_ID"] = ""
os.environ["FEISHU_APP_SECRET"] = ""
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_feishu_webhook_routes_message_event() -> None:
    payload = {
        "schema": "2.0",
        "header": {"event_type": "im.message.receive_v1"},
        "event": {
            "sender": {"sender_id": {"open_id": "ou_test"}},
            "message": {
                "chat_id": "oc_test",
                "message_type": "text",
                "content": json.dumps({"text": "risk"}),
            },
        },
    }
    response = client.post("/api/v1/integrations/feishu/webhook", json=payload)
    assert response.status_code == 200
    data = response.json()
    assert data["handled"] is True
    assert data["result"]["command"] == "risk_summary"


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_new_ledgers_reports_and_risk_events() -> None:
    domains_response = client.get("/api/v1/business/domains", headers=headers)
    assert domains_response.status_code == 200
    domains = domains_response.json()["domains"]
    assert "attendance-records" in domains
    assert "work-reports" in domains
    assert "risk-events" in domains

    today = date.today()
    attendance_response = client.post(
        "/api/v1/business/attendance-records",
        headers=headers,
        json={
            "actor": "pytest",
            "data": {
                "code": "ATT-SMOKE-001",
                "employee_name": "Tester",
                "department": "QA",
                "work_date": today.isoformat(),
                "status": "正常",
            },
        },
    )
    assert attendance_response.status_code == 200

    task_response = client.post(
        "/api/v1/business/tasks",
        headers=headers,
        json={
            "actor": "pytest",
            "data": {
                "code": "TASK-RISK-001",
                "title": "Overdue smoke task",
                "owner": "tester",
                "status": "待办",
                "due_date": (today - timedelta(days=1)).isoformat(),
            },
        },
    )
    assert task_response.status_code == 200

    attendance_summary = client.get("/api/v1/reports/attendance-summary", headers=headers)
    assert attendance_summary.status_code == 200
    assert attendance_summary.json()["total"] >= 1

    report_response = client.post(
        "/api/v1/reports/work-reports/generate",
        headers=headers,
        json={"report_type": "daily", "reporter": "pytest", "actor": "pytest"},
    )
    assert report_response.status_code == 200
    assert report_response.json()["data"]["report_type"] == "daily"

    risk_response = client.post(
        "/api/v1/risks/events/generate?actor=pytest",
        headers=headers,
    )
    assert risk_response.status_code == 200
    assert risk_response.json()["created"] >= 1

    events_response = client.get("/api/v1/risks/events?status=open", headers=headers)
    assert events_response.status_code == 200
    assert any(item["risk_type"] == "overdue_task" for item in events_response.json()["items"])


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"