aps-agent/tests/golden/test_drawing_links.py

from __future__ import annotations

import time
import uuid
from copy import deepcopy
from pathlib import Path

from server.agent_core import harness
from server.agent_core.approval_store import utc_now
from server.aps_domain.drawing_understanding import build_drawing_link_evidence_records
from server.aps_domain.workflow import execute_confirmed
from server.auth.context import get_identity
from server.state.seed import seed_world
from server.state.store import WorldStore


def _selected_candidates() -> list[dict]:
    return [
        {"candidateType": "MATERIAL", "candidateId": "m-1", "code": "CTRL-A",
         "name": "Controller A", "revision": "e", "confidence": 0.9},
        {"candidateType": "BOM_REFERENCE", "candidateId": "b-1",
         "componentReference": "PCB-001", "parentMaterialCode": "CTRL-A",
         "confidence": 0.72, "revision": "e"},
        {"candidateType": "ROUTING_OPERATION", "candidateId": "r-1",
         "operationCode": "ASSEMBLY", "productCode": "CTRL-A",
         "confidence": 0.63, "revision": "e"},
    ]


def _stage_drawing_confirmation(store: WorldStore, selected: list[dict]) -> str:
    params = {
        "drawingId": "drawing-1",
        "sourceSha256": "0" * 64,
        "selectedCandidates": selected,
        "candidateIds": [row["candidateId"] for row in selected],
        "executionSupported": True,
    }
    identity = get_identity()
    confirm_id = f"drawing-link-{uuid.uuid4().hex[:12]}"
    now = time.time()
    record = {
        "confirmId": confirm_id,
        "sessionId": "drawing-test",
        "action": "drawing.master.apply",
        "power": "P2",
        "params": deepcopy(params),
        "paramsHash": harness._params_fingerprint(params),
        "tenantUuid": store.tenant_uuid,
        "ownerUserId": identity.user_id,
        "delegateUserId": None,
        "projectId": "default",
        "worldKey": store.world_key,
        "evidenceRefs": [],
        "beforeSnapshot": None,
        "beforeFingerprint": None,
        "requester": {"userId": identity.user_id, "username": identity.username},
        "approvals": [],
        "approvalStep": 0,
        "requiredApprovals": 1,
        "createdAt": utc_now(now),
        "createdAtEpoch": now,
        "expiresAt": utc_now(now + 3600),
        "expiresAtEpoch": now + 3600,
    }
    record["envelopeHash"] = harness._confirmation_envelope_fingerprint(record)
    assert harness._approval_store.stage(record, now_epoch=now)
    return confirm_id


def test_link_evidence_builder_adds_bom_routing_order_without_master_write():
    world = seed_world()
    before = deepcopy(world)
    counter = {"n": 0}

    def next_id(kind: str) -> int:
        counter["n"] += 1
        return counter["n"]

    records = build_drawing_link_evidence_records(
        world,
        drawing_id="drawing-1",
        source_sha256="0" * 64,
        selected_candidates=_selected_candidates(),
        actor="tester",
        confirm_id="confirm-1",
        next_id=next_id,
    )

    assert world == before
    assert {row["linkType"] for row in records} == {"BOM", "ORDER", "ROUTING"}
    bom = next(row for row in records if row["linkType"] == "BOM")
    routing = next(row for row in records if row["linkType"] == "ROUTING")
    orders = [row for row in records if row["linkType"] == "ORDER"]
    assert bom["status"] == "PENDING_REVIEW"
    assert bom["masterCommitted"] is False
    assert bom["confidence"] == 0.72
    assert bom["evidence"]["confirmId"] == "confirm-1"
    assert routing["status"] == "PENDING_REVIEW"
    assert routing["masterCommitted"] is False
    assert orders
    assert all(row["status"] == "CONFIRMED" for row in orders)
    assert all(row["masterCommitted"] is True for row in orders)


def test_drawing_master_apply_persists_extended_drawing_link_evidence(tmp_path: Path):
    store = WorldStore(path=tmp_path / "world.json")
    selected = _selected_candidates()
    before_master = {
        key: deepcopy(store.data.get(key, []))
        for key in ("boms", "bomItems", "routings", "routingSteps")
    }
    confirm_id = _stage_drawing_confirmation(store, selected)

    message = execute_confirmed(store, confirm_id, True, actor="tester")

    assert "Drawing review applied:" in message
    assert "links=" in message
    after_master = {
        key: deepcopy(store.data.get(key, [])) for key in before_master
    }
    assert after_master == before_master
    links = store.data.get("drawingLinks", [])
    assert {"BOM", "MATERIAL", "ROUTING"} <= {row["linkType"] for row in links}
    material = next(row for row in links if row["linkType"] == "MATERIAL")
    bom = next(row for row in links if row["linkType"] == "BOM")
    routing = next(row for row in links if row["linkType"] == "ROUTING")
    assert material["masterCommitted"] is True
    assert material["evidence"]["contractVersion"] == "drawing-link-evidence.v1"
    assert bom["masterCommitted"] is False
    assert bom["status"] == "PENDING_REVIEW"
    assert routing["masterCommitted"] is False
    assert routing["status"] == "PENDING_REVIEW"