school_compare/backend/tests/test_sixth_form_flag.py

"""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