""" Regression tests for the DEMO ingestion failure (`NotNullViolation` on `exchanges.country_id`). ## The bug these tests exist to prevent coming back Running `PYTHONPATH=. python3 scripts/seed_demo.py` — or calling `ingest_security(db, adapter, "AAPL")` — died with: psycopg.errors.NotNullViolation: null value in column "country_id" of relation "exchanges" violates not-null constraint Failing row: mic=XNAS, name=XNAS, country_id=NULL, timezone=UTC preceded by `ingest.exchange.country_unknown mic=XNAS ticker=AAPL`, and the database was left completely empty afterwards. Three separate defects combined to produce that, and each is pinned by tests below. 1. **`exchanges.country_id` and `companies.country_id` are NOT NULL, but `get_or_create_company()` wrote `country_id=.id if else None`.** The code's own "country unknown" branch could never succeed — it handed Postgres a NULL for a NOT NULL column, so a one-line reference-data gap surfaced as a constraint violation several statements later, with a traceback pointing at `db.flush()`. `test_no_not_null_foreign_key_is_assigned_a_conditional_none` is the guard. 2. **`XSOF` is not an ISO 10383 MIC.** The demo profile for `4SBK` (First Investment Bank, Sofia) carried it, while `EXCHANGE_COUNTRY_BY_MIC` correctly listed the real code, `XBUL`. So that one company could not resolve a venue country no matter how the reference data was seeded. Note the *shape* of this bug: the demo universe and the MIC table were free to disagree, and nothing noticed until an INSERT failed in production. `test_every_demo_universe_mic_resolves_to_a_country` closes that gap for the whole universe rather than for one ticker. 3. **`seed_demo_universe()` caught an exception per ticker without rolling back.** The first failure left the session in a failed transaction, so every subsequent ticker raised `PendingRollbackError` — which is why the run produced an *empty* database and nineteen identical-looking failures rather than 19 successes and 1 failure. `test_seed_demo_universe_rolls_back_a_failed_ticker` is the guard. ## Why these tests are written against source text rather than a live database `sqlalchemy`, `psycopg` and `alembic` are not installed in this environment and PyPI is blocked by egress policy (HTTP 403), so `app.workers.ingest` and `scripts.seed_demo` cannot be imported at all — both pull in SQLAlchemy at module scope. A test that cannot run is worth nothing, so the invariants that need the ORM are asserted against the source and the migration as text. That is a genuine limitation and it is stated rather than hidden: these tests prove the *code no longer contains the defect*. They do **not** prove ingestion works against a real PostgreSQL — nothing in this environment can. The end-to-end check belongs on the deployment host and is listed as such in `docs/FINAL_REPORT.md`. The MIC and country-coverage tests, by contrast, are real behavioural tests: `app.engines.identity` and `app.adapters.demo` are dependency-free and are genuinely executed here. """ from __future__ import annotations import ast import pathlib import re from app.adapters.demo import DEMO_SEED_PROFILES from app.engines.identity import EXCHANGE_COUNTRY_BY_MIC, exchange_country_for_mic from app.engines.reference_data import ( COUNTRIES, DEFAULT_INDUSTRY_CODE, DEFAULT_SECTOR_CODE, EXCHANGES, MIC_ALIASES, ReferencePlan, TaxonomyConflictError, build_taxonomy, canonical_mic, exchange_for_mic, reference_plan, rows_to_insert, ) _BACKEND = pathlib.Path(__file__).resolve().parent.parent _INGEST = _BACKEND / "app" / "workers" / "ingest.py" _SEED = _BACKEND / "scripts" / "seed_demo.py" _MIGRATION = _BACKEND / "alembic" / "versions" / "0001_initial_schema.py" def _demo_pairs(): return [(p.sector_code, p.industry_code) for p in DEMO_SEED_PROFILES] def _plan() -> ReferencePlan: """Exactly what `seed_reference_data()` will insert into an empty database.""" return reference_plan(_demo_pairs()) def _seeded_country_iso2() -> set[str]: return {c.iso2 for c in _plan().countries} # --- 1. The reported failure, pinned exactly ------------------------------------------------- def test_xnas_resolves_to_the_united_states(): """The literal reported case: AAPL -> XNAS -> US. `XNAS` was in `EXCHANGE_COUNTRY_BY_MIC` all along, which is why the fix was not "add the MIC". Pinned anyway: it is the case the bug report names, and a future edit to the table must not quietly drop it. """ fact = exchange_country_for_mic("XNAS") assert fact.known, fact.reason assert fact.iso2 == "US" def test_the_united_states_is_seeded_as_a_country(): """`XNAS -> US` is only useful if a `countries` row with iso2 `US` exists to point at.""" assert "US" in _seeded_country_iso2() def test_mic_lookup_is_case_and_whitespace_insensitive(): """Providers are not consistent about casing. `xnas` and ` XNAS ` must not become UNKNOWN.""" for spelling in ("xnas", " XNAS ", "Xnas"): assert exchange_country_for_mic(spelling).iso2 == "US", spelling # --- 2. The systemic version: the whole demo universe, not one ticker ------------------------ def test_every_demo_universe_mic_resolves_to_a_country(): """Every demo profile's MIC must be in `EXCHANGE_COUNTRY_BY_MIC`. This is the test that would have caught the real defect. `4SBK` carried `XSOF`, which is not an ISO 10383 MIC — the Bulgarian Stock Exchange is `XBUL` — so it alone could never resolve a venue country, and nothing checked until a production INSERT failed. """ unresolved = { p.ticker: p.exchange_mic for p in DEMO_SEED_PROFILES if not exchange_country_for_mic(p.exchange_mic).known } assert not unresolved, ( f"demo profiles whose MIC is not in EXCHANGE_COUNTRY_BY_MIC: {unresolved}. " f"Add the MIC with its real ISO 3166-1 alpha-2 country, or correct the demo profile — " f"do not invent a MIC." ) def test_no_demo_profile_uses_xsof_which_is_not_a_real_mic(): """The specific fabricated code, named so it cannot drift back in. ISO 10383 assigns the Bulgarian Stock Exchange the operating MIC `XBUL` (segment MICs `ABUL`, `IBUL`, `GBUL`, `PBUL`). `XSOF` is not among them. """ assert "XSOF" not in {p.exchange_mic for p in DEMO_SEED_PROFILES} assert "XSOF" not in EXCHANGE_COUNTRY_BY_MIC assert EXCHANGE_COUNTRY_BY_MIC["XBUL"] == "BG" def test_every_country_the_demo_universe_needs_is_seeded(): """Both country facts per company must have a `countries` row waiting for them. `companies.country_id` needs the company's own country; `exchanges.country_id` needs the listing venue's country. They are different questions (§18) and either one missing is a NOT NULL violation at ingest time. """ seeded = _seeded_country_iso2() missing: dict[str, set[str]] = {} for p in DEMO_SEED_PROFILES: needed = {p.country_iso2, exchange_country_for_mic(p.exchange_mic).iso2} gap = {c for c in needed if c and c not in seeded} if gap: missing[p.ticker] = gap assert not missing, ( f"demo profiles referring to countries that seed_reference_data() does not insert: " f"{missing}" ) def test_mic_table_values_are_iso3166_alpha2_shaped(): """A typo like `USA` or `us` in the table would fail the `countries.iso2` lookup at runtime with a message about the exchange, not about the typo.""" bad = {mic: iso for mic, iso in EXCHANGE_COUNTRY_BY_MIC.items() if not (isinstance(iso, str) and len(iso) == 2 and iso.isupper() and iso.isalpha())} assert not bad, bad def test_mic_table_keys_are_uppercase(): """Lookup upper-cases its input, so a lower-case key would be unreachable.""" bad = [mic for mic in EXCHANGE_COUNTRY_BY_MIC if mic != mic.upper()] assert not bad, bad # --- 3. NOT NULL columns must never be handed a conditional None ----------------------------- def _not_null_fk_columns() -> dict[str, set[str]]: """`{table: {column, ...}}` for NOT NULL ForeignKey columns, read from the initial migration.""" text = _MIGRATION.read_text() out: dict[str, set[str]] = {} for table_match in re.finditer(r'op\.create_table\(\s*"(\w+)",(.*?)\n \)', text, re.S): table, body = table_match.group(1), table_match.group(2) cols = set() for col in re.finditer( r'sa\.Column\(\s*"(\w+)",[^)]*?sa\.ForeignKey\([^)]*\)([^)]*)\)', body ): name, tail = col.group(1), col.group(2) if "nullable=True" not in tail: cols.add(name) if cols: out[table] = cols return out def test_the_migration_really_does_declare_these_columns_not_null(): """Guards the guard: if the parse above silently matched nothing, the next test would pass vacuously and prove nothing.""" not_null = _not_null_fk_columns() assert not_null.get("exchanges") == {"country_id"}, not_null.get("exchanges") assert {"country_id", "sector_id", "industry_id"} <= not_null.get("companies", set()) def _conditional_none_keyword_assignments(path: pathlib.Path, columns: set[str]) -> list[str]: """Keyword arguments named after a NOT NULL column whose value can evaluate to None. Parsed with `ast`, not searched with a regex: the module's own prose describes the defect, and a text search cannot tell an explanation from an occurrence. `ast` sees only real code. Catches both spellings of the same mistake — `x if cond else None` and `a.b if a else None` — by looking for a conditional expression with a bare `None` in either branch. """ tree = ast.parse(path.read_text()) offenders: list[str] = [] for node in ast.walk(tree): if not isinstance(node, ast.Call): continue for kw in node.keywords: if kw.arg not in columns or not isinstance(kw.value, ast.IfExp): continue branches = (kw.value.body, kw.value.orelse) if any(isinstance(b, ast.Constant) and b.value is None for b in branches): offenders.append(f"line {kw.value.lineno}: {kw.arg}={ast.unparse(kw.value)}") return offenders def test_no_not_null_foreign_key_is_assigned_a_conditional_none(): """The exact shape of the original defect. `country_id=exchange_country.id if exchange_country else None` type-checks, reads as careful defensive coding, and is a guaranteed `NotNullViolation`. Any conditional expression that can yield None, assigned to a NOT NULL foreign key in the ingestion path, is rejected here. Note what this caught when it was first run: not only the `country_id` site the bug report named, but `exchange_id=exchange.id if exchange else None` on `Security` as well — `securities.exchange_id` is NOT NULL too, so a provider profile with no exchange MIC would have failed in exactly the same way, on a different table. Scans the whole `app/` tree rather than `ingest.py` alone: the defect is a pattern, not a location, and ten NOT NULL foreign keys in this schema can all be written by any worker. """ columns = {c for cols in _not_null_fk_columns().values() for c in cols} offenders: list[str] = [] for path in sorted((_BACKEND / "app").rglob("*.py")): offenders += [ f"{path.relative_to(_BACKEND)} {o}" for o in _conditional_none_keyword_assignments(path, columns) ] assert not offenders, ( "NOT NULL foreign keys assigned a conditional None: " f"{offenders}. Raise ReferenceDataError instead — the column cannot hold NULL, so the " "fallback branch cannot succeed and only defers the failure to db.flush()." ) def test_the_conditional_none_guard_actually_detects_the_original_code(): """Guards the guard. A detector that silently matches nothing would let the defect back in while reporting success, so it is run here against the code as it was before the fix.""" import tempfile original = ( "Exchange(mic=m, name=m, country_id=c.id if c else None, timezone='UTC')\n" "Security(company_id=x, exchange_id=e.id if e else None, ticker=t)\n" ) with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh: fh.write(original) tmp = pathlib.Path(fh.name) try: found = _conditional_none_keyword_assignments(tmp, {"country_id", "exchange_id"}) finally: tmp.unlink() assert len(found) == 2, found def test_ingest_raises_reference_data_error_rather_than_writing_null(): """The replacement behaviour is present, and names the remedy in the message.""" source = _INGEST.read_text() assert "class ReferenceDataError" in source assert source.count("raise ReferenceDataError(") >= 3, ( "expected the unknown-MIC, unseeded-exchange-country and unseeded-company-country paths " "each to raise" ) assert "EXCHANGE_COUNTRY_BY_MIC" in source, "the error should name the table to edit" def test_reference_data_is_validated_before_anything_is_added_to_the_session(): """The validation must run before the first `db.add()`, or a failing ticker leaves pending rows and poisons the seed loop's session even though the loop rolls back.""" source = _INGEST.read_text() body = source.split("def get_or_create_company(", 1)[1] validate_at = body.index("_resolve_required_reference_data(") first_add = body.index("db.add(") assert validate_at < first_add, ( "get_or_create_company() adds rows to the session before validating reference data" ) # --- 4. The seed loop must survive a failing ticker ------------------------------------------- def test_seed_demo_universe_rolls_back_a_failed_ticker(): """Without `db.rollback()` the session stays in a failed transaction and every later ticker dies with `PendingRollbackError` — one bad ticker costs all twenty.""" tree = ast.parse(_SEED.read_text()) func = next( n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == "seed_demo_universe" ) handlers = [h for n in ast.walk(func) if isinstance(n, ast.Try) for h in n.handlers] assert handlers, "seed_demo_universe() no longer has an exception handler" rolls_back = any( isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "rollback" for h in handlers for n in ast.walk(h) ) assert rolls_back, "seed_demo_universe() catches per-ticker errors without db.rollback()" def test_seed_reports_a_partial_run_as_a_failure(): """A seed that ingests 3 of 20 and exits 0 lets a container start-up report success over an almost-empty database.""" source = _SEED.read_text() assert "succeeded = seed_demo_universe(db)" in source assert "return 1" in source, "main() should exit non-zero when not every company ingested" # --- 5. The reference seed must actually populate all four tables ---------------------------- # # Production result that prompted these: `seed_reference_data()` exited 0 having written # countries: 19 exchanges: 0 sectors: 0 industries: 0 # so the reference layer looked seeded while three of its four tables were empty. def test_the_plan_populates_all_four_reference_tables(): """The headline regression: a plan with any empty table is the bug that was reported.""" counts = _plan().counts() empty = [table for table, n in counts.items() if n == 0] assert not empty, f"reference plan leaves these tables empty: {empty} (counts: {counts})" def test_the_plan_creates_exchanges(): exchanges = _plan().exchanges assert len(exchanges) >= len({p.exchange_mic for p in DEMO_SEED_PROFILES}) assert all(e.mic and e.name and e.country_iso2 and e.timezone for e in exchanges) def test_the_plan_creates_sectors(): sectors = _plan().sectors assert sectors assert all(code and name for code, name in sectors) assert DEFAULT_SECTOR_CODE in {code for code, _ in sectors}, ( "ingestion falls back to the DEFAULT sector code when a profile carries none; it must " "exist rather than be created mid-ingest" ) def test_the_plan_creates_industries(): industries = _plan().industries assert industries assert all(code and name and sector for code, name, sector in industries) assert DEFAULT_INDUSTRY_CODE in {code for code, _, _ in industries} def test_xnas_is_seeded_as_an_exchange_pointing_at_the_united_states(): """The reported case, now at the reference-data level: `XNAS: NOT FOUND` in production.""" venue = exchange_for_mic("XNAS") assert venue is not None, "XNAS is not in EXCHANGES, so seeding cannot create it" assert venue.country_iso2 == "US" assert venue in _plan().exchanges def test_xbul_is_seeded_as_an_exchange_pointing_at_bulgaria(): """The other production `NOT FOUND`, and the venue whose MIC was wrong in the demo universe.""" venue = exchange_for_mic("XBUL") assert venue is not None assert venue.country_iso2 == "BG" assert venue in _plan().exchanges def test_every_demo_profile_has_a_company_country_in_the_plan(): seeded = {c.iso2 for c in _plan().countries} missing = {p.ticker: p.country_iso2 for p in DEMO_SEED_PROFILES if p.country_iso2 not in seeded} assert not missing, missing def test_every_demo_profile_has_an_exchange_in_the_plan(): seeded = {e.mic for e in _plan().exchanges} missing = {p.ticker: p.exchange_mic for p in DEMO_SEED_PROFILES if canonical_mic(p.exchange_mic) not in seeded} assert not missing, missing def test_every_demo_profile_has_a_sector_in_the_plan(): seeded = {code for code, _ in _plan().sectors} missing = {p.ticker: p.sector_code for p in DEMO_SEED_PROFILES if p.sector_code.upper() not in seeded} assert not missing, missing def test_every_demo_profile_has_an_industry_in_the_plan(): seeded = {code for code, _, _ in _plan().industries} missing = {p.ticker: p.industry_code for p in DEMO_SEED_PROFILES if p.industry_code.upper() not in seeded} assert not missing, missing def test_every_not_null_foreign_key_in_the_plan_is_resolvable(): """`exchanges.country_id` and `industries.sector_id` are NOT NULL. Every row the seed writes must have its parent in the same plan, or the INSERT fails on a foreign key instead of a missing reference row — which is the failure mode this whole series of fixes is about.""" plan = _plan() countries = {c.iso2 for c in plan.countries} sectors = {code for code, _ in plan.sectors} dangling_exchange = {e.mic: e.country_iso2 for e in plan.exchanges if e.country_iso2 not in countries} dangling_industry = {code: sector for code, _, sector in plan.industries if sector not in sectors} assert not dangling_exchange, f"exchanges.country_id unresolvable: {dangling_exchange}" assert not dangling_industry, f"industries.sector_id unresolvable: {dangling_industry}" # --- 6. Idempotency --------------------------------------------------------------------------- def test_reseeding_inserts_nothing(): """`rows_to_insert()` is the whole of the seed's idempotency, so running it against the keys the first run created must yield an empty plan. Asserted on the real function rather than on the presence of an `if ... .first()` guard in the source.""" plan = _plan() second = rows_to_insert( plan, {c.iso2 for c in plan.countries}, {e.mic for e in plan.exchanges}, {code for code, _ in plan.sectors}, {code for code, _, _ in plan.industries}, ) assert second.is_empty, second.counts() assert second.counts() == {"countries": 0, "exchanges": 0, "sectors": 0, "industries": 0} def test_the_first_seed_inserts_everything(): """Guards the guard: if `rows_to_insert()` returned nothing regardless of input, the idempotency test above would pass while the seed inserted nothing at all — which is exactly the production symptom.""" plan = _plan() first = rows_to_insert(plan, set(), set(), set(), set()) assert first.counts() == plan.counts() def test_a_partial_existing_database_inserts_only_the_gap(): """The realistic case on Azeroth: countries already seeded, the other three tables empty.""" plan = _plan() partial = rows_to_insert( plan, {c.iso2 for c in plan.countries}, set(), set(), set() ) assert partial.counts()["countries"] == 0 assert partial.counts()["exchanges"] == len(plan.exchanges) assert partial.counts()["sectors"] == len(plan.sectors) assert partial.counts()["industries"] == len(plan.industries) def test_no_duplicate_natural_keys_in_the_plan(): """Every one of these is UNIQUE in the schema; a duplicate in the source data is an IntegrityError on the first run, not on a re-run.""" plan = _plan() for label, keys in ( ("countries.iso2", [c.iso2 for c in plan.countries]), ("countries.iso3", [c.iso3 for c in plan.countries]), ("exchanges.mic", [e.mic for e in plan.exchanges]), ("sectors.code", [code for code, _ in plan.sectors]), ("industries.code", [code for code, _, _ in plan.industries]), ): assert len(keys) == len(set(keys)), f"duplicate {label}: {sorted(keys)}" # --- 7. The reference data itself must be well formed ----------------------------------------- def test_every_exchange_timezone_is_a_real_iana_zone(): """Executed against the running tzdata, so a typo or an invented zone fails here rather than being written to the database as fact.""" from zoneinfo import available_timezones zones = available_timezones() bad = {e.mic: e.timezone for e in EXCHANGES if e.timezone not in zones} assert not bad, bad def test_country_codes_are_well_formed(): bad = [c for c in COUNTRIES if len(c.iso2) != 2 or len(c.iso3) != 3 or not (c.iso2.isupper() and c.iso3.isupper()) or len(c.currency) != 3 or not c.currency.isupper()] assert not bad, bad def test_bulgaria_is_recorded_in_euro(): """Bulgaria adopted the euro on 1 January 2026. The seed carried `BGN` until this pass, which would have written a stale currency into a reference table on first deploy.""" bg = next(c for c in COUNTRIES if c.iso2 == "BG") assert bg.currency == "EUR", bg def test_provider_short_names_resolve_to_the_canonical_mic(): """FMP returns `exchangeShortName` ("NASDAQ"), EODHD returns a MIC ("XNAS"). If both were taken at face value the same venue would become two `exchanges` rows, silently splitting peer groups and country rollups.""" assert canonical_mic("NASDAQ") == "XNAS" assert canonical_mic("nasdaq") == "XNAS" assert canonical_mic("NYSE") == "XNYS" assert exchange_for_mic("NASDAQ") is exchange_for_mic("XNAS") def test_no_alias_is_also_seeded_as_a_venue(): """An alias that also had its own `EXCHANGES` row would defeat the point: two rows, one venue.""" venues = {e.mic for e in EXCHANGES} overlap = set(MIC_ALIASES) & venues assert not overlap, overlap unresolvable = {a: m for a, m in MIC_ALIASES.items() if m not in venues} assert not unresolvable, unresolvable def test_an_industry_claimed_by_two_sectors_is_rejected(): """Peer groups are built by industry and then by sector, so an industry filed under the wrong sector changes every percentile in it. Whichever company ingested first used to decide.""" try: build_taxonomy([("TECHNOLOGY", "BANKS"), ("FINANCIALS", "BANKS")]) except TaxonomyConflictError: return raise AssertionError("build_taxonomy accepted an industry under two different sectors") def test_a_profile_with_no_classification_falls_back_to_default(): taxonomy = build_taxonomy([(None, None)]) assert DEFAULT_SECTOR_CODE in {code for code, _ in taxonomy.sectors} assert (DEFAULT_INDUSTRY_CODE, "Default", DEFAULT_SECTOR_CODE) in taxonomy.industries ALL_TESTS = [v for k, v in sorted(globals().items()) if k.startswith("test_")] if __name__ == "__main__": passed = failed = 0 for t in ALL_TESTS: try: t() print(f"PASS {t.__name__}") passed += 1 except AssertionError as e: print(f"FAIL {t.__name__}: {e}") failed += 1 print(f"\n{passed}/{passed + failed} passed") raise SystemExit(1 if failed else 0)