vent_ai_platform/tests/test_business_api.py

from io import BytesIO
import re

from fastapi.testclient import TestClient
from openpyxl import load_workbook

from app.core.config import settings
from app.db.repositories import document as document_repo
from app.db.init_db import init_db
from app.db.seed import seed_demo_data
from app.db.session import SessionLocal
from app.main import app
from app.services.document_business_templates import resolve_document_template


def test_seed_dashboard_diagnosis_and_evaluation(monkeypatch) -> None:
    init_db()
    with SessionLocal() as db:
        seed_demo_data(db)
        matched = resolve_document_template(
            db,
            title="上湾煤矿2025年7月份配风计划7.15(1)",
        )
        assert matched.code == "monthly_air_distribution_plan"
        capacity = resolve_document_template(
            db,
            title="上湾煤矿2025年通风能力核定",
        )
        assert capacity.code == "ventilation_capacity_verification_report"
        dust_plan = resolve_document_template(
            db,
            title="附件：2025年采掘作业场所粉尘分散度、游离二氧化硅含量监测",
        )
        assert dust_plan.code == "dust_dispersion_free_silica_monitoring_ledger"
        air_door_record = resolve_document_template(
            db,
            title="上湾煤矿行车风门检查记录",
        )
        assert air_door_record.code == "vehicle_air_door_inspection_record"

    with TestClient(app) as client:
        mines = client.get("/api/v1/mines")
        assert mines.status_code == 200
        mine_id = mines.json()[0]["id"]

        dashboard = client.get("/api/v1/dashboard/overview", params={"mine_id": mine_id})
        assert dashboard.status_code == 200
        assert dashboard.json()["mine_name"] == "上湾煤矿"

        documents = client.get(
            "/api/v1/documents",
            params={"mine_id": mine_id, "page": 1, "page_size": 10},
        )
        assert documents.status_code == 200
        document_id = documents.json()["items"][0]["id"]

        async def fake_process_document_run(run_id: str) -> None:
            return None

        monkeypatch.setattr(
            "app.api.v1.endpoints.documents.process_document_run",
            fake_process_document_run,
        )
        template_id = client.get("/api/v1/documents/templates").json()[0]["id"]
        reprocess = client.post(
            f"/api/v1/documents/{document_id}/process",
            json={
                "force": True,
                "parse_mode": "lightweight",
                "template_id": template_id,
            },
        )
        assert reprocess.status_code == 202
        assert reprocess.json()["document_id"] == document_id
        assert reprocess.json()["status"] == "pending"
        processing_tasks = client.get(
            "/api/v1/documents/processing-tasks",
            params={"mine_id": mine_id, "limit": 10},
        )
        assert processing_tasks.status_code == 200
        assert "processing_seconds" in processing_tasks.json()[0]
        assert "started_at" in processing_tasks.json()[0]
        assert any(
            item["document_id"] == document_id
            for item in processing_tasks.json()
        )

        with SessionLocal() as db:
            document_repo.create_processing_run(
                db,
                {
                    "document_id": document_id,
                    "status": "completed",
                    "stage": "completed",
                    "progress": 100,
                    "document_type": "",
                    "parser_engine": "pytest",
                    "parsed_markdown_path": "",
                    "parsed_markdown": "PYTEST_FULLTEXT_TOKEN 用于验证解析全文检索。",
                    "model_output": {},
                    "error": "",
                    "metadata_json": {},
                },
            )
        fulltext_search = client.get(
            "/api/v1/documents/search",
            params={"mine_id": mine_id, "q": "PYTEST_FULLTEXT_TOKEN"},
        )
        assert fulltext_search.status_code == 200
        fulltext_search = fulltext_search.json()
        assert fulltext_search["total"] >= 1
        assert fulltext_search["items"][0]["document"]["id"] == document_id
        assert any(
            snippet["source"] == "parsed_text"
            for snippet in fulltext_search["items"][0]["snippets"]
        )
        export = client.get(
            "/api/v1/documents/export",
            params={"mine_id": mine_id},
        )
        assert export.status_code == 200
        assert export.content.startswith(b"PK")
        assert "spreadsheetml" in export.headers["content-type"]
        assert "filename*=UTF-8''" in export.headers["content-disposition"]
        workbook = load_workbook(BytesIO(export.content), data_only=True)
        sheet = workbook["资料清单"]
        assert "上湾煤矿业务资料清单" in str(sheet["A1"].value)
        assert sheet["A3"].value == "序号"
        assert sheet["B3"].value == "资料名称"
        assert sheet["E3"].value == "资料子类"
        assert all(
            sheet.cell(row_index, 5).value
            for row_index in range(4, sheet.max_row + 1)
        )
        document_types = [
            sheet.cell(row_index, 6).value
            for row_index in range(4, sheet.max_row + 1)
        ]
        assert all(
            not value or re.fullmatch(r"[a-z0-9_]+", str(value)) is None
            for value in document_types
        )
        workbook.close()

        preview_file = settings.upload_path / "pytest_preview_source.csv"
        preview_file.parent.mkdir(parents=True, exist_ok=True)
        preview_file.write_text("column_a,column_b\nvalue_1,value_2\n", encoding="utf-8")
        preview_document = client.post(
            "/api/v1/documents",
            json={
                "mine_id": mine_id,
                "title": "pytest preview document",
                "category": "TEST",
                "file_name": "pytest_preview_source.csv",
                "file_path": str(preview_file),
                "file_ext": ".csv",
                "file_size": preview_file.stat().st_size,
            },
        )
        assert preview_document.status_code == 201
        preview_document_id = preview_document.json()["id"]
        preview = client.get(f"/api/v1/documents/{preview_document_id}/preview")
        assert preview.status_code == 200
        preview_data = preview.json()
        assert preview_data["preview_type"] == "text"
        assert preview_data["source_available"] is True
        assert preview_data["can_download"] is True
        assert "column_a,column_b" in preview_data["text"]
        inline_file = client.get(f"/api/v1/documents/{preview_document_id}/file")
        assert inline_file.status_code == 200
        assert "inline" in inline_file.headers["content-disposition"]
        assert inline_file.content == preview_file.read_bytes()
        download_file = client.get(
            f"/api/v1/documents/{preview_document_id}/file",
            params={"download": "true"},
        )
        assert download_file.status_code == 200
        assert "attachment" in download_file.headers["content-disposition"]
        assert client.delete(f"/api/v1/documents/{preview_document_id}").status_code == 200

        statistics = client.get(
            "/api/v1/documents/statistics",
            params={"mine_id": mine_id},
        )
        assert statistics.status_code == 200
        statistics_data = statistics.json()
        assert statistics_data["total_documents"] >= 1
        assert statistics_data["extracted_fields"] >= 2
        assert 0 <= statistics_data["field_completeness"] <= 100
        assert 0 <= statistics_data["extraction_accuracy"] <= 100
        assert statistics_data["pending_tasks"] >= 1
        assert statistics_data["estimated_minutes"] >= 0
        assert statistics_data["storage_used_bytes"] >= 0
        assert statistics_data["storage_capacity_bytes"] > 0
        assert 0 <= statistics_data["storage_usage_percent"] <= 100
        assert statistics_data["expected_fields"] >= statistics_data["extracted_fields"]
        assert (
            statistics_data["valid_fields"]
            + statistics_data["invalid_fields"]
            + statistics_data["missing_fields"]
            == statistics_data["expected_fields"]
        )
        expected_completeness = round(
            statistics_data["valid_fields"]
            / max(statistics_data["expected_fields"], 1)
            * 100,
            1,
        )
        assert statistics_data["field_completeness"] == expected_completeness
        document_detail = client.get(f"/api/v1/documents/{document_id}")
        assert document_detail.status_code == 200
        assert document_detail.json()["id"] == document_id

        metadata_update = client.patch(
            f"/api/v1/documents/{document_id}",
            json={
                "title": "更新后的资料名称",
                "code": "UPDATED-001",
                "category": "瓦斯防治",
            },
        )
        assert metadata_update.status_code == 200
        assert metadata_update.json()["title"] == "更新后的资料名称"
        assert metadata_update.json()["code"] == "UPDATED-001"
        assert metadata_update.json()["category"] == "瓦斯防治"

        extraction = client.post(
            f"/api/v1/documents/{document_id}/extractions",
            json={
                "document_id": document_id,
                "field_key": "test_field",
                "field_name": "测试字段",
                "field_value": "原始值",
                "normalized_value": "原始值",
            },
        )
        assert extraction.status_code == 201
        extraction = extraction.json()
        assert extraction["edited"] is False
        assert extraction["extraction_correct"] is True
        updated_extraction = client.patch(
            f"/api/v1/documents/{document_id}/extractions/{extraction['id']}",
            json={
                "field_value": "前端编辑值",
                "normalized_value": "前端编辑值",
            },
        )
        assert updated_extraction.status_code == 200
        updated_extraction = updated_extraction.json()
        assert updated_extraction["normalized_value"] == "前端编辑值"
        assert updated_extraction["edited"] is True
        assert updated_extraction["original_value"] == "原始值"
        assert updated_extraction["edit_judgement_status"] == "pending"
        assert updated_extraction["extraction_correct"] is True

        statistics_after_edit = client.get(
            "/api/v1/documents/statistics",
            params={"mine_id": mine_id},
        )
        assert statistics_after_edit.status_code == 200
        statistics_after_edit = statistics_after_edit.json()
        assert statistics_after_edit["edited_fields"] >= 1
        assert statistics_after_edit["incorrect_fields"] >= 1

        reverted_extraction = client.patch(
            f"/api/v1/documents/{document_id}/extractions/{extraction['id']}",
            json={
                "field_value": "原始值",
                "normalized_value": "原始值",
            },
        )
        assert reverted_extraction.status_code == 200
        reverted_extraction = reverted_extraction.json()
        assert reverted_extraction["edited"] is False
        assert reverted_extraction["edit_judgement_status"] == "completed"
        assert reverted_extraction["edit_judgement_source"] == "reverted"
        assert reverted_extraction["extraction_correct"] is True

        statistics_after_revert = client.get(
            "/api/v1/documents/statistics",
            params={"mine_id": mine_id},
        )
        assert statistics_after_revert.status_code == 200
        statistics_after_revert = statistics_after_revert.json()
        assert statistics_after_revert["edited_fields"] == 0
        assert statistics_after_revert["incorrect_fields"] == 0

        deleted = client.delete(f"/api/v1/documents/{document_id}")
        assert deleted.status_code == 200
        assert client.get(f"/api/v1/documents/{document_id}").status_code == 404

        templates = client.get("/api/v1/documents/templates")
        assert templates.status_code == 200
        assert {item["code"] for item in templates.json()} == {
            "ground_main_fan_inspection_report",
            "main_fan_external_leakage_rate_report",
            "monthly_air_distribution_plan",
            "mine_ventilation_resistance_report",
            "ventilation_capacity_verification_report",
            "airflow_measurement_decadal_report",
            "gas_grade_appraisal_report",
            "coal_dust_explosion_spontaneous_combustion_report",
            "safety_monitoring_system_ledger",
            "dust_dispersion_free_silica_monitoring_ledger",
            "vehicle_air_door_inspection_record",
        }

        diagnosis = client.post(
            "/api/v1/diagnosis/run",
            json={"mine_id": mine_id, "scene": "all", "requested_by": "pytest"},
        )
        assert diagnosis.status_code == 201
        assert diagnosis.json()["status"] == "completed"

        evaluation = client.post(
            "/api/v1/evaluations/run",
            json={"mine_id": mine_id, "requested_by": "pytest"},
        )
        assert evaluation.status_code == 201
        assert evaluation.json()["grade"] in {"A", "B", "C", "D"}