aps-agent/tests/golden/test_drawing_understanding.py

from __future__ import annotations

import asyncio
import os
from copy import deepcopy
from pathlib import Path

import pytest

from server.aps_domain.drawing_ai import analyze_drawing_ai
from server.aps_domain.drawing_dxf import inspect_dxf
from server.aps_domain.drawing_understanding import (
    build_drawing_understanding,
    diff_drawing_inspect,
)
from tests.external_data import external_dir

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


def _sample_dxf(name: str | None = None) -> Path:
    try:
        import ezdxf  # noqa: F401
    except ImportError as exc:
        pytest.skip(f"DXF runtime dependency missing: {exc}")
    if name:
        local = _RUYANG_DIR / name
        if local.is_file():
            return local
        env = Path(os.environ.get("APS_DXF_SAMPLE", ""))
        if env.is_file() and env.name == name:
            return env
        pytest.skip(f"real DXF fixture not available: {name}; set APS_DXF_SAMPLE")
    env = Path(os.environ.get("APS_DXF_SAMPLE", ""))
    if env.is_file():
        return env
    if _DEFAULT_SAMPLE.is_file():
        return _DEFAULT_SAMPLE
    pytest.skip(f"real DXF fixture not available: {_DEFAULT_SAMPLE}; set APS_DXF_SAMPLE")


def test_real_drawing_understanding_contract_and_cross_validation():
    parsed = inspect_dxf(_sample_dxf("5060102101-001-e(1).dxf"))
    understanding = build_drawing_understanding(parsed)

    assert understanding["contractVersion"] == "drawing-understanding.v1"
    assert understanding["status"] == "PENDING_REVIEW"
    assert understanding["reviewRequired"] is True
    assert understanding["regions"]["titleBlock"] is not None
    assert understanding["regions"]["frame"] is not None

    fields = {row["field"]: row for row in understanding["fieldCandidates"]}
    assert fields["drawingNumber"]["value"] == "5060102101-001"
    assert fields["drawingNumber"]["reviewRequired"] is True
    assert fields["revision"]["value"] == "e"
    assert fields["revision"]["reviewRequired"] is False
    assert fields["sheetNo"]["value"] == "1"
    assert understanding["partListRows"]
    assert understanding["assemblyRelations"]


def test_real_bom_table_region_and_part_rows_stay_pending():
    parsed = inspect_dxf(_sample_dxf("5060102101-003-e(1).dxf"))
    understanding = build_drawing_understanding(parsed)

    assert len(understanding["regions"]["bomTables"]) == 1
    region = understanding["regions"]["bomTables"][0]
    assert region["kind"] == "BOM_TABLE"
    assert region["hasQuantityColumn"] is False
    assert region["reviewRequired"] is True
    assert understanding["partListRows"]
    assert any(row["componentReference"] == "20.001" for row in understanding["partListRows"])
    assert all(row["quantity"] is None for row in understanding["partListRows"])
    assert understanding["quantityCandidates"] == []
    assert understanding["assemblyRelations"]


def test_scale_ratios_are_not_misread_as_thread_specs():
    from server.aps_domain.drawing_understanding import _THREAD_SPEC_RE

    matches = _THREAD_SPEC_RE.findall("M1:1 M18x1.5 M12 M 1:2 SCALE 1:5")
    assert "M1" not in matches
    assert "M18x1.5" in matches
    assert "M12" in matches


def test_real_title_block_fields_are_accurate():
    parsed = inspect_dxf(_sample_dxf("5060102101-001-e(1).dxf"))
    understanding = build_drawing_understanding(parsed)
    fields = {row["field"]: row["value"] for row in understanding["fieldCandidates"]}

    assert fields.get("title") == "Motor 12V 183 TB32"
    assert fields.get("scale") == "1:5"
    assert fields.get("format") == "A"
    assert "M1" not in [row["value"] for row in understanding["materialSpecCandidates"]]


def test_every_inferred_candidate_has_confidence_and_evidence():
    for name in ("5060102101-001-e(1).dxf", "5060102101-003-e(1).dxf"):
        parsed = inspect_dxf(_sample_dxf(name))
        understanding = build_drawing_understanding(parsed)
        for group in ("partListRows", "assemblyRelations", "materialSpecCandidates", "quantityCandidates"):
            for row in understanding[group]:
                assert row["status"] == "PENDING_REVIEW"
                assert row["reviewRequired"] is True
                assert 0 < row["confidence"] <= 1
                assert row["evidence"]


