"""Tests for the GIAS-driven has_sixth_form flag (spec 2026-07-07 §3). The filter and payloads must use dim_school.has_sixth_form, not the old age_range-contains-"18" substring heuristic. The key regression case is a 16-19 sixth-form college: flag true, but "16-19" contains no "18". """ import numpy as np import pandas as pd import pytest from fastapi.testclient import TestClient def _schools_df() -> pd.DataFrame: """Latest-year snapshot rows as produced by load_latest_school_data.""" base = { "local_authority": "Testshire", "school_type": "Academy", "phase": "Secondary", "address": "1 Test Street", "town": "Testtown", "postcode": "TS1 1AA", "religious_denomination": None, "gender": "Mixed", "admissions_policy": None, "ofsted_grade": np.nan, "ofsted_date": None, "ofsted_framework": None, "latitude": 51.5, "longitude": -0.1, "year": 202425, "total_pupils": 1000, "rwm_expected_pct": np.nan, "attainment_8_score": 50.0, } return pd.DataFrame( [ # 11-18 school WITH a registered sixth form {**base, "urn": 100001, "school_name": "Alpha High", "age_range": "11-18", "has_sixth_form": True}, # 16-19 college: old heuristic said NO ("16-19" has no "18"), # GIAS flag says YES — must appear in the yes-filter results {**base, "urn": 100002, "school_name": "Beta Sixth Form College", "age_range": "16-19", "has_sixth_form": True}, # 11-18 age range on paper but NO registered sixth form: # old heuristic said YES, GIAS flag says NO {**base, "urn": 100003, "school_name": "Gamma Academy", "age_range": "11-18", "has_sixth_form": False}, # Missing flag (pipeline not yet re-run) — must not crash, # must not match the yes-filter {**base, "urn": 100004, "school_name": "Delta School", "age_range": "11-16", "has_sixth_form": None}, ] ) @pytest.fixture() def client(monkeypatch): from backend import app as app_module monkeypatch.setattr(app_module, "load_latest_school_data", _schools_df) monkeypatch.setattr(app_module, "load_school_data", _schools_df) monkeypatch.setattr(app_module, "get_supplementary_data", lambda db, urn: {}) return TestClient(app_module.app, raise_server_exceptions=False) def _urns(resp): return sorted(s["urn"] for s in resp.json()["schools"]) def test_filter_yes_uses_flag_not_age_range(client): resp = client.get("/api/schools?has_sixth_form=yes") assert resp.status_code == 200, resp.text # 16-19 college included; 11-18-without-sixth-form excluded assert _urns(resp) == [100001, 100002] def test_filter_no_uses_flag_not_age_range(client): resp = client.get("/api/schools?has_sixth_form=no") assert resp.status_code == 200, resp.text # Gamma (flag false) and Delta (flag missing => not true) assert _urns(resp) == [100003, 100004] def test_list_payload_includes_flag(client): resp = client.get("/api/schools") assert resp.status_code == 200, resp.text by_urn = {s["urn"]: s for s in resp.json()["schools"]} assert by_urn[100002]["has_sixth_form"] is True assert by_urn[100003]["has_sixth_form"] is False assert by_urn[100004]["has_sixth_form"] is None def test_detail_payload_includes_flag(client): resp = client.get("/api/schools/100002") assert resp.status_code == 200, resp.text assert resp.json()["school_info"]["has_sixth_form"] is True def test_detail_payload_serializes_numpy_bool(monkeypatch): """Once the pipeline has run, has_sixth_form is a real bool dtype column (dbt not_null test guarantees no NULLs), so row access yields numpy.bool_ rather than a Python bool. convert_to_native must handle it — otherwise FastAPI's jsonable_encoder raises ValueError and the detail endpoint 500s (C2).""" from backend import app as app_module df = _schools_df() # Drop the row with a None flag — this fixture models the post-pipeline # state where the column is a genuine, fully-populated bool dtype. df = df[df["has_sixth_form"].notna()].reset_index(drop=True) df["has_sixth_form"] = df["has_sixth_form"].astype(bool) assert df["has_sixth_form"].dtype == bool monkeypatch.setattr(app_module, "load_school_data", lambda: df) monkeypatch.setattr(app_module, "get_supplementary_data", lambda db, urn: {}) client = TestClient(app_module.app, raise_server_exceptions=False) resp = client.get("/api/schools/100002") assert resp.status_code == 200, resp.text assert resp.json()["school_info"]["has_sixth_form"] is True def test_load_school_data_survives_missing_has_sixth_form_column(monkeypatch): """Real prod state until the nightly pipeline first rebuilds the mart: marts.dim_school lacks has_sixth_form entirely. The first query raises UndefinedColumn; load_school_data_as_dataframe must retry without the column (synthesizing it as None) rather than swallow the error and return (and then have load_school_data cache) an empty DataFrame (C1).""" import sqlalchemy.exc from backend import data_loader data_loader._df_cache = None data_loader._df_latest_cache = None good_df = pd.DataFrame( [ { "urn": 1, "school_name": "Fallback School", "school_type": "Academy", "has_sixth_form": None, } ] ) calls = [] def fake_read_sql(query, con): calls.append(query) if len(calls) == 1: # The statement text still contains phase_code, school_type_code, # etc. (it's the full _MAIN_QUERY SELECT list) — that's exactly # the collision this test guards against: matching must be done # against exc.orig (the DBAPI error), not str(exc)/the statement. raise sqlalchemy.exc.ProgrammingError( statement=str(data_loader._MAIN_QUERY), params=None, orig=Exception( "(psycopg2.errors.UndefinedColumn) column s.has_sixth_form " "does not exist" ), ) return good_df.copy() monkeypatch.setattr(data_loader.pd, "read_sql", fake_read_sql) try: df = data_loader.load_school_data_as_dataframe() finally: data_loader._df_cache = None data_loader._df_latest_cache = None assert len(calls) == 2, "must retry with the no-sixth-form query variant" assert calls[1] is data_loader._MAIN_QUERY_NO_SIXTH_FORM assert not df.empty assert "has_sixth_form" in df.columns assert df["has_sixth_form"].iloc[0] is None