Spaces:
Sleeping
Sleeping
File size: 6,386 Bytes
6ca3a1d 887f013 6ca3a1d 887f013 f278c3f 887f013 f278c3f 887f013 f278c3f 887f013 f278c3f 887f013 f278c3f 887f013 6ca3a1d 887f013 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | from __future__ import annotations
from collections.abc import Iterator
import pytest
from fastapi.testclient import TestClient
from chokepoint.api import deps
@pytest.fixture(autouse=True, scope="module")
def _real_graph_backend() -> Iterator[None]:
"""Run this module against the real NetworkX graph and the stub LLM, regardless of .env.
``get_graph_store`` is lru_cache'd and ``DEMO_MODE=true`` would silently swap in the
stub, which returns three canned nodes for *any* event and would make these tests
pass or fail for the wrong reason. ``LLM_PROVIDER`` is pinned to ``stub`` so a
developer's ``LLM_PROVIDER=hf`` never turns this module into a billed network call.
"""
with pytest.MonkeyPatch.context() as patch:
patch.setenv("DEMO_MODE", "false")
patch.setenv("GRAPH_BACKEND", "networkx")
patch.setenv("LLM_PROVIDER", "stub")
deps.get_settings.cache_clear()
deps.get_graph_store.cache_clear()
deps.get_llm_provider.cache_clear()
yield
deps.get_settings.cache_clear()
deps.get_graph_store.cache_clear()
deps.get_llm_provider.cache_clear()
MINIMAL_EVENT_PAYLOAD = {
"event": {
"event_type": "strike",
"locations": [{"raw": "Hamburg Port", "node_id": "port_hamburg"}],
"affected_goods": ["auto parts"],
"severity": 4,
"confidence": 0.8,
"summary": "test",
}
}
def test_simulate_with_event_returns_valid_result(client: TestClient) -> None:
resp = client.post("/api/v1/simulate", json=MINIMAL_EVENT_PAYLOAD)
assert resp.status_code == 200
body = resp.json()
assert body["epicentre_node_ids"] == ["port_hamburg"]
assert len(body["impacted"]) > 0
assert body["degraded"] is True # only stubs exist in this slice
assert body["unresolved_entities"] == []
assert body["runtime_ms"] >= 0
def test_simulate_with_text_uses_stub_llm_provider(client: TestClient) -> None:
resp = client.post("/api/v1/simulate", json={"text": "A strike hit Hamburg Port today."})
assert resp.status_code == 200
body = resp.json()
assert body["event"]["event_type"] == "strike"
assert body["degraded"] is True
def test_simulate_impacted_nodes_are_sorted_desc_by_impact(client: TestClient) -> None:
resp = client.post("/api/v1/simulate", json=MINIMAL_EVENT_PAYLOAD)
scores = [node["impact_score"] for node in resp.json()["impacted"]]
assert scores == sorted(scores, reverse=True)
def test_simulate_without_text_doc_id_or_event_returns_422(client: TestClient) -> None:
resp = client.post("/api/v1/simulate", json={})
assert resp.status_code == 422
def test_simulate_with_doc_id_returns_422_in_this_slice(client: TestClient) -> None:
# No document store is wired in yet (arrives with Person 1's ingestion PR).
resp = client.post("/api/v1/simulate", json={"doc_id": "abc123"})
assert resp.status_code == 422
def test_simulate_severity_override_changes_impact_scores(client: TestClient) -> None:
low = client.post(
"/api/v1/simulate",
json={**MINIMAL_EVENT_PAYLOAD, "severity_override": 1},
).json()
high = client.post(
"/api/v1/simulate",
json={**MINIMAL_EVENT_PAYLOAD, "severity_override": 5},
).json()
assert high["impacted"][0]["impact_score"] >= low["impacted"][0]["impact_score"]
def test_simulate_rejects_out_of_range_max_hops(client: TestClient) -> None:
resp = client.post("/api/v1/simulate", json={**MINIMAL_EVENT_PAYLOAD, "max_hops": 0})
assert resp.status_code == 422
# ── the real graph: Suez / Red Sea and Black Sea grain corridors ───────────────────────
def _event_payload(node_id: str, raw: str, event_type: str, severity: int) -> dict[str, object]:
return {
"event": {
"event_type": event_type,
"locations": [{"raw": raw, "node_id": node_id}],
"severity": severity,
"confidence": 0.9,
"summary": "test",
}
}
def test_simulate_suez_corridor_produces_impact(client: TestClient) -> None:
resp = client.post(
"/api/v1/simulate", json=_event_payload("chokepoint_suez", "Suez Canal", "blockade", 5)
)
assert resp.status_code == 200
body = resp.json()
assert len(body["impacted"]) > 0
assert body["epicentre_node_ids"] == ["chokepoint_suez"]
assert body["unresolved_entities"] == []
# Not vacuous: the stub returns three canned Hamburg-corridor nodes for ANY event, so pin
# ids that only exist in the real Suez corridor (and its cross-link into corridor 1).
ids = {node["node_id"] for node in body["impacted"]}
assert {"port_piraeus", "com_electronics", "ind_electronics_eu", "mkt_eu_retail"} <= ids
assert "com_containers" in ids
assert all("weighted propagation" in node["explanation"] for node in body["impacted"])
assert not any("Stub" in node["explanation"] for node in body["impacted"])
def test_simulate_odesa_grain_cascade_reaches_egypt(client: TestClient) -> None:
resp = client.post(
"/api/v1/simulate", json=_event_payload("port_odesa", "Port of Odesa", "conflict", 4)
)
assert resp.status_code == 200
impacted = resp.json()["impacted"]
ids = [node["node_id"] for node in impacted]
assert {"com_grain", "ind_agri_ua", "mkt_egypt_wheat", "mkt_mena_food"} <= set(ids)
hops = {node["node_id"]: node["hops"] for node in impacted}
assert hops["com_grain"] == 1 and hops["mkt_egypt_wheat"] == 2 # a genuine multi-hop cascade
def test_simulate_severity_override_scales_the_suez_cascade(client: TestClient) -> None:
base = _event_payload("chokepoint_suez", "Suez Canal", "blockade", 3)
low = client.post("/api/v1/simulate", json={**base, "severity_override": 1}).json()
high = client.post("/api/v1/simulate", json={**base, "severity_override": 5}).json()
assert low["impacted"] and high["impacted"]
assert high["impacted"][0]["impact_score"] > low["impacted"][0]["impact_score"]
def test_simulate_node_absent_from_the_graph_is_a_clean_empty_result(client: TestClient) -> None:
resp = client.post(
"/api/v1/simulate", json=_event_payload("port_atlantis", "Atlantis", "strike", 5)
)
assert resp.status_code == 200 # never a 500 for an unmodelled place
assert resp.json()["impacted"] == []
|