163 lines
7.4 KiB
Python
163 lines
7.4 KiB
Python
"""
|
|
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
|