diff --git a/changelog.d/batched-exit-status.fixed.md b/changelog.d/batched-exit-status.fixed.md new file mode 100644 index 00000000000..5d9ba9142a3 --- /dev/null +++ b/changelog.d/batched-exit-status.fixed.md @@ -0,0 +1 @@ +Decide a test batch by the child process exit status rather than its printed summary, and enumerate `.yml` test files alongside `.yaml`. diff --git a/changelog.d/cpuc-housing-subsidy-source.changed.md b/changelog.d/cpuc-housing-subsidy-source.changed.md new file mode 100644 index 00000000000..c657f58570c --- /dev/null +++ b/changelog.d/cpuc-housing-subsidy-source.changed.md @@ -0,0 +1 @@ +Count the housing subsidy a household reports receiving toward CA CPUC CARE/FERA income, in place of the SPM's capped housing subsidy. PG&E Form 01-9077 defines total gross annual household income as "all taxable and non-taxable revenues from all people living in the home, from whatever sources derived, including, but not limited to, ... housing and military subsidies", with no netting against rent and no cap; the SPM cap made CARE eligibility depend on which SPM geography a request selected, and the ungated `housing_assistance` would have counted a voucher the model imputes to any income-eligible renter. diff --git a/changelog.d/reported-housing-assistance.added.md b/changelog.d/reported-housing-assistance.added.md new file mode 100644 index 00000000000..7f930933786 --- /dev/null +++ b/changelog.d/reported-housing-assistance.added.md @@ -0,0 +1 @@ +Add `reported_housing_assistance`, the housing assistance of a unit that reports receiving it, for income definitions that count subsidies received rather than modelled take-up. diff --git a/changelog.d/spm-dataset-poverty-alias-guard.fixed.md b/changelog.d/spm-dataset-poverty-alias-guard.fixed.md new file mode 100644 index 00000000000..dcbc91a6109 --- /dev/null +++ b/changelog.d/spm-dataset-poverty-alias-guard.fixed.md @@ -0,0 +1 @@ +Reject derived poverty aliases such as `in_poverty` from stored datasets. diff --git a/changelog.d/spm-default-dataset-hash.fixed.md b/changelog.d/spm-default-dataset-hash.fixed.md new file mode 100644 index 00000000000..45d28e64013 --- /dev/null +++ b/changelog.d/spm-default-dataset-hash.fixed.md @@ -0,0 +1 @@ +Verify the default dataset content hash so a moved build id cannot silently change every standalone-country result. diff --git a/policyengine_us/parameters/gov/states/ca/cpuc/income_sources.yaml b/policyengine_us/parameters/gov/states/ca/cpuc/income_sources.yaml index 2f43cdfca8a..bf329c42cd2 100644 --- a/policyengine_us/parameters/gov/states/ca/cpuc/income_sources.yaml +++ b/policyengine_us/parameters/gov/states/ca/cpuc/income_sources.yaml @@ -1,13 +1,16 @@ description: >- Income sources counted toward CA CPUC CARE/FERA eligibility. Total gross annual household income includes all taxable and - nontaxable revenues from all people living in the home, from + non-taxable revenues from all people living in the home, from whatever sources derived, including, but not limited to, wages, salaries, interest, dividends, spousal and child support payments, public assistance payments, Social Security and pensions, housing and military subsidies, rental income, income from self-employment and all employment-related, non-cash income. -# PG&E Form 01-9077, CARE/FERA Program Application. +# Quoted verbatim from the footnote to the CARE and FERA Income Guidelines +# tables on PG&E Form 01-9077, CARE/FERA Program Application (Rev. 6.26); the +# same footnote appears on the CPUC-filed tariff sheet. Each entry below names +# the phrase it implements. values: 0000-01-01: # Wages, salaries @@ -36,7 +39,35 @@ values: - pension_income - retirement_distributions # Housing subsidies - - spm_unit_capped_housing_subsidy + # + # The form lists "housing and military subsidies" among the revenues that + # make up total gross annual household income, with no netting against rent, + # no cap and no poverty-measurement valuation anywhere on it - so the entry + # is the housing subsidy itself, at the amount the household receives. + # + # reported_housing_assistance is housing_assistance gated on + # receives_housing_assistance, an input defaulting to false. The ungated + # variable is not a report of receipt: is_eligible_for_housing_assistance is + # receives_housing_assistance | (is_renter & is_income_eligible) and + # takes_up_housing_assistance_if_eligible defaults to true, so counting it + # would attach an imputed voucher to every income-eligible renter who + # supplies a rent: a Los Angeles renter earning $35,000 and paying $24,000 + # of rent, sending receives_housing_assistance: false, carries $11,580 of + # modelled assistance against the payment standard the model derives for the + # county, and $13,500 against a supplied $36,000 standard - either way out + # of CARE, for a subsidy nobody receives. The form asks an applicant what + # the household receives, so the entry counts receipt. + # + # It was spm_unit_capped_housing_subsidy, the Census SPM's housing subsidy + # capped at the housing portion of the SPM threshold less the tenant + # payment. That cap belongs to the poverty measure, not to this form, and it + # made a utility discount depend on which SPM geography the request + # selected: a Los Angeles renter earning $40,700 with $23,790 of housing + # assistance had a $0 capped subsidy under national selection and $310.88 + # under county selection, moving countable income across the CARE limit and + # flipping ca_care_eligible from true to false. Requesting CARE with no SPM + # configuration at all raised SPM_GEOGRAPHY_REQUIRED. + - reported_housing_assistance # Military subsidies - veterans_benefits - military_service_income @@ -57,8 +88,14 @@ metadata: period: year label: CA CPUC CARE/FERA income sources reference: - - title: PG&E Form 01-9077, CARE/FERA Program Application - href: https://bha.berkeleyca.gov/sites/default/files/2022-04/PGE%20CARE%20FERA%20Application.pdf + - title: PG&E Form 01-9077, CARE/FERA Program Application (Rev. 6.26) + href: https://www.pge.com/assets/pge/localized/en/docs/account/billing-and-assistance/care-fera-application.pdf + - title: >- + Revised Cal. P.U.C. Sheet No. 61313-E, PG&E Electric Sample Form + No. 79-1055, CARE/FERA Program Application for Sub-Metered + Residential Customers (Large Print), Advice 7905-E, effective + June 1, 2026 - carries the same income footnote as a CPUC filing + href: https://www.pge.com/tariffs/assets/pdf/tariffbook/ELEC_FORMS_79-1055.pdf - title: CA PUC 739.1 (CARE) href: https://leginfo.legislature.ca.gov/faces/codes_displaySection.xhtml?lawCode=PUC§ionNum=739.1 - title: CA PUC 739.12 (FERA) diff --git a/policyengine_us/spm.py b/policyengine_us/spm.py index bedc086632e..f208add1d13 100644 --- a/policyengine_us/spm.py +++ b/policyengine_us/spm.py @@ -34,6 +34,35 @@ COUNTY_FIPS_PATTERN = re.compile(r"[0-9]{5}") +# Poverty outputs this country derives from the calculator's measurement. +# +# ``FORMULA_OWNED_INPUTS`` names what the calculator owns: the thresholds, the +# SPM resource total, and the canonical indicators ``poverty_line``, +# ``poverty_gap``, ``spm_unit_is_in_spm_poverty`` and +# ``spm_unit_is_in_deep_spm_poverty``. These names are country-side functions +# of those same values - ``in_poverty`` of ``poverty_gap``, +# ``deep_poverty_line`` of ``poverty_line``, ``deep_poverty_gap`` of +# ``deep_poverty_line`` and ``spm_unit_net_income``, ``in_deep_poverty`` of +# ``deep_poverty_gap``, and ``person_in_poverty`` of ``in_poverty`` - so a +# dataset column for any of them is an alias for a formula-owned output under a +# name the calculator does not police. Accepting one lets the poverty +# calculation documented in ``docs/usage/microsimulation.md`` and the canonical +# indicator publish different poverty rates for the same population: a dataset +# storing ``in_poverty=[False, False]`` alongside a population every unit of +# which is below its threshold reports 0% and 100% at once. +DERIVED_POVERTY_OUTPUTS = frozenset( + { + "deep_poverty_gap", + "deep_poverty_line", + "in_deep_poverty", + "in_poverty", + "person_in_poverty", + } +) + +# Every name a dataset must not store, from either side of the boundary. +REJECTED_DATASET_INPUTS = frozenset(FORMULA_OWNED_INPUTS) | DERIVED_POVERTY_OUTPUTS + COUNTY_INPUT_FIX = ( 'send county_fips as a five-digit string (for example "06037"), or select ' 'geography_kind="national" in the spm configuration' @@ -347,7 +376,7 @@ def set_input(self, variable_name, period, value): # public consumers validate their household input contract separately. if ( getattr(self, "is_over_dataset", False) - and variable_name in FORMULA_OWNED_INPUTS + and variable_name in REJECTED_DATASET_INPUTS ): raise ValueError( f"Dataset supplies formula-owned SPM output {variable_name}. " diff --git a/policyengine_us/system.py b/policyengine_us/system.py index 6b83b8b3acd..f7798fd3cd6 100644 --- a/policyengine_us/system.py +++ b/policyengine_us/system.py @@ -1,3 +1,5 @@ +import hashlib +from functools import lru_cache from pathlib import Path from policyengine_core.taxbenefitsystems import TaxBenefitSystem from policyengine_us.entities import * @@ -53,6 +55,16 @@ # Populace ships from a Hugging Face *dataset* repo, hence the `hf://datasets/` # prefix handled in `_resolve_dataset_path`. DEFAULT_DATASET = "hf://datasets/policyengine/populace-us/populace_us_2024.h5@populace-us-2024-spm-20260909" +# Content hash of the certified build the default URI must resolve to. +# +# A Hugging Face tag is a mutable pointer: re-tagging the repository, or a +# truncated or substituted download, would hand the model another schema-valid +# H5 and change every standalone-country result with no rejection anywhere. +# Only the country's own default is checked - an explicit `dataset=` argument +# names the caller's own artifact and remains the caller's responsibility. +DEFAULT_DATASET_SHA256 = ( + "6496cc4393d4d3c6574f76eca231de5898c803b9067645591fd5c4d3e65aee84" +) class CountryTaxBenefitSystem(TaxBenefitSystem): @@ -312,6 +324,44 @@ def _resolve_dataset_path(dataset_str): raise FileNotFoundError(f"Dataset file not found: {dataset_str}") +def _file_sha256(file_path): + digest = hashlib.sha256() + with open(file_path, "rb") as handle: + for block in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +@lru_cache(maxsize=None) +def _checked_default_dataset_digest(file_path, size, mtime_ns): + """Digest the resolved default build once per process. + + The certified build is ~830 MB, so a per-construction digest would add + about a third of a second to every default `Microsimulation`. Key the + digest on the resolved path together with its size and modification time, + so a replaced or re-downloaded cache entry is digested again rather than + inheriting the previous file's verdict. + """ + return _file_sha256(file_path) + + +def _verify_default_dataset(file_path): + """Reject a default build whose bytes are not the certified ones.""" + stat = Path(file_path).stat() + actual = _checked_default_dataset_digest( + str(file_path), stat.st_size, stat.st_mtime_ns + ) + if actual != DEFAULT_DATASET_SHA256: + raise ValueError( + f"The default dataset {DEFAULT_DATASET} resolved to {file_path}, " + f"whose content is not the certified build: expected sha256 " + f"{DEFAULT_DATASET_SHA256}, got {actual}. The build id in the " + "default URI points at different bytes than this model version " + "was certified against; upgrade policyengine-us, or pass an " + "explicit dataset= argument to use that artifact deliberately." + ) + + def _is_hdfstore_format(file_path): """Check if an HDF5 file uses entity-level HDFStore format. @@ -362,6 +412,9 @@ def __init__(self, *args, **kwargs): ) dataset = kwargs.get("dataset") + # Only the build this model version chose for itself is content-checked + # below; a caller-supplied dataset= is the caller's own artifact. + uses_country_default = dataset is None if dataset is None: # Route the class default through the same interception below as an # explicit dataset, so an entity-level (HDFStore) default such as @@ -381,6 +434,10 @@ def __init__(self, *args, **kwargs): # entity-level datasets, making this interception unnecessary. if dataset is not None and isinstance(dataset, str): local_path = _resolve_dataset_path(dataset) + if uses_country_default and dataset == DEFAULT_DATASET: + # A subclass may point `default_dataset` elsewhere; only the + # shipped default is certified by DEFAULT_DATASET_SHA256. + _verify_default_dataset(local_path) if _is_hdfstore_format(local_path): from policyengine_us.data.economic_assumptions import ( extend_single_year_dataset, diff --git a/policyengine_us/tests/code_health/test_batched_exit_status.py b/policyengine_us/tests/code_health/test_batched_exit_status.py new file mode 100644 index 00000000000..10d5421f9c8 --- /dev/null +++ b/policyengine_us/tests/code_health/test_batched_exit_status.py @@ -0,0 +1,298 @@ +"""The batch runner's verdict must come from the runner's own exit status. + +`test_batched.py` used to decide a batch by parsing the pytest summary line: +absent a `"N failed"` count it called the batch passed, then terminated the +child without ever reading its status. pytest prints no failure count for an +errored run, so a real child producing + + 1 passed, 1 error in 0.01s (actual exit status: 1) + +was reported as a passing batch, and `make test-yaml-*` exited 0 on it. + +The status cannot simply be waited for: policyengine-core takes tens of seconds +to tear down after pytest.main() returns, which is why the child is terminated +as soon as it reports. So the command runs through a shim that writes pytest's +return value to stdout the instant it has one, and the runner reads that. + +Each test here drives the real `run_batch` against a real subprocess whose +output and status are scripted, so the runner's own Popen/select/terminate path +is what is under test, not a stand-in for it. +""" + +import importlib.util +import sys +from pathlib import Path + +import pytest +import yaml + +BATCHER = Path(__file__).resolve().parents[1] / "test_batched.py" +SPEC = importlib.util.spec_from_file_location("test_batched", BATCHER) +batched = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(batched) + + +def scripted_child( + summary: str, + exit_code: int, + then_hang: bool = False, + report: bool = True, +): + """A command that prints a pytest-shaped summary, then finishes as told. + + With report=True it also writes the shim's exit marker, standing in for + the real runner; with report=False it stands in for a child that died + before pytest could hand back a status. + """ + body = f"import sys, time\nprint({summary!r}, flush=True)\n" + if report: + body += ( + f"print('{batched.BATCH_EXIT_MARKER} ' + str({exit_code}), flush=True)\n" + ) + if then_hang: + body += "time.sleep(600)\n" + body += f"sys.exit({exit_code})\n" + return [sys.executable, "-u", "-c", body] + + +PASSING_SUMMARY = "============ 2 passed in 0.01s ============" +# pytest prints no failure count for an errored run: this is the exact string +# from the reproduction, and the one the old parser scored as a pass. +ERRORED_SUMMARY = "============ 1 passed, 1 error in 0.01s ============" +FAILING_SUMMARY = "============ 1 failed, 1 passed in 0.01s ============" + + +def run_scripted(summary: str, exit_code: int, **kwargs): + return batched.run_batch( + ["unused/path.yaml"], + "Batch 1", + stream=False, + command=scripted_child(summary, exit_code, **kwargs), + ) + + +def test_summary_without_a_failure_count_does_not_override_an_error_exit(): + """The reproduced defect: "1 passed, 1 error" with child exit 1.""" + result = run_scripted(ERRORED_SUMMARY, 1) + assert result["returncode"] == 1 + assert result["status"] == "failed" + + +def test_a_clean_summary_still_fails_on_a_nonzero_exit(): + """Interrupted, internal-error and usage-error runs print no count either.""" + for exit_code in (2, 3, 4): + result = run_scripted(PASSING_SUMMARY, exit_code) + assert result["returncode"] == exit_code + assert result["status"] == "failed", exit_code + + +def test_no_tests_collected_is_a_failure_for_the_batch_runner(): + """This runner enumerates the suite, so an empty batch is broken enumeration. + + `run_selective_tests.py` deliberately skips pytest's exit 5 because it + selects paths from a changed-file list, where a changed test helper + legitimately collects nothing. Nothing selects paths here but this file. + """ + result = run_scripted("============ no tests ran in 0.01s ============", 5) + assert result["returncode"] == batched.PYTEST_NO_TESTS_COLLECTED + assert result["status"] == "failed" + assert "collected no tests" in batched.describe_exit_status(5) + + +def test_a_passing_batch_still_passes(): + """The guard must not turn the ordinary green run red.""" + result = run_scripted(PASSING_SUMMARY, 0) + assert result["returncode"] == 0 + assert result["status"] == "passed" + + +def test_a_reported_failure_stays_a_failure(): + result = run_scripted(FAILING_SUMMARY, 1) + assert result["status"] == "failed" + + +def test_a_slow_teardown_after_a_clean_run_still_passes(): + """The real runner takes tens of seconds to exit; that is not a failure. + + Waiting the child out instead of reading its report turned every large + green batch red: measured on this suite, a single small state took 18.2s + to exit after its summary and the large state batches took over 30s. + """ + result = run_scripted(PASSING_SUMMARY, 0, then_hang=True) + assert result["returncode"] == 0 + assert result["status"] == "passed" + + +def test_a_slow_teardown_after_a_failed_run_still_fails(): + result = run_scripted(FAILING_SUMMARY, 1, then_hang=True) + assert result["returncode"] == 1 + assert result["status"] == "failed" + + +def test_a_child_killed_before_reporting_is_not_credited_a_pass(): + """No report and a nonzero status: an OOM kill mid-run, not a pass.""" + result = run_scripted(PASSING_SUMMARY, 137, report=False) + assert result["returncode"] == 137 + assert result["status"] == "failed" + + +def test_the_shim_reports_the_real_runner_status(): + """End to end against policyengine-core, not a scripted stand-in. + + An empty directory collects nothing, so pytest exits 5 - a status the + summary line never carries and the runner now reports. + """ + import tempfile + + with tempfile.TemporaryDirectory() as empty: + result = batched.run_batch([empty], "Batch 1", stream=False) + assert batched.BATCH_EXIT_MARKER in result["output"] + assert result["returncode"] == batched.PYTEST_NO_TESTS_COLLECTED + assert result["status"] == "failed" + + +def test_a_child_that_never_prints_a_summary_is_judged_by_its_exit_status(): + """Crashes before the summary line (OOM kill, import error) still count.""" + crashing = [sys.executable, "-u", "-c", "import sys; sys.exit(1)"] + result = batched.run_batch( + ["unused/path.yaml"], "Batch 1", stream=False, command=crashing + ) + assert result["status"] == "failed" + + clean = [sys.executable, "-u", "-c", "pass"] + assert ( + batched.run_batch(["unused/path.yaml"], "Batch 1", stream=False, command=clean)[ + "status" + ] + == "passed" + ) + + +@pytest.mark.parametrize("returncode", [1, 2, 3, 4, 5, -9, -15, None]) +def test_batch_status_passes_nothing_but_a_zero_exit(returncode): + assert batched.batch_status(returncode) == "failed" + + +def test_batch_status_passes_a_zero_exit(): + assert batched.batch_status(0) == "passed" + + +# --- .yml enumeration -------------------------------------------------------- +# +# policyengine-core collects both suffixes (tools/test_runner.py:413), so a +# `.yml` test omitted from a file-level batch would never run, and the batch it +# belonged to would still report green. + + +def write_cases(path: Path, reforms: list) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + yaml.safe_dump( + [ + { + "name": f"{path.stem} {reform}", + "period": 2026, + "reforms": reform, + "input": {"people": {"person": {"age": 30}}}, + "output": {"age": 30}, + } + for reform in reforms + ] + ) + ) + + +def test_yaml_enumeration_covers_both_suffixes(tmp_path): + write_cases(tmp_path / "a.yaml", ["r1"]) + write_cases(tmp_path / "nested" / "b.yml", ["r2"]) + assert batched.yaml_files(tmp_path) == [ + tmp_path / "a.yaml", + tmp_path / "nested" / "b.yml", + ] + assert batched.root_yaml_files(tmp_path) == [tmp_path / "a.yaml"] + assert batched.count_yaml_files(tmp_path) == 2 + + +def test_heavy_subdir_splitting_includes_yml_files(tmp_path): + """The split that packs a heavy folder file by file must see every file.""" + heavy = tmp_path / "heavy" + write_cases(heavy / "a.yaml", ["r1", "r2", "r3", "r4"]) + write_cases(heavy / "b.yml", ["r5", "r6"]) + batches = batched.subdir_batches(heavy) + packed = {path for batch in batches for path in batch} + assert str(heavy / "b.yml") in packed + assert str(heavy / "a.yaml") in packed + + +def test_heavy_subdir_weight_counts_yml_files(tmp_path): + """A folder pushed over budget by `.yml` files alone must still split.""" + heavy = tmp_path / "heavy" + write_cases(heavy / "a.yml", ["r1", "r2", "r3", "r4"]) + write_cases(heavy / "b.yml", ["r5", "r6"]) + assert batched.subdir_batches(heavy) != [[str(heavy)]] + + +def test_per_file_mode_includes_yml_files(tmp_path): + write_cases(tmp_path / "a.yaml", ["r1"]) + write_cases(tmp_path / "b.yml", ["r2"]) + batches = batched.split_into_batches(tmp_path, 2, mode="per-file") + assert batches == [[str(tmp_path / "a.yaml")], [str(tmp_path / "b.yml")]] + + +def test_per_subdir_mode_collects_root_yml_files(tmp_path): + write_cases(tmp_path / "light" / "a.yaml", ["r1"]) + write_cases(tmp_path / "root.yml", ["r2"]) + batches = batched.split_into_batches(tmp_path, 2, mode="per-subdir") + assert batches == [[str(tmp_path / "light")], [str(tmp_path / "root.yml")]] + + +def test_default_split_and_exclude_branches_include_yml_files(tmp_path): + write_cases(tmp_path / "keep" / "a.yml", ["r1"]) + write_cases(tmp_path / "drop" / "b.yaml", ["r2"]) + write_cases(tmp_path / "root.yml", ["r3"]) + assert batched.split_into_batches(tmp_path, 1, exclude=["drop"]) == [ + [str(tmp_path / "keep"), str(tmp_path / "root.yml")] + ] + assert batched.split_into_batches(tmp_path, 2) == [ + [str(tmp_path / "drop" / "b.yaml"), str(tmp_path / "keep" / "a.yml")], + [str(tmp_path / "root.yml")], + ] + + +# --- The monitoring loop is bounded ------------------------------------------ +# +# The loop ends when the child exits or reports, and nothing else. The 30-minute +# batch budget sat below it, where the child had already gone and the wait it +# guarded returned instantly, so it never fired. Reading the status from the +# runner rather than terminating a second after the summary also opened a +# second window: pytest's session-finish hooks, between the summary line and +# pytest.main() returning. Both are bounded inside the loop now. + + +def test_a_child_that_never_reports_after_its_summary_is_abandoned(monkeypatch): + """Not a hang: terminated, and reported as the failure it is.""" + monkeypatch.setattr(batched, "MARKER_GRACE_SECONDS", 1) + result = run_scripted(PASSING_SUMMARY, 0, then_hang=True, report=False) + assert result["status"] == "failed" + assert "No status" in result["output"] + + +def test_a_child_that_never_says_anything_hits_the_batch_budget(monkeypatch): + """The 30-minute budget the runner documents now actually fires.""" + monkeypatch.setattr(batched, "BATCH_TIMEOUT_SECONDS", 1) + silent = [sys.executable, "-u", "-c", "import time; time.sleep(600)"] + result = batched.run_batch( + ["unused/path.yaml"], "Batch 1", stream=False, command=silent + ) + assert result["status"] == "timeout" + assert "Timeout" in result["output"] + # main() counts anything but "passed" as a failure. + assert result["status"] != "passed" + + +def test_the_bounds_do_not_fire_on_an_ordinary_run(): + """A batch that reports promptly is untouched by either bound.""" + result = run_scripted(PASSING_SUMMARY, 0) + assert result["status"] == "passed" + assert "Timeout" not in result["output"] + assert "No status" not in result["output"] diff --git a/policyengine_us/tests/core/test_spm_system.py b/policyengine_us/tests/core/test_spm_system.py index 2b60bcd08ed..4c113ab1011 100644 --- a/policyengine_us/tests/core/test_spm_system.py +++ b/policyengine_us/tests/core/test_spm_system.py @@ -1,7 +1,10 @@ """Construction and dataset contracts that require the Python simulation API.""" +import ast import hashlib +import inspect import json +import textwrap import numpy as np import pandas as pd @@ -15,8 +18,20 @@ from policyengine_us import Microsimulation, Simulation from policyengine_us.data.dataset_schema import USMultiYearDataset, USSingleYearDataset from policyengine_us.entities import Person -from policyengine_us.spm import create_spm_provider -from policyengine_us.system import DEFAULT_DATASET, _resolve_dataset_path, system +from policyengine_us.spm import ( + DERIVED_POVERTY_OUTPUTS, + REJECTED_DATASET_INPUTS, + create_spm_provider, +) +from policyengine_us import system as system_module +from policyengine_us.system import ( + DEFAULT_DATASET, + DEFAULT_DATASET_SHA256, + _file_sha256, + _resolve_dataset_path, + _verify_default_dataset, + system, +) def single_person_situation(): @@ -474,3 +489,219 @@ def test_microsimulation_rejects_stored_measurements_without_mutating_dataset( pd.testing.assert_frame_equal(original, current) if dataset_format == "hdfstore": assert path.read_bytes() == content + + +@pytest.mark.parametrize("name", sorted(DERIVED_POVERTY_OUTPUTS)) +def test_dataset_storing_a_derived_poverty_output_is_refused(name): + """A poverty alias the calculator does not own must still be refused. + + ``FORMULA_OWNED_INPUTS`` stops at the calculator's own outputs, so a + dataset supplying ``in_poverty=[False, False]`` was accepted unchanged + while every unit in the population sat below its threshold: the poverty + calculation documented in ``docs/usage/microsimulation.md`` reported 0% + and ``spm_unit_is_in_spm_poverty`` reported 100% for the same population. + """ + variable = system.variables[name] + source = small_dataset() + table = getattr(source, variable.entity.key) + table[name] = [False, False] if variable.value_type == bool else [0.0, 0.0] + before = [frame.copy(deep=True) for frame in source.tables] + with pytest.raises(ValueError) as error: + Microsimulation(dataset=source) + message = str(error.value) + assert f"formula-owned SPM output {name}" in message + assert "Use primitive inputs" in message + for original, current in zip(before, source.tables): + pd.testing.assert_frame_equal(original, current) + + +def test_accepted_population_publishes_one_poverty_rate(): + """The two published indicators agree once the aliases cannot be stored.""" + simulation = Microsimulation(dataset=small_dataset()) + np.testing.assert_array_equal( + simulation.calculate("in_poverty", 2024), + simulation.calculate("spm_unit_is_in_spm_poverty", 2024), + ) + np.testing.assert_array_equal( + simulation.calculate("in_deep_poverty", 2024), + simulation.calculate("spm_unit_is_in_deep_spm_poverty", 2024), + ) + + +def test_rejection_set_extends_rather_than_replaces_the_calculator_contract(): + for name in FORMULA_OWNED_INPUTS: + assert name in REJECTED_DATASET_INPUTS + for name in DERIVED_POVERTY_OUTPUTS: + # A rejected name must be computed, never a legitimate stored input. + assert system.variables[name].formula is not None + + +ENTITY_KEYS = frozenset( + {"person", "marital_unit", "tax_unit", "family", "spm_unit", "household"} +) +# policyengine-core helpers that name the variables they read as string +# literals, exactly as an entity call does, so the constant scan below sees +# their reads too (policyengine_core/commons/formulas.py: for_each_variable +# and the aggregators built on it). +NAME_LISTING_HELPERS = frozenset( + {"add", "and_", "or_", "max_", "min_", "for_each_variable", "sum_of_variables"} +) + + +def _static_variable_reads(variable): + """Variable names a registered variable reads, or None when not static. + + Over-collects rather than under-collects: every string constant in the + class body that names a registered variable counts, so a helper this scan + does not model cannot make a variable look like a pure function of the + poverty chain when it is not. + + A formula is only trusted when its reads are visibly named — through an + entity call or a name-listing helper. A variable with no formula at all is + fully described by its ``adds``/``subtracts`` list, which is how most of + the poverty chain is written: ``poverty_line`` is ``adds = + ["spm_unit_spm_threshold"]`` and nothing else. Requiring an entity call of + those too discarded 884 of the system's variables, this measurement's own + aliases among them, leaving the closure below blind to the very idiom the + next alias would most likely use. + """ + for attribute in ("adds", "subtracts"): + if isinstance(getattr(variable, attribute, None), str): + # A parameter path: its contents are not statically known here. + return None + reads = { + item + for attribute in ("adds", "subtracts") + for item in (getattr(variable, attribute, None) or []) + } + try: + source = textwrap.dedent(inspect.getsource(type(variable))) + except (OSError, TypeError): + # Calculator-built variables have no country source file. + return None + tree = ast.parse(source) + named_reads = False + has_formula = False + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + if node.name.startswith("formula"): + has_formula = True + if isinstance(node, ast.Call): + function = node.func + name = getattr(function, "id", None) or getattr(function, "attr", None) + if name in ENTITY_KEYS or name in NAME_LISTING_HELPERS: + named_reads = True + if ( + isinstance(node, ast.Constant) + and isinstance(node.value, str) + and node.value in system.variables + ): + reads.add(node.value) + if has_formula and not named_reads: + # A formula reaching for its inputs some other way; what it reads is + # not knowable from the source. + return None + return reads - {variable.name} + + +def test_every_pure_function_of_the_poverty_chain_is_rejected(): + """The rejection set is closed: no further alias can be stored. + + A variable whose every read is already rejected and which reads at least + one threshold or poverty output is another name for the same measurement. + ``in_poverty`` was exactly that, and this closure is what stops the next + one from shipping unpoliced. + """ + chain = { + name + for name in REJECTED_DATASET_INPUTS + if "poverty" in name or "threshold" in name + } + unrejected_aliases = [] + checked = [] + for name, variable in sorted(system.variables.items()): + reads = _static_variable_reads(variable) + if not reads or not reads <= REJECTED_DATASET_INPUTS or not reads & chain: + continue + checked.append(name) + if name not in REJECTED_DATASET_INPUTS: + unrejected_aliases.append(f"{name} reads {sorted(reads)}") + assert not unrejected_aliases, ( + "These variables are pure functions of rejected SPM poverty outputs " + "but a dataset may still store them; add them to " + "DERIVED_POVERTY_OUTPUTS in policyengine_us/spm.py:\n" + + "\n".join(unrejected_aliases) + ) + # Guard the guard: a scan that silently matched nothing proves nothing. + assert set(DERIVED_POVERTY_OUTPUTS) <= set(checked) + + +@pytest.fixture +def decoy_default_dataset(monkeypatch, tmp_path): + """Resolve the default URI to a valid but uncertified build. + + The decoy loads and calculates, so anything that rejects it rejects it on + content rather than on shape. + """ + path = tmp_path / "decoy.h5" + small_dataset().save(path) + monkeypatch.setattr( + system_module, + "_resolve_dataset_path", + lambda dataset_str: str(path), + ) + return path + + +def test_default_dataset_rejects_content_that_is_not_the_certified_build( + decoy_default_dataset, +): + """A moved build id must fail loudly instead of changing every result. + + The default pins a Hugging Face tag, which is a mutable pointer: without + this check, re-tagging the repository silently substitutes another + schema-valid population. + """ + with pytest.raises(ValueError) as error: + Microsimulation() + message = str(error.value) + assert DEFAULT_DATASET in message + assert str(decoy_default_dataset) in message + assert DEFAULT_DATASET_SHA256 in message + assert _file_sha256(decoy_default_dataset) in message + + +def test_default_dataset_accepts_the_certified_digest( + decoy_default_dataset, monkeypatch +): + """The check passes exactly the certified bytes and obstructs nothing else.""" + monkeypatch.setattr( + system_module, + "DEFAULT_DATASET_SHA256", + _file_sha256(decoy_default_dataset), + ) + simulation = Microsimulation() + assert simulation.calculate("spm_measurement_adults", 2024).tolist() == [1, 1] + + +def test_explicit_dataset_argument_is_not_content_checked(decoy_default_dataset): + """A caller naming its own artifact owns that artifact's provenance.""" + simulation = Microsimulation(dataset=str(decoy_default_dataset)) + assert simulation.calculate("spm_measurement_adults", 2024).tolist() == [1, 1] + + +def test_default_dataset_is_digested_once_per_resolved_file(monkeypatch, tmp_path): + """The certified build is ~830 MB; construction must not re-digest it.""" + path = tmp_path / "counted.h5" + path.write_bytes(b"populace") + digests = [] + + def counting_sha256(file_path): + digests.append(str(file_path)) + return "0" * 64 + + monkeypatch.setattr(system_module, "_file_sha256", counting_sha256) + monkeypatch.setattr(system_module, "DEFAULT_DATASET_SHA256", "0" * 64) + _verify_default_dataset(path) + _verify_default_dataset(path) + assert digests == [str(path)] diff --git a/policyengine_us/tests/policy/baseline/gov/hud/reported_housing_assistance.yaml b/policyengine_us/tests/policy/baseline/gov/hud/reported_housing_assistance.yaml new file mode 100644 index 00000000000..44fcbda1c84 --- /dev/null +++ b/policyengine_us/tests/policy/baseline/gov/hud/reported_housing_assistance.yaml @@ -0,0 +1,30 @@ +- name: Reported receipt passes the assistance through in full. + period: 2026 + input: + receives_housing_assistance: true + housing_assistance: 1_200 + output: + reported_housing_assistance: 1_200 + +- name: An unreported voucher is not counted, whatever the model computes. + period: 2026 + input: + receives_housing_assistance: false + housing_assistance: 1_200 + output: + reported_housing_assistance: 0 + +- name: Receipt defaults to false, so the default is zero. + period: 2026 + input: + housing_assistance: 1_200 + output: + reported_housing_assistance: 0 + +- name: A reporting unit with no assistance has nothing to count. + period: 2026 + input: + receives_housing_assistance: true + housing_assistance: 0 + output: + reported_housing_assistance: 0 diff --git a/policyengine_us/tests/policy/baseline/gov/states/ca/cpuc/ca_cpuc_countable_income.yaml b/policyengine_us/tests/policy/baseline/gov/states/ca/cpuc/ca_cpuc_countable_income.yaml index aa9eda413ec..40ffe85d5d9 100644 --- a/policyengine_us/tests/policy/baseline/gov/states/ca/cpuc/ca_cpuc_countable_income.yaml +++ b/policyengine_us/tests/policy/baseline/gov/states/ca/cpuc/ca_cpuc_countable_income.yaml @@ -40,3 +40,78 @@ state_code: CA output: ca_cpuc_countable_income: 68_000 + +# PG&E Form 01-9077 counts "housing and military subsidies" among the revenues +# making up total gross annual household income, with no netting against rent +# and no cap on the form. The entry is reported_housing_assistance: the form +# asks an applicant what the household receives, so a subsidy the model imputes +# to an income-eligible renter who does not report one is not counted. This +# used to count the SPM's capped housing subsidy, which made the discount +# depend on the SPM geography a request selected. +- name: CPUC countable income includes the housing subsidy received in full + period: 2026 + input: + people: + person: + employment_income: 20_000 + spm_units: + spm_unit: + members: [person] + receives_housing_assistance: true + housing_assistance: 12_000 + # Pinned so the housing entry is what the total tests, not the SNAP + # this earnings level happens to qualify for. + snap: 0 + households: + household: + members: [person] + state_code: CA + output: + ca_cpuc_countable_income: 32_000 + +- name: CPUC countable income excludes a housing subsidy not reported as received + period: 2026 + input: + people: + person: + employment_income: 20_000 + spm_units: + spm_unit: + members: [person] + # The amount is supplied directly, so the gate is what this case + # tests: receives_housing_assistance defaults to false, and without + # it nothing is counted. In the household path the amount does not + # have to be supplied - housing_assistance is eligibility times + # take-up, and take-up defaults to true, so the model carries one for + # any income-eligible renter who supplies a rent. + housing_assistance: 12_000 + snap: 0 + households: + household: + members: [person] + state_code: CA + output: + reported_housing_assistance: 0 + ca_cpuc_countable_income: 20_000 + +- name: CPUC countable income includes military subsidies alongside housing + period: 2026 + input: + people: + person: + employment_income: 20_000 + veterans_benefits: 5_000 + spm_units: + spm_unit: + members: [person] + receives_housing_assistance: true + housing_assistance: 12_000 + # Pinned so the housing entry is what the total tests, not the SNAP + # this earnings level happens to qualify for. + snap: 0 + households: + household: + members: [person] + state_code: CA + output: + ca_cpuc_countable_income: 37_000 diff --git a/policyengine_us/tests/test_batched.py b/policyengine_us/tests/test_batched.py index bf84fbee3f3..54b73ff1f8d 100644 --- a/policyengine_us/tests/test_batched.py +++ b/policyengine_us/tests/test_batched.py @@ -21,11 +21,33 @@ import yaml +# policyengine-core collects both YAML suffixes +# (policyengine_core/tools/test_runner.py accepts ".yaml" and ".yml"), so any +# batch built by listing files has to list both. Enumerating only "*.yaml" +# would hand the subprocess a path list that silently omits a ".yml" test — +# invisibly, because the batch it belonged to still passes. +YAML_SUFFIXES = (".yaml", ".yml") + + +def yaml_files(directory: Path) -> List[Path]: + """Every YAML test file under a directory, recursively, in a stable order.""" + return sorted( + path for suffix in YAML_SUFFIXES for path in directory.rglob(f"*{suffix}") + ) + + +def root_yaml_files(directory: Path) -> List[Path]: + """YAML test files directly in a directory, not in its subdirectories.""" + return sorted( + path for suffix in YAML_SUFFIXES for path in directory.glob(f"*{suffix}") + ) + + def count_yaml_files(directory: Path) -> int: """Count YAML files in a directory recursively.""" if not directory.exists(): return 0 - return len(list(directory.rglob("*.yaml"))) + return len(yaml_files(directory)) def batch_cost(batch_paths: List[str]) -> int: @@ -178,7 +200,7 @@ def subdir_batches(subdir: Path) -> List[List[str]]: minutes before CI killed it. Pack such a folder's files by combo weight instead. """ - files = sorted(subdir.rglob("*.yaml")) + files = yaml_files(subdir) combos: set = set() for file in files: combos |= set(file_reform_combos(file)) @@ -214,7 +236,7 @@ def split_into_batches( # Explicit modes — bypass the per-path auto heuristics so new files # added to the target auto-route without a Makefile edit. if mode == "per-file": - return [[str(f)] for f in sorted(base_path.rglob("*.yaml"))] + return [[str(f)] for f in yaml_files(base_path)] if mode == "per-subdir": subdirs = sorted( @@ -222,7 +244,7 @@ def split_into_batches( for item in base_path.iterdir() if item.is_dir() and item.name not in exclude ) - root_files = sorted(base_path.glob("*.yaml")) + root_files = root_yaml_files(base_path) batches = [batch for s in subdirs for batch in subdir_batches(s)] if root_files: batches.append([str(f) for f in root_files]) @@ -264,7 +286,7 @@ def split_into_batches( for item in base_path.iterdir() if item.is_dir() and item.name not in exclude ) - root_files = sorted(base_path.glob("*.yaml")) + root_files = root_yaml_files(base_path) # One batch per file for per-file folders, then one batch per # heavy subdir (if present). @@ -272,7 +294,7 @@ def split_into_batches( [str(file)] for subdir in subdirs if subdir.name in PER_FILE - for file in sorted(subdir.rglob("*.yaml")) + for file in yaml_files(subdir) ] batches += [[str(subdir)] for subdir in subdirs if subdir.name in HEAVY] @@ -303,10 +325,10 @@ def split_into_batches( subdirs = sorted([item for item in base_path.iterdir() if item.is_dir()]) batches = [] for subdir in subdirs: - batches.extend(pack_files_by_combo_weight(sorted(subdir.rglob("*.yaml")))) + batches.extend(pack_files_by_combo_weight(yaml_files(subdir))) # Also include any root-level YAML files as trailing batches - root_files = sorted(base_path.glob("*.yaml")) + root_files = root_yaml_files(base_path) if root_files: batches.extend(pack_files_by_combo_weight(root_files)) @@ -319,7 +341,7 @@ def split_into_batches( # shutdown signal". A fresh subprocess per file frees each peak between # files. if "reform" in str(base_path): - return [[str(f)] for f in sorted(base_path.rglob("*.yaml"))] + return [[str(f)] for f in yaml_files(base_path)] # Special handling for states directory - support excluding specific states # and splitting into multiple sequential batches for memory management @@ -334,7 +356,7 @@ def split_into_batches( # Root-level YAML files (e.g. cross-state filing-status test) are # state-agnostic and would be invisible to subdir-based batching # otherwise — collect them into a dedicated trailing batch. - root_files = sorted(base_path.glob("*.yaml")) + root_files = root_yaml_files(base_path) if not subdirs and not root_files: return [] @@ -395,7 +417,7 @@ def collect_gov_paths(folder_names): if gov_item.is_dir(): if gov_item.name not in heavy_set and gov_item.name != "states": remaining.append(str(gov_item)) - elif gov_item.suffix == ".yaml": + elif gov_item.suffix in YAML_SUFFIXES: remaining.append(str(gov_item)) # Add non-gov directories and root YAML files @@ -404,7 +426,7 @@ def collect_gov_paths(folder_names): if item.name in ["household", "contrib", "gov"]: continue remaining.append(str(item)) - elif item.suffix == ".yaml": + elif item.suffix in YAML_SUFFIXES: remaining.append(str(item)) # Build batches (only include non-empty ones) @@ -435,10 +457,11 @@ def collect_gov_paths(folder_names): paths = sorted( str(item) for item in base_path.iterdir() - if (item.is_dir() or item.suffix == ".yaml") and item.name not in exclude + if (item.is_dir() or item.suffix in YAML_SUFFIXES) + and item.name not in exclude ) else: - paths = sorted(str(f) for f in base_path.rglob("*.yaml")) + paths = [str(f) for f in yaml_files(base_path)] if not paths: return [] @@ -459,7 +482,113 @@ def collect_gov_paths(folder_names): return batches -def run_batch(test_paths: List[str], batch_name: str, stream: bool = True) -> Dict: +# pytest exits 5 when a path collects no tests (see pytest.ExitCode). +PYTEST_NO_TESTS_COLLECTED = 5 +# How long a child gets to die after being asked to, before it is killed. +TERMINATE_GRACE_SECONDS = 5 +# Overall budget for one batch, enforced inside the monitoring loop below. +BATCH_TIMEOUT_SECONDS = 1800 +# How long the runner may take to hand back a status after printing its +# summary. That gap is pytest's own session-finish hooks and unconfigure, not +# the slow interpreter teardown described below, which happens after the +# marker: measured, the marker arrives in the same read as the summary line. +# This is deliberately generous, and exists only so a runner that stops +# responding right at the end is reported instead of hanging the job. +MARKER_GRACE_SECONDS = 300 + +# The runner's process exit status is not reachable in bounded time. Once +# policyengine-core's pytest.main() returns, interpreter teardown frees the +# tax-benefit systems the run cached, which is slow: measured on this suite, a +# single small state took 18.2s to exit after printing its summary, and the +# large state batches took over 30s. That is why the runner is terminated as +# soon as it reports rather than waited out - and why its status used to be +# discarded along with it, leaving the printed summary as the only evidence. +# +# So ask for the status at the only moment it is both known and free: run the +# command through a shim that writes pytest's own return value to stdout the +# instant pytest.main() hands it back, before any teardown. Measured at the +# same moment as the summary line, so this costs no wall clock, and the child +# is still terminated exactly as promptly as before. +BATCH_EXIT_MARKER = "__POLICYENGINE_BATCH_EXIT__" + +RUNNER_SHIM = f""" +import runpy +import sys + +code = 0 +try: + runpy.run_module( + "policyengine_core.scripts.policyengine_command", run_name="__main__" + ) +except SystemExit as requested: + code = requested.code +sys.stdout.write("\\n{BATCH_EXIT_MARKER} %s\\n" % (0 if code is None else code)) +sys.stdout.flush() +sys.exit(code) +""" + +BATCH_EXIT_PATTERN = re.compile(rf"{BATCH_EXIT_MARKER} (-?\d+)") + + +def build_test_command(test_paths: List[str], python_exe: str) -> List[str]: + """The policyengine-core command one batch runs in its own subprocess. + + `python -c` stops consuming options after the program text, so everything + after RUNNER_SHIM reaches policyengine_command's own argument parser + unchanged. + """ + return ( + [python_exe, "-c", RUNNER_SHIM, "test"] + test_paths + ["-c", "policyengine_us"] + ) + + +def describe_exit_status(returncode: Optional[int]) -> str: + """Human-readable form of a batch subprocess's exit status.""" + if returncode is None: + return "no exit status (killed after it stopped responding)" + if returncode == 0: + return "exit 0" + if returncode < 0: + return f"killed by signal {-returncode}" + if returncode == PYTEST_NO_TESTS_COLLECTED: + return f"exit {returncode} (pytest collected no tests)" + return f"exit {returncode}" + + +def batch_status(returncode: Optional[int]) -> str: + """Batch outcome, decided by the runner's exit status and nothing else. + + policyengine-core's `test` command is `sys.exit(pytest.main(...))` + (scripts/run_test.py), so the status is pytest's own ExitCode and only 0 is + a pass. Reading the printed summary instead loses every outcome pytest does + not express as a failure count: + + * "1 passed, 1 error" prints no failed count at all, so the old parse read + an errored batch as success while the child exited 1. + * 2 (interrupted) prints a count for whatever ran before the interrupt and + no failure count, which the old parse also read as success: under + policyengine-core's argv, a session whose second test raises + KeyboardInterrupt exits 2 beneath "1 passed in 0.10s" (measured). + * 3 and 4 (internal error, usage error) print no summary count at all, so + the old parse fell through to a status it had already discarded. + * 5 (no tests collected) is a failure *for this runner*. It differs from + run_selective_tests.py, which skips 5 deliberately: that runner selects + paths from a changed-file list, where a changed test helper legitimately + collects nothing. This runner enumerates the YAML suite itself, so a + batch that collects nothing means the suite or this enumeration is + broken — a result to report, not to pass over. + * None means the child never reported a status, having been killed after + it stopped responding. An unreported batch is not a passing batch. + """ + return "passed" if returncode == 0 else "failed" + + +def run_batch( + test_paths: List[str], + batch_name: str, + stream: bool = True, + command: Optional[List[str]] = None, +) -> Dict: """Run a batch of tests in an isolated subprocess. With stream=True output is echoed to stdout in real time (sequential @@ -485,17 +614,10 @@ def emit(text: str, end: str = "\n") -> None: peak_rss_mb = None last_rss_sample = 0.0 - # Build command - direct policyengine-core with timeout protection - cmd = ( - [ - python_exe, - "-m", - "policyengine_core.scripts.policyengine_command", - "test", - ] - + test_paths - + ["-c", "policyengine_us"] - ) + # Build command - direct policyengine-core with timeout protection. + # `command` is an injection point for the code-health tests, which drive a + # real subprocess with scripted output and exit statuses. + cmd = command or build_test_command(test_paths, python_exe) emit(f" Running {batch_name}...") emit(f" Paths: {len(test_paths)} items") @@ -520,10 +642,41 @@ def sample_rss() -> None: try: test_completed = False - test_passed = False + summary_reported_pass = False + abandoned: Optional[str] = None + reported_returncode: Optional[int] = None output_lines = [] output_text = "" + summary_at: Optional[float] = None + + def stop_process() -> None: + """End a child, whether or not it told us how it finished.""" + process.terminate() + try: + process.wait(timeout=TERMINATE_GRACE_SECONDS) + except subprocess.TimeoutExpired: + emit(" Force killing process...") + process.kill() + process.wait() + + def overdue() -> Optional[str]: + """Why this batch should stop being waited on, if it should. + + The monitoring loop is otherwise unbounded, in this revision and + the one before it: it ends only when the child exits or reports, + and the 30-minute budget below it was only ever consulted after + the child had already gone. Enforce that budget here, where it can + fire, and bound the window this file opened by reading the status + from the runner instead of terminating a second after its summary. + """ + now = time.time() + if now - start_time >= BATCH_TIMEOUT_SECONDS: + return "timeout" + if summary_at is not None and now - summary_at >= MARKER_GRACE_SECONDS: + return "unreported" + return None + # Monitor output line by line while True: ready, _, _ = select.select([process.stdout], [], [], 0.1) @@ -535,6 +688,20 @@ def sample_rss() -> None: if poll_result is not None: # Process terminated break + give_up = overdue() + if give_up is not None: + abandoned = give_up + emit( + "\n ⏱️ " + + ( + "Timeout" + if give_up == "timeout" + else f"No status {MARKER_GRACE_SECONDS}s after the summary" + ) + + " - terminating process..." + ) + stop_process() + break now = time.time() # In buffered (concurrent) mode the shared heartbeat in # run_batches_concurrently reports liveness instead. @@ -572,41 +739,38 @@ def sample_rss() -> None: # Detect pytest completion # Look for patterns like "====== 5638 passed in 491.24s ======" # or "====== 2 failed, 5636 passed in 500s ======" - if re.search( + if not test_completed and re.search( r"=+.*\d+\s+(passed|failed).*in\s+[\d.]+s.*=+", output_text, ): test_completed = True - # Check if tests passed by parsing actual failure count + summary_at = time.time() + # The summary line only decides how loudly a disagreement is + # reported below; batch_status() reads the runner's status. failed_match = re.search(r"(\d+) failed", output_text) - if failed_match: - failed_count = int(failed_match.group(1)) - test_passed = failed_count == 0 - else: - # No "X failed" in line means all passed - test_passed = True - - emit(f"\n Tests completed, terminating process...") + summary_reported_pass = not failed_match or ( + int(failed_match.group(1)) == 0 + ) - # Give 1 second grace period for cleanup - time.sleep(1) + # The runner has reported pytest's return value and is now only + # tearing down, which takes tens of seconds. Stop it here, as + # before, but keep the status it just gave us. + marker_match = BATCH_EXIT_PATTERN.search(output_text) + if marker_match: + reported_returncode = int(marker_match.group(1)) + emit( + "\n Runner reported " + f"{describe_exit_status(reported_returncode)}; " + "terminating process..." + ) # Final sample while the pid still exists — captures the # end-of-run high-water mark. sample_rss() - - # Terminate the process - process.terminate() - try: - process.wait(timeout=5) - except subprocess.TimeoutExpired: - # Force kill if it won't terminate - emit(f" Force killing process...") - process.kill() - process.wait() + stop_process() break - # If we didn't detect completion, wait for process with timeout - if not test_completed: + # If the runner never reported, wait for the process with a timeout + if reported_returncode is None and process.poll() is None: try: # Wait up to 30 minutes total elapsed = time.time() - start_time @@ -626,36 +790,53 @@ def sample_rss() -> None: return { "elapsed": elapsed, "status": "timeout", + "returncode": process.poll(), "peak_rss_mb": peak_rss_mb, "output": "".join(buf), } elapsed = time.time() - start_time - if test_completed: + # The runner's own report wins: the process status after a terminate is + # the signal we sent it, not how the tests went. + returncode = ( + reported_returncode if reported_returncode is not None else process.poll() + ) + + if abandoned == "timeout": emit( - f"\n Batch completed in {elapsed:.1f}s " + f"\n Batch abandoned after {elapsed:.1f}s " f"(peak RSS: {_format_rss(peak_rss_mb)})" ) return { "elapsed": elapsed, - "status": "passed" if test_passed else "failed", + "status": "timeout", + "returncode": returncode, "peak_rss_mb": peak_rss_mb, "output": "".join(buf), } - else: - # Process ended without detecting test completion - returncode = process.poll() + + status = batch_status(returncode) + if test_completed and summary_reported_pass and status != "passed": + # The case that used to be reported as a pass: pytest printed no + # failure count (e.g. "1 passed, 1 error") yet finished nonzero. emit( - f"\n Batch completed in {elapsed:.1f}s " - f"(peak RSS: {_format_rss(peak_rss_mb)})" + f" ❌ {batch_name} printed a summary with no failures but " + f"finished with {describe_exit_status(returncode)}; " + "the runner's status decides." ) - return { - "elapsed": elapsed, - "status": "passed" if returncode == 0 else "failed", - "peak_rss_mb": peak_rss_mb, - "output": "".join(buf), - } + emit( + f"\n Batch completed in {elapsed:.1f}s " + f"(peak RSS: {_format_rss(peak_rss_mb)}, " + f"{describe_exit_status(returncode)})" + ) + return { + "elapsed": elapsed, + "status": status, + "returncode": returncode, + "peak_rss_mb": peak_rss_mb, + "output": "".join(buf), + } except Exception as e: elapsed = time.time() - start_time @@ -675,6 +856,7 @@ def sample_rss() -> None: return { "elapsed": elapsed, "status": "error", + "returncode": process.poll(), "peak_rss_mb": peak_rss_mb, "output": "".join(buf), } @@ -743,6 +925,7 @@ def run_one(idx: int, batch_paths: List[str]) -> Dict: result = { "elapsed": 0.0, "status": "error", + "returncode": None, "peak_rss_mb": None, "output": f"Worker crashed: {str(e)[:200]}\n", } @@ -754,6 +937,7 @@ def run_one(idx: int, batch_paths: List[str]) -> Dict: f"\n[Batch {idx}/{total}] finished in " f"{result['elapsed']:.1f}s | peak RSS: " f"{_format_rss(result.get('peak_rss_mb'))} | " + f"{describe_exit_status(result.get('returncode'))} | " f"{result['status']}", flush=True, ) @@ -897,7 +1081,7 @@ def main(): print(" Cleaning up memory...") gc.collect() - all_failed = any(result["status"] != "passed" for _, result in results) + any_failed = any(result["status"] != "passed" for _, result in results) total_elapsed = sum(result["elapsed"] for _, result in results) # Final summary @@ -909,6 +1093,7 @@ def main(): print( f" {name}: {result['elapsed']:.1f}s | " f"peak RSS: {_format_rss(result.get('peak_rss_mb'))} | " + f"{describe_exit_status(result.get('returncode'))} | " f"{result['status']}" ) print(f"Workers: {args.workers}") @@ -917,7 +1102,7 @@ def main(): print(f"Wall time: {time.time() - run_start:.1f}s") # Exit code - sys.exit(1 if all_failed else 0) + sys.exit(1 if any_failed else 0) if __name__ == "__main__": diff --git a/policyengine_us/tests/unit/test_spm_integration_contract.py b/policyengine_us/tests/unit/test_spm_integration_contract.py index e03e00f18a1..42dd64da715 100644 --- a/policyengine_us/tests/unit/test_spm_integration_contract.py +++ b/policyengine_us/tests/unit/test_spm_integration_contract.py @@ -419,3 +419,199 @@ def test_generic_benefit_consumers_do_not_use_spm_measurement_composition( simulation.calculate("spm_measurement_children", YEAR), [1, 0] ) assert simulation.spm_provenance()["years"] == {} + + +# --- CPUC CARE/FERA countable income ----------------------------------------- +# +# CARE and FERA are utility discounts, not poverty measurement. PG&E Form +# 01-9077 counts "housing and military subsidies" among the revenues making up +# total gross annual household income, with no netting against rent and no cap +# anywhere on the form, so gov.states.ca.cpuc.income_sources counts the housing +# assistance the household reports receiving - reported_housing_assistance, +# which is housing_assistance gated on receives_housing_assistance. It used to +# count spm_unit_capped_housing_subsidy, which made the discount depend on the +# SPM geography the request selected. + + +def assisted_la_renter(earnings, *, county=None): + situation = household(earnings=earnings, county=county) + situation["spm_units"]["spm_unit"]["pre_subsidy_electricity_expense"] = { + YEAR: 1_800 + } + # Pin the categorical route off so income is what decides, and the + # equal-eligibility assertions below cannot pass for an unrelated reason. + situation["households"]["household"]["ca_care_categorically_eligible"] = { + YEAR: False + } + return situation + + +CARE_OUTPUTS = ("ca_cpuc_countable_income", "ca_care_income_eligible", "ca_care") + + +@pytest.mark.parametrize( + "earnings,eligible", + [ + # $0 earnings, ~$35,975 of assistance: $36,559 countable, inside the + # 2024 one-person CARE limit of 2 x $20,440. + (0, True), + # The reported case: a Los Angeles renter earning $40,700 with $23,790 + # of assistance. $64,490 countable puts it outside the limit under + # every geography, where the capped subsidy flipped it at $310.88. + (40_700, False), + ], +) +def test_care_eligibility_is_the_same_under_every_spm_geography(earnings, eligible): + """An assisted household's utility discount must not move with SPM geography. + + The capped subsidy is a poverty-measurement construct: it nets the + assistance against the housing portion of the SPM threshold, which is + geographically adjusted. Counting it made one Los Angeles renter + CARE-eligible nationally and ineligible under county selection. + """ + national = Simulation( + situation=assisted_la_renter(earnings), + spm={"geography_kind": "national"}, + ) + county = Simulation( + situation=assisted_la_renter(earnings, county="06037"), + spm={"geography_kind": "county"}, + ) + unconfigured = Simulation(situation=assisted_la_renter(earnings)) + + # Guard the guard: the capped subsidy this parameter used to count still + # differs between the two selections, so equal CARE outputs below are the + # income definition's doing and not a quiet loss of geographic adjustment. + assert ( + national.calculate("spm_unit_capped_housing_subsidy", YEAR)[0] + != (county.calculate("spm_unit_capped_housing_subsidy", YEAR)[0]) + ) + + for variable in CARE_OUTPUTS: + values = [sim.calculate(variable, YEAR)[0] for sim in (national, county)] + assert values[0] == values[1], variable + assert bool(national.calculate("ca_care_eligible", YEAR)[0]) is eligible + + # And the same answer with no SPM configuration and no county at all: this + # is what raised SPM_GEOGRAPHY_REQUIRED for partners requesting CARE. + for variable in (*CARE_OUTPUTS, "ca_care_eligible"): + assert ( + unconfigured.calculate(variable, YEAR)[0] + == national.calculate(variable, YEAR)[0] + ), variable + # Nothing was looked up, so no measurement was received. + assert unconfigured.spm_provenance()["years"] == {} + + +def test_cpuc_countable_income_counts_the_full_housing_assistance(): + """Non-vacuity: the subsidy is in countable income at its full amount. + + Equal answers across geographies would also hold if the entry had simply + been dropped, so pin the arithmetic the form describes. + """ + simulation = Simulation(situation=assisted_la_renter(40_700)) + assistance = simulation.calculate("housing_assistance", YEAR)[0] + assert assistance > 0 + assert simulation.calculate("ca_cpuc_countable_income", YEAR)[0] == pytest.approx( + 40_700 + assistance + ) + + +def income_eligible_la_renter(*, receives, county="06037"): + """A renter HUD would income-qualify, so the imputation has something to bite. + + Earnings of $35,000 against $24,000 of rent leave this household inside the + HUD low-income limit for Los Angeles County, so it is eligible for housing + assistance whatever it reports, and `takes_up_housing_assistance_if_eligible` + defaults to true. + """ + situation = household(earnings=35_000, county=county, rent=24_000) + situation["spm_units"]["spm_unit"]["receives_housing_assistance"] = {YEAR: receives} + situation["spm_units"]["spm_unit"]["pre_subsidy_electricity_expense"] = { + YEAR: 1_800 + } + # Pin the categorical route off so income is what decides. + situation["households"]["household"]["ca_care_categorically_eligible"] = { + YEAR: False + } + return situation + + +def test_cpuc_ignores_an_imputed_subsidy_a_household_does_not_report_receiving(): + """Only reported receipt counts, so imputed take-up cannot decide CARE. + + `housing_assistance` is eligibility times take-up: is_eligible is + ``receives_housing_assistance | (is_renter & is_income_eligible)`` and + `takes_up_housing_assistance_if_eligible` defaults to true, so this + household - which sends ``receives_housing_assistance: false`` - is still + modelled as holding a voucher. Counting that modelled amount would put a + subsidy nobody receives into CARE income and take the discount away. + """ + simulation = Simulation( + situation=income_eligible_la_renter(receives=False), + spm={"geography_kind": "national"}, + ) + assert not simulation.calculate("receives_housing_assistance", YEAR)[0] + # The imputation is live: eligible, taking up, and a positive amount. + assert simulation.calculate("is_eligible_for_housing_assistance", YEAR)[0] + assert simulation.calculate("takes_up_housing_assistance_if_eligible", YEAR)[0] + imputed = simulation.calculate("housing_assistance", YEAR)[0] + assert imputed > 0 + # None of it is counted. + assert simulation.calculate("reported_housing_assistance", YEAR)[0] == 0 + assert simulation.calculate("ca_cpuc_countable_income", YEAR)[0] == pytest.approx( + 35_000 + ) + # Earnings alone decide, and $35,000 is inside the 2024 one-person CARE + # limit of 2 x $20,440; the imputed $13,500 would have carried it outside. + assert imputed == pytest.approx(13_500) + # The capped entry this replaced also contributed nothing here, so the + # defect this test guards is new: it arrives with the uncapped amount. + assert simulation.calculate("spm_unit_capped_housing_subsidy", YEAR)[0] == 0 + assert bool(simulation.calculate("ca_care_income_eligible", YEAR)[0]) + assert bool(simulation.calculate("ca_care_eligible", YEAR)[0]) + assert simulation.calculate("ca_care", YEAR)[0] == pytest.approx(0.325 * 1_800) + + +@pytest.mark.parametrize("spm", [None, {"geography_kind": "national"}]) +@pytest.mark.parametrize("county", ["06037", None]) +def test_cpuc_countable_income_for_a_non_recipient_needs_no_geography(spm, county): + """The unreported case computes under every selection, county input or not. + + Without a county the HUD income limits do not resolve, so the household is + not even modelled as voucher-eligible; with one it is. Either way nothing + consults SPM geography, so no configuration is required and no measurement + is received. + """ + simulation = Simulation( + situation=income_eligible_la_renter(receives=False, county=county), + spm=spm, + ) + assert simulation.calculate("reported_housing_assistance", YEAR)[0] == 0 + assert simulation.calculate("ca_cpuc_countable_income", YEAR)[0] == pytest.approx( + 35_000 + ) + assert bool(simulation.calculate("ca_care_eligible", YEAR)[0]) + assert simulation.spm_provenance()["years"] == {} + + +def test_cpuc_counts_the_housing_assistance_a_household_reports_receiving(): + """Reported receipt counts in full, and takes this household out of CARE. + + The same household as above, now reporting the voucher: the assistance is + counted at the amount the model computes for it, with no netting and no + cap, and $35,000 of earnings plus that subsidy sits outside the CARE limit. + """ + simulation = Simulation( + situation=income_eligible_la_renter(receives=True), + spm={"geography_kind": "national"}, + ) + assistance = simulation.calculate("housing_assistance", YEAR)[0] + assert assistance > 0 + assert simulation.calculate("reported_housing_assistance", YEAR)[0] == assistance + assert simulation.calculate("ca_cpuc_countable_income", YEAR)[0] == pytest.approx( + 35_000 + assistance + ) + assert not bool(simulation.calculate("ca_care_income_eligible", YEAR)[0]) + assert not bool(simulation.calculate("ca_care_eligible", YEAR)[0]) + assert simulation.calculate("ca_care", YEAR)[0] == 0 diff --git a/policyengine_us/variables/gov/hud/housing_assistance.py b/policyengine_us/variables/gov/hud/housing_assistance.py index 4060700f6cb..a6644f626eb 100644 --- a/policyengine_us/variables/gov/hud/housing_assistance.py +++ b/policyengine_us/variables/gov/hud/housing_assistance.py @@ -6,7 +6,16 @@ class housing_assistance(Variable): entity = SPMUnit label = "Housing assistance" unit = USD - documentation = "Housing assistance" + documentation = """ + Housing assistance the model computes for this unit: HUD's housing + assistance payment where the unit is eligible and takes up. + + Eligibility is `receives_housing_assistance | (is_renter & + is_income_eligible)` and take-up defaults to true, so this is a modelled + amount rather than a report of receipt. An income definition that counts + only subsidies a household reports receiving wants + `reported_housing_assistance`. + """ definition_period = YEAR defined_for = "is_eligible_for_housing_assistance" diff --git a/policyengine_us/variables/gov/hud/reported_housing_assistance.py b/policyengine_us/variables/gov/hud/reported_housing_assistance.py new file mode 100644 index 00000000000..099d878c787 --- /dev/null +++ b/policyengine_us/variables/gov/hud/reported_housing_assistance.py @@ -0,0 +1,40 @@ +from policyengine_us.model_api import * + + +class reported_housing_assistance(Variable): + value_type = float + entity = SPMUnit + label = "Housing assistance received" + unit = USD + definition_period = YEAR + documentation = """ + Housing assistance for a unit that reports receiving it, and zero otherwise. + + PG&E Form 01-9077, the CARE/FERA application, counts "housing and military + subsidies" among the revenues making up total gross annual household + income. The form asks an applicant what the household receives, so the + subsidy it counts is one the household actually holds. + + `housing_assistance` is not that: `is_eligible_for_housing_assistance` is + `receives_housing_assistance | (is_renter & is_income_eligible)` and + `takes_up_housing_assistance_if_eligible` defaults to true, so any + income-eligible renter who supplies rent is modelled as holding a voucher. + Gating on `receives_housing_assistance` - an input, defaulting to false, + and already this model's report of Section 8 receipt for FCC Lifeline, + MA LIHEAP, CT TFA, RI Works and Orange County general relief - counts the + reported receipt and leaves imputed take-up out of the income definitions + that ask for it. + + This is the model's amount, gated on a reported flag, not an amount the + applicant reports: a caller who sends a `housing_assistance` value without + also sending `receives_housing_assistance: true` is not reporting receipt, + and nothing is counted here. Everything else follows `housing_assistance`, + including a reform abolishing HUD assistance and a unit that reports + receipt while `takes_up_housing_assistance_if_eligible` is false - the + model holds no subsidy for it, so neither does this. + """ + reference = "https://www.pge.com/assets/pge/localized/en/docs/account/billing-and-assistance/care-fera-application.pdf" + + def formula(spm_unit, period, parameters): + receives = spm_unit("receives_housing_assistance", period) + return where(receives, spm_unit("housing_assistance", period), 0)