aps-agent/tests/golden/test_drawing_manage.py

# ============================================================
# 图纸管理黄金测试：清除单张/清空项目 + AI 精确解析（fail-closed）。
# ============================================================
from __future__ import annotations

import asyncio
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from server.aps_domain.drawing_ai import analyze_drawing_ai
from server.gateway.app import create_app
from tests.auth_provider import install_test_auth
from tests.external_data import external_dir

_RUIYANG_DIR = external_dir("RUIYANG_DEMO_DIR", "ruiyang")
_DEFAULT_SAMPLE = _RUIYANG_DIR / "5060102101-001-e(1).dxf"
_SAMPLE_003 = _RUIYANG_DIR / "5060102101-003-e(1).dxf"


def _sample(path: Path) -> Path:
    if not path.is_file():
        pytest.skip(f"real DXF fixture not available: {path}")
    return path


class _FakeProvider:
    enabled = True

    def __init__(self, payload):
        self.payload = payload

    async def chat_json(self, system, user, timeout=20.0):
        if isinstance(self.payload, Exception):
            raise self.payload
        return self.payload


def test_ai_analysis_uses_llm_json_when_provider_available():
    parsed = {
        "asset": {"sha256": "0" * 64, "filename": "AI-001-a.dxf"},
        "drawing": {"drawingNumber": "AI-001", "revision": "a"},
        "fieldCandidates": [{"field": "drawingNumber", "value": "AI-001", "confidence": 0.9}],
        "texts": [{"text": "DREHEN Ø20 TOLERANZ ±0.01", "evidence": {"source": "drawingText"}}],
        "itemReferences": [],
    }
    payload = {
        "drawingNumber": "AI-001", "revision": "a", "title": "Test Part",
        "processSteps": [{"code": "TURNING", "name": "车削", "description": "DREHEN Ø20", "confidence": 0.9}],
        "details": {"tolerance": "±0.01"}, "confidence": 0.9, "warnings": [],
    }
    result = asyncio.run(analyze_drawing_ai(parsed, provider=_FakeProvider(payload)))
    assert result["status"] == "AI_ANALYZED"
    assert result["deterministic"] is False
    assert result["analysis"]["drawingNumber"] == "AI-001"


def test_ai_analysis_falls_back_when_provider_missing_or_invalid():
    parsed = {
        "asset": {"sha256": "0" * 64, "filename": "AI-001-a.dxf"},
        "drawing": {"drawingNumber": "AI-001", "revision": "a"},
        "fieldCandidates": [], "texts": [], "itemReferences": [],
    }
    missing = asyncio.run(analyze_drawing_ai(parsed, provider=None))
    assert missing["status"] == "AI_UNAVAILABLE"
    assert missing["deterministic"] is True

    invalid = asyncio.run(analyze_drawing_ai(parsed, provider=_FakeProvider("not json")))
    assert invalid["status"] == "AI_UNAVAILABLE"
    assert "message" in invalid


def test_ai_analysis_parses_double_encoded_and_markdown_json():
    parsed = {
        "asset": {"sha256": "0" * 64, "filename": "AI-001-a.dxf"},
        "drawing": {"drawingNumber": "AI-001", "revision": "a"},
        "fieldCandidates": [], "texts": [], "itemReferences": [],
    }
    double_encoded = asyncio.run(
        analyze_drawing_ai(parsed, provider=_FakeProvider('{"drawingNumber": "AI-001", "warnings": []}'))
    )
    assert double_encoded["status"] == "AI_ANALYZED"
    assert double_encoded["analysis"]["drawingNumber"] == "AI-001"

    markdown = asyncio.run(
        analyze_drawing_ai(parsed, provider=_FakeProvider('```json\n{"drawingNumber": "AI-001", "warnings": []}\n```'))
    )
    assert markdown["status"] == "AI_ANALYZED"


class _TextFallbackProvider(_FakeProvider):
    def __init__(self, text):
        super().__init__(None)
        self.text = text

    async def chat_text(self, system, user, timeout=45.0):
        return self.text


def test_ai_analysis_retries_with_chat_text_when_json_invalid():
    parsed = {
        "asset": {"sha256": "0" * 64, "filename": "AI-001-a.dxf"},
        "drawing": {"drawingNumber": "AI-001", "revision": "a"},
        "fieldCandidates": [], "texts": [], "itemReferences": [],
    }
    provider = _TextFallbackProvider('```json\n{"drawingNumber": "AI-001", "warnings": []}\n```')
    result = asyncio.run(analyze_drawing_ai(parsed, provider=provider))
    assert result["status"] == "AI_ANALYZED"
    assert result["analysis"]["drawingNumber"] == "AI-001"


