59 lines
2.4 KiB
Python
59 lines
2.4 KiB
Python
"""
|
|
Verifies the P010 provenance additions (BEFORE_STATE.md §3: source,
|
|
retrieval timestamp, transform, version, producer were absent from the real
|
|
dataflow) land correctly on every real connector's output — mocked HTTP,
|
|
real parsing code.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from sources.connectors import brreg_by_orgnr, make_provenance, CONNECTOR_VERSIONS
|
|
|
|
|
|
class TestMakeProvenanceFactory:
|
|
def test_populates_all_new_fields_honestly_and_leaves_the_rest_none(self):
|
|
p = make_provenance("name", "brreg", 0.97, 3)
|
|
assert p.field_name == "name"
|
|
assert p.source_id == "brreg"
|
|
assert p.retrieved_at is not None
|
|
assert p.freshness == p.retrieved_at[:10] # date prefix of the same instant
|
|
assert p.transform == "raw_fetch"
|
|
assert p.connector_version == CONNECTOR_VERSIONS["brreg"]
|
|
assert p.producer == "0pt-engine"
|
|
# Never guessed — these are filled by later pipeline stages, not at fetch time.
|
|
assert p.quality is None
|
|
assert p.entity_resolution is None
|
|
assert p.artifact_ref is None
|
|
|
|
def test_transform_override_is_honored(self):
|
|
p = make_provenance("name", "companies_house", 0.0, 0, transform="unavailable_no_key")
|
|
assert p.transform == "unavailable_no_key"
|
|
|
|
def test_unknown_source_id_has_no_fabricated_version(self):
|
|
p = make_provenance("x", "some_future_source", 0.5, 1)
|
|
assert p.connector_version is None # honestly unknown, not guessed
|
|
|
|
|
|
class TestBrregConnectorProvenance:
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_real_parsing_path_produces_fully_populated_provenance(self):
|
|
respx.get("https://data.brreg.no/enhetsregisteret/api/enheter/923609016").mock(
|
|
return_value=httpx.Response(200, json={
|
|
"navn": "Volvo Personvagnar Norge AS", "konkurs": False,
|
|
"forretningsadresse": {"adresse": ["Storgata 1"]},
|
|
})
|
|
)
|
|
res = await brreg_by_orgnr("923609016")
|
|
assert res is not None
|
|
assert len(res.provenance) == 3 # name, status, address
|
|
for p in res.provenance:
|
|
assert p.source_id == "brreg"
|
|
assert p.connector_version == "brreg-connector@1"
|
|
assert p.producer == "0pt-engine"
|
|
assert p.retrieved_at is not None
|
|
assert p.transform == "raw_fetch"
|