company-ai-platform/tests/test_smoke.py

import json
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["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_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"