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

"""
Tests for workers/quality_bridge.py (the Python half of the P010 TS/Python
bridge) and its wiring into workers/pipeline.run_job. Verifies the
fail-open contract end to end: disabled, not configured, timeout,
non-2xx, malformed body, and success — and, critically, that `delivered`
rows from run_job are IDENTICAL regardless of which of those the bridge
call resolves to (the bridge is advisory, never a second gate).
"""
from __future__ import annotations

import httpx
import pytest
import respx

from workers.pipeline import run_job
from workers.quality_bridge import run_quality_bridge

BRIDGE_URL = "http://edge.local/api/0pt/quality/bridge-run"
BRREG_URL = "https://data.brreg.no/enhetsregisteret/api/enheter/923609016"

VALID_BRREG_BODY = {
    "navn": "Volvo Personvagnar Norge AS",
    "konkurs": False,
    "forretningsadresse": {"adresse": ["Storgata 1"]},
}


@pytest.fixture(autouse=True)
def _clean_env(monkeypatch):
    monkeypatch.delenv("OPT_QUALITY_BRIDGE_ENABLED", raising=False)
    monkeypatch.delenv("OPT_QUALITY_BRIDGE_URL", raising=False)
    monkeypatch.delenv("EDGE_URL", raising=False)
    monkeypatch.delenv("INTERNAL_TOKEN", raising=False)


class TestBridgeDirectly:
    @pytest.mark.asyncio
    async def test_disabled_by_default_makes_zero_network_calls(self, monkeypatch):
        # No respx mock registered at all — if this made an HTTP call it would raise.
        result = await run_quality_bridge(tenant_id="t_a", project_id="p_1", dataset_id="ds_1", rows=[])
        assert result.status == "disabled"
        assert result.score is None

    @pytest.mark.asyncio
    async def test_enabled_but_no_url_configured(self, monkeypatch):
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_ENABLED", "true")
        result = await run_quality_bridge(tenant_id="t_a", project_id="p_1", dataset_id="ds_1", rows=[])
        assert result.status == "not_configured"

    @pytest.mark.asyncio
    @respx.mock
    async def test_successful_run_is_parsed(self, monkeypatch):
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_ENABLED", "true")
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_URL", "http://edge.local")
        respx.post(BRIDGE_URL).mock(return_value=httpx.Response(201, json={
            "run": {"run_id": "dqrun_1", "status": "succeeded", "score": 100},
            "issues": [],
        }))
        result = await run_quality_bridge(
            tenant_id="t_a", project_id="p_1", dataset_id="ds_1", rows=[{"name": "X", "confidence": 0.9}]
        )
        assert result.status == "ok"
        assert result.score == 100
        assert result.run_id == "dqrun_1"
        assert result.issue_count == 0

    @pytest.mark.asyncio
    @respx.mock
    async def test_timeout_fails_open(self, monkeypatch):
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_ENABLED", "true")
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_URL", "http://edge.local")
        respx.post(BRIDGE_URL).mock(side_effect=httpx.TimeoutException("timed out"))
        result = await run_quality_bridge(tenant_id="t_a", project_id="p_1", dataset_id="ds_1", rows=[])
        assert result.status == "unavailable"
        assert result.detail == "timeout"

    @pytest.mark.asyncio
    @respx.mock
    async def test_non_2xx_fails_open(self, monkeypatch):
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_ENABLED", "true")
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_URL", "http://edge.local")
        respx.post(BRIDGE_URL).mock(return_value=httpx.Response(500, json={"error": "boom"}))
        result = await run_quality_bridge(tenant_id="t_a", project_id="p_1", dataset_id="ds_1", rows=[])
        assert result.status == "unavailable"
        assert "500" in result.detail

    @pytest.mark.asyncio
    @respx.mock
    async def test_malformed_body_fails_open_not_raises(self, monkeypatch):
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_ENABLED", "true")
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_URL", "http://edge.local")
        respx.post(BRIDGE_URL).mock(return_value=httpx.Response(201, content=b"not json"))
        result = await run_quality_bridge(tenant_id="t_a", project_id="p_1", dataset_id="ds_1", rows=[])
        assert result.status == "unavailable"

    @pytest.mark.asyncio
    @respx.mock
    async def test_connection_refused_fails_open(self, monkeypatch):
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_ENABLED", "true")
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_URL", "http://edge.local")
        respx.post(BRIDGE_URL).mock(side_effect=httpx.ConnectError("refused"))
        result = await run_quality_bridge(tenant_id="t_a", project_id="p_1", dataset_id="ds_1", rows=[])
        assert result.status == "unavailable"


