admin
5137bf87a1
CI (SIAX Cloud) / integration (push) Skipped
CI (SIAX Cloud) / sonar (push) Skipped
masterplan-lock / check (push) Successful in 4s
CI (SIAX Cloud) / sonar (pull_request) Skipped
CI (SIAX Cloud) / security (push) Successful in 15s
masterplan-lock / check (pull_request) Successful in 7s
Security Scan (SIAX Cloud) / security (push) Successful in 18s
Security Scan (SIAX Cloud) / security (pull_request) Successful in 17s
CI (SIAX Cloud) / security (pull_request) Successful in 20s
CI (SIAX Cloud) / integration (pull_request) Successful in 35s
CI (SIAX Cloud) / quality (pull_request) Successful in 39s
CI (SIAX Cloud) / quality (push) Successful in 48s
54 lines
2.4 KiB
Python
54 lines
2.4 KiB
Python
"""A failed or malformed source must never become a verified empty result."""
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from fastapi.testclient import TestClient
|
|
from orchestrator import main
|
|
from sources.connectors import opensanctions_screen, rdw_vehicle
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("status", [403, 429, 500])
|
|
async def test_source_http_failure_raises(status):
|
|
with respx.mock:
|
|
respx.get("https://api.opensanctions.org/search/default").respond(status)
|
|
respx.get("https://opendata.rdw.nl/resource/m9d7-ebf2.json").respond(status)
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
await opensanctions_screen("example")
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
await rdw_vehicle("AB123")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_malformed_and_valid_empty_are_distinct():
|
|
with respx.mock:
|
|
screen = respx.get("https://api.opensanctions.org/search/default").respond(200, json={})
|
|
vehicle = respx.get("https://opendata.rdw.nl/resource/m9d7-ebf2.json").respond(200, json={})
|
|
with pytest.raises(ValueError):
|
|
await opensanctions_screen("example")
|
|
with pytest.raises(ValueError):
|
|
await rdw_vehicle("AB123")
|
|
screen.respond(200, json={"results": []})
|
|
vehicle.respond(200, json=[])
|
|
assert (await opensanctions_screen("example"))["matches"] == []
|
|
assert await rdw_vehicle("AB123") is None
|
|
|
|
|
|
def test_routes_return_503_on_source_failure_and_mark_verified_empty(monkeypatch):
|
|
async def failed(*args):
|
|
raise RuntimeError("fixture failure")
|
|
async def empty_screen(*args):
|
|
return {"source": "opensanctions", "matches": []}
|
|
async def empty_vehicle(*args):
|
|
return None
|
|
monkeypatch.setattr(main, "_INTERNAL_TOKEN", "")
|
|
monkeypatch.setattr(main, "opensanctions_screen", failed)
|
|
monkeypatch.setattr(main, "rdw_vehicle", failed)
|
|
client = TestClient(main.app)
|
|
assert client.get('/screen?q=example').status_code == 503
|
|
assert client.get('/vehicles/AB123').status_code == 503
|
|
monkeypatch.setattr(main, "opensanctions_screen", empty_screen)
|
|
monkeypatch.setattr(main, "rdw_vehicle", empty_vehicle)
|
|
assert client.get('/screen?q=example').json() == {"source": "opensanctions", "source_status": "ok", "matches": []}
|
|
assert client.get('/vehicles/AB123').json() == {"vehicle": None, "source_status": "ok"}
|