241 lines
8.2 KiB
Python
241 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
import struct
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import APIRouter, HTTPException
|
|
from fastapi.routing import APIRoute
|
|
|
|
from k1link.web import l34f_adjudication_api
|
|
from k1link.web.l34f_adjudication_api import (
|
|
L34FCreateRequest,
|
|
L34FFrameRequest,
|
|
L34FFreezeRequest,
|
|
L34FObjectRequest,
|
|
L34FSaveRequest,
|
|
build_l34f_adjudication_router,
|
|
)
|
|
|
|
|
|
def _endpoint(router: APIRouter, path: str, method: str = "GET") -> object:
|
|
for route in router.routes:
|
|
if (
|
|
isinstance(route, APIRoute)
|
|
and route.path == path
|
|
and route.methods is not None
|
|
and method in route.methods
|
|
):
|
|
return route.endpoint
|
|
raise AssertionError(f"{method} {path} route is missing")
|
|
|
|
|
|
def _png_header(width: int, height: int) -> bytes:
|
|
return (
|
|
b"\x89PNG\r\n\x1a\n" + struct.pack(">I", 13) + b"IHDR" + struct.pack(">II", width, height)
|
|
)
|
|
|
|
|
|
def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
diagnostic_id = f"l34e-self-review-diagnostic-{'e' * 64}"
|
|
diagnostic_root = tmp_path / "diagnostics"
|
|
result_root = diagnostic_root / diagnostic_id
|
|
result_root.mkdir(parents=True)
|
|
image = tmp_path / "frame.png"
|
|
image.write_bytes(_png_header(800, 600))
|
|
cases = []
|
|
for sequence in range(1, 33):
|
|
cases.append(
|
|
{
|
|
"truth_island_sequence": sequence,
|
|
"image_id": sequence,
|
|
"frame_index": sequence * 10,
|
|
"group_id": f"group-{sequence}",
|
|
"session_seconds": float(sequence),
|
|
"source_image_sha256": "a" * 64,
|
|
"predictions": [
|
|
{
|
|
"prediction_index": 1,
|
|
"category": "car",
|
|
"score": 0.91,
|
|
"box_xyxy": [10.0, 20.0, 100.0, 120.0],
|
|
"diagnostic_verdict": "localization_disagreement",
|
|
"associated_object_id": f"self-{sequence:02d}-01",
|
|
}
|
|
],
|
|
"references": [
|
|
{
|
|
"object_id": f"self-{sequence:02d}-01",
|
|
"category": "car",
|
|
"proposed_label": None,
|
|
"box_xyxy": [12.0, 22.0, 102.0, 122.0],
|
|
"occluded": False,
|
|
"truncated": False,
|
|
}
|
|
],
|
|
"diagnostic_summary": {"severity_score": 1},
|
|
}
|
|
)
|
|
diagnostic = {
|
|
"result_id": diagnostic_id,
|
|
"result_root": result_root,
|
|
"manifest": {
|
|
"identity": {"self_review": {"session_id": f"l34-annotation-session-{'b' * 64}"}}
|
|
},
|
|
"report": {"source_session_id": "RAVNOVES00"},
|
|
"cases": tuple(cases),
|
|
}
|
|
monkeypatch.setattr(
|
|
l34f_adjudication_api,
|
|
"read_l34e_self_review_diagnostic",
|
|
lambda path: diagnostic,
|
|
)
|
|
|
|
def resolve(**kwargs: object):
|
|
sequence = int(kwargs["sequence"])
|
|
return diagnostic, cases[sequence - 1], image
|
|
|
|
monkeypatch.setattr(l34f_adjudication_api, "resolve_l34e_case_source", resolve)
|
|
router = build_l34f_adjudication_router(
|
|
diagnostic_root_provider=lambda: diagnostic_root,
|
|
adjudication_root_provider=lambda: tmp_path / "adjudication",
|
|
result_root_provider=lambda: tmp_path / "frozen",
|
|
)
|
|
return router, diagnostic_id, diagnostic
|
|
|
|
|
|
def test_l34f_case_separates_candidate_and_hides_scores(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
router, diagnostic_id, _ = _fixture(tmp_path, monkeypatch)
|
|
case = _endpoint(
|
|
router,
|
|
"/api/v1/laboratory/l34f/diagnostics/{diagnostic_id}/cases/{sequence}",
|
|
)(diagnostic_id=diagnostic_id, sequence=1) # type: ignore[operator]
|
|
|
|
assert case["model_scores_included"] is False
|
|
assert case["ground_truth"] is False
|
|
assert case["candidate_objects"][0]["category"] == "car"
|
|
assert "score" not in case["candidate_objects"][0]
|
|
assert "references" not in case
|
|
|
|
|
|
def test_l34f_session_is_revisioned_and_freezes_only_at_32_of_32(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
router, diagnostic_id, diagnostic = _fixture(tmp_path, monkeypatch)
|
|
created = _endpoint(
|
|
router,
|
|
"/api/v1/laboratory/l34f/diagnostics/{diagnostic_id}/sessions",
|
|
"POST",
|
|
)(
|
|
diagnostic_id=diagnostic_id,
|
|
request=L34FCreateRequest(idempotency_key="create-l34f-1"),
|
|
) # type: ignore[operator]
|
|
assert created["revision"] == 0
|
|
assert created["progress"]["reviewed_frame_count"] == 0
|
|
assert len(created["frames"]) == 32
|
|
assert created["frames"][0]["objects"][0]["origin"] == "self_review_seed"
|
|
assert created["authority"]["ground_truth"] is False
|
|
|
|
frames = []
|
|
for source in diagnostic["cases"]:
|
|
reference = source["references"][0]
|
|
frames.append(
|
|
L34FFrameRequest(
|
|
truth_island_sequence=source["truth_island_sequence"],
|
|
reviewed=source["truth_island_sequence"] != 32,
|
|
objects=[
|
|
L34FObjectRequest(
|
|
object_id=reference["object_id"],
|
|
category="car",
|
|
proposed_label=None,
|
|
origin="self_review_seed",
|
|
box_xyxy=reference["box_xyxy"],
|
|
occluded=False,
|
|
truncated=False,
|
|
)
|
|
],
|
|
)
|
|
)
|
|
save = _endpoint(
|
|
router,
|
|
"/api/v1/laboratory/l34f/diagnostics/{diagnostic_id}/sessions/{session_id}",
|
|
"PUT",
|
|
)
|
|
saved = save( # type: ignore[operator]
|
|
diagnostic_id=diagnostic_id,
|
|
session_id=created["session_id"],
|
|
request=L34FSaveRequest(
|
|
expected_revision=0,
|
|
idempotency_key="save-l34f-1",
|
|
title=created["title"],
|
|
frames=frames,
|
|
),
|
|
)
|
|
assert saved["revision"] == 1
|
|
assert saved["progress"]["reviewed_frame_count"] == 31
|
|
|
|
freeze = _endpoint(
|
|
router,
|
|
("/api/v1/laboratory/l34f/diagnostics/{diagnostic_id}/sessions/{session_id}/freeze"),
|
|
"POST",
|
|
)
|
|
with pytest.raises(HTTPException) as caught:
|
|
freeze( # type: ignore[operator]
|
|
diagnostic_id=diagnostic_id,
|
|
session_id=created["session_id"],
|
|
request=L34FFreezeRequest(expected_revision=1),
|
|
)
|
|
assert caught.value.status_code == 422
|
|
|
|
frames[-1].reviewed = True
|
|
completed = save( # type: ignore[operator]
|
|
diagnostic_id=diagnostic_id,
|
|
session_id=created["session_id"],
|
|
request=L34FSaveRequest(
|
|
expected_revision=1,
|
|
idempotency_key="save-l34f-2",
|
|
title=created["title"],
|
|
frames=frames,
|
|
),
|
|
)
|
|
assert completed["progress"]["complete"] is True
|
|
|
|
def build(**kwargs: object):
|
|
assert Path(kwargs["adjudication_session_path"]).is_file()
|
|
return {
|
|
"result_id": f"l34f-adjudicated-reference-{'f' * 64}",
|
|
"manifest": {
|
|
"identity": {
|
|
"l34e_diagnostic": {"result_id": diagnostic_id},
|
|
"adjudication_session": {"session_id": created["session_id"]},
|
|
}
|
|
},
|
|
"report": {
|
|
"created_at_utc": "2026-08-03T00:00:00.000Z",
|
|
"status": "completed-candidate-visible-adjudication-not-truth",
|
|
"source_session_id": "RAVNOVES00",
|
|
"camera_source_id": "sensor.camera.right",
|
|
"metrics": {"frame_count": 32},
|
|
"decision": {"candidate_accepted": False},
|
|
"limitations": [],
|
|
"authority": completed["authority"],
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
l34f_adjudication_api,
|
|
"build_l34f_adjudicated_reference",
|
|
build,
|
|
)
|
|
frozen = freeze( # type: ignore[operator]
|
|
diagnostic_id=diagnostic_id,
|
|
session_id=created["session_id"],
|
|
request=L34FFreezeRequest(expected_revision=2),
|
|
)
|
|
assert frozen["ground_truth"] is False
|
|
assert frozen["decision"]["candidate_accepted"] is False
|