class TestBridgeWiredIntoThePipeline:
    """The part that matters most: whatever the bridge does, `delivered`
    (the rows the customer actually receives) must not change."""

    @pytest.mark.asyncio
    @respx.mock
    async def test_missing_tenant_id_skips_the_bridge_honestly(self):
        respx.get(BRREG_URL).mock(return_value=httpx.Response(200, json=VALID_BRREG_BODY))
        result = await run_job("order_notenant", {"markets": ["NO"], "confidence": 50, "seeds": {"NO": ["923609016"]}})
        assert result["metrics"]["quality_bridge"]["status"] == "not_configured"
        assert len(result["records"]) == 1  # delivery unaffected

    @pytest.mark.asyncio
    @respx.mock
    async def test_bridge_disabled_delivers_identically_to_bridge_enabled_and_failing(self, monkeypatch):
        respx.get(BRREG_URL).mock(return_value=httpx.Response(200, json=VALID_BRREG_BODY))
        config = {
            "markets": ["NO"], "confidence": 50, "seeds": {"NO": ["923609016"]},
            "tenant_id": "t_bridge_pipeline", "project_id": "p_1",
        }

        monkeypatch.delenv("OPT_QUALITY_BRIDGE_ENABLED", raising=False)
        disabled_result = await run_job("order_bridge_off", config)

        monkeypatch.setenv("OPT_QUALITY_BRIDGE_ENABLED", "true")
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_URL", "http://edge.local")
        respx.post(BRIDGE_URL).mock(side_effect=httpx.ConnectError("edge unreachable"))
        failing_result = await run_job("order_bridge_fail", config)

        assert disabled_result["metrics"]["rows"] == failing_result["metrics"]["rows"] == 1
        assert [r["name"] for r in disabled_result["records"]] == [r["name"] for r in failing_result["records"]]
        assert disabled_result["metrics"]["quality_bridge"]["status"] == "disabled"
        assert failing_result["metrics"]["quality_bridge"]["status"] == "unavailable"

    @pytest.mark.asyncio
    @respx.mock
    async def test_bridge_success_stamps_quality_provenance_without_dropping_or_adding_rows(self, monkeypatch):
        respx.get(BRREG_URL).mock(return_value=httpx.Response(200, json=VALID_BRREG_BODY))
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_ENABLED", "true")
        monkeypatch.setenv("OPT_QUALITY_BRIDGE_URL", "http://edge.local")
        respx.post(BRIDGE_URL).mock(return_value=httpx.Response(201, json={
            "run": {"run_id": "dqrun_9", "status": "succeeded", "score": 100},
            "issues": [],
        }))
        result = await run_job(
            "order_bridge_ok",
            {
                "markets": ["NO"], "confidence": 50, "seeds": {"NO": ["923609016"]},
                "tenant_id": "t_bridge_ok", "project_id": "p_1",
            },
        )
        assert result["metrics"]["rows"] == 1
        assert result["metrics"]["quality_bridge"]["status"] == "ok"
        assert result["metrics"]["quality_bridge"]["score"] == 100
        for p in result["records"][0]["provenance"]:
            assert p["quality"] == {
                "score": 100, "run_id": "dqrun_9", "run_status": "succeeded", "checked_by": "0pt-quality-bridge",
            }
            assert p["artifact_ref"] == "0pt:pipeline_delivery:order_bridge_ok"