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

"""
Fault-injection tests for workers/pipeline.py's run_job — the REAL order
pipeline (not a mock of it). BEFORE_STATE.md §4 found schema-evolution,
CDC-interruption, backfill, and restart entirely untested. This file closes
what can be closed against code that actually exists, and says explicitly
why the rest can't be, instead of fabricating a harness for infrastructure
this repo doesn't have.

All external HTTP is mocked with respx (never hits real registries) so this
suite is deterministic and offline.
"""
from __future__ import annotations

import httpx
import pytest
import respx

from workers.pipeline import run_job

BRREG_URL = "https://data.brreg.no/enhetsregisteret/api/enheter/923609016"
PRH_URL = "https://avoindata.prh.fi/opendata-ytj-api/v3/companies?businessId=2077474-0"

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


class TestRestartMidFetch:
    """A worker crash/restart mid-job is not literally simulate-able (no
    checkpointing exists to resume from — the pipeline is a single async
    function with no persisted intermediate state). What this DOES test,
    honestly: a connector raising mid-job (network reset, upstream 5xx,
    connection drop — the kind of fault a real restart would also surface
    as) never crashes the whole order; the job completes and delivers
    whatever it validly could."""

    @pytest.mark.asyncio
    @respx.mock
    async def test_a_connector_exception_does_not_crash_the_order(self):
        respx.get(BRREG_URL).mock(side_effect=httpx.ConnectError("connection reset"))
        respx.get(PRH_URL).mock(return_value=httpx.Response(200, json={
            "companies": [{"names": [{"name": "Suomi Oy"}], "status": "active"}],
        }))
        result = await run_job(
            "order_restart_1",
            {"markets": ["NO", "FI"], "confidence": 50, "seeds": {"NO": ["923609016"], "FI": ["2077474-0"]}},
        )
        assert result["state"] == "done"
        # NO failed entirely (connector raised); FI still delivered.
        assert result["metrics"]["per_market"]["NO"] == 0
        assert result["metrics"]["per_market"]["FI"] == 1
        assert len(result["records"]) == 1
        assert result["records"][0]["market"] == "FI"

    @pytest.mark.asyncio
    @respx.mock
    async def test_every_connector_failing_delivers_zero_rows_not_a_crash(self):
        respx.get(BRREG_URL).mock(side_effect=httpx.ConnectError("down"))
        respx.get(PRH_URL).mock(side_effect=httpx.ConnectError("down"))
        result = await run_job(
            "order_restart_2",
            {"markets": ["NO", "FI"], "confidence": 50, "seeds": {"NO": ["923609016"], "FI": ["2077474-0"]}},
        )
        assert result["state"] == "done"
        assert result["records"] == []
        assert result["metrics"]["rows"] == 0


class TestSchemaEvolution:
    """Upstream registries adding/removing/renaming fields without notice is
    a real risk (BEFORE_STATE.md flagged this as wholly untested). Verified
    against the actual HTTP-parsing boundary in sources/connectors.py, via
    respx-mocked responses shaped like plausible upstream drift — not a
    fabricated schema-registry harness (this repo has no schema
    versioning/CDC system to test against; see test_known_gaps.py)."""

    @pytest.mark.asyncio
    @respx.mock
    async def test_upstream_adding_unexpected_new_fields_does_not_break_parsing(self):
        drifted_body = {
            **VALID_BRREG_BODY,
            # Simulated schema drift: brreg adds fields this connector never asked for.
            "stiftelsesdato": "2020-01-01",
            "naeringskode1": {"kode": "45.11", "beskrivelse": "Handel"},
            "nested_new_thing": {"a": {"b": [1, 2, 3]}},
        }
        respx.get(BRREG_URL).mock(return_value=httpx.Response(200, json=drifted_body))
        result = await run_job(
            "order_schema_1",
            {"markets": ["NO"], "confidence": 50, "seeds": {"NO": ["923609016"]}},
        )
        assert result["state"] == "done"
        assert len(result["records"]) == 1
        assert result["records"][0]["name"] == "Volvo Personvagnar Norge AS"

    @pytest.mark.asyncio
    @respx.mock
    async def test_upstream_removing_an_expected_field_degrades_honestly_not_a_crash(self):
        # Real drift case: upstream drops "navn" (renamed/moved). The
        # connector must not crash, and must not fabricate a name — the
        # record is simply not produced (pipeline's existing `if res and
        # res.name` gate), which is the correct "level 0 / not available"
        # behavior, not a bug to work around.
        stripped_body = {k: v for k, v in VALID_BRREG_BODY.items() if k != "navn"}
        respx.get(BRREG_URL).mock(return_value=httpx.Response(200, json=stripped_body))
        result = await run_job(
            "order_schema_2",
            {"markets": ["NO"], "confidence": 50, "seeds": {"NO": ["923609016"]}},
        )
        assert result["state"] == "done"
        assert result["records"] == []  # honestly nothing delivered, not a fabricated name

    @pytest.mark.asyncio
    @respx.mock
    async def test_upstream_returning_a_wholly_different_shape_does_not_crash(self):
        # Worst case: upstream returns something structurally alien (e.g. a
        # maintenance-page JSON blob instead of a company record).
        respx.get(BRREG_URL).mock(return_value=httpx.Response(200, json={"error": "maintenance", "retry_after": 60}))
        result = await run_job(
            "order_schema_3",
            {"markets": ["NO"], "confidence": 50, "seeds": {"NO": ["923609016"]}},
        )
        assert result["state"] == "done"
        assert result["records"] == []


class TestBackfillReplay:
    """This pipeline has NO persistence layer (BEFORE_STATE.md — every
    run_job call recomputes from scratch; there is no delivery store to
    dedupe a backfill against). A true "backfill doesn't create duplicate
    deliveries" test would need that store and is NOT fabricated here. What
    IS real and testable: replaying the identical order config (the exact
    scenario a backfill re-run performs) is idempotent AT THE COMPUTE LEVEL
    — same input, byte-identical output — so a backfill re-run at least
    doesn't silently drift or become order-dependent across replays."""

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

        first = await run_job("order_backfill", config)
        second = await run_job("order_backfill", config)

        # Strip the per-run artifact_ref/timestamps (legitimately differ —
        # every replay IS a new delivery event) before comparing the
        # substantive content.
        def strip_run_identity(result):
            recs = []
            for r in result["records"]:
                rec = {k: v for k, v in r.items()}
                rec["provenance"] = [
                    {k: v for k, v in p.items() if k not in ("retrieved_at", "artifact_ref")} for p in rec["provenance"]
                ]
                recs.append(rec)
            return recs

        assert strip_run_identity(first) == strip_run_identity(second)
        assert first["metrics"]["rows"] == second["metrics"]["rows"] == 1