tests/test_api_contract.py

from __future__ import annotations
import os,unittest
from unittest.mock import patch
from fastapi.testclient import TestClient
import app
from api_models import CursorError,decode_cursor,encode_cursor

REVISION="d2b8896cabf2489d1b2f82694376f90a3b38049c"
READY={"status":"ready","ready":True,"dataset_revision":REVISION,"artifact_count":25,"cached_bytes":310651359,"cache_dir":"/not-used-by-contract-tests","error":None}
class ApiContractTests(unittest.TestCase):
 def setUp(self):
  self.stack=patch.multiple(app.CACHE,status=lambda:dict(READY));self.stack.start();self.env=patch.dict(os.environ,{"DILUTIONRISK_BUILD_SHA":"b"*40});self.env.start();self.client=TestClient(app.app)
 def tearDown(self):self.env.stop();self.stack.stop()
 def test_openapi_exposes_bounded_typed_contract(self):
  schema=self.client.get("/openapi.json").json();self.assertIn("/api/v1/companies",schema["paths"]);self.assertIn("/api/v1/companies/{ticker}",schema["paths"]);self.assertIn("CompanyPage",schema["components"]["schemas"]);self.assertIn("CompanySnapshot",schema["components"]["schemas"]);operation=schema["paths"]["/api/v1/companies"]["get"];limit=next(x for x in operation["parameters"] if x["name"]=="limit");self.assertEqual(limit["schema"]["maximum"],100);self.assertEqual(operation["responses"]["422"]["content"]["application/json"]["schema"]["$ref"],"#/components/schemas/ErrorResponse")
 def test_company_snapshot_combines_metrics_instruments_and_revision(self):
  company={"issuer_id":"ASX:BHP","exchange":"ASX","ticker":"BHP","name":"BHP Group","sector":None,"industry":None,"as_of_date":"2026-07-17","needs_review":False,"basic_shares":5078913175,"fd_ratio":1.003,"worst_case_dilution_pct":0.3,"expiry_pct_lt_3m":0.0,"expiry_pct_lt_6m":0.0,"expiry_pct_lt_12m":0.0,"cash_runway_months":None};instrument={"instrument_id":"ASX:BHP:OPT","ticker":"BHP","instrument_type":"option","quantity":100,"series_name":"Options","asx_code":None,"exercise_price_aud":"1.0","expiry":"2027-01-01","holder_type":"broker","vesting_condition":"none","status":"active","as_of_date":"2026-07-17","document_ids":["ASX:BHP:1"]}
  with patch("app.tool_context",return_value=(dict(READY),{"dataset_revision":REVISION,"published_at":"2026-07-17T12:49:16Z","source_watermark":{"provider":"ASX"},"dataset_citation":"https://example.test/dataset"})),patch("app.query_companies",return_value=([company],False,{})),patch("app.dilution_instruments",return_value=[instrument]):
   response=self.client.get("/api/v1/companies/BHP");self.assertEqual(response.status_code,200);payload=response.json();self.assertEqual(payload["company"]["ticker"],"BHP");self.assertEqual(payload["instruments"][0]["document_ids"],["ASX:BHP:1"]);self.assertEqual(payload["dataset_revision"],REVISION)
  with patch("app.tool_context",return_value=(dict(READY),{"dataset_revision":REVISION,"published_at":"2026-07-17T12:49:16Z","source_watermark":{"provider":"ASX"},"dataset_citation":"https://example.test/dataset"})),patch("app.company_at_date",return_value=dict(company)),patch("app.dilution_instruments",return_value=[instrument]) as instruments:
   response=self.client.get("/api/v1/companies/BHP",params={"as_of_date":"2026-07-17"});self.assertEqual(response.status_code,200);instruments.assert_called_once_with(READY["cache_dir"],"BHP",100,"2026-07-17",positions_only=True,exchange="ASX")
 def test_document_page_carries_batched_public_source_links(self):
  document={"document_id":"ASX:CLA:03115626","exchange":"ASX","ticker":"CLA","source_document_id":"03115626","announcement_date":"2026-07-17","announced_at":"2026-07-17T01:00:00Z","title":"Capital raising","doc_type":"Announcement","doc_type_source":"rule","doc_type_confidence":1.0,"raw_artifact_key":None,"raw_size_bytes":None,"markdown_artifact_key":"artifacts/v1/asx/CLA/03115626.md","markdown_size_bytes":10,"publication_state":"public"}
  with patch("app.query_documents",return_value=([document],None)),patch("app.document_citations",return_value={"ASX:CLA:03115626":"https://www.asx.com.au/asxpdf/example.pdf"}):
   response=self.client.get("/api/v1/documents",params={"ticker":"CLA","limit":1});self.assertEqual(response.status_code,200);self.assertEqual(response.json()["items"][0]["source_link"],"https://www.asx.com.au/asxpdf/example.pdf")
 def test_status_carries_exact_revision(self):
  response=self.client.get("/api/v1/status");self.assertEqual(response.status_code,200);self.assertEqual(response.json()["dataset_revision"],REVISION)
 def test_invalid_inputs_are_structured_4xx(self):
  for url in ("/api/v1/companies?limit=0","/api/v1/companies?limit=101","/api/v1/companies?q="+"x"*121,"/api/v1/documents?ticker=../etc"):
   response=self.client.get(url);self.assertEqual(response.status_code,422);self.assertEqual(response.json()["error"]["code"],"invalid_request")
 def test_cursor_is_revision_and_endpoint_aware(self):
  cursor=encode_cursor(REVISION,"companies",["ASX:BHP"]);self.assertEqual(decode_cursor(cursor,REVISION,"companies"),["ASX:BHP"])
  self.assertEqual(self.client.get("/api/v1/documents",params={"cursor":cursor}).status_code,400)
  stale=encode_cursor("a"*40,"companies",["ASX:BHP"]);response=self.client.get("/api/v1/companies",params={"cursor":stale});self.assertEqual(response.status_code,409);self.assertEqual(response.json()["error"]["code"],"stale_cursor")
if __name__=="__main__":unittest.main()