0pt/services/api/tests/test_resolve_determinism.py

"""Regression tests for the Python adapter to canonical TypeScript ER.

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; hard negative
identity evidence constrains the whole cluster; bridge failure leaves records
unmerged.
"""
from __future__ import annotations

import copy

import httpx
import pytest
import respx

from workers.resolve import _apply_matches, resolve


def _prov(field_name: str, source_id: str, confidence: float) -> dict:
    return {
        "field_name": field_name,
        "source_id": source_id,
        "confidence": confidence,
        "entity_resolution": None,
    }


def _rec(name: str, source: str, confidence: float = 0.9) -> dict:
    return {
        "market": "SE",
        "name": name,
        "status": "active",
        "lei": None,
        "org_nr": None,
        "source": source,
        "confidence": confidence,
        "provenance": [_prov("name", source, confidence)],
    }


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": score,
        "status": status,
        "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)]
    assert len(_apply_matches(copy.deepcopy(records), [])) == 2


@pytest.mark.asyncio
@respx.mock
async def test_canonical_match_merges_and_stamps_provenance(monkeypatch):
    monkeypatch.setenv("OPT_ER_BRIDGE_URL", "https://edge.example.test")
    monkeypatch.setenv("INTERNAL_TOKEN", "test-internal-token")
    records = [_rec("Acme AB", "source-a", 0.91), _rec("Acme AB", "source-b", 0.96)]

    route = respx.post("https://edge.example.test/api/0pt/internal/er/candidates")

    def responder(request: httpx.Request) -> httpx.Response:
        body = __import__("json").loads(request.content)
        assert request.headers["X-Internal-Token"] == "test-internal-token"
        assert body["tenant_id"] == "tenant-a"
        assert body["blocking_fields"] == ["market", "block_initial"]
        a, b = body["records"][0]["record_id"], body["records"][1]["record_id"]
        # Endast hard-rule-beslut (decided_by != score/None) far skapa en
        # canonical identity-lank - consumenten ignorerar medvetet score/"match".
        return httpx.Response(
            200,
            json={"pairs": [_pair(a, b, "match", "0pt.company.stable-identifiers.v3", 0.99)], "review_queue": []},
        )

    route.side_effect = responder
    result = await resolve(records, tenant_id="tenant-a", project_id="project-a")

    assert len(result) == 1
    assert result[0]["agreed_sources"] == 2
    assert result[0]["confidence"] == pytest.approx(0.99)
    assert set(result[0]["merged_from"]) <= {"source-a", "source-b"}
    for provenance in result[0]["provenance"]:
        er = provenance["entity_resolution"]
        assert er["method"] == "canonical_ts_entity_resolution"
        assert set(er["merged_sources"]) == {"source-a", "source-b"}


@pytest.mark.asyncio
@respx.mock
async def test_pending_review_never_auto_merges(monkeypatch):
    monkeypatch.setenv("OPT_ER_BRIDGE_URL", "https://edge.example.test")
    monkeypatch.setenv("INTERNAL_TOKEN", "test-internal-token")
    records = [_rec("Acme AB", "source-a"), _rec("Acme Group AB", "source-b")]

    def responder(request: httpx.Request) -> httpx.Response:
        body = __import__("json").loads(request.content)
        a, b = body["records"][0]["record_id"], body["records"][1]["record_id"]
        pair = _pair(a, b, "pending_review")
        return httpx.Response(200, json={"pairs": [pair], "review_queue": [pair]})

    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 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", "0pt.company.stable-identifiers.v3", 0.98),
                    _pair(b, c, "match", "0pt.company.stable-identifiers.v3", 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):
    monkeypatch.setenv("OPT_ER_BRIDGE_URL", "https://edge.example.test")
    monkeypatch.setenv("INTERNAL_TOKEN", "test-internal-token")
    records = [_rec("Acme AB", "source-a"), _rec("Acme AB", "source-b")]
    respx.post("https://edge.example.test/api/0pt/internal/er/candidates").mock(side_effect=httpx.ConnectError("offline"))

    result = await resolve(records, tenant_id="tenant-a", project_id="project-a")
    assert len(result) == 2
    assert all(not r.get("merged_from") for r in result)


@pytest.mark.asyncio
async def test_missing_internal_configuration_makes_zero_network_calls(monkeypatch):
    monkeypatch.delenv("OPT_ER_BRIDGE_URL", raising=False)
    monkeypatch.delenv("EDGE_URL", raising=False)
    monkeypatch.delenv("INTERNAL_TOKEN", raising=False)
    records = [_rec("Acme AB", "source-a"), _rec("Acme AB", "source-b")]
    result = await resolve(records, tenant_id="tenant-a", project_id="project-a")
    assert len(result) == 2