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

"""
P014: integration proof that a source signaling a block (HTTP 401/403/429/451)
during a real run_job() call (a) never crashes or fabricates a result, (b) is
recorded on the process-wide adaptive_router so the engine backs off that
source, and (c) the pipeline still delivers from the next compliant source in
the plan -- exactly the "source-racing several authorized sources" behavior
the siax-0pt skill documents, now with an honest memory of blocks instead of
silently re-trying a rate-limited source with none.

Uses the real default_router (module-level singleton pipeline.py actually
uses), cleared before and after every test in this file so this suite never
leaks state into -- or picks up state from -- any other test module that
imports workers.pipeline.
"""
from __future__ import annotations

import httpx
import pytest
import respx

from workers.adaptive_router import default_router
from workers.pipeline import run_job

COMPANIES_HOUSE_URL = "https://api.company-information.service.gov.uk/company/00000006"
GLEIF_URL = "https://api.gleif.org/api/v1/lei-records/00000006"


@pytest.fixture(autouse=True)
def _clean_router():
    default_router.clear()
    yield
    default_router.clear()


class TestSourceBlockedDuringARealRun:
    @pytest.mark.asyncio
    @respx.mock
    async def test_a_429_from_one_source_is_recorded_as_blocked_and_the_run_still_delivers_from_the_next_source(
        self, monkeypatch
    ):
        monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key")
        respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(429, json={"error": "rate limited"}))
        respx.get(GLEIF_URL).mock(
            return_value=httpx.Response(
                200,
                json={"data": {"attributes": {"entity": {"legalName": {"name": "Fallback Ltd"}, "status": "ACTIVE", "legalAddress": {}}}}},
            )
        )

        assert default_router.is_blocked("GB", "companies_house") is False  # sanity: clean slate

        result = await run_job("order_block_1", {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006"]}})

        assert result["state"] == "done"
        assert result["metrics"]["rows"] == 1
        assert result["records"][0]["name"] == "Fallback Ltd"
        assert result["records"][0]["source"] == "gleif"  # companies_house was blocked, gleif answered

        # THE regression this test guards: the block was actually recorded,
        # not silently discarded the way a bare `except Exception: res = None`
        # would (that's still the fallback for genuine errors, but a block is
        # now distinguishable from one).
        assert default_router.is_blocked("GB", "companies_house") is True

    @pytest.mark.asyncio
    @respx.mock
    async def test_a_blocked_source_is_excluded_from_the_plan_on_the_very_next_key_in_the_same_run(self, monkeypatch):
        monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key")
        gb_second_url = "https://api.company-information.service.gov.uk/company/11111111"
        gleif_second_url = "https://api.gleif.org/api/v1/lei-records/11111111"

        companies_house_calls = respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(429, json={}))
        # No mock registered for gb_second_url at all -- if the pipeline tried
        # companies_house AGAIN for the second key after it was just blocked,
        # respx would raise on the unmocked request and this test would fail
        # with a connection error, not a clean assertion. That absence IS the
        # assertion: the plan excluded the just-blocked source.
        respx.get(GLEIF_URL).mock(
            return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "First"}, "legalAddress": {}}}}})
        )
        respx.get(gleif_second_url).mock(
            return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "Second"}, "legalAddress": {}}}}})
        )

        result = await run_job(
            "order_block_2",
            {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006", "11111111"]}},
        )

        assert result["state"] == "done"
        assert result["metrics"]["rows"] == 2
        names = {r["name"] for r in result["records"]}
        assert names == {"First", "Second"}
        # companies_house was hit exactly once (for the first key, where it
        # got blocked) -- never retried for the second key in this run.
        assert companies_house_calls.call_count == 1

    @pytest.mark.asyncio
    @respx.mock
    async def test_a_plain_5xx_is_not_treated_as_a_block(self, monkeypatch):
        # A genuine server error is a different signal from a compliance
        # block -- it should NOT be recorded via record_blocked (that would
        # incorrectly suppress a source that's just having a bad moment, not
        # one that told the engine to stop).
        monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key")
        respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(500, json={}))
        respx.get(GLEIF_URL).mock(
            return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "X"}, "legalAddress": {}}}}})
        )

        result = await run_job("order_block_3", {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006"]}})

        assert result["state"] == "done"
        assert default_router.is_blocked("GB", "companies_house") is False