diff --git a/services/api/tests/test_resolve_determinism.py b/services/api/tests/test_resolve_determinism.py index a3f2a14..0a522f9 100644 --- a/services/api/tests/test_resolve_determinism.py +++ b/services/api/tests/test_resolve_determinism.py @@ -2,8 +2,9 @@ The removed Python scorer is deliberately not tested because it no longer exists. These tests pin the load-bearing adapter semantics instead: -canonical match edges merge; review/non-match edges do not; merge application -is deterministic; bridge failure leaves records unmerged. +canonical match edges merge; review/non-match edges do not; hard negative +identity evidence constrains the whole cluster; bridge failure leaves records +unmerged. """ from __future__ import annotations @@ -38,22 +39,19 @@ def _rec(name: str, source: str, confidence: float = 0.9) -> dict: } -def _pair(a: str, b: str, status: str = "match") -> dict: +def _pair(a: str, b: str, status: str = "match", decided_by: str = "score", score: float = 0.97) -> dict: return { "candidate_id": f"c_{a}_{b}", "record_a": a, "record_b": b, - "score": 0.97, + "score": score, "status": status, - "decided_by": "score", + "decided_by": decided_by, } def test_apply_matches_is_deterministic_and_uses_canonical_edges_only(): records = [_rec("Acme AB", "source-a", 0.91), _rec("ACME Aktiebolag", "source-b", 0.96)] - # The helper's ids are private implementation detail. Capture them by - # asking the live adapter body in the async test below; here we verify the - # no-edge contract and review-band safety directly. assert len(_apply_matches(copy.deepcopy(records), [])) == 2 @@ -106,6 +104,40 @@ async def test_pending_review_never_auto_merges(monkeypatch): assert all(not r.get("merged_from") for r in result) +@pytest.mark.asyncio +@respx.mock +async def test_hard_non_match_blocks_transitive_cluster_merge(monkeypatch): + """A-B and B-C matches may never imply A=C when canonical hard evidence says A!=C.""" + monkeypatch.setenv("OPT_ER_BRIDGE_URL", "https://edge.example.test") + monkeypatch.setenv("INTERNAL_TOKEN", "test-internal-token") + records = [ + _rec("Acme AB", "source-a", 0.95), + _rec("Acme AB", "source-b", 0.94), + _rec("Acme AB", "source-c", 0.93), + ] + + def responder(request: httpx.Request) -> httpx.Response: + body = __import__("json").loads(request.content) + a, b, c = [item["record_id"] for item in body["records"]] + return httpx.Response( + 200, + json={ + "pairs": [ + _pair(a, b, "match", "score", 0.98), + _pair(b, c, "match", "score", 0.97), + _pair(a, c, "non_match", "0pt.company.stable-identifiers.v2", 0.0), + ], + "review_queue": [], + }, + ) + + respx.post("https://edge.example.test/api/0pt/internal/er/candidates").side_effect = responder + result = await resolve(records, tenant_id="tenant-a", project_id="project-a") + assert len(result) == 2 + assert max(r.get("agreed_sources", 1) for r in result) == 2 + assert not any(set(r.get("merged_from", [])) >= {"source-b", "source-c"} for r in result) + + @pytest.mark.asyncio @respx.mock async def test_bridge_outage_fails_closed_to_unmerged_records(monkeypatch):