company-ai-platform/tests/test_smoke.py

import json
import os
import tempfile
from datetime import date, timedelta
from pathlib import Path

import pytest
from fastapi import HTTPException

from app.modules.ai_agent.constants import AIProviderName, AIResponseKey
from app.modules.business.constants import StatusValue

_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["APPROVAL_API_KEY"] = "approval-key"
os.environ["APPROVAL_API_ACTOR"] = "approval-manager"
os.environ["FEISHU_APP_ID"] = ""
os.environ["FEISHU_APP_SECRET"] = ""
os.environ["FEISHU_VERIFICATION_TOKEN"] = "test-feishu-token"
os.environ["MODEL_PROVIDER"] = AIProviderName.NOOP
os.environ["SCHEDULER_ENABLED"] = "false"

from fastapi.testclient import TestClient

from app.core.config import get_settings
from app.core.database import Base, engine
from app.core.security import require_api_key, require_approval_api_key
from app.main import app
from app.modules.legacy_mysql.service import LegacyMySQLService
from app.modules.reports.constants import (
    LifecycleAttentionKey,
    LifecycleResponseKey,
    LifecycleSection,
    MetricKey,
    ReportTitle,
    ReportType,
)


Base.metadata.create_all(bind=engine)
client = TestClient(app)
headers = {"X-API-Key": "test-key"}
approval_headers = {"X-API-Key": "test-key", "X-Approval-API-Key": "approval-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", "token": "test-feishu-token"},
        "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_api_key_and_feishu_webhook_fail_closed(monkeypatch) -> None:
    monkeypatch.setenv("API_KEY", "")
    get_settings.cache_clear()
    try:
        with pytest.raises(HTTPException) as exc_info:
            require_api_key("test-key")
        assert exc_info.value.status_code == 503

        monkeypatch.setenv("API_KEY", "test-key")
        get_settings.cache_clear()

        response = client.post(
            "/api/v1/integrations/feishu/webhook",
            json={"schema": "2.0", "header": {"event_type": "im.message.receive_v1"}},
        )
        assert response.status_code == 401

        monkeypatch.setenv("APPROVAL_API_KEY", "")
        get_settings.cache_clear()
        with pytest.raises(HTTPException) as approval_exc_info:
            require_approval_api_key("approval-key")
        assert approval_exc_info.value.status_code == 503
    finally:
        monkeypatch.setenv("API_KEY", "test-key")
        monkeypatch.setenv("APPROVAL_API_KEY", "approval-key")
        get_settings.cache_clear()


