File size: 5,571 Bytes
23d337e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """Unit tests for normalization/merger.py."""
from __future__ import annotations
from confidence.engine import ConfidenceEngine
from confidence.conflicts import ConflictDetector
from models.providers import ProviderCapability
from normalization.merger import ReportMerger
from providers.base import ProviderResult
def _make_result(provider: str, capability: ProviderCapability,
normalized: dict, success: bool = True) -> ProviderResult:
return ProviderResult(
provider=provider,
capability=capability,
success=success,
elapsed_ms=10.0,
normalized=normalized,
)
class TestReportMerger:
def test_merge_empty_results(self):
engine = ConfidenceEngine()
cd = ConflictDetector()
merger = ReportMerger(engine, cd)
report = merger.merge({}, "hash", "job-id", 100.0, "detection")
assert report.metadata.image_hash == "hash"
assert report.metadata.job_id == "job-id"
assert report.metadata.total_elapsed_ms == 100.0
assert report.detections == []
assert report.matches == []
assert report.evidence == []
assert report.conflicts == []
def test_merge_detection_results(self):
engine = ConfidenceEngine()
cd = ConflictDetector()
merger = ReportMerger(engine, cd)
results = {
"haar": _make_result(
"haar", ProviderCapability.DETECTION,
{"boxes": [{"x": 10, "y": 20, "w": 100, "h": 120}],
"confidences": [0.95], "landmarks": None, "num_faces": 1},
),
}
report = merger.merge(results, "hash", "job-id", 50.0, "detection")
assert len(report.detections) == 1
assert report.detections[0].box == {"x": 10, "y": 20, "w": 100, "h": 120}
assert "haar" in report.detections[0].detected_by
assert report.detections[0].confidence.overall > 0.0
assert len(report.evidence) == 1
assert report.evidence[0].provider == "haar"
def test_merge_preserves_failed_results_as_evidence(self):
engine = ConfidenceEngine()
cd = ConflictDetector()
merger = ReportMerger(engine, cd)
results = {
"dnn": ProviderResult(
provider="dnn", capability=ProviderCapability.DETECTION,
success=False, elapsed_ms=5.0, error="model not loaded",
error_type="RuntimeError",
),
}
report = merger.merge(results, "hash", "job-id", 5.0, "detection")
# Failed provider shouldn't contribute detections
assert report.detections == []
# But should be preserved as evidence
assert len(report.evidence) == 1
assert report.evidence[0].success is False
assert report.evidence[0].error == "model not loaded"
assert "dnn" in report.metadata.providers_failed
def test_merge_image_analysis_results(self):
engine = ConfidenceEngine()
cd = ConflictDetector()
merger = ReportMerger(engine, cd)
results = {
"image_quality": _make_result(
"image_quality", ProviderCapability.IMAGE_ANALYSIS,
{"quality_score": 0.85, "brightness": 128.0, "contrast": 50.0,
"sharpness": 200.0, "noise_level": 5.0, "width": 200, "height": 200,
"channels": 3, "color_profile": "BGR", "dominant_colors": ["#ff0000"]},
),
}
report = merger.merge(results, "hash", "job-id", 30.0, "image_analysis")
assert len(report.image_analyses) == 1
assert report.image_analyses[0].provider == "image_quality"
assert report.image_analyses[0].quality_score == 0.85
assert report.image_analyses[0].confidence is not None
def test_merge_metadata_results(self):
engine = ConfidenceEngine()
cd = ConflictDetector()
merger = ReportMerger(engine, cd)
results = {
"exif": _make_result(
"exif", ProviderCapability.METADATA,
{"format": "JPEG", "exif": {"Make": "Canon"}, "camera_make": "Canon"},
),
}
report = merger.merge(results, "hash", "job-id", 20.0, "metadata")
assert len(report.metadata_extractions) == 1
assert report.metadata_extractions[0].format == "JPEG"
assert report.metadata_extractions[0].camera_make == "Canon"
def test_merge_forensics_results(self):
engine = ConfidenceEngine()
cd = ConflictDetector()
merger = ReportMerger(engine, cd)
results = {
"image_integrity": _make_result(
"image_integrity", ProviderCapability.FORENSICS,
{"integrity_score": 0.95, "manipulation_indicators": [],
"details": {"sha256": "abc123"}},
),
}
report = merger.merge(results, "hash", "job-id", 15.0, "forensics")
assert len(report.forensics) == 1
assert report.forensics[0].integrity_score == 0.95
def test_merge_collects_limitations(self):
engine = ConfidenceEngine()
cd = ConflictDetector()
merger = ReportMerger(engine, cd)
results = {
"dnn": ProviderResult(
provider="dnn", capability=ProviderCapability.DETECTION,
success=False, elapsed_ms=5.0, error="model not loaded",
),
}
report = merger.merge(results, "hash", "job-id", 5.0, "detection")
assert any("dnn" in lim for lim in report.metadata.limitations)
|