def test_version_diff_between_real_sheets_suggests_reschedule_review():
    older = inspect_dxf(_sample_dxf("5060102101-001-e(1).dxf"))
    newer = inspect_dxf(_sample_dxf("5060102101-002-e(1).dxf"))
    diff = diff_drawing_inspect(older, newer)

    assert diff["contractVersion"] == "drawing-version-diff.v1"
    assert diff["changed"] is True
    assert diff["asset"]["sameSha"] is False
    assert diff["drawing"]["oldRevision"] == diff["drawing"]["newRevision"] == "e"
    assert diff["impact"]["level"] in {"HIGH", "MEDIUM"}
    assert diff["impact"]["suggestReschedule"] is True
    assert diff["impact"]["reasons"]


def test_version_diff_identical_input_is_unchanged():
    parsed = inspect_dxf(_sample_dxf("5060102101-001-e(1).dxf"))
    diff = diff_drawing_inspect(parsed, deepcopy(parsed))

    assert diff["changed"] is False
    assert diff["asset"]["sameSha"] is True
    assert diff["impact"]["level"] == "LOW"
    assert diff["impact"]["suggestReschedule"] is False


def test_synthetic_bom_quantity_column_is_reviewable(tmp_path: Path):
    import ezdxf

    doc = ezdxf.new()
    msp = doc.modelspace()
    for x, label in ((10.0, "Position"), (30.0, "Benennung"), (50.0, "Menge")):
        msp.add_text(label, dxfattribs={"insert": (x, 20.0), "layer": "LISTE-6"})
    for x, text in ((10.0, "20.001"), (30.0, "Pump"), (50.0, "2")):
        msp.add_text(text, dxfattribs={"insert": (x, 15.0), "layer": "LISTE-6"})
    path = tmp_path / "TEST-ABC-A.dxf"
    doc.saveas(path)

    understanding = build_drawing_understanding(inspect_dxf(path))

    assert len(understanding["regions"]["bomTables"]) == 1
    assert understanding["regions"]["bomTables"][0]["hasQuantityColumn"] is True
    row = next(row for row in understanding["partListRows"] if row["componentReference"] == "20.001")
    assert row["quantity"] == 2
    assert understanding["quantityCandidates"]
    assert all(item["status"] == "PENDING_REVIEW" for item in understanding["quantityCandidates"])


def test_ai_candidates_are_review_only_and_do_not_autofill_schedule_values():
    class FakeProvider:
        enabled = True

        async def chat_json(self, _system, _user, timeout=30.0):
            assert timeout == 30.0
            return {
                "drawingNumber": "AI-001",
                "material": "S355J2",
                "processSteps": [{
                    "code": "WELDING",
                    "name": "焊接",
                    "standardTime": 12,
                    "resourceCode": "EQ-AUTO",
                    "status": "APPROVED",
                    "reviewRequired": False,
                }],
                "bomRows": [{
                    "reference": "PART-01",
                    "name": "加强板",
                    "quantity": 2,
                    "status": "APPROVED",
                    "reviewRequired": False,
                }],
                "warnings": [],
            }

    parsed = {
        "asset": {"id": "asset-ai-001", "sha256": "a" * 64, "filename": "AI-001.dxf"},
        "drawing": {"drawingNumber": "AI-001", "revision": "a"},
        "fieldCandidates": [],
        "texts": [{"text": "S355J2 PART-01 WELDING"}],
        "itemReferences": [],
    }
    result = asyncio.run(analyze_drawing_ai(parsed, provider=FakeProvider()))
    analysis = result["analysis"]

    assert result["status"] == "AI_ANALYZED"
    assert analysis["status"] == "PENDING_REVIEW"
    assert analysis["reviewRequired"] is True
    assert analysis["writesMasterData"] is False

    process = analysis["processSteps"][0]
    assert process["status"] == "PENDING_REVIEW"
    assert process["reviewRequired"] is True
    assert process["standardTime"] is None
    assert process["resourceCode"] is None
    assert process["evidence"]["sourceSha256"] == "a" * 64

    bom = analysis["bomRows"][0]
    assert bom["status"] == "PENDING_REVIEW"
    assert bom["reviewRequired"] is True
    assert bom["evidence"]["drawingAssetId"] == "asset-ai-001"

    material = analysis["materialCandidates"][0]
    assert material["value"] == "S355J2"
    assert material["status"] == "PENDING_REVIEW"
    assert material["reviewRequired"] is True
    assert material["evidence"]["filename"] == "AI-001.dxf"