202 lines
9.0 KiB
Python
202 lines
9.0 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from app import create_app
|
|
from app.prices import PriceSnapshotStore
|
|
from app.vintages import VintageStore
|
|
|
|
|
|
class ApiTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.snapshot = {
|
|
"as_of": "2026-08-21",
|
|
"source": {
|
|
"source_id": "fixture.tourism",
|
|
"source_url": "https://example.invalid/tourism",
|
|
"published_at": "2026-08-21T08:00:00Z",
|
|
"retrieved_at": "2026-08-21T08:05:00Z",
|
|
"vintage_id": "fixture-1",
|
|
},
|
|
"observations": [
|
|
{"metric_key": "arrivals_yoy", "value": 12, "expected": 8, "scale": 2, "unit": "percent"},
|
|
],
|
|
"exposures": [
|
|
{"symbol": "AOT", "coefficient": 1.0, "confidence": 0.95, "evidence": "airport"},
|
|
{"symbol": "PTT", "coefficient": -0.2, "confidence": 0.60, "evidence": "control"},
|
|
],
|
|
}
|
|
self.app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_WRITE_TOKEN": "test-token"})
|
|
self.client = self.app.test_client()
|
|
|
|
def _login_paper(self):
|
|
response = self.client.post("/api/v1/auth/paper", json={"token": "test-token"})
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_health_reports_research_mode(self):
|
|
response = self.client.get("/api/v1/health")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.get_json()["mode"], "research")
|
|
|
|
def test_summary_contains_lineage_and_signal_counts(self):
|
|
response = self.client.get("/api/v1/dashboard/summary")
|
|
body = response.get_json()
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(body["data_health"]["vintage_id"], "fixture-1")
|
|
self.assertEqual(body["data_health"]["status"], "fixture")
|
|
self.assertEqual(body["signal_summary"]["total"], 2)
|
|
self.assertEqual(body["signal_summary"]["long"], 1)
|
|
self.assertEqual(body["signal_summary"]["short"], 1)
|
|
|
|
def test_data_health_reports_lineage_and_replayability(self):
|
|
response = self.client.get("/api/v1/data-health")
|
|
body = response.get_json()
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(body["status"], "fixture")
|
|
self.assertEqual(body["source_id"], "fixture.tourism")
|
|
self.assertFalse(body["replayable"])
|
|
self.assertIn("vintage_id", body)
|
|
|
|
def test_replay_endpoint_returns_frozen_snapshot_result(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
(root / "fixture-1.json").write_text(json.dumps(self.snapshot), encoding="utf-8")
|
|
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "SNAPSHOT_DIR": root})
|
|
response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1")
|
|
body = response.get_json()
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(body["vintage_id"], "fixture-1")
|
|
self.assertEqual(body["result"]["theme"], "tourism")
|
|
self.assertEqual(body["result"]["signals"][0]["symbol"], "AOT")
|
|
|
|
def test_replay_endpoint_rejects_path_traversal(self):
|
|
response = self.client.get("/api/v1/replay/tourism?vintage_id=../secret")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_bot_source_mode_uses_injected_adapter(self):
|
|
class FakeAdapter:
|
|
def fetch(self, *, exposures):
|
|
snapshot = dict(self_snapshot)
|
|
snapshot["source"] = {**snapshot["source"], "source_id": "bot.ec_ei_028_s2"}
|
|
snapshot["data_quality"] = "provisional"
|
|
snapshot["exposures"] = exposures
|
|
return snapshot
|
|
|
|
self_snapshot = self.snapshot
|
|
app = create_app(
|
|
{
|
|
"TESTING": True,
|
|
"SNAPSHOT": self.snapshot,
|
|
"TOURISM_SOURCE": "bot",
|
|
"TOURISM_ADAPTER": FakeAdapter(),
|
|
}
|
|
)
|
|
body = app.test_client().get("/api/v1/data-health").get_json()
|
|
self.assertEqual(body["source_id"], "bot.ec_ei_028_s2")
|
|
self.assertEqual(body["status"], "provisional")
|
|
|
|
def test_backtest_readiness_blocks_without_enough_vintages(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
store = VintageStore(Path(temp_dir))
|
|
store.persist(b"fixture raw", self.snapshot)
|
|
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": store, "PRICE_STORE": PriceSnapshotStore(Path(temp_dir) / "prices")})
|
|
response = app.test_client().get("/api/v1/backtest/tourism?min_events=12")
|
|
body = response.get_json()
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertEqual(body["status"], "blocked")
|
|
self.assertEqual(body["reason"], "insufficient_vintages")
|
|
self.assertEqual(body["available_events"], 1)
|
|
self.assertEqual(body["price_snapshot"]["status"], "missing")
|
|
|
|
def test_backtest_readiness_blocks_revised_or_missing_prices(self):
|
|
class FakeVintageStore:
|
|
def list_vintages(self, _as_of=None):
|
|
return [{"vintage_id": f"vintage-{index}"} for index in range(12)]
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
app = create_app(
|
|
{
|
|
"TESTING": True,
|
|
"VINTAGE_STORE": FakeVintageStore(),
|
|
"PRICE_STORE": PriceSnapshotStore(Path(temp_dir)),
|
|
}
|
|
)
|
|
response = app.test_client().get("/api/v1/backtest/tourism?min_events=12")
|
|
body = response.get_json()
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertEqual(body["status"], "blocked")
|
|
self.assertEqual(body["reason"], "price_series_not_point_in_time")
|
|
|
|
def test_prices_health_reports_missing_snapshot(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
app = create_app({"TESTING": True, "PRICE_STORE": PriceSnapshotStore(Path(temp_dir))})
|
|
response = app.test_client().get("/api/v1/prices/health")
|
|
body = response.get_json()
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertFalse(body["available"])
|
|
self.assertEqual(body["status"], "missing")
|
|
|
|
def test_backtest_readiness_rejects_invalid_min_events(self):
|
|
response = self.client.get("/api/v1/backtest/tourism?min_events=bad")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_vintages_endpoint_filters_by_publication_timestamp(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
store = VintageStore(Path(temp_dir))
|
|
store.persist(b"fixture raw", self.snapshot)
|
|
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": store})
|
|
client = app.test_client()
|
|
before = client.get("/api/v1/vintages?as_of=2026-08-20T00:00:00Z")
|
|
after = client.get("/api/v1/vintages?as_of=2026-08-22T00:00:00Z")
|
|
self.assertEqual(before.status_code, 200)
|
|
self.assertEqual(before.get_json()["count"], 0)
|
|
self.assertEqual(after.status_code, 200)
|
|
self.assertEqual(after.get_json()["count"], 1)
|
|
|
|
def test_vintages_endpoint_rejects_invalid_as_of(self):
|
|
response = self.client.get("/api/v1/vintages?as_of=not-a-timestamp")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_paper_ledger_requires_token(self):
|
|
response = self.client.post(
|
|
"/api/v1/paper/ledger",
|
|
json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 10},
|
|
)
|
|
self.assertEqual(response.status_code, 401)
|
|
|
|
def test_paper_ledger_rejects_invalid_paper_token(self):
|
|
response = self.client.post("/api/v1/auth/paper", json={"token": "wrong-token"})
|
|
self.assertEqual(response.status_code, 401)
|
|
|
|
def test_paper_ledger_rejects_non_finite_price(self):
|
|
self._login_paper()
|
|
response = self.client.post(
|
|
"/api/v1/paper/ledger",
|
|
json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": "NaN"},
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_paper_ledger_rejects_unknown_symbol(self):
|
|
self._login_paper()
|
|
response = self.client.post(
|
|
"/api/v1/paper/ledger",
|
|
json={"symbol": "UNKNOWN", "target_weight": 0.1, "assumed_price": 10},
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_paper_ledger_records_valid_entry(self):
|
|
self._login_paper()
|
|
response = self.client.post(
|
|
"/api/v1/paper/ledger",
|
|
json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 60},
|
|
)
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.get_json()["entry"]["symbol"], "AOT")
|
|
ledger = self.client.get("/api/v1/paper/ledger").get_json()["entries"]
|
|
self.assertEqual(len(ledger), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|