def _auth_client(monkeypatch: pytest.MonkeyPatch, tenant: str) -> TestClient:
    install_test_auth(monkeypatch, tenant)
    client = TestClient(create_app())
    login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
    assert login.status_code == 200, login.text
    return client


def _create_project(client: TestClient, name: str, work_dir: Path) -> str:
    created = client.post("/api/projects", json={"name": name, "workDir": str(work_dir)})
    assert created.status_code == 200, created.text
    return created.json()["project"]["id"]


def _inspect(client: TestClient, pid: str, path: Path) -> str:
    resp = client.post("/api/drawings/inspect", json={"path": str(path), "projectId": pid})
    assert resp.status_code == 200, resp.text
    return (resp.json().get("parsed") or resp.json()["drawing"])["asset"]["id"]


def test_delete_drawing_clears_asset_and_candidates(monkeypatch, tmp_path: Path):
    sample = _sample(_DEFAULT_SAMPLE)
    client = _auth_client(monkeypatch, "tenant-drawing-delete")
    pid = _create_project(client, "图纸删除验收", _DEFAULT_SAMPLE.parent)
    drawing_id = _inspect(client, pid, sample)

    resp = client.delete(f"/api/drawings/{drawing_id}")
    assert resp.status_code == 200, resp.text
    assert resp.json()["deleted"] is True
    listing = client.get(f"/api/projects/{pid}/drawings")
    assert listing.status_code == 200
    assert listing.json()["count"] == 0


def test_clear_project_drawings_removes_all(monkeypatch, tmp_path: Path):
    first = _sample(_DEFAULT_SAMPLE)
    second = _sample(_SAMPLE_003)
    client = _auth_client(monkeypatch, "tenant-drawing-clear")
    pid = _create_project(client, "图纸清空验收", _SAMPLE_003.parent)
    _inspect(client, pid, first)
    _inspect(client, pid, second)

    resp = client.delete(f"/api/projects/{pid}/drawings")
    assert resp.status_code == 200, resp.text
    assert resp.json()["deleted"] == 2
    listing = client.get(f"/api/projects/{pid}/drawings")
    assert listing.json()["count"] == 0


def test_two_mb_upload_is_not_blocked_by_app_limit(monkeypatch, tmp_path: Path):
    client = _auth_client(monkeypatch, "tenant-large-upload")
    pid = _create_project(client, "大文件上传验收", tmp_path)
    payload = b"0\nSECTION\n2\nHEADER\n0\nENDSEC\n0\nEOF\n" + b"x" * 2_500_000
    resp = client.post(
        f"/api/projects/{pid}/files/upload",
        files=[("files", ("large-001-a.dxf", payload, "application/octet-stream"))],
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["saved"] == ["large-001-a.dxf"]
    assert (tmp_path / "large-001-a.dxf").stat().st_size == len(payload)


def test_process_text_analysis_endpoint(monkeypatch):
    client = _auth_client(monkeypatch, "tenant-process-text")
    resp = client.post("/api/drawings/analyze-text", json={"text": "烟囱(9005)分段在上建场地拼板装焊，完成焊接后翻身形成组立。"})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["status"] == "PENDING_REVIEW"
    assert any(row["blockCode"] == "9005" for row in body["blocks"])
    assert any(row["processName"] == "装焊" for row in body["operations"])
    assert body["sequenceSteps"]


def test_drawing_ai_api_endpoint(monkeypatch, tmp_path: Path):
    sample = _sample(_DEFAULT_SAMPLE)
    client = _auth_client(monkeypatch, "tenant-drawing-ai")
    pid = _create_project(client, "图纸 AI 验收", _DEFAULT_SAMPLE.parent)
    drawing_id = _inspect(client, pid, sample)
    monkeypatch.setattr(
        "server.agent_core.providers.get_provider",
        lambda: _FakeProvider({"drawingNumber": "5060102101-001", "revision": "e", "confidence": 0.9, "warnings": []}),
    )
    resp = client.post(f"/api/drawings/{drawing_id}/ai-analyze")
    assert resp.status_code == 200, resp.text
    assert resp.json()["status"] == "AI_ANALYZED"
    assert resp.json()["analysis"]["drawingNumber"] == "5060102101-001"

    monkeypatch.setattr("server.agent_core.providers.get_provider", lambda: _FakeProvider(None))
    fallback = client.post(f"/api/drawings/{drawing_id}/ai-analyze")
    assert fallback.status_code == 200, fallback.text
    assert fallback.json()["status"] == "AI_UNAVAILABLE"
    assert fallback.json()["deterministic"] is True