def test_approval_gate_for_high_risk_update() -> None:
    create_payload = {
        "code": "FUND-SMOKE-001",
        "name": "Main Account",
        "current_balance": 1000,
        "safety_line": 500,
    }
    blocked_create_response = client.post(
        "/api/v1/business/fund-accounts",
        headers=headers,
        json={
            "actor": "spoofed-user",
            "data": {
                "code": "FUND-SMOKE-BLOCKED",
                "name": "Blocked Account",
            },
        },
    )
    assert blocked_create_response.status_code == 409

    create_approval_response = client.post(
        "/api/v1/approvals",
        headers=headers,
        json={
            "domain": "fund-accounts",
            "action": "create:fund-accounts",
            "applicant": "spoofed-user",
            "reason": "Smoke test account creation",
            "payload": create_payload,
        },
    )
    assert create_approval_response.status_code == 200
    assert create_approval_response.json()["applicant"] == "api"
    create_ticket_id = create_approval_response.json()["ticket_id"]

    approve_create_response = client.post(
        f"/api/v1/approvals/{create_ticket_id}/approve",
        headers=approval_headers,
        json={"approver": "spoofed-manager", "comment": "ok"},
    )
    assert approve_create_response.status_code == 200
    assert approve_create_response.json()["approver"] == "approval-manager"

    create_response = client.post(
        "/api/v1/business/fund-accounts",
        headers=headers,
        json={
            "actor": "spoofed-user",
            "approval_ticket_id": create_ticket_id,
            "data": create_payload,
        },
    )
    assert create_response.status_code == 200
    record_id = create_response.json()["data"]["id"]

    reuse_create_response = client.post(
        "/api/v1/business/fund-accounts",
        headers=headers,
        json={
            "approval_ticket_id": create_ticket_id,
            "data": {
                "code": "FUND-SMOKE-REUSE",
                "name": "Reuse Account",
            },
        },
    )
    assert reuse_create_response.status_code == 403

    blocked_response = client.patch(
        f"/api/v1/business/fund-accounts/{record_id}",
        headers=headers,
        json={"actor": "spoofed-user", "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": "spoofed-user",
            "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": "spoofed-user",
            "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=approval_headers,
        json={"approver": "spoofed-manager", "comment": "ok"},
    )
    assert approve_response.status_code == 200
    assert approve_response.json()["status"] == "approved"
    assert approve_response.json()["approver"] == "approval-manager"

    mismatch_response = client.patch(
        f"/api/v1/business/fund-accounts/{record_id}",
        headers=headers,
        json={
            "actor": "spoofed-user",
            "approval_ticket_id": ticket_id,
            "data": {"current_balance": 101},
        },
    )
    assert mismatch_response.status_code == 403

    update_response = client.patch(
        f"/api/v1/business/fund-accounts/{record_id}",
        headers=headers,
        json={
            "actor": "spoofed-user",
            "approval_ticket_id": ticket_id,
            "data": {"current_balance": 100},
        },
    )
    assert update_response.status_code == 200
    assert update_response.json()["data"]["current_balance"] == 100.0

    reuse_update_response = client.patch(
        f"/api/v1/business/fund-accounts/{record_id}",
        headers=headers,
        json={
            "actor": "spoofed-user",
            "approval_ticket_id": ticket_id,
            "data": {"current_balance": 100},
        },
    )
    assert reuse_update_response.status_code == 403


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": ReportType.DAILY, "reporter": "pytest", "actor": "pytest"},
    )
    assert report_response.status_code == 200
    assert report_response.json()["data"]["report_type"] == ReportType.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_project_lifecycle_report_summarizes_progress_cost_and_risk() -> None:
    today = date.today()
    project_code = "P-LIFECYCLE-001"

    project_response = client.post(
        "/api/v1/business/projects",
        headers=headers,
        json={
            "actor": "pytest",
            "data": {
                "code": project_code,
                "name": "Lifecycle Project",
                "owner": "lifecycle-owner",
                "status": "执行中",
                "progress_percent": 40,
                "budget_amount": 1000,
                "actual_amount": 1500,
                "due_date": (today - timedelta(days=1)).isoformat(),
            },
        },
    )
    assert project_response.status_code == 200

    task_response = client.post(
        "/api/v1/business/tasks",
        headers=headers,
        json={
            "actor": "pytest",
            "data": {
                "code": "TASK-LIFECYCLE-001",
                "title": "Lifecycle overdue task",
                "project_code": project_code,
                "owner": "lifecycle-owner",
                "status": "待办",
                "due_date": (today - timedelta(days=1)).isoformat(),
                "blocker": "waiting for decision",
            },
        },
    )
    assert task_response.status_code == 200

    procurement_response = client.post(
        "/api/v1/business/procurements",
        headers=headers,
        json={
            "actor": "pytest",
            "data": {
                "code": "PROC-LIFECYCLE-001",
                "name": "Lifecycle procurement",
                "project_code": project_code,
                "expected_amount": 300,
                "actual_amount": 100,
                "approval_status": StatusValue.PENDING_APPROVAL,
                "delivery_status": StatusValue.UNDELIVERED,
                "payment_status": StatusValue.UNPAID,
            },
        },
    )
    assert procurement_response.status_code == 200

    expense_response = client.post(
        "/api/v1/business/expenses",
        headers=headers,
        json={
            "actor": "pytest",
            "data": {
                "code": "EXP-LIFECYCLE-001",
                "expense_type": "差旅",
                "amount": 80,
                "project_code": project_code,
                "approval_status": StatusValue.PENDING_APPROVAL,
                "payment_status": StatusValue.UNPAID,
            },
        },
    )
    assert expense_response.status_code == 200

    attendance_response = client.post(
        "/api/v1/business/attendance-records",
        headers=headers,
        json={
            "actor": "pytest",
            "data": {
                "code": "ATT-LIFECYCLE-001",
                "employee_name": "Lifecycle Tester",
                "project_code": project_code,
                "work_date": today.isoformat(),
                "status": StatusValue.MISSING_PUNCH,
            },
        },
    )
    assert attendance_response.status_code == 200

    response = client.get(
        f"/api/v1/reports/project-lifecycle?project_code={project_code}",
        headers=headers,
    )
    assert response.status_code == 200
    data = response.json()
    assert data[LifecycleResponseKey.TITLE] == ReportTitle.PROJECT_LIFECYCLE
    assert data[LifecycleResponseKey.METRICS][LifecycleSection.PROJECTS][MetricKey.TOTAL] == 1
    assert data[LifecycleResponseKey.METRICS][LifecycleSection.PROJECTS][MetricKey.DELAYED] == 1
    assert (
        data[LifecycleResponseKey.METRICS][LifecycleSection.PROJECTS][MetricKey.OVER_BUDGET]
        == 1
    )
    assert data[LifecycleResponseKey.METRICS][LifecycleSection.TASKS][MetricKey.OVERDUE] == 1
    assert (
        data[LifecycleResponseKey.METRICS][LifecycleSection.PROCUREMENTS][
            MetricKey.PENDING_APPROVAL
        ]
        == 1
    )
    assert (
        data[LifecycleResponseKey.METRICS][LifecycleSection.EXPENSES][
            MetricKey.PENDING_APPROVAL
        ]
        == 1
    )
    assert data[LifecycleResponseKey.METRICS][LifecycleSection.ATTENDANCE][MetricKey.ABNORMAL] == 1
    assert (
        data[LifecycleResponseKey.ATTENTION][LifecycleAttentionKey.DELAYED_PROJECTS][0]["code"]
        == project_code
    )
    assert "生命周期健康分" in data[LifecycleResponseKey.CONTENT]
    assert data[LifecycleResponseKey.RECOMMENDATIONS]

    ai_response = client.get(
        f"/api/v1/reports/project-lifecycle?project_code={project_code}&include_ai=true",
        headers=headers,
    )
    assert ai_response.status_code == 200
    ai_data = ai_response.json()
    assert ai_data[LifecycleResponseKey.AI_ANALYSIS][AIResponseKey.OK] is True
    assert (
        ai_data[LifecycleResponseKey.AI_ANALYSIS][AIResponseKey.PROVIDER]
        == AIProviderName.NOOP
    )


def test_work_report_counts_pending_approval_backlog_outside_period() -> None:
    today = date.today()
    project_code = "P-BACKLOG-001"
    old_created_at = (today - timedelta(days=30)).isoformat() + "T00:00:00"

    procurement_response = client.post(
        "/api/v1/business/procurements",
        headers=headers,
        json={
            "data": {
                "code": "PROC-BACKLOG-001",
                "name": "Backlog procurement",
                "project_code": project_code,
                "approval_status": StatusValue.PENDING_APPROVAL,
                "created_at": old_created_at,
            },
        },
    )
    assert procurement_response.status_code == 200

    expense_response = client.post(
        "/api/v1/business/expenses",
        headers=headers,
        json={
            "data": {
                "code": "EXP-BACKLOG-001",
                "expense_type": "办公",
                "amount": 50,
                "project_code": project_code,
                "approval_status": StatusValue.PENDING_APPROVAL,
                "created_at": old_created_at,
            },
        },
    )
    assert expense_response.status_code == 200

    report_response = client.post(
        "/api/v1/reports/work-reports/generate",
        headers=headers,
        json={
            "report_type": ReportType.DAILY,
            "project_code": project_code,
            "period_start": today.isoformat(),
            "period_end": today.isoformat(),
            "persist": False,
        },
    )
    assert report_response.status_code == 200
    metrics = report_response.json()["report"]["metrics"]
    assert metrics["procurements_pending"] == 1
    assert metrics["expenses_pending"] == 1


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()[AIResponseKey.PROVIDER] == AIProviderName.NOOP


def test_legacy_project_payload_does_not_create_legacy_none_code() -> None:
    payload = LegacyMySQLService(None)._project_payload({"name": "Missing Id"}, {})

    assert payload["code"] is None
    assert payload["external_id"] is None


def test_legacy_readonly_query_requires_allowlist(monkeypatch) -> None:
    monkeypatch.delenv("LEGACY_PROJECT_QUERY", raising=False)
    get_settings.cache_clear()
    try:
        with pytest.raises(HTTPException) as blocked_exc_info:
            LegacyMySQLService(None).execute_readonly("SELECT id FROM secret_projects")
        assert blocked_exc_info.value.status_code == 403

        monkeypatch.setenv("LEGACY_PROJECT_QUERY", "SELECT id FROM projects")
        get_settings.cache_clear()

        def unavailable_engine():
            raise HTTPException(status_code=503, detail="legacy unavailable")

        monkeypatch.setattr(
            LegacyMySQLService,
            "_ensure_engine",
            staticmethod(unavailable_engine),
        )
        with pytest.raises(HTTPException) as engine_exc_info:
            LegacyMySQLService(None).execute_readonly("SELECT id FROM projects")
        assert engine_exc_info.value.status_code == 503
    finally:
        monkeypatch.delenv("LEGACY_PROJECT_QUERY", raising=False)
        get_settings.cache_clear()