geointel/backend/tests/test_postprocessing_parity_with_training_gate.py

"""What is promoted must be what the workbench then runs.

The candidate evaluation freezes its post-processing before the protected test
— NMS IoU and a containment threshold selected during calibration. The runtime
applied its own hardcoded containment value, so a model gated at one setting
was served at another and suppressed detections the gate had counted. The
difference is invisible in both reports.

Containment is therefore configuration, recorded with every run, and two runs
that post-processed differently are not comparable however good their numbers
look.
"""

from __future__ import annotations

import pytest
from shapely.geometry import box

from app.core.config import Settings
from app.services.detection_comparison_service import DetectionComparisonService
from app.services.detection_service import DetectionService


def _candidate(name: str, geometry, confidence: float):
    return {
        "class_name": "building",
        "confidence": confidence,
        "geometry": geometry,
        "bbox": [0.0, 0.0, 1.0, 1.0],
        "source_tile_path": f"/tiles/{name}.tif",
        "properties": {"name": name},
    }


class TestConfigurableContainment:
    def test_the_runtime_threshold_comes_from_settings(self) -> None:
        assert Settings(_env_file=None).yolo_containment_nms_threshold == pytest.approx(0.85)
        assert Settings(
            _env_file=None, yolo_containment_nms_threshold=1.0
        ).yolo_containment_nms_threshold == pytest.approx(1.0)

    def test_a_strict_threshold_suppresses_only_a_fully_nested_box(self) -> None:
        outer = _candidate("outer", box(0, 0, 10, 10), 0.9)
        # 90% of the smaller box lies inside the larger one, but their IoU is
        # only 0.09 — so only the containment rule can act on this pair.
        mostly_nested = _candidate("mostly", box(8.2, 1, 10.2, 6), 0.5)

        kept = DetectionService._suppress_duplicate_candidates(
            [outer, mostly_nested], iou_threshold=0.5, containment_threshold=1.0
        )

        assert [item["properties"]["name"] for item in kept] == ["outer", "mostly"]

    def test_a_looser_threshold_suppresses_it(self) -> None:
        outer = _candidate("outer", box(0, 0, 10, 10), 0.9)
        mostly_nested = _candidate("mostly", box(8.2, 1, 10.2, 6), 0.5)

        kept = DetectionService._suppress_duplicate_candidates(
            [outer, mostly_nested], iou_threshold=0.5, containment_threshold=0.7
        )

        assert [item["properties"]["name"] for item in kept] == ["outer"]

    def test_the_default_matches_the_documented_runtime_value(self) -> None:
        outer = _candidate("outer", box(0, 0, 10, 10), 0.9)
        nested = _candidate("nested", box(1, 1, 9, 9), 0.5)

        kept = DetectionService._suppress_duplicate_candidates([outer, nested], iou_threshold=0.5)

        assert [item["properties"]["name"] for item in kept] == ["outer"]


class TestComparabilityOfPostProcessing:
    def _entry(self, *, dataset_id, reference_id, containment: float, duplicate_iou: float = 0.5):
        from uuid import uuid4

        return {
            "analysis_run_id": uuid4(),
            "dataset_id": dataset_id,
            "model_id": "yolo-configured",
            "model_asset_id": "asset",
            "reference_dataset_id": reference_id,
            "coverage_mode": "persisted_tile_manifest_union",
            "reference_evaluated_count": 100,
            "containment_suppression_threshold": containment,
            "duplicate_iou_threshold": duplicate_iou,
        }

    def test_runs_with_the_same_post_processing_stay_comparable(self) -> None:
        from uuid import uuid4

        dataset_id, reference_id = uuid4(), uuid4()
        report = DetectionComparisonService.assess_comparability(
            [
                self._entry(dataset_id=dataset_id, reference_id=reference_id, containment=0.85),
                self._entry(dataset_id=dataset_id, reference_id=reference_id, containment=0.85),
            ]
        )

        assert report["comparable"] is True

    def test_a_different_containment_threshold_blocks_the_comparison(self) -> None:
        from uuid import uuid4

        dataset_id, reference_id = uuid4(), uuid4()
        report = DetectionComparisonService.assess_comparability(
            [
                self._entry(dataset_id=dataset_id, reference_id=reference_id, containment=0.85),
                self._entry(dataset_id=dataset_id, reference_id=reference_id, containment=1.0),
            ]
        )

        assert report["comparable"] is False
        assert "different_post_processing" in report["blocking_reasons"]

    def test_a_different_duplicate_iou_blocks_the_comparison(self) -> None:
        from uuid import uuid4

        dataset_id, reference_id = uuid4(), uuid4()
        report = DetectionComparisonService.assess_comparability(
            [
                self._entry(dataset_id=dataset_id, reference_id=reference_id, containment=0.85, duplicate_iou=0.5),
                self._entry(dataset_id=dataset_id, reference_id=reference_id, containment=0.85, duplicate_iou=0.7),
            ]
        )

        assert report["comparable"] is False
        assert "different_post_processing" in report["blocking_reasons"]

    def test_runs_from_before_the_setting_existed_do_not_block(self) -> None:
        """Older runs recorded no threshold; absence is not a difference."""

        from uuid import uuid4

        dataset_id, reference_id = uuid4(), uuid4()
        entries = [
            self._entry(dataset_id=dataset_id, reference_id=reference_id, containment=0.85),
            self._entry(dataset_id=dataset_id, reference_id=reference_id, containment=0.85),
        ]
        for entry in entries:
            entry.pop("containment_suppression_threshold")
            entry.pop("duplicate_iou_threshold")

        assert DetectionComparisonService.assess_comparability(entries)["comparable"] is True