"""Unit tests for cores.correlation — matchers + graph builder.""" from __future__ import annotations import numpy as np from cores.correlation import ( CorrelationGraphBuilder, Node, EdgeType, match_faces, match_objects, match_locations, match_cameras, match_hashes, match_embeddings, match_metadata, match_timestamps, ) class TestMatchFaces: def test_identical_embeddings_match(self): emb = np.random.randn(128).astype(np.float32) matches = match_faces([("f1", emb), ("f2", emb)], threshold=0.9) assert len(matches) == 1 assert matches[0][0] == "f1" assert matches[0][1] == "f2" assert matches[0][2] >= 0.9 def test_different_embeddings_no_match(self): emb1 = np.random.randn(128).astype(np.float32) emb2 = -emb1 matches = match_faces([("f1", emb1), ("f2", emb2)], threshold=0.9) assert len(matches) == 0 def test_empty_list(self): assert match_faces([]) == [] class TestMatchObjects: def test_same_class_overlapping(self): objects = [ ("o1", "car", {"x": 0, "y": 0, "w": 100, "h": 100}), ("o2", "car", {"x": 10, "y": 10, "w": 100, "h": 100}), ] matches = match_objects(objects, iou_threshold=0.5) assert len(matches) == 1 def test_different_class_no_match(self): objects = [ ("o1", "car", {"x": 0, "y": 0, "w": 100, "h": 100}), ("o2", "truck", {"x": 10, "y": 10, "w": 100, "h": 100}), ] matches = match_objects(objects, iou_threshold=0.5) assert len(matches) == 0 def test_non_overlapping_no_match(self): objects = [ ("o1", "car", {"x": 0, "y": 0, "w": 50, "h": 50}), ("o2", "car", {"x": 200, "y": 200, "w": 50, "h": 50}), ] matches = match_objects(objects, iou_threshold=0.5) assert len(matches) == 0 class TestMatchLocations: def test_nearby_coordinates_match(self): locs = [ ("l1", {"lat": 48.85, "lon": 2.35}), ("l2", {"lat": 48.8501, "lon": 2.3501}), ] matches = match_locations(locs, distance_threshold=0.01) assert len(matches) == 1 def test_far_coordinates_no_match(self): locs = [ ("l1", {"lat": 48.85, "lon": 2.35}), ("l2", {"lat": 40.71, "lon": -74.01}), ] matches = match_locations(locs, distance_threshold=0.01) assert len(matches) == 0 class TestMatchCameras: def test_fingerprint_match(self): cameras = [ ("c1", "Canon", "EOS 5D", "abc123fingerprint"), ("c2", "Canon", "EOS 5D", "abc123fingerprint"), ] matches = match_cameras(cameras) assert len(matches) == 1 assert matches[0][2] >= 0.9 def test_make_model_match(self): cameras = [ ("c1", "Nikon", "D850", "fp1"), ("c2", "Nikon", "D850", "fp2"), ] matches = match_cameras(cameras) assert len(matches) == 1 assert 0.5 < matches[0][2] < 0.9 def test_different_cameras_no_match(self): cameras = [ ("c1", "Canon", "EOS", "fp1"), ("c2", "Nikon", "D850", "fp2"), ] matches = match_cameras(cameras) assert len(matches) == 0 class TestMatchHashes: def test_sha256_exact_match(self): hashes = [ ("h1", "abc123def456", "phash1"), ("h2", "abc123def456", "phash2"), ] matches = match_hashes(hashes) assert len(matches) == 1 assert matches[0][2] == 1.0 def test_phash_near_match(self): # Two pHashes that differ by 2 bits phash1 = "ffff" # all 1s phash2 = "fffd" # differs by 1 hex digit = up to 4 bits hashes = [ ("h1", "sha1", phash1), ("h2", "sha2", phash2), ] matches = match_hashes(hashes) assert len(matches) == 1 class TestMatchTimestamps: def test_close_timestamps_match(self): timestamps = [ ("t1", "2023-01-15T14:30:45Z"), ("t2", "2023-01-15T14:31:00Z"), # 15 seconds apart ] matches = match_timestamps(timestamps, tolerance_seconds=60) assert len(matches) == 1 def test_far_timestamps_no_match(self): timestamps = [ ("t1", "2023-01-15T14:30:45Z"), ("t2", "2023-06-15T14:30:45Z"), # months apart ] matches = match_timestamps(timestamps, tolerance_seconds=60) assert len(matches) == 0 class TestGraphBuilder: def test_build_empty(self): builder = CorrelationGraphBuilder() graph = builder.build() assert graph.num_nodes == 0 assert graph.num_edges == 0 def test_build_with_nodes_and_edges(self): builder = CorrelationGraphBuilder() builder.add_node(Node(id="f1", node_type="face", label="Face 1")) builder.add_node(Node(id="f2", node_type="face", label="Face 2")) builder.add_matches(EdgeType.SAME_FACE, [ ("f1", "f2", 0.95, "cosine=0.95"), ]) graph = builder.build() assert graph.num_nodes == 2 assert graph.num_edges == 1 assert graph.edges[0].edge_type == "same_face" assert graph.edges[0].confidence == 0.95 def test_deduplicates_nodes(self): builder = CorrelationGraphBuilder() builder.add_node(Node(id="f1", node_type="face")) builder.add_node(Node(id="f1", node_type="face")) # dup graph = builder.build() assert graph.num_nodes == 1 def test_deduplicates_edges(self): builder = CorrelationGraphBuilder() builder.add_matches(EdgeType.SAME_FACE, [ ("f1", "f2", 0.9, "a"), ("f2", "f1", 0.9, "b"), # reverse dup ]) graph = builder.build() assert graph.num_edges == 1