diff --git a/changelog.d/606.fixed.md b/changelog.d/606.fixed.md new file mode 100644 index 00000000..993d683e --- /dev/null +++ b/changelog.d/606.fixed.md @@ -0,0 +1 @@ +A UK `Simulation.run()` now stamps its output with the policyengine-core and policyengine-uk versions that calculated it, an output format revision, and the stamp of the year file its input data was cut into, if any. A UK dataset read from a year file, or returned by `create_datasets`, keeps the file's stamp in `metadata`. `Simulation.load()` refuses a saved UK output, and `Simulation.ensure()` runs it again, unless that stamp is the one a run of the simulation's dataset would record now: when other package versions are installed, or the simulation's dataset is read from another year file or another cut of it. Once `ensure_datasets` has cut a UK year file again, because the bundle re-pinned its dataset, its local source changed or a package was upgraded, a simulation over the new cut no longer reuses outputs calculated from the old one. The error names the years, sources or versions that differ. A UK output is now written beside its path and moved into place with its stamp, so a save that fails leaves the previous output whole, and two processes saving to one path cannot leave one run's tables under another's stamp. Loading a UK year file or output checks that the stamp and the tables were read from one unchanged file, and reads again if another process replaced it meanwhile. `PolicyEngineUKDataset.load()` now raises `ValueError` for a dataset with no filepath. UK outputs saved before the stamp existed are calculated again once. diff --git a/docs/microsim.md b/docs/microsim.md index ac5a0500..3a91801a 100644 --- a/docs/microsim.md +++ b/docs/microsim.md @@ -351,11 +351,123 @@ are moved into place whole. Three more apply: - a multi-year source is read as stored, without the projection. A year file opened directly, as `PolicyEngineUKDataset(filepath=...)`, is not -checked, and saving it again with `save()` drops its stamp. A saved UK -simulation output records the data year its run anchored on; -`Simulation.load()` refuses an output saved without one, and -`Simulation.ensure()` runs it again. A saved UK output does not record the -stamp of the year file it was calculated from. +checked, and saving it again with `save()` drops its stamp. A dataset read +from a year file, or returned by `create_datasets`, keeps the file's stamp as +`metadata["year_file_stamp"]`, so an output calculated from it records the +stamp (see below). + +Loading reads the stamp and the tables through separate opens of the file, +and another process can move another cut onto the path in between. So +loading holds the file open while it reads, and afterwards checks that the +path still names that file, with the size and modification time it had +before. If not, it reads again, and raises `RuntimeError` after three +attempts. The stamp a dataset keeps is therefore the stamp of the cut its +tables came from. + +### Saved UK outputs + +A UK `Simulation.run()` records a stamp in its output's +`metadata["output_stamp"]`: + +```json +{ + "packages": {"policyengine-core": "3.33.0", "policyengine-uk": "2.126.12"}, + "revision": 1, + "year_file": { + "packages": {"policyengine-core": "3.33.0", "policyengine-uk": "2.126.12"}, + "revision": 1, + "source": { + "sha256": "5ba399ea5ab9a6c7d4e9aaf6c623080beb34d9ad5f330f6244dbad72b8d960b6", + "uri": "hf://policyengine/policyengine-uk-data-private/enhanced_frs_2024_25.h5@1.58.0" + }, + "year": 2026 + } +} +``` + +- `packages`: the policyengine-core and policyengine-uk versions installed + when the run started, which calculated it; +- `revision`: the UK output format revision; +- `year_file`: the stamp the input dataset keeps of the year file it was cut + into, or `null` when it keeps none (a dataset built in memory, for + example). It names the tables the run read, and the versions that cut + them. Those need not be the versions that calculated the run: a year file + opened directly after an upgrade is run with the new ones. + +`save()` writes the tables, the data year and then the stamp to a temporary +file beside the output's path (`.{id}.…partial`), and moves that onto the +path in one step, as `create_datasets` writes a year file. The tables and the +stamp at the path therefore always come from one save, even when two +processes save an output to the same path, and a save that fails leaves the +output already there as it was. The stamp is stored as UTF-8 JSON in +`uk.datasets.OUTPUT_STAMP_H5_DATASET` (`policyengine_uk_output_stamp`), and +`load()` restores it. UK year files and US outputs store their stamps under +other names, so none passes as another. + +`Simulation.load()` refuses a saved UK output unless its stamp is the one a +run of the simulation's dataset would record now, +`uk.datasets.output_stamp(uk.datasets.input_year_file_stamp(simulation.dataset))`. +`Simulation.ensure()` then runs the simulation again and saves the new +output. `uk.datasets.output_stamp_problem(path, year_file)` returns the +reason, or `None` for a current output. It reads only the output's stamp and +whether the output has a data year, never its tables or the year file. A +saved UK output is out of date when: + +- it has no stamp: it was saved before the stamp existed, and is calculated + again once; +- it has no data year. It was saved before runs anchored on the observed + data year, or not saved completely, and may have taken a projected year as + observed data and uprated the State Pension by CPI; +- it was calculated with other policyengine-core or policyengine-uk versions + than are installed, whatever its input was; +- it was calculated from another year file than the simulation's dataset was + read from, or from another cut of it: the two year-file stamps differ. + `ensure_datasets` cuts a year file again, with another stamp, when the + bundle re-pins its dataset, a local source is edited or a package is + upgraded, so an output calculated from the old cut is not reused for a + simulation over the new one. Nor is an output reused for another year's + file in the same folder, or between a dataset that keeps a stamp and one + that keeps none; +- the stamp its dataset kept of the year file is unreadable or malformed. + Such a stamp names no cut, so the output is calculated again every time; +- it has another output format revision, or its stamp cannot be read. + +The reason names the years, sources and package versions that differ. A +missing output file raises `FileNotFoundError`, as before, so `ensure()` runs +the simulation. An output file that cannot be opened for reasons unrelated +to its contents (no permission, a directory, a lock held by another process) +raises that error rather than counting as out of date. + +Loading checks the stamp and reads the tables through separate opens of the +output, so it holds the file open and checks afterwards that the path still +names it unchanged, as loading a year file does. If another process moved +another output onto the path meanwhile, the check is made again on that +output. + +The check has these limits: + +- it compares the output with the simulation's dataset, not with the source + requested now. A year file that is out of date but not yet cut again, and + opened directly, is run and its outputs are reused: they are what a run of + that file gives. `load_datasets` refuses such a year file, and + `ensure_datasets` cuts it again; +- the stamp records policyengine-core and policyengine-uk, not + policyengine.py's own version. A change to how policyengine.py runs a UK + simulation is marked by raising the output format revision; +- the versions are those installed when the run starts, read from the + installed distributions' metadata. As for year files, restart the process + after upgrading a package; +- a dataset built in memory, and a year file cut before the stamp existed + and opened directly, keep no stamp. Outputs calculated from them record + `null`, so two such datasets saved under one simulation id in one folder + are not told apart; +- editing a dataset's tables in memory does not change the stamp it keeps; +- within one process, `ensure()` returns the output it already holds in + memory for a simulation id without checking its stamp; +- as for year files, moving an output into place replaces the file at its + path rather than rewriting it, so the save needs write access to the + folder, and a process killed mid-save can leave a temporary file behind, + which is never read and can be deleted. ## Simulations diff --git a/src/policyengine/provenance/year_file_stamp.py b/src/policyengine/provenance/year_file_stamp.py index 61d3af3c..99598f57 100644 --- a/src/policyengine/provenance/year_file_stamp.py +++ b/src/policyengine/provenance/year_file_stamp.py @@ -27,7 +27,12 @@ - each requested dataset's year-file stem and source identity, refusing two sources that would share year files (:func:`year_file_sources`); - writing a year file beside its path and moving it into place only once it - is stamped (:func:`temporary_year_file`, :func:`publish_year_file`). + is stamped (:func:`temporary_year_file`, :func:`publish_year_file`); +- reading a file's stamp and its tables from one file, though each is read + through its own open of the path (:func:`read_one_file`). + +A saved UK simulation output is written, stamped and read the same way (see +``tax_benefit_models.uk.datasets.OUTPUT_STAMP_H5_DATASET``). """ from __future__ import annotations @@ -35,10 +40,10 @@ import json import os import secrets -from collections.abc import Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping from importlib import metadata as importlib_metadata from pathlib import Path -from typing import Any, Optional +from typing import Any, Optional, TypeVar from .dataset_materialization import ( describe_dataset_source_identity, @@ -180,6 +185,57 @@ def publish_year_file( os.replace(temporary, path) +_Read = TypeVar("_Read") + + +def read_one_file( + path: str | Path, read: Callable[[str], _Read], attempts: int = 3 +) -> _Read: + """Return ``read(path)``, having checked that it read one unchanged file. + + ``read`` may open ``path`` several times, for example once for a stamp + and once for the tables. Another process can move another file onto the + path in between (:func:`publish_year_file` does), so what one open reads + need not belong with what the next reads. + + The file at ``path`` is therefore held open while ``read`` runs. After it + returns, ``path`` must still name that file (the same device and inode, + which the file system does not give to another file while this one is + held open), and the file must have the size and modification time it had + before. If not, ``read`` runs again, up to ``attempts`` times in all, and + then ``RuntimeError`` is raised. A file moved onto the path is another + file, even when its contents are those of an earlier one. + + Two changes are not noticed: a file rewritten in place to the same size + within one tick of the file system's modification time, and a file + replaced and then put back by moving another hard link to it onto the + path. Neither ``create_datasets`` nor ``save_output`` writes that way: + each moves a newly written file onto the path. + + Whatever ``read`` raises is raised, as is whatever opening ``path`` + raises (``FileNotFoundError`` for a missing file). + """ + path = str(path) + for _ in range(attempts): + with open(path, "rb") as held: + before = os.fstat(held.fileno()) + result = read(path) + after = os.fstat(held.fileno()) + try: + named = os.stat(path) + except FileNotFoundError: + continue + if os.path.samestat(named, before) and ( + after.st_size, + after.st_mtime_ns, + ) == (before.st_size, before.st_mtime_ns): + return result + raise RuntimeError( + f"{path} was replaced or modified while it was read, {attempts} times; " + "read it again once it is no longer being written" + ) + + def is_version_map(value: Any) -> bool: """Whether ``value`` maps package names to versions, ``None`` for absent.""" return isinstance(value, dict) and all( @@ -213,19 +269,40 @@ def source_and_package_differences( "requested source is " f"{describe_dataset_source_identity(current['source'])}" ) - stamped_packages, packages = stamp["packages"], current["packages"] - for name in sorted(stamped_packages.keys() | packages.keys()): + reasons.extend(package_differences(stamp["packages"], current["packages"])) + return reasons + + +def package_differences( + stamped: Mapping[str, Optional[str]], + packages: Mapping[str, Optional[str]], + done: str = "cut", +) -> list[str]: + """Return the clauses naming the package versions a stamp differs in. + + ``stamped`` is the version map a stamp records and ``packages`` the one a + stamp made now would (each as :func:`is_version_map` accepts). ``done`` + says what the stamped file had done to it with those versions: a year + file was "cut" with them, an output "calculated". One clause, starting + "it", names each package whose version differs, in name order. A package + ``stamped`` records that ``packages`` does not is named too, and a + package recorded in both with one version is not. + + Returns no clause exactly when the two maps are equal. + """ + reasons = [] + for name in sorted(stamped.keys() | packages.keys()): if name not in packages: reasons.append( - f"it records the {name} version it was cut with, which this " + f"it records the {name} version it was {done} with, which this " "version of policyengine.py does not check" ) continue - if name in stamped_packages and stamped_packages[name] == packages[name]: + if name in stamped and stamped[name] == packages[name]: continue - cut_with = ( - f"{name} {stamped_packages[name]}" - if stamped_packages.get(name) is not None + done_with = ( + f"{name} {stamped[name]}" + if stamped.get(name) is not None else f"no recorded {name} version" ) installed = ( @@ -233,7 +310,7 @@ def source_and_package_differences( if packages[name] is not None else f"{name} is not installed" ) - reasons.append(f"it was cut with {cut_with}, and {installed}") + reasons.append(f"it was {done} with {done_with}, and {installed}") return reasons diff --git a/src/policyengine/tax_benefit_models/common/model_version.py b/src/policyengine/tax_benefit_models/common/model_version.py index e10d3330..6701076f 100644 --- a/src/policyengine/tax_benefit_models/common/model_version.py +++ b/src/policyengine/tax_benefit_models/common/model_version.py @@ -388,6 +388,28 @@ def save(self, simulation: Simulation) -> None: "inputs were mapped when it was calculated; run again " "before saving" ) + if self.country_code == "uk": + from policyengine.tax_benefit_models.uk.datasets import ( + OUTPUT_STAMP_KEY, + save_output, + ) + + # The stamp the run recorded, not one made now: the year file the + # output was calculated from, and the package versions it was + # calculated with. + output: Any = simulation.output_dataset + uk_stamp = (getattr(output, "metadata", None) or {}).get(OUTPUT_STAMP_KEY) + if uk_stamp is None: + raise ValueError( + "This UK output does not record what it was calculated " + "from and with; run again before saving" + ) + # The tables and the stamp reach the output's path in one step, + # so neither a save that stops part-way nor another process + # saving to the same path leaves one run's tables under another + # run's stamp. + save_output(output, uk_stamp) + return simulation.output_dataset.save() if serialized_spm is not None: # Store UTF-8 JSON in a dataset rather than an attribute: the @@ -443,29 +465,63 @@ def load(self, simulation: Simulation) -> None: "stored inputs (it has no record of them); run it again" ) + def hydrate(path: str, **kept: Any) -> Any: + return self._dataset_class( + id=simulation.id, + name=simulation.dataset.name, + description=simulation.dataset.description, + filepath=path, + year=simulation.dataset.year, + is_output_dataset=True, + **kept, + ) + if self.country_code == "uk": + from policyengine.provenance.year_file_stamp import read_one_file from policyengine.tax_benefit_models.uk.datasets import ( - _year_file_records_data_year, + OUTPUT_STAMP_KEY, + input_year_file_stamp, + output_stamp, + output_stamp_problem, ) - # UK outputs saved before runs anchored on the observed data year - # (PolicyEngine/policyengine.py#556) took a projected year as - # observed and uprated the State Pension by CPI, so they are not - # reused. ``Simulation.ensure()`` runs such a simulation again. - if not _year_file_records_data_year(Path(filepath)): - raise ValueError( - "Saved UK simulation predates anchoring on the observed " - "data year (it records none); run it again" + # The stamp a run of the simulation's dataset would record now: + # the installed package versions and the year-file stamp the + # dataset keeps. + expected = output_stamp(input_year_file_stamp(simulation.dataset)) + + def read(path: str) -> Any: + # Outputs calculated with other package versions, from + # another year file than the simulation's dataset is read + # from, or from another cut of it (cut again after the + # bundle re-pinned its dataset, its local source changed or + # a package was upgraded), hold other results. So do outputs + # saved before runs anchored on the observed data year + # (PolicyEngine/policyengine.py#556), which took a projected + # year as observed and uprated the State Pension by CPI. + # None is reused: ``Simulation.ensure()`` runs such a + # simulation again. A missing file raises + # ``FileNotFoundError`` here. + problem = output_stamp_problem( + path, expected["year_file"], expected["packages"] ) - - simulation.output_dataset = self._dataset_class( - id=simulation.id, - name=simulation.dataset.name, - description=simulation.dataset.description, - filepath=filepath, - year=simulation.dataset.year, - is_output_dataset=True, - ) + if problem is not None: + raise ValueError( + f"Saved UK simulation output {path} is out of date: " + f"{problem}. Run the simulation again " + "(Simulation.ensure() does so)." + ) + # The stored stamp equals ``expected``; kept so that saving + # the loaded output again writes it. + return hydrate(path, metadata={OUTPUT_STAMP_KEY: expected}) + + # The stamp is checked and the tables are read through separate + # opens of the file, and another process can move another output + # onto the path in between. ``read_one_file`` checks that both + # read one unchanged file, and reads again if not. + simulation.output_dataset = read_one_file(filepath, read) + else: + simulation.output_dataset = hydrate(filepath) if self.country_code == "us": from policyengine.core.spm import SPMSelection diff --git a/src/policyengine/tax_benefit_models/uk/datasets.py b/src/policyengine/tax_benefit_models/uk/datasets.py index 4efe38e3..6ed23bff 100644 --- a/src/policyengine/tax_benefit_models/uk/datasets.py +++ b/src/policyengine/tax_benefit_models/uk/datasets.py @@ -1,3 +1,4 @@ +import copy import logging from collections.abc import Mapping from pathlib import Path @@ -5,7 +6,7 @@ import pandas as pd from microdf import MicroDataFrame -from pydantic import ConfigDict +from pydantic import ConfigDict, Field from policyengine.core import Dataset, YearData from policyengine.provenance import year_file_stamp as _shared_stamp @@ -14,6 +15,7 @@ dataset_source_change, dataset_source_identity, dataset_source_snapshot, + describe_dataset_source_identity, is_dataset_source_identity, materialize_dataset, ) @@ -21,10 +23,13 @@ from policyengine.provenance.year_file_stamp import ( ABSENT, NOT_HDF5, + UNREADABLE, UNRELATED_TO_CONTENTS, is_version_map, + package_differences, publish_year_file, read_json_dataset, + read_one_file, read_stored_json, source_and_package_differences, temporary_year_file, @@ -58,8 +63,50 @@ #: #: Revision 1 is the first: UK year files cut before it have no stamp, so #: they are cut again once. +#: +#: A saved output records the stamp of the year file it was calculated from, +#: so once the year files are cut again under a new revision, outputs +#: calculated from the old ones are out of date too. YEAR_FILE_STAMP_REVISION = 1 +#: Key under which a UK dataset read from, or cut into, a year file keeps that +#: file's stamp in its ``metadata``, as the JSON value stored (see +#: :func:`year_file_stamp_metadata`). A run records it in its output's stamp. +YEAR_FILE_STAMP_KEY = "year_file_stamp" + +#: H5 dataset stamping a saved UK simulation output, as UTF-8 JSON +#: ``{"packages": {...}, "revision": int, "year_file": ...}``: the installed +#: versions of the packages that calculated it, its output format revision, +#: and the stamp of the year file it was calculated from (``null`` when the +#: input dataset kept none). See :func:`output_stamp`. +#: +#: A UK run records the stamp in its output's ``metadata`` (under +#: :data:`OUTPUT_STAMP_KEY`) when it is calculated. Saving the output writes +#: its tables, its data year and then the stamp to a temporary file, and +#: moves that onto the output's path in one step (see :func:`save_output`), so +#: the tables and the stamp at the path always come from one run. +#: ``Simulation.load()`` refuses an output whose stamp is not the one a run of +#: the simulation's dataset would record now (see +#: :func:`output_stamp_problem`), and ``Simulation.ensure()`` then calculates +#: it again. +#: +#: Its name differs from :data:`YEAR_FILE_STAMP_H5_DATASET`'s and from the US +#: output stamp's (``us.legacy_inputs.OUTPUT_STAMP_H5_DATASET``), so an output +#: never passes as a year file, nor as an output of the other country. +OUTPUT_STAMP_H5_DATASET = "policyengine_uk_output_stamp" + +#: Format revision of saved UK outputs, stored in their stamp. Raise it when a +#: change to how policyengine.py runs a UK simulation means that outputs +#: calculated before it must be calculated again, for a reason the rest of +#: the stamp does not capture. +#: +#: Revision 1 is the first: UK outputs saved before it have no stamp, so they +#: are calculated again once. +OUTPUT_STAMP_REVISION = 1 + +#: Key under which a UK run's output keeps its stamp in ``metadata``. +OUTPUT_STAMP_KEY = "output_stamp" + #: Distributions whose installed versions a UK year file's stamp records. #: ``create_datasets`` stores each year's tables as policyengine-uk projects #: the source dataset. policyengine-uk extends a single-year source to later @@ -70,6 +117,13 @@ #: generator, seeded by the year; numpy's version is not recorded.) YEAR_FILE_PACKAGES = ("policyengine-core", "policyengine-uk") +#: Distributions whose installed versions a UK output's stamp records: those +#: that calculate a run. They are the ones a year file's stamp records, but +#: an output records them itself: a run can be calculated with other versions +#: than cut the year file it reads (a year file opened directly after an +#: upgrade, for example). +OUTPUT_PACKAGES = YEAR_FILE_PACKAGES + #: Prefix of the HDF5 keys holding the data year's entity tables in a year #: file whose year comes after its data year. DATA_YEAR_TABLE_PREFIX = "data_year_" @@ -120,6 +174,15 @@ class PolicyEngineUKDataset(Dataset): instead of the triple lock. """ + metadata: dict[str, Any] = Field(default_factory=dict) + """Provenance kept with the tables. + + A dataset read from a year file, or returned by ``create_datasets``, + keeps the file's stamp under :data:`YEAR_FILE_STAMP_KEY`, and a run + records it in its output's stamp. A run's output keeps that stamp under + :data:`OUTPUT_STAMP_KEY`, and saving the output writes it. + """ + def model_post_init(self, __context): """Called after Pydantic initialization. @@ -185,7 +248,10 @@ def save(self) -> None: filepath = Path(self.filepath) if not filepath.parent.exists(): filepath.parent.mkdir(parents=True, exist_ok=True) + self._write(filepath) + def _write(self, filepath: Path) -> None: + """Write the tables and the data year to ``filepath``, replacing it.""" # Mode "w" truncates the file first, so saving a year file again # drops the stamp ``create_datasets`` wrote after it. with pd.HDFStore(filepath, mode="w") as store: @@ -197,21 +263,51 @@ def save(self) -> None: store.put(DATA_YEAR_KEY, pd.Series([int(self.data_year)])) def load(self) -> None: - """Load dataset from HDF5 file into this instance.""" - filepath = self.filepath - with pd.HDFStore(filepath, mode="r") as store: - keys = set(store.keys()) - self.data = _read_year_data(store) - self.data_year = ( - int(store[DATA_YEAR_KEY].iloc[0]) - if f"/{DATA_YEAR_KEY}" in keys - else None - ) - self.data_year_data = ( - _read_year_data(store, DATA_YEAR_TABLE_PREFIX) - if f"/{DATA_YEAR_TABLE_PREFIX}person" in keys - else None + """Load dataset from HDF5 file into this instance. + + The stamp of the year file the tables were cut into, if the file is + one, is kept in ``metadata`` under :data:`YEAR_FILE_STAMP_KEY` (see + :func:`year_file_stamp_metadata`), and dropped when the file has none. + A run records it in its output's stamp, so an output calculated from a + year file that is cut again is not reused. + + The stamp and the tables are each read through their own open of the + file, and ``create_datasets`` in another process can move another cut + onto the path in between. ``year_file_stamp.read_one_file`` checks + that both were read from one unchanged file, and reads again if not + (``RuntimeError`` after three attempts), so the stamp a dataset keeps + is the stamp of the cut its tables came from. + """ + if not self.filepath: + raise ValueError( + "Cannot load a PolicyEngineUKDataset with no filepath. This " + "is an in-memory dataset; set `.filepath` to a file to load." ) + + def read(filepath: str): + stamp = year_file_stamp_metadata(filepath) + with pd.HDFStore(filepath, mode="r") as store: + keys = set(store.keys()) + data = _read_year_data(store) + data_year = ( + int(store[DATA_YEAR_KEY].iloc[0]) + if f"/{DATA_YEAR_KEY}" in keys + else None + ) + data_year_data = ( + _read_year_data(store, DATA_YEAR_TABLE_PREFIX) + if f"/{DATA_YEAR_TABLE_PREFIX}person" in keys + else None + ) + return stamp, data, data_year, data_year_data + + stamp, self.data, self.data_year, self.data_year_data = read_one_file( + self.filepath, read + ) + if stamp is None: + self.metadata.pop(YEAR_FILE_STAMP_KEY, None) + else: + self.metadata[YEAR_FILE_STAMP_KEY] = stamp self._check_data_year() def __repr__(self) -> str: @@ -452,16 +548,7 @@ def year_file_stamp_problem( "cut from and with, so it was cut by an older policyengine.py or " "not written completely" ) - # ``type(...) is int`` rather than ``isinstance``: JSON ``true`` loads as - # ``True``, which equals 1. - if not ( - isinstance(stamp, dict) - and set(stamp) == {"packages", "revision", "source", "year"} - and type(revision) is int - and type(stamp["year"]) is int - and is_version_map(stamp["packages"]) - and is_dataset_source_identity(stamp["source"]) - ): + if not _is_year_file_stamp(stamp): return "its stamp of what it was cut from and with is malformed" if stamp == current: return None @@ -491,6 +578,364 @@ def _records_data_year(file: Any) -> bool: return False +def _is_year_file_stamp(stamp: Any) -> bool: + """Whether ``stamp`` has the shape :func:`year_file_stamp` gives.""" + # ``type(...) is int`` rather than ``isinstance``: JSON ``true`` loads as + # ``True``, which equals 1. + return ( + isinstance(stamp, dict) + and set(stamp) == {"packages", "revision", "source", "year"} + and type(stamp["revision"]) is int + and type(stamp["year"]) is int + and is_version_map(stamp["packages"]) + and is_dataset_source_identity(stamp["source"]) + ) + + +#: What a dataset keeps under :data:`YEAR_FILE_STAMP_KEY` when its year file's +#: stamp cannot be read, or is ``null`` (see :func:`year_file_stamp_metadata`). +UNREADABLE_YEAR_FILE_STAMP = "unreadable" + + +def year_file_stamp_metadata(path: str | Path) -> Any | None: + """Return the year-file stamp a UK dataset read from ``path`` keeps. + + A dataset keeps it in ``metadata`` under :data:`YEAR_FILE_STAMP_KEY`, and + a run records it in its output's stamp (see :func:`output_stamp`). + + A missing file raises ``FileNotFoundError``, and an error that says + nothing about the file's contents (no permission, a directory, a lock) is + raised too, as by :func:`year_file_stamp_problem`. + + Returns: + ``None`` when the file stores no stamp or is not an H5 file; the + stored JSON value when it is readable and not ``null``; otherwise + :data:`UNREADABLE_YEAR_FILE_STAMP`. No release stamps a year file + with a string, so an output calculated from a file whose stamp is + unreadable is never current. + """ + import h5py + + try: + with h5py.File(path, "r") as file: + stamp = read_json_dataset(file, YEAR_FILE_STAMP_H5_DATASET) + except (FileNotFoundError, *UNRELATED_TO_CONTENTS): + raise + except OSError: + return None + if stamp is ABSENT: + return None + if stamp is UNREADABLE or stamp is None: + return UNREADABLE_YEAR_FILE_STAMP + return stamp + + +def input_year_file_stamp(dataset: Any) -> Any | None: + """Return the year-file stamp a run of ``dataset`` records in its output. + + That is the stamp ``dataset`` keeps under :data:`YEAR_FILE_STAMP_KEY`, or + ``None`` when it keeps none (a dataset built in memory, for example). A + run loads a dataset whose tables are not loaded first, so for one with a + file it is the stamp loading the file would keep. + """ + filepath = getattr(dataset, "filepath", None) + if getattr(dataset, "data", None) is None and filepath: + return year_file_stamp_metadata(filepath) + metadata = getattr(dataset, "metadata", None) or {} + return copy.deepcopy(metadata.get(YEAR_FILE_STAMP_KEY)) + + +def installed_output_packages() -> dict[str, Optional[str]]: + """Return the installed version of each of :data:`OUTPUT_PACKAGES`. + + ``None`` stands for a package that is not installed. The versions come + from the installed distributions' metadata, so no package is imported. + """ + return _shared_stamp.installed_package_versions(OUTPUT_PACKAGES) + + +def output_stamp( + year_file: Any = None, + packages: Optional[Mapping[str, Optional[str]]] = None, +) -> dict[str, Any]: + """Return the stamp for a UK output calculated now. + + A run records it in its output's ``metadata`` under + :data:`OUTPUT_STAMP_KEY`, and saving the output writes it as UTF-8 JSON in + :data:`OUTPUT_STAMP_H5_DATASET`:: + + {"packages": {"policyengine-core": str | null, + "policyengine-uk": str | null}, + "revision": int, + "year_file": {...} | null} + + ``packages`` defaults to :func:`installed_output_packages`, read when this + is called: the versions that calculate the run. ``revision`` is + :data:`OUTPUT_STAMP_REVISION`. ``year_file`` is the year-file stamp the + input dataset keeps (see :func:`input_year_file_stamp`), or ``None`` when + it keeps none. It names the tables the run reads, and records the + versions that cut them, which need not be the ones that calculate the + run. + """ + if packages is None: + packages = installed_output_packages() + return { + "packages": dict(sorted(packages.items())), + "revision": OUTPUT_STAMP_REVISION, + "year_file": copy.deepcopy(year_file), + } + + +def read_output_stamp(path: str | Path) -> Any | None: + """Return the JSON value of a saved UK output's stamp, well formed or not. + + ``None`` means the file stores no readable stamp (see + :data:`OUTPUT_STAMP_H5_DATASET`) or is not an H5 file. + """ + return read_stored_json(path, OUTPUT_STAMP_H5_DATASET) + + +def write_output_stamp(path: str | Path, stamp: Mapping[str, Any]) -> None: + """Stamp a saved UK output (see :func:`output_stamp`), replacing any stamp.""" + write_json_dataset(path, OUTPUT_STAMP_H5_DATASET, dict(stamp)) + + +def save_output(dataset: PolicyEngineUKDataset, stamp: Mapping[str, Any]) -> None: + """Save a run's output and its stamp to ``dataset.filepath`` in one step. + + The tables and the data year are written to a temporary file beside the + path, which is then stamped and moved onto the path, as + ``create_datasets`` writes a year file (see + ``year_file_stamp.temporary_year_file`` and ``publish_year_file``). The + tables and the stamp at the path therefore always come from one save, + even when two processes save an output to the same path at once, and a + save that fails leaves the output already there as it was. + """ + if not dataset.filepath: + raise ValueError( + "Cannot save a PolicyEngineUKDataset with no filepath. This " + "is an in-memory dataset (e.g. a region-scoped copy); set " + "`.filepath` to a destination before calling save()." + ) + temporary = temporary_year_file(dataset.filepath) + try: + dataset._write(temporary) + publish_year_file(temporary, dataset.filepath, OUTPUT_STAMP_H5_DATASET, stamp) + finally: + temporary.unlink(missing_ok=True) + + +#: What a UK run without a data year would have done wrong. +_WITHOUT_AN_OUTPUT_DATA_YEAR = ( + "may have taken a projected year as observed data and uprated the State " + "Pension by CPI rather than the triple lock" +) + +#: Why an unstamped UK output without a recorded data year is out of date. +OUTPUT_WITHOUT_DATA_YEAR = ( + "it records none of the observed data year its run anchored on, so it was " + "saved before runs anchored on the observed data year " + "(PolicyEngine/policyengine.py#556), or not saved completely, and " + f"{_WITHOUT_AN_OUTPUT_DATA_YEAR}" +) + +#: Why a stamped UK output without a recorded data year is out of date. +STAMPED_OUTPUT_WITHOUT_DATA_YEAR = ( + "it is stamped but records no data year, which every output a UK run " + f"saves records, so it {_WITHOUT_AN_OUTPUT_DATA_YEAR}" +) + +#: Why a UK output with a data year but no stamp is out of date. +UNSTAMPED_OUTPUT = ( + "it has no stamp of what it was calculated from and with, so it was saved " + "by an older policyengine.py" +) + +#: Why a UK output whose stamp is unreadable or of another shape is out of +#: date. +MALFORMED_OUTPUT_STAMP = ( + "its stamp of what it was calculated from and with is malformed" +) + +#: Why a UK output calculated from a year file whose stamp is unreadable or +#: of another shape is out of date: such a stamp names no cut. +MALFORMED_OUTPUT_YEAR_FILE = ( + "it was calculated from a year file whose stamp is unreadable or malformed" +) + + +def output_stamp_problem( + path: str | Path, + year_file: Any, + packages: Optional[Mapping[str, Optional[str]]] = None, +) -> Optional[str]: + """Return why a saved UK simulation output is out of date, or ``None``. + + ``year_file`` is the year-file stamp a run of the simulation's dataset + would record now (see :func:`input_year_file_stamp`), and ``packages`` + defaults to :func:`installed_output_packages`. A saved output is current + only when all of these hold: + + - it records its data year (``DATA_YEAR_KEY``), as every output a UK run + saves does; + - its stamp equals ``output_stamp(year_file, packages)``, the one a run of + the simulation's dataset would record now: this release's output format + revision, the installed package versions, and the year-file stamp the + dataset keeps. An output calculated with other versions holds other + results, and so does one calculated from another year file, or from + another cut of the same one. ``create_datasets`` cuts a year file again + with another stamp when the bundle re-pins its dataset, its local + source changes or a package is upgraded, so an output is not reused + once the simulation's dataset is read from such a cut; + - ``year_file`` is ``None`` or a well-formed year-file stamp. One that + is unreadable, or of another shape, names no cut, so an output + calculated from such a year file is never current. + + Whether the year file is current for the source requested now is not + checked here: ``load_datasets`` refuses, and ``ensure_datasets`` cuts + again, a year file that is not (see :func:`year_file_stamp_problem`). + + Only the output's stamp and the presence of its data year are read, never + its tables or the year file. A file that cannot be read as HDF5 is out of + date. A missing file raises ``FileNotFoundError``, so + ``Simulation.ensure()`` calculates the output without reporting an + unexpected error. An error that says nothing about the file's contents + (no permission, a directory, a file another process has locked) is raised + too, since the output could not be saved again either. + + Returns: + ``None`` for a current output; otherwise a clause, starting "it" or + "its", that says why the output is out of date. It names the years, + format revisions, sources and package versions that differ between + the year file the output was calculated from and the simulation's + dataset, and the package versions the output was calculated with + that are not the ones installed. + """ + import h5py + + current = output_stamp(year_file, packages) + try: + with h5py.File(path, "r") as file: + stamp = read_json_dataset(file, OUTPUT_STAMP_H5_DATASET) + records_data_year = _records_data_year(file) + except (FileNotFoundError, *UNRELATED_TO_CONTENTS): + raise + except OSError: + return NOT_HDF5 + + # The revision comes first: a later revision may shape its stamp, or + # record the data year, differently. + revision = stamp.get("revision") if isinstance(stamp, dict) else None + if type(revision) is int and revision != OUTPUT_STAMP_REVISION: + return ( + f"it was saved with output format revision {revision}, and this " + f"version of policyengine.py reads only revision " + f"{OUTPUT_STAMP_REVISION}" + ) + if not records_data_year: + return ( + OUTPUT_WITHOUT_DATA_YEAR + if stamp is ABSENT + else STAMPED_OUTPUT_WITHOUT_DATA_YEAR + ) + if stamp is ABSENT: + return UNSTAMPED_OUTPUT + # ``type(...) is int`` rather than ``isinstance``: JSON ``true`` loads as + # ``True``, which equals 1. + if not ( + isinstance(stamp, dict) + and set(stamp) == {"packages", "revision", "year_file"} + and type(revision) is int + and is_version_map(stamp["packages"]) + ): + return MALFORMED_OUTPUT_STAMP + reasons = _output_year_file_reasons(stamp["year_file"], current["year_file"]) + reasons.extend( + package_differences(stamp["packages"], current["packages"], "calculated") + ) + return "; ".join(reasons) or None + + +def _output_year_file_reasons(recorded: Any, current: Any) -> list[str]: + """Say why an output calculated from the year file stamped ``recorded`` is + out of date, for a simulation whose dataset keeps ``current``. + + Returns no reason exactly when ``recorded`` equals ``current`` and both + are ``None`` or well-formed year-file stamps. They must be well formed to + count as equal: Python equality takes JSON ``true`` for 1, and a + well-formed stamp holds only strings, integers and ``null``, which + compare as JSON does. + """ + if recorded != current: + return [_year_file_mismatch(recorded, current)] + if recorded is None: + return [] + if not (_is_year_file_stamp(recorded) and _is_year_file_stamp(current)): + return [MALFORMED_OUTPUT_YEAR_FILE] + return [] + + +def _year_file_mismatch(recorded: Any, current: Any) -> str: + """Say how the year file an output was calculated from (stamped + ``recorded``) differs from the simulation's dataset (``current``). + + ``recorded`` and ``current`` differ. Every year, format revision, source + and package version that differs is named when both are well formed. + """ + if recorded is None: + return ( + "it was calculated from data that kept no year-file stamp, and the " + "simulation's dataset keeps one" + ) + if current is None: + return ( + "it was calculated from a year file, and the simulation's dataset " + "keeps no year-file stamp" + ) + if not (_is_year_file_stamp(recorded) and _is_year_file_stamp(current)): + return ( + "it was calculated from another year file, or another cut of it, " + "than the simulation's dataset was read from" + ) + reasons = [] + if recorded["year"] != current["year"]: + reasons.append( + f"it was calculated from the year file for {recorded['year']}, and " + f"the simulation's dataset is the one for {current['year']}" + ) + if recorded["revision"] != current["revision"]: + reasons.append( + "it was calculated from a year file of year-file format revision " + f"{recorded['revision']}, and the simulation's dataset is of " + f"revision {current['revision']}" + ) + if recorded["source"] != current["source"]: + reasons.append( + "it was calculated from a year file cut from " + f"{describe_dataset_source_identity(recorded['source'])}, and the " + "simulation's dataset was cut from " + f"{describe_dataset_source_identity(current['source'])}" + ) + for name in sorted(recorded["packages"].keys() | current["packages"].keys()): + cut_with = _cut_with(recorded["packages"], name) + dataset_cut_with = _cut_with(current["packages"], name) + if cut_with != dataset_cut_with: + reasons.append( + f"it was calculated from a year file cut {cut_with}, and the " + f"simulation's dataset was cut {dataset_cut_with}" + ) + return "; ".join(reasons) + + +def _cut_with(packages: Mapping[str, Optional[str]], name: str) -> str: + """Describe the ``name`` version a year-file stamp's ``packages`` record.""" + if name not in packages: + return f"without a recorded {name} version" + if packages[name] is None: + return f"without {name} installed" + return f"with {name} {packages[name]}" + + def _year_file_sources( datasets: list[str], allow_unmanaged: bool = True ) -> list[tuple[str, dict[str, Any]]]: @@ -624,13 +1069,14 @@ def create_datasets( ) # Stamp last, then move each year file onto its path in one step. for temporary, filepath, year, dataset_key in cut: + stamp = year_file_stamp(identity, year, packages) publish_year_file( - temporary, - filepath, - YEAR_FILE_STAMP_H5_DATASET, - year_file_stamp(identity, year, packages), + temporary, filepath, YEAR_FILE_STAMP_H5_DATASET, stamp ) result[dataset_key].filepath = filepath + # Keep it, as loading the file would, so a run over the + # returned dataset records it in its output's stamp. + result[dataset_key].metadata[YEAR_FILE_STAMP_KEY] = copy.deepcopy(stamp) finally: for temporary, *_ in cut: temporary.unlink(missing_ok=True) diff --git a/src/policyengine/tax_benefit_models/uk/model.py b/src/policyengine/tax_benefit_models/uk/model.py index 4aafcc0a..b66fbe0a 100644 --- a/src/policyengine/tax_benefit_models/uk/model.py +++ b/src/policyengine/tax_benefit_models/uk/model.py @@ -17,7 +17,13 @@ output_dataset_filepath as _output_dataset_filepath, ) -from .datasets import PolicyEngineUKDataset, UKYearData +from .datasets import ( + OUTPUT_STAMP_KEY, + PolicyEngineUKDataset, + UKYearData, + input_year_file_stamp, + output_stamp, +) if TYPE_CHECKING: from policyengine.core.simulation import Simulation @@ -181,6 +187,13 @@ def run(self, simulation: "Simulation") -> "Simulation": # autosave removal in datasets.py). if dataset.data is None: dataset.load() + # The installed package versions, which calculate this run, and the + # stamp of the year file the input data was cut into, if it is one. + # Both are stamped into the output when it is saved, and ``load()`` + # refuses the output once other versions are installed or the + # simulation's dataset is read from another cut (see + # ``uk.datasets.OUTPUT_STAMP_H5_DATASET``). + stamp = output_stamp(input_year_file_stamp(dataset)) # Apply regional scoping if specified if simulation.scoping_strategy: @@ -291,6 +304,7 @@ def run(self, simulation: "Simulation") -> "Simulation": benunit=data["benunit"], household=data["household"], ), + metadata={OUTPUT_STAMP_KEY: stamp}, ) diff --git a/tests/test_uk_output_stamp.py b/tests/test_uk_output_stamp.py new file mode 100644 index 00000000..0badf93d --- /dev/null +++ b/tests/test_uk_output_stamp.py @@ -0,0 +1,1943 @@ +"""Unit and property tests for the stamp on saved UK simulation outputs. + +A UK run records a stamp in its output (``uk.datasets.OUTPUT_STAMP_H5_DATASET`` +once saved): the installed versions of the packages that calculated it, an +output format revision, and the stamp of the year file its input data was cut +into, or ``null``. ``Simulation.load()`` refuses an output, and +``Simulation.ensure()`` calculates it again, unless the stamp is the one a run +of the simulation's dataset would record now. These tests state the check's +invariants over generated stamps, data-year nodes, package versions and input +year-file stamps, without a country engine: + +- **Exactness.** An output is current exactly when it records its data year, + its stamp has this release's output format revision and the installed + package versions, its year-file stamp equals the simulation's dataset's, + and that is ``null`` or a well-formed year-file stamp. +- **Totality.** Whatever the stamp, the data year and the dataset's stamp + hold, the check returns a reason or ``None``, and raises only for errors + unrelated to the file's contents (a missing file, no permission, a + directory, a lock). +- **Round trip.** A stamp reads back as written, and writing again replaces it. +- **Named differences.** The reason names every year, format revision, + source and package version that differs between the year file the output + was calculated from and the simulation's dataset, and every package + version the output was calculated with that is not the one installed, and + nothing that agrees. +- **No partial output is current.** Every truncation of a current output is + out of date. +- **Distinct stamps.** An output never passes as a year file, nor a year file + as an output, nor a US output as a UK one. +- **One save per output.** The tables and the stamp at an output's path come + from one save: a save that fails leaves the output already there whole, + and two saves to one path never leave one's tables under the other's + stamp. +- **One file per load.** A dataset loaded from a year file keeps the stamp of + the cut its tables came from, even when the file is replaced, replaced and + put back, or modified while it is read. + +``test_uk_output_stamp_integration.py`` runs, saves, loads and ensures outputs +over a real policyengine-uk engine, on a tiny dataset. +""" + +from __future__ import annotations + +import copy +import functools +import json +import os +import shutil +import tempfile +from pathlib import Path + +import h5py +import numpy as np +import pandas as pd +import pytest +from hypothesis import event, given, settings +from hypothesis import strategies as st +from microdf import MicroDataFrame + +from policyengine.provenance import year_file_stamp as shared_stamp +from policyengine.provenance.dataset_materialization import ( + describe_dataset_source_identity, +) +from policyengine.provenance.year_file_stamp import ( + NOT_HDF5, + package_differences, + read_one_file, +) +from policyengine.tax_benefit_models.uk import datasets as uk_datasets +from policyengine.tax_benefit_models.uk.datasets import ( + DATA_YEAR_KEY, + MALFORMED_OUTPUT_STAMP, + MALFORMED_OUTPUT_YEAR_FILE, + OUTPUT_PACKAGES, + OUTPUT_STAMP_H5_DATASET, + OUTPUT_STAMP_KEY, + OUTPUT_STAMP_REVISION, + OUTPUT_WITHOUT_DATA_YEAR, + STAMPED_OUTPUT_WITHOUT_DATA_YEAR, + UNREADABLE_YEAR_FILE_STAMP, + UNSTAMPED_OUTPUT, + YEAR_FILE_STAMP_H5_DATASET, + YEAR_FILE_STAMP_KEY, + YEAR_FILE_STAMP_REVISION, + PolicyEngineUKDataset, + UKYearData, + input_year_file_stamp, + output_stamp, + output_stamp_problem, + read_output_stamp, + read_year_file_stamp, + save_output, + write_output_stamp, + year_file_stamp, + year_file_stamp_metadata, + year_file_stamp_problem, +) + +REVISION = OUTPUT_STAMP_REVISION +YEAR_FILE_REVISION = YEAR_FILE_STAMP_REVISION +DATA_YEAR = 2024 +YEAR = 2026 +SOURCE = "hf://fixture/repo/tiny.h5" +IDENTITY = {"uri": SOURCE} +#: The installed versions every test in this module sees (see +#: ``_installed_packages``), so no expectation depends on the environment. +PACKAGES = {"policyengine-core": "3.33.0", "policyengine-uk": "2.126.12"} +#: The real reader, kept before ``_installed_packages`` replaces it. +REAL_INSTALLED_OUTPUT_PACKAGES = uk_datasets.installed_output_packages +#: The US output stamp's H5 dataset (``us.legacy_inputs.OUTPUT_STAMP_H5_DATASET``, +#: PolicyEngine/policyengine.py#594), named here so this module does not +#: depend on that change. +US_OUTPUT_STAMP_H5_DATASET = "policyengine_output_stamp" + +#: Small alphabets of source identities in each of their three shapes (see +#: ``dataset_source_identity``), so generated stamps often agree. +URIS = st.sampled_from([SOURCE, f"{SOURCE}@r1", f"{SOURCE}@r2"]) +SOURCES = st.one_of( + st.builds(lambda uri: {"uri": uri}, URIS), + st.builds( + lambda uri, sha256: {"uri": uri, "sha256": sha256}, + URIS, + st.sampled_from(["a" * 64, "b" * 64]), + ), + st.builds( + lambda path, size, mtime_ns: {"path": path, "size": size, "mtime_ns": mtime_ns}, + st.sampled_from(["/data/tiny.h5", "/elsewhere/tiny.h5"]), + st.sampled_from([10, 11]), + st.sampled_from([0, 1_760_000_000_123_456_789]), + ), +) +PACKAGE_NAMES = st.sampled_from([*OUTPUT_PACKAGES, "policyengine-us", "extra"]) +VERSIONS = st.one_of(st.none(), st.sampled_from(["1.0", "1.1"])) +PACKAGE_MAPS = st.one_of( + st.just(PACKAGES), st.dictionaries(PACKAGE_NAMES, VERSIONS, max_size=3) +) +YEARS = st.sampled_from([YEAR, YEAR + 1]) +YEAR_FILE_REVISIONS = st.sampled_from([YEAR_FILE_REVISION, YEAR_FILE_REVISION + 1]) +#: Well-formed year-file stamps, often this release's. +YEAR_FILE_STAMPS = st.builds( + lambda revision, source, packages, year: { + "packages": dict(packages), + "revision": revision, + "source": dict(source), + "year": year, + }, + YEAR_FILE_REVISIONS, + SOURCES, + PACKAGE_MAPS, + YEARS, +) +#: Raw node contents: ``_ABSENT`` for no node, a JSON value, or a node the +#: reader cannot parse as JSON. +_ABSENT = object() +JSON_VALUES = st.recursive( + st.none() | st.booleans() | st.integers() | st.floats() | st.text(max_size=8), + lambda children: ( + st.lists(children, max_size=3) + | st.dictionaries(st.text(max_size=8), children, max_size=3) + ), + max_leaves=6, +) +#: Year-file stamps a dataset may keep: none, well formed, unreadable, or any +#: JSON value, including well-formed keys holding values of the wrong type. +DATASET_STAMPS = st.one_of( + st.none(), + YEAR_FILE_STAMPS, + st.just(UNREADABLE_YEAR_FILE_STAMP), + JSON_VALUES, + st.builds( + lambda stamp, key, value: {**stamp, key: value}, + YEAR_FILE_STAMPS, + st.sampled_from(["packages", "revision", "source", "year", "extra"]), + st.one_of(st.booleans(), st.floats(), JSON_VALUES), + ), +) +#: Text stored as it is, not as JSON. HDF5 variable-length strings cannot +#: hold a NUL character, and a lone surrogate cannot be encoded as UTF-8. +RAW_TEXT = st.text( + alphabet=st.characters(blacklist_categories=("Cs",), blacklist_characters="\x00"), + max_size=12, +) +#: How an output's data year can be stored: ``True`` as ``save()`` writes it +#: (a group), ``False`` for none, or a node that is not a data year. +DATA_YEAR_NODES = ("dataset", "dangling link", "link cycle") + + +@pytest.fixture(autouse=True, scope="module") +def _installed_packages(): + """Pin the installed package versions a stamp records to ``PACKAGES``. + + Module-scoped, so Hypothesis tests can run under it. + """ + with pytest.MonkeyPatch.context() as patcher: + patcher.setattr( + uk_datasets, "installed_package_versions", lambda: dict(PACKAGES) + ) + patcher.setattr( + uk_datasets, "installed_output_packages", lambda: dict(PACKAGES) + ) + yield + + +def _frames(age=70) -> dict[str, pd.DataFrame]: + return { + "person": pd.DataFrame( + { + "person_id": [1, 2], + "person_benunit_id": [1, 1], + "person_household_id": [1, 1], + "person_weight": [1.0, 1.0], + "age": [age, 68], + } + ), + "benunit": pd.DataFrame({"benunit_id": [1], "benunit_weight": [1.0]}), + "household": pd.DataFrame( + {"household_id": [1], "household_weight": [1.0], "region": ["LONDON"]} + ), + } + + +def _tables(age=70) -> UKYearData: + return UKYearData( + **{ + entity: MicroDataFrame(frame, weights=f"{entity}_weight") + for entity, frame in _frames(age).items() + } + ) + + +def _output_dataset(path, data_year=DATA_YEAR, age=70) -> PolicyEngineUKDataset: + """A two-person UK output for ``YEAR``, as a run makes one.""" + return PolicyEngineUKDataset( + name="output", + description="Uncertified two-person UK output fixture", + filepath=str(path), + year=YEAR, + is_output_dataset=True, + data=_tables(age), + data_year=data_year, + ) + + +def _year_dataset(path, age=70) -> PolicyEngineUKDataset: + """A two-person UK year file for ``YEAR`` projected from ``DATA_YEAR``.""" + return PolicyEngineUKDataset( + name="year-file", + description="Uncertified two-person UK year-file fixture", + filepath=str(path), + year=YEAR, + data=_tables(age), + data_year=DATA_YEAR, + data_year_data=_tables(age), + ) + + +def _write_node(file, name, content) -> None: + """Store ``content`` under ``name``: JSON, or a node kind to test.""" + if content is _ABSENT: + return + if isinstance(content, tuple): + kind, value = content + if kind == "text": + file.create_dataset(name, data=value, dtype=h5py.string_dtype("utf-8")) + elif kind == "bytes": + file.create_dataset(name, data=np.bytes_(value)) + elif kind == "number": + file.create_dataset(name, data=value) + elif kind == "array": + file.create_dataset(name, data=value, dtype=h5py.string_dtype("utf-8")) + elif kind == "group": + file.create_group(name) + else: + raise AssertionError(kind) + return + file.create_dataset( + name, data=json.dumps(content), dtype=h5py.string_dtype("utf-8") + ) + + +def _link_cycle(file, name) -> None: + """Point ``name`` into a cycle of soft links.""" + file[name] = h5py.SoftLink(f"/{name}_loop") + file[f"{name}_loop"] = h5py.SoftLink(f"/{name}") + + +def _write_data_year_node(file, kind) -> None: + if kind == "dataset": + file.create_dataset(DATA_YEAR_KEY, data=DATA_YEAR) + elif kind == "dangling link": + file[DATA_YEAR_KEY] = h5py.SoftLink("/nowhere") + elif kind == "link cycle": + _link_cycle(file, DATA_YEAR_KEY) + else: + raise AssertionError(kind) + + +def _write_output( + path, *, stamp=_ABSENT, data_year=True, year_file_stamp=_ABSENT, age=70 +): + """Write an output's tables and data year, then its stamps, as given. + + ``stamp`` (the output stamp) and ``year_file_stamp`` are JSON values, + ``_ABSENT``, or a raw node (see :func:`_write_node`). ``data_year`` is + ``True`` for the data year as ``save()`` writes it, ``False`` for none, + or one of :data:`DATA_YEAR_NODES`. + """ + Path(path).parent.mkdir(parents=True, exist_ok=True) + _output_dataset(path, DATA_YEAR if data_year is True else None, age).save() + with h5py.File(path, "a") as file: + if data_year not in (True, False): + _write_data_year_node(file, data_year) + _write_node(file, YEAR_FILE_STAMP_H5_DATASET, year_file_stamp) + _write_node(file, OUTPUT_STAMP_H5_DATASET, stamp) + return Path(path) + + +def _write_year_file(path, stamp, age=70) -> Path: + """Write a year file's tables and data year, then its stamp.""" + Path(path).parent.mkdir(parents=True, exist_ok=True) + _year_dataset(path, age).save() + with h5py.File(path, "a") as file: + _write_node(file, YEAR_FILE_STAMP_H5_DATASET, stamp) + return Path(path) + + +def _write_stamps(path, *, stamp=_ABSENT, data_year=True, year_file_stamp=_ABSENT): + """Write what the checks read of an output or a year file, and no tables. + + The checks read the stamps and whether there is a data year, never the + tables (``test_the_decision_reads_only_the_stamp_and_the_data_year``), and + decide alike with and without them + (``test_a_file_without_tables_is_checked_as_one_with_them``). The property + tests write one file per example, so they write these, a fortieth of the + size of :func:`_write_output`'s. Arguments are as for that function. + """ + Path(path).parent.mkdir(parents=True, exist_ok=True) + with h5py.File(path, "w") as file: + if data_year is True: + # As pandas stores the data year: a group. + file.create_group(DATA_YEAR_KEY) + elif data_year is not False: + _write_data_year_node(file, data_year) + _write_node(file, YEAR_FILE_STAMP_H5_DATASET, year_file_stamp) + _write_node(file, OUTPUT_STAMP_H5_DATASET, stamp) + return Path(path) + + +def _age(path) -> int: + """The first person's age in the tables stored at ``path``.""" + with pd.HDFStore(path, mode="r") as store: + return int(store["person"]["age"].iloc[0]) + + +#: What :func:`_stored` gives for a node no reader can parse as JSON. +_UNREADABLE = object() + + +def _stored(content): + """Decode what :func:`_write_node` stores, as the checker reads it back. + + An oracle independent of the checker's reader: raw text is parsed as + JSON; raw bytes are fixed-length strings, which read back without + trailing NULs; any other raw node, or text that is not JSON, is + unreadable. A JSON value reads back as its JSON round trip. + """ + if content is _ABSENT: + return _ABSENT + if isinstance(content, tuple): + kind, value = content + try: + if kind == "text": + return json.loads(value) + if kind == "bytes": + return json.loads(value.rstrip(b"\x00").decode("utf-8")) + except (RecursionError, ValueError): + pass + return _UNREADABLE + return json.loads(json.dumps(content)) + + +def _same_json(left, right) -> bool: + """Equal as JSON: ``True`` is not ``1``, and ``1.0`` is not ``1``.""" + return json.dumps(left, sort_keys=True) == json.dumps(right, sort_keys=True) + + +def _well_formed_year_file_stamp(value) -> bool: + """The year-file stamp's shape, stated directly (see ``year_file_stamp``).""" + if not ( + isinstance(value, dict) + and set(value) == {"packages", "revision", "source", "year"} + and type(value["revision"]) is int + and type(value["year"]) is int + and isinstance(value["packages"], dict) + and all( + isinstance(name, str) and (version is None or isinstance(version, str)) + for name, version in value["packages"].items() + ) + and isinstance(value["source"], dict) + ): + return False + source = value["source"] + shapes = { + frozenset({"uri"}): {"uri": str}, + frozenset({"uri", "sha256"}): {"uri": str, "sha256": str}, + frozenset({"path", "size", "mtime_ns"}): { + "path": str, + "size": int, + "mtime_ns": int, + }, + } + types = shapes.get(frozenset(source)) + return types is not None and all( + type(source[key]) is kind for key, kind in types.items() + ) + + +def _is_current(stamp, data_year, dataset_stamp, packages=PACKAGES) -> bool: + """The exactness invariant, stated directly on the stored values. + + Only a data year written as ``save()`` writes it (a group) counts. + """ + if dataset_stamp is not None and not _well_formed_year_file_stamp(dataset_stamp): + return False + return ( + data_year is True + and isinstance(stamp, dict) + and set(stamp) == {"packages", "revision", "year_file"} + and type(stamp["revision"]) is int + and stamp["revision"] == REVISION + and _same_json(stamp["packages"], dict(packages)) + and _same_json(stamp["year_file"], dataset_stamp) + ) + + +def _year_stamp( + revision=YEAR_FILE_REVISION, source=IDENTITY, packages=PACKAGES, year=YEAR +): + return { + "packages": dict(packages), + "revision": revision, + "source": dict(source), + "year": year, + } + + +def _current_year_stamp(year=YEAR): + return year_file_stamp(IDENTITY, year) + + +def _stamp(year_file=None, packages=PACKAGES, revision=REVISION): + """An output stamp, stated directly rather than through ``output_stamp``.""" + return { + "packages": dict(packages), + "revision": revision, + "year_file": copy.deepcopy(year_file), + } + + +# --- The stamp ---------------------------------------------------------- + + +def test_the_stamp_holds_the_packages_the_revision_and_the_year_file(): + assert output_stamp() == { + "packages": PACKAGES, + "revision": REVISION, + "year_file": None, + } + assert output_stamp(_current_year_stamp()) == { + "packages": PACKAGES, + "revision": REVISION, + "year_file": _current_year_stamp(), + } + assert OUTPUT_STAMP_REVISION == 1 + assert list(output_stamp()["packages"]) == sorted(PACKAGES) + + +def test_the_stamp_reads_the_packages_and_revision_when_called(monkeypatch): + monkeypatch.setattr( + uk_datasets, + "installed_output_packages", + lambda: {"policyengine-uk": "9.0.0", "policyengine-core": None}, + ) + monkeypatch.setattr(uk_datasets, "OUTPUT_STAMP_REVISION", 7) + + assert output_stamp() == { + "packages": {"policyengine-core": None, "policyengine-uk": "9.0.0"}, + "revision": 7, + "year_file": None, + } + + +def test_explicit_packages_override_the_installed_ones(): + assert output_stamp(None, {"policyengine-uk": "1.0"})["packages"] == { + "policyengine-uk": "1.0" + } + + +def test_the_stamp_records_the_packages_that_calculate_a_run(): + assert OUTPUT_PACKAGES == ("policyengine-core", "policyengine-uk") + + +def test_installed_output_packages_reads_the_distributions(monkeypatch): + """The real reader asks for the output's packages, importing none.""" + seen = [] + + def recording(names): + seen.append(tuple(names)) + return {name: "1.0" for name in names} + + monkeypatch.setattr(shared_stamp, "installed_package_versions", recording) + + assert REAL_INSTALLED_OUTPUT_PACKAGES() == { + "policyengine-core": "1.0", + "policyengine-uk": "1.0", + } + assert seen == [OUTPUT_PACKAGES] + + +def test_the_stamp_copies_the_year_file_stamp(): + year_file = _current_year_stamp() + stamp = output_stamp(year_file) + year_file["source"]["uri"] = "changed" + year_file["packages"]["policyengine-uk"] = "changed" + + assert stamp["year_file"] == _current_year_stamp() + + +def test_the_stamp_is_stored_as_utf8_json(tmp_path): + path = _write_output(tmp_path / "output.h5") + write_output_stamp(path, output_stamp(_current_year_stamp())) + + with h5py.File(path, "r") as file: + node = file[OUTPUT_STAMP_H5_DATASET] + assert node.shape == () + assert h5py.check_string_dtype(node.dtype).encoding == "utf-8" + assert json.loads(node.asstr()[()]) == output_stamp(_current_year_stamp()) + + +STAMPS = st.fixed_dictionaries( + { + "packages": PACKAGE_MAPS, + "revision": st.integers(), + "year_file": st.one_of(st.none(), YEAR_FILE_STAMPS), + } +) + + +@settings(max_examples=50, deadline=None) +@given(first=STAMPS, second=STAMPS) +def test_an_output_stamp_reads_back_as_written(first, second): + with tempfile.TemporaryDirectory() as directory: + path = _write_stamps(Path(directory) / "output.h5") + write_output_stamp(path, first) + assert read_output_stamp(path) == first + write_output_stamp(path, second) + assert read_output_stamp(path) == second + + +def test_a_missing_or_foreign_file_has_no_output_stamp(tmp_path): + assert read_output_stamp(tmp_path / "missing.h5") is None + foreign = tmp_path / "foreign.h5" + foreign.write_text("not an H5 file") + assert read_output_stamp(foreign) is None + assert read_output_stamp(_write_output(tmp_path / "unstamped.h5")) is None + + +def test_the_stamp_does_not_disturb_pandas_or_loading(tmp_path): + path = _write_output(tmp_path / "output.h5", stamp=output_stamp()) + + with pd.HDFStore(path, mode="r") as store: + assert {"/person", "/benunit", "/household", f"/{DATA_YEAR_KEY}"} <= set( + store.keys() + ) + loaded = PolicyEngineUKDataset( + name="output", + description="reload", + filepath=str(path), + year=YEAR, + is_output_dataset=True, + ) + assert loaded.data_year == DATA_YEAR + assert len(loaded.data.person) == 2 + + +# --- Exactness ---------------------------------------------------------- + + +def _weighted(draw, *choices): + """Draw one of ``choices``, each a ``(weight, strategy)`` pair.""" + total = sum(weight for weight, _ in choices) + pick = draw(st.integers(0, total - 1)) + for weight, strategy in choices: + if pick < weight: + return draw(strategy) + pick -= weight + raise AssertionError + + +@st.composite +def stamped_outputs(draw): + """An output stamped as a run would, near the simulation's dataset. + + Most draws stamp the output from the dataset's own stamp, or one a single + field away, with the installed packages, so current outputs and near + misses are both common. + """ + dataset_stamp = draw(DATASET_STAMPS) + near = st.one_of(YEAR_FILE_STAMPS, DATASET_STAMPS) + if _well_formed_year_file_stamp(dataset_stamp): + near = st.one_of( + near, + st.builds( + lambda key, value: {**dataset_stamp, key: value}, + st.sampled_from(["packages", "revision", "source", "year"]), + st.one_of(PACKAGE_MAPS, YEAR_FILE_REVISIONS, SOURCES, YEARS), + ), + ) + year_file = _weighted( + draw, + (5, st.just(copy.deepcopy(dataset_stamp))), + (3, near), + (1, st.none()), + ) + packages = _weighted( + draw, + (6, st.just(dict(PACKAGES))), + (2, PACKAGE_MAPS), + (1, st.one_of(st.booleans(), JSON_VALUES)), + ) + stamp = _weighted( + draw, + ( + 7, + st.just( + {"packages": packages, "revision": REVISION, "year_file": year_file} + ), + ), + ( + 1, + st.fixed_dictionaries( + { + "packages": st.just(packages), + "revision": st.one_of( + st.sampled_from([REVISION + 1, 0]), st.booleans(), st.floats() + ), + "year_file": st.just(year_file), + } + ), + ), + (1, st.just(_ABSENT)), + (1, JSON_VALUES), + ) + data_year = draw(st.sampled_from([True, True, True, True, False, *DATA_YEAR_NODES])) + return stamp, data_year, dataset_stamp + + +@settings(max_examples=400, deadline=None) +@given(case=stamped_outputs()) +def test_an_output_is_current_only_under_the_exact_stamp_its_run_would_record(case): + stamp, data_year, dataset_stamp = case + with tempfile.TemporaryDirectory() as directory: + path = _write_stamps( + Path(directory) / "output.h5", stamp=stamp, data_year=data_year + ) + problem = output_stamp_problem(path, dataset_stamp) + + expected = _is_current(_stored(stamp), data_year, dataset_stamp) + event(f"current: {expected}") + assert (problem is None) == expected, problem + if problem is not None: + assert problem.startswith(("it ", "its ")), problem + + +NODES = st.one_of( + st.just(_ABSENT), + JSON_VALUES, + st.fixed_dictionaries( + { + "packages": st.one_of(PACKAGE_MAPS, JSON_VALUES), + "revision": st.one_of(st.just(REVISION), st.booleans(), JSON_VALUES), + "year_file": DATASET_STAMPS, + } + ), + st.tuples(st.just("text"), RAW_TEXT), + st.tuples(st.just("bytes"), st.binary(min_size=1, max_size=12)), + st.tuples(st.just("number"), st.integers(-5, 5)), + st.tuples(st.just("array"), st.lists(RAW_TEXT, min_size=1, max_size=3)), + st.tuples(st.just("group"), st.none()), +) + + +@settings(max_examples=400, deadline=None) +@given( + stamp=NODES, + data_year=st.sampled_from([True, False, *DATA_YEAR_NODES]), + dataset_stamp=DATASET_STAMPS, + packages=PACKAGE_MAPS, +) +def test_the_output_problem_is_total_and_exact_for_any_content( + stamp, data_year, dataset_stamp, packages +): + with tempfile.TemporaryDirectory() as directory: + path = _write_stamps( + Path(directory) / "output.h5", stamp=stamp, data_year=data_year + ) + problem = output_stamp_problem(path, dataset_stamp, packages) + + assert problem is None or isinstance(problem, str) + expected = _is_current(_stored(stamp), data_year, dataset_stamp, packages) + assert (problem is None) == expected, problem + if problem is not None: + assert problem.startswith(("it ", "its ")), problem + + +def test_an_output_stamped_with_what_its_run_would_record_is_current(tmp_path): + for dataset_stamp in (None, _current_year_stamp()): + path = _write_output( + tmp_path / f"{dataset_stamp is None}.h5", + stamp=_stamp(dataset_stamp), + ) + assert output_stamp_problem(path, dataset_stamp) is None + + +@pytest.mark.parametrize("data_year", [True, False, *DATA_YEAR_NODES]) +@pytest.mark.parametrize( + "stamp", + [ + _ABSENT, + None, + {"revision": REVISION + 1}, + {"packages": PACKAGES, "revision": REVISION, "year_file": None}, + { + "packages": {"policyengine-uk": "1.0"}, + "revision": REVISION, + "year_file": None, + }, + { + "packages": PACKAGES, + "revision": REVISION, + "year_file": { + "packages": {}, + "revision": 1, + "source": {"uri": "u"}, + "year": 1, + }, + }, + ("text", "not json"), + ("group", None), + ], +) +def test_a_file_without_tables_is_checked_as_one_with_them(tmp_path, stamp, data_year): + """The property tests' files hold no tables (see ``_write_stamps``).""" + full = _write_output(tmp_path / "full.h5", stamp=stamp, data_year=data_year) + light = _write_stamps(tmp_path / "light.h5", stamp=stamp, data_year=data_year) + + for dataset_stamp in (None, _current_year_stamp()): + assert output_stamp_problem(light, dataset_stamp) == output_stamp_problem( + full, dataset_stamp + ) + assert light.stat().st_size < full.stat().st_size / 10 + year_file = _write_year_file(tmp_path / "year.h5", stamp) + light_year_file = _write_stamps(tmp_path / "light_year.h5", year_file_stamp=stamp) + assert year_file_stamp_metadata(light_year_file) == year_file_stamp_metadata( + year_file + ) + assert year_file_stamp_problem( + light_year_file, IDENTITY, YEAR + ) == year_file_stamp_problem(year_file, IDENTITY, YEAR) + + +def test_outputs_without_a_stamp_or_data_year_are_out_of_date_with_distinct_reasons( + tmp_path, +): + no_data_year = _write_output(tmp_path / "a.h5", data_year=False) + stamped = _write_output(tmp_path / "b.h5", stamp=_stamp(), data_year=False) + unstamped = _write_output(tmp_path / "c.h5") + + assert output_stamp_problem(no_data_year, None) == OUTPUT_WITHOUT_DATA_YEAR + assert output_stamp_problem(stamped, None) == STAMPED_OUTPUT_WITHOUT_DATA_YEAR + assert output_stamp_problem(unstamped, None) == UNSTAMPED_OUTPUT + reasons = { + OUTPUT_WITHOUT_DATA_YEAR, + STAMPED_OUTPUT_WITHOUT_DATA_YEAR, + UNSTAMPED_OUTPUT, + MALFORMED_OUTPUT_STAMP, + MALFORMED_OUTPUT_YEAR_FILE, + } + assert len(reasons) == 5 + # The wording the data-year refusal has always matched. + assert "records none" in OUTPUT_WITHOUT_DATA_YEAR + + +@pytest.mark.parametrize( + "stamp", + [ + None, + [], + "stamp", + {}, + {"revision": REVISION}, + {"revision": REVISION, "year_file": None}, + {"packages": PACKAGES, "revision": REVISION}, + {"packages": PACKAGES, "year_file": None}, + {"packages": PACKAGES, "revision": REVISION, "year_file": None, "extra": 1}, + {"packages": PACKAGES, "revision": True, "year_file": None}, + {"packages": PACKAGES, "revision": 1.0, "year_file": None}, + {"packages": PACKAGES, "revision": "1", "year_file": None}, + {"packages": None, "revision": REVISION, "year_file": None}, + {"packages": [], "revision": REVISION, "year_file": None}, + {"packages": {"policyengine-uk": 2}, "revision": REVISION, "year_file": None}, + ("text", "not json"), + ("number", 3), + ("group", None), + ], +) +def test_a_malformed_stamp_is_out_of_date(tmp_path, stamp): + path = _write_output(tmp_path / "output.h5", stamp=stamp) + + assert output_stamp_problem(path, None) == MALFORMED_OUTPUT_STAMP + + +@pytest.mark.parametrize("revision", [0, 2, -1]) +def test_an_output_from_another_format_revision_is_out_of_date(tmp_path, revision): + # Even without a data year: a later revision may record it differently. + path = _write_output( + tmp_path / "output.h5", + stamp={"revision": revision, "shape": "of another revision"}, + data_year=False, + ) + + assert output_stamp_problem(path, None) == ( + f"it was saved with output format revision {revision}, and this version " + f"of policyengine.py reads only revision {REVISION}" + ) + + +def test_raising_the_output_revision_makes_older_outputs_out_of_date( + tmp_path, monkeypatch +): + path = _write_output(tmp_path / "output.h5", stamp=_stamp()) + assert output_stamp_problem(path, None) is None + + monkeypatch.setattr(uk_datasets, "OUTPUT_STAMP_REVISION", REVISION + 1) + + assert output_stamp_problem(path, None) == ( + f"it was saved with output format revision {REVISION}, and this version " + f"of policyengine.py reads only revision {REVISION + 1}" + ) + + +def test_an_outputs_year_file_need_not_be_this_releases(tmp_path, monkeypatch): + """Whether a year file is current is ``load_datasets``' check: an output + calculated from a year file cut with other versions, or another year-file + format revision, is what a run of that year file would save now.""" + old = _year_stamp( + revision=YEAR_FILE_REVISION + 1, + packages={**PACKAGES, "policyengine-uk": "2.126.11"}, + ) + path = _write_output(tmp_path / "output.h5", stamp=_stamp(old)) + + assert year_file_stamp(IDENTITY, YEAR) != old + assert output_stamp_problem(path, old) is None + monkeypatch.setattr(uk_datasets, "YEAR_FILE_STAMP_REVISION", YEAR_FILE_REVISION + 5) + assert output_stamp_problem(path, old) is None + + +# --- Named differences -------------------------------------------------- + + +def _cut_with(packages, name): + if name not in packages: + return f"without a recorded {name} version" + if packages[name] is None: + return f"without {name} installed" + return f"with {name} {packages[name]}" + + +def _year_file_clauses(recorded, current) -> list[str]: + """Each difference between two well-formed year-file stamps, in order.""" + expected = [] + if recorded["year"] != current["year"]: + expected.append( + f"it was calculated from the year file for {recorded['year']}, and " + f"the simulation's dataset is the one for {current['year']}" + ) + if recorded["revision"] != current["revision"]: + expected.append( + "it was calculated from a year file of year-file format revision " + f"{recorded['revision']}, and the simulation's dataset is of " + f"revision {current['revision']}" + ) + if recorded["source"] != current["source"]: + expected.append( + "it was calculated from a year file cut from " + f"{describe_dataset_source_identity(recorded['source'])}, and the " + "simulation's dataset was cut from " + f"{describe_dataset_source_identity(current['source'])}" + ) + for name in sorted(recorded["packages"].keys() | current["packages"].keys()): + if _cut_with(recorded["packages"], name) != _cut_with( + current["packages"], name + ): + expected.append( + f"it was calculated from a year file cut " + f"{_cut_with(recorded['packages'], name)}, and the simulation's " + f"dataset was cut {_cut_with(current['packages'], name)}" + ) + return expected + + +def _package_clauses(calculated_with, installed) -> list[str]: + """Each package version an output was calculated with that is not the one + installed, in name order.""" + expected = [] + for name in sorted(calculated_with.keys() | installed.keys()): + if name not in installed: + expected.append( + f"it records the {name} version it was calculated with, which " + "this version of policyengine.py does not check" + ) + elif calculated_with.get(name, _ABSENT) != installed[name]: + with_version = ( + f"{name} {calculated_with[name]}" + if calculated_with.get(name) is not None + else f"no recorded {name} version" + ) + now = ( + f"{name} {installed[name]} is installed" + if installed[name] is not None + else f"{name} is not installed" + ) + expected.append(f"it was calculated with {with_version}, and {now}") + return expected + + +@settings(max_examples=300, deadline=None) +@given( + recorded=YEAR_FILE_STAMPS, + current=YEAR_FILE_STAMPS, + calculated_with=PACKAGE_MAPS, + installed=PACKAGE_MAPS, +) +def test_a_stale_output_names_each_difference( + recorded, current, calculated_with, installed +): + with tempfile.TemporaryDirectory() as directory: + path = _write_stamps( + Path(directory) / "output.h5", stamp=_stamp(recorded, calculated_with) + ) + problem = output_stamp_problem(path, current, installed) + + expected = _year_file_clauses(recorded, current) + _package_clauses( + calculated_with, installed + ) + assert (problem.split("; ") if problem else []) == expected + assert (problem is None) == (recorded == current and calculated_with == installed) + + +@settings(max_examples=200, deadline=None) +@given(stamped=PACKAGE_MAPS, packages=PACKAGE_MAPS) +def test_package_differences_name_what_was_done_with_them(stamped, packages): + """The shared clauses, for a year file ("cut") and an output + ("calculated"), differ only in that word.""" + calculated = package_differences(stamped, packages, "calculated") + + assert calculated == _package_clauses(stamped, packages) + assert package_differences(stamped, packages) == [ + clause.replace("calculated with", "cut with") for clause in calculated + ] + assert (calculated == []) == (dict(stamped) == dict(packages)) + + +@pytest.mark.parametrize( + "recorded, current, reason", + [ + ( + None, + _year_stamp(), + "it was calculated from data that kept no year-file stamp, and the " + "simulation's dataset keeps one", + ), + ( + _year_stamp(), + None, + "it was calculated from a year file, and the simulation's dataset " + "keeps no year-file stamp", + ), + ( + _year_stamp(), + UNREADABLE_YEAR_FILE_STAMP, + "it was calculated from another year file, or another cut of it, " + "than the simulation's dataset was read from", + ), + ( + {**_year_stamp(), "year": True}, + _year_stamp(), + "it was calculated from another year file, or another cut of it, " + "than the simulation's dataset was read from", + ), + ( + UNREADABLE_YEAR_FILE_STAMP, + UNREADABLE_YEAR_FILE_STAMP, + MALFORMED_OUTPUT_YEAR_FILE, + ), + ( + {**_year_stamp(), "year": 2026.0}, + {**_year_stamp(), "year": 2026.0}, + MALFORMED_OUTPUT_YEAR_FILE, + ), + ], +) +def test_outputs_from_other_or_unreadable_year_files_are_out_of_date( + tmp_path, recorded, current, reason +): + path = _write_output(tmp_path / "output.h5", stamp=_stamp(recorded)) + + assert output_stamp_problem(path, current) == reason + assert MALFORMED_OUTPUT_YEAR_FILE == ( + "it was calculated from a year file whose stamp is unreadable or malformed" + ) + + +def test_a_stamp_equal_in_python_but_not_in_json_is_not_current(tmp_path): + # JSON ``true`` loads as ``True``, which equals 1: neither stamp may pass + # for the other. + stamp = _current_year_stamp() + tampered = {**stamp, "revision": True} + assert tampered == stamp + + for recorded, current in ((stamp, tampered), (tampered, stamp)): + path = _write_output(tmp_path / "output.h5", stamp=_stamp(recorded)) + assert output_stamp_problem(path, current) == MALFORMED_OUTPUT_YEAR_FILE + + +def test_an_output_from_another_source_names_both(tmp_path): + old = _year_stamp(source={"uri": f"{SOURCE}@1.57.0", "sha256": "a" * 64}) + new = _year_stamp(source={"uri": f"{SOURCE}@1.58.0", "sha256": "b" * 64}) + path = _write_output(tmp_path / "output.h5", stamp=_stamp(old)) + + assert output_stamp_problem(path, new) == ( + f"it was calculated from a year file cut from {SOURCE}@1.57.0 (sha256 " + f"{'a' * 64}), and the simulation's dataset was cut from " + f"{SOURCE}@1.58.0 (sha256 {'b' * 64})" + ) + + +def test_an_output_calculated_with_another_policyengine_uk_names_both(tmp_path): + """After an upgrade, before and after the year file is cut again.""" + before = {**PACKAGES, "policyengine-uk": "2.126.11"} + old = _year_stamp(packages=before) + path = _write_output(tmp_path / "output.h5", stamp=_stamp(old, before)) + + # Before the year file is cut again: the versions that calculated the + # output are not the ones installed. + assert output_stamp_problem(path, old) == ( + "it was calculated with policyengine-uk 2.126.11, and policyengine-uk " + "2.126.12 is installed" + ) + # After: the simulation's dataset is the new cut, too. + assert output_stamp_problem(path, _current_year_stamp()) == ( + "it was calculated from a year file cut with policyengine-uk 2.126.11, " + "and the simulation's dataset was cut with policyengine-uk 2.126.12; it " + "was calculated with policyengine-uk 2.126.11, and policyengine-uk " + "2.126.12 is installed" + ) + + +def test_an_output_records_the_versions_that_calculated_it_not_its_year_files( + tmp_path, +): + """A year file cut with one policyengine-uk and run with another: the + output is that other version's, whatever the year file's stamp says.""" + cut_with = {**PACKAGES, "policyengine-uk": "2.126.11"} + year_file = _year_stamp(packages=cut_with) + # Calculated under the installed versions, from the old cut. + path = _write_output(tmp_path / "output.h5", stamp=_stamp(year_file, PACKAGES)) + + assert output_stamp_problem(path, year_file) is None + # With the year file's versions installed again, it is not their output. + assert output_stamp_problem(path, year_file, cut_with) == ( + "it was calculated with policyengine-uk 2.126.12, and policyengine-uk " + "2.126.11 is installed" + ) + + +def test_an_output_from_data_built_in_memory_is_out_of_date_after_an_upgrade( + tmp_path, +): + path = _write_output(tmp_path / "output.h5", stamp=_stamp(None)) + + assert output_stamp_problem(path, None) is None + assert output_stamp_problem( + path, None, {**PACKAGES, "policyengine-core": "3.34.0"} + ) == ( + "it was calculated with policyengine-core 3.33.0, and policyengine-core " + "3.34.0 is installed" + ) + + +def test_explicit_packages_override_the_installed_ones_in_the_check(tmp_path): + other = {"policyengine-core": "9.0.0", "policyengine-uk": "9.0.0"} + path = _write_output(tmp_path / "output.h5", stamp=_stamp(None, other)) + + assert output_stamp_problem(path, None, other) is None + assert output_stamp_problem(path, None) is not None + assert output_stamp_problem(path, None, PACKAGES) == output_stamp_problem( + path, None + ) + + +@pytest.mark.parametrize( + "recorded, current, cut, dataset_cut", + [ + ( + {"policyengine-uk": "1.0"}, + {"policyengine-uk": "1.1"}, + "with policyengine-uk 1.0", + "with policyengine-uk 1.1", + ), + ( + {"policyengine-uk": None}, + {"policyengine-uk": "1.1"}, + "without policyengine-uk installed", + "with policyengine-uk 1.1", + ), + ( + {}, + {"policyengine-uk": None}, + "without a recorded policyengine-uk version", + "without policyengine-uk installed", + ), + ( + {"policyengine-uk": "1.0"}, + {}, + "with policyengine-uk 1.0", + "without a recorded policyengine-uk version", + ), + ], +) +def test_a_version_the_year_file_lacked_or_did_not_record_is_named( + tmp_path, recorded, current, cut, dataset_cut +): + path = _write_output( + tmp_path / "output.h5", stamp=_stamp(_year_stamp(packages=recorded)) + ) + + assert output_stamp_problem(path, _year_stamp(packages=current)) == ( + f"it was calculated from a year file cut {cut}, and the simulation's " + f"dataset was cut {dataset_cut}" + ) + + +@pytest.mark.parametrize( + "calculated_with, installed, reason", + [ + ( + {"policyengine-uk": "1.0"}, + {"policyengine-uk": "1.1"}, + "it was calculated with policyengine-uk 1.0, and policyengine-uk 1.1 " + "is installed", + ), + ( + {"policyengine-uk": None}, + {"policyengine-uk": "1.1"}, + "it was calculated with no recorded policyengine-uk version, and " + "policyengine-uk 1.1 is installed", + ), + ( + {"policyengine-uk": "1.0"}, + {"policyengine-uk": None}, + "it was calculated with policyengine-uk 1.0, and policyengine-uk is " + "not installed", + ), + ( + {"policyengine-uk": "1.0", "extra": "2"}, + {"policyengine-uk": "1.0"}, + "it records the extra version it was calculated with, which this " + "version of policyengine.py does not check", + ), + ( + {}, + {"policyengine-uk": "1.1"}, + "it was calculated with no recorded policyengine-uk version, and " + "policyengine-uk 1.1 is installed", + ), + ], +) +def test_a_version_the_output_was_calculated_with_is_named( + tmp_path, calculated_with, installed, reason +): + path = _write_output(tmp_path / "output.h5", stamp=_stamp(None, calculated_with)) + + assert output_stamp_problem(path, None, installed) == reason + + +def test_an_output_of_another_year_names_both(tmp_path): + path = _write_output( + tmp_path / "output.h5", stamp=_stamp(_current_year_stamp(YEAR)) + ) + + assert output_stamp_problem(path, _current_year_stamp(YEAR + 1)) == ( + f"it was calculated from the year file for {YEAR}, and the simulation's " + f"dataset is the one for {YEAR + 1}" + ) + + +# --- Partial and foreign files ------------------------------------------- + + +@functools.cache +def _current_output_bytes() -> bytes: + with tempfile.TemporaryDirectory() as directory: + path = _write_stamps( + Path(directory) / "output.h5", stamp=_stamp(_current_year_stamp()) + ) + assert output_stamp_problem(path, _current_year_stamp()) is None + return path.read_bytes() + + +@settings(max_examples=100, deadline=None) +@given(data=st.data()) +def test_a_truncated_output_is_never_current(data): + content = _current_output_bytes() + length = data.draw(st.integers(0, len(content) - 1)) + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "output.h5" + path.write_bytes(content[:length]) + + assert output_stamp_problem(path, _current_year_stamp()) is not None + + +def test_a_file_that_is_not_hdf5_is_out_of_date(tmp_path): + path = tmp_path / "output.h5" + path.write_text("not an H5 file") + + assert output_stamp_problem(path, None) == NOT_HDF5 + + +def test_saving_the_tables_again_drops_the_stamp(tmp_path): + path = _write_output(tmp_path / "output.h5", stamp=_stamp()) + + PolicyEngineUKDataset( + name="output", + description="reload", + filepath=str(path), + year=YEAR, + is_output_dataset=True, + ).save() + + assert read_output_stamp(path) is None + assert output_stamp_problem(path, None) == UNSTAMPED_OUTPUT + + +def test_a_dangling_link_in_place_of_the_stamp_reads_as_absent(tmp_path): + path = _write_output(tmp_path / "output.h5") + with h5py.File(path, "a") as file: + file[OUTPUT_STAMP_H5_DATASET] = h5py.SoftLink("/nowhere") + + assert output_stamp_problem(path, None) == UNSTAMPED_OUTPUT + + +def test_a_stamp_link_that_leads_back_to_itself_is_unreadable(tmp_path): + path = _write_output(tmp_path / "output.h5") + with h5py.File(path, "a") as file: + _link_cycle(file, OUTPUT_STAMP_H5_DATASET) + + assert output_stamp_problem(path, None) == MALFORMED_OUTPUT_STAMP + + +def test_json_nested_too_deeply_is_out_of_date_not_an_error(tmp_path): + path = _write_output( + tmp_path / "output.h5", stamp=("text", "[" * 100_000 + "0" + "]" * 100_000) + ) + + assert output_stamp_problem(path, None) == MALFORMED_OUTPUT_STAMP + + +# --- Distinct stamps ---------------------------------------------------- + + +def test_the_stamps_have_distinct_names_and_keys(): + names = { + OUTPUT_STAMP_H5_DATASET, + YEAR_FILE_STAMP_H5_DATASET, + US_OUTPUT_STAMP_H5_DATASET, + } + assert len(names) == 3 + assert set(output_stamp()) != set(_current_year_stamp()) + # The US output stamp records a register of renamed inputs. + assert set(output_stamp()) != {"register", "revision", "year_file"} + + +@settings(max_examples=100, deadline=None) +@given(year_file=st.one_of(st.none(), YEAR_FILE_STAMPS)) +def test_an_output_never_passes_as_a_year_file_nor_the_reverse(year_file): + with tempfile.TemporaryDirectory() as directory: + directory = Path(directory) + # An output stamped as current carries no year-file stamp. + output = _write_stamps(directory / "output.h5", stamp=_stamp(year_file)) + assert output_stamp_problem(output, year_file) is None + assert read_year_file_stamp(output) is None + assert year_file_stamp_metadata(output) is None + assert year_file_stamp_problem(output, IDENTITY, YEAR) is not None + + # A current year file carries no output stamp. + year = _write_stamps( + directory / "year.h5", year_file_stamp=_current_year_stamp() + ) + assert year_file_stamp_problem(year, IDENTITY, YEAR) is None + assert read_output_stamp(year) is None + assert output_stamp_problem(year, _current_year_stamp()) == UNSTAMPED_OUTPUT + + # Nor does a US output stamp pass as a UK one. + us_output = _write_stamps(directory / "us.h5") + with h5py.File(us_output, "a") as file: + _write_node(file, US_OUTPUT_STAMP_H5_DATASET, _stamp(year_file)) + assert output_stamp_problem(us_output, year_file) == UNSTAMPED_OUTPUT + + +# --- Errors unrelated to the file's contents ---------------------------- + + +def test_a_missing_output_raises_file_not_found(tmp_path): + with pytest.raises(FileNotFoundError): + output_stamp_problem(tmp_path / "missing.h5", None) + with pytest.raises(FileNotFoundError): + year_file_stamp_metadata(tmp_path / "missing.h5") + + +def _unreadable_by_permission(path) -> Path: + path = _write_output(path, stamp=_stamp()) + os.chmod(path, 0) + return path + + +def _a_directory(path) -> Path: + path.mkdir() + return path + + +@pytest.mark.parametrize( + "make, error", + [ + (_unreadable_by_permission, PermissionError), + (_a_directory, IsADirectoryError), + ], +) +@pytest.mark.parametrize( + "check", + [lambda path: output_stamp_problem(path, None), year_file_stamp_metadata], +) +def test_errors_unrelated_to_the_contents_are_raised(tmp_path, make, error, check): + if error is PermissionError and hasattr(os, "geteuid") and os.geteuid() == 0: + pytest.skip("root can read a file without permission bits") + path = make(tmp_path / "output.h5") + try: + with pytest.raises(error): + check(path) + finally: + if path.is_file(): + os.chmod(path, 0o644) + + +ERRORS = [ + PermissionError(13, "Permission denied"), + IsADirectoryError(21, "Is a directory"), + BlockingIOError(35, "Unable to lock file"), +] + + +@pytest.mark.parametrize("error", ERRORS) +@pytest.mark.parametrize( + "check", + [lambda path: output_stamp_problem(path, None), year_file_stamp_metadata], +) +def test_errors_unrelated_to_the_contents_are_raised_on_opening( + tmp_path, monkeypatch, error, check +): + path = _write_output(tmp_path / "output.h5", stamp=_stamp()) + + def fail(*args, **kwargs): + raise error + + monkeypatch.setattr(h5py, "File", fail) + with pytest.raises(type(error)): + check(path) + + +@pytest.mark.parametrize("error", ERRORS) +def test_errors_unrelated_to_the_contents_are_raised_while_reading( + tmp_path, monkeypatch, error +): + path = _write_output( + tmp_path / "output.h5", + stamp=_stamp(), + year_file_stamp=_current_year_stamp(), + ) + + def fail(self, *args, **kwargs): + raise error + + monkeypatch.setattr(h5py.Dataset, "asstr", fail) + with pytest.raises(type(error)): + output_stamp_problem(path, None) + with pytest.raises(type(error)): + year_file_stamp_metadata(path) + + +@pytest.mark.parametrize("error", ERRORS) +def test_errors_unrelated_to_the_contents_are_raised_while_finding_the_data_year( + tmp_path, monkeypatch, error +): + path = _write_output(tmp_path / "output.h5", stamp=_stamp()) + get = h5py.Group.get + + def fail_for_the_data_year(self, name, *args, **kwargs): + if name == DATA_YEAR_KEY: + raise error + return get(self, name, *args, **kwargs) + + monkeypatch.setattr(h5py.Group, "get", fail_for_the_data_year) + with pytest.raises(type(error)): + output_stamp_problem(path, None) + + +def test_the_decision_reads_only_the_stamp_and_the_data_year(tmp_path, monkeypatch): + path = _write_output(tmp_path / "output.h5", stamp=_stamp(_current_year_stamp())) + read = [] + get = h5py.Group.get + + def recording_get(self, name, *args, **kwargs): + read.append(name) + return get(self, name, *args, **kwargs) + + def no_tables(*args, **kwargs): + raise AssertionError("the tables were opened") + + monkeypatch.setattr(h5py.Group, "get", recording_get) + monkeypatch.setattr(pd, "HDFStore", no_tables) + monkeypatch.setattr(pd, "read_hdf", no_tables) + + assert output_stamp_problem(path, _current_year_stamp()) is None + assert set(read) == {OUTPUT_STAMP_H5_DATASET, DATA_YEAR_KEY} + + +# --- The year-file stamp a dataset keeps -------------------------------- + + +def _load(path) -> PolicyEngineUKDataset: + return PolicyEngineUKDataset( + name="year-file", description="load", filepath=str(path), year=YEAR + ) + + +@pytest.mark.parametrize( + "stored, kept", + [ + (_ABSENT, None), + (_year_stamp(), _year_stamp()), + ({"any": "json"}, {"any": "json"}), + (None, UNREADABLE_YEAR_FILE_STAMP), + (("text", "not json"), UNREADABLE_YEAR_FILE_STAMP), + (("number", 3), UNREADABLE_YEAR_FILE_STAMP), + (("group", None), UNREADABLE_YEAR_FILE_STAMP), + ], +) +def test_a_dataset_keeps_its_year_files_stamp(tmp_path, stored, kept): + path = _write_year_file(tmp_path / "year.h5", stored) + + assert year_file_stamp_metadata(path) == kept + dataset = _load(path) + assert dataset.metadata.get(YEAR_FILE_STAMP_KEY) == kept + assert (YEAR_FILE_STAMP_KEY in dataset.metadata) == (kept is not None) + assert input_year_file_stamp(dataset) == kept + + +def test_a_file_that_is_not_hdf5_keeps_no_year_file_stamp(tmp_path): + path = tmp_path / "year.h5" + path.write_text("not an H5 file") + + assert year_file_stamp_metadata(path) is None + + +def test_reloading_a_resaved_year_file_drops_the_stamp_it_kept(tmp_path): + path = _write_year_file(tmp_path / "year.h5", _current_year_stamp()) + dataset = _load(path) + assert dataset.metadata[YEAR_FILE_STAMP_KEY] == _current_year_stamp() + + dataset.save() + # Saving writes no stamp, so the file no longer holds the cut it named. + dataset.load() + + assert YEAR_FILE_STAMP_KEY not in dataset.metadata + assert input_year_file_stamp(dataset) is None + + +def test_a_dataset_built_in_memory_keeps_no_year_file_stamp(): + dataset = PolicyEngineUKDataset( + name="memory", description="memory", year=YEAR, data=_tables() + ) + + assert dataset.metadata == {} + assert input_year_file_stamp(dataset) is None + + +def test_loading_a_dataset_with_no_filepath_is_refused(): + dataset = PolicyEngineUKDataset( + name="memory", description="memory", year=YEAR, data=_tables() + ) + + with pytest.raises(ValueError, match="no filepath"): + dataset.load() + + +def test_a_dataset_whose_tables_are_not_loaded_gives_its_files_stamp(tmp_path): + path = _write_year_file(tmp_path / "year.h5", _current_year_stamp()) + dataset = _load(path) + # A run loads a dataset whose tables are not loaded, so the stamp it + # records is the one the file holds, not one kept from before. + dataset.data = None + dataset.metadata[YEAR_FILE_STAMP_KEY] = "kept from before" + + assert input_year_file_stamp(dataset) == _current_year_stamp() + + +def test_a_loaded_dataset_keeps_the_stamp_of_the_tables_it_holds(tmp_path): + """A year file cut again after it was loaded: the dataset still holds the + old cut's tables, so a run of it records the old cut's stamp.""" + path, replacement, old, new = _two_cuts(tmp_path) + dataset = _load(path) + + os.replace(replacement, path) + + assert year_file_stamp_metadata(path) == new + assert pd.DataFrame(dataset.data.person)["age"].iloc[0] == 70 + assert input_year_file_stamp(dataset) == old + dataset.load() + assert pd.DataFrame(dataset.data.person)["age"].iloc[0] == 99 + assert input_year_file_stamp(dataset) == new + + +def test_the_stamp_a_run_records_is_a_copy(tmp_path): + path = _write_year_file(tmp_path / "year.h5", _current_year_stamp()) + dataset = _load(path) + + stamp = input_year_file_stamp(dataset) + stamp["year"] = 0 + stamp["source"]["uri"] = "changed" + stamp["packages"]["policyengine-uk"] = "changed" + + assert dataset.metadata[YEAR_FILE_STAMP_KEY] == _current_year_stamp() + + +def test_the_metadata_a_dataset_keeps_survives_a_json_round_trip(): + dataset = PolicyEngineUKDataset( + name="memory", + description="memory", + year=YEAR, + metadata={ + YEAR_FILE_STAMP_KEY: _current_year_stamp(), + OUTPUT_STAMP_KEY: _stamp(_current_year_stamp()), + }, + ) + + restored = PolicyEngineUKDataset.model_validate_json(dataset.model_dump_json()) + + assert restored.metadata == dataset.metadata + assert dataset.model_dump(mode="json")["metadata"] == dataset.metadata + + +def test_the_metadata_keys_are_the_documented_ones(): + assert OUTPUT_STAMP_KEY == "output_stamp" + assert YEAR_FILE_STAMP_KEY == "year_file_stamp" + assert OUTPUT_STAMP_H5_DATASET == "policyengine_uk_output_stamp" + + +# --- One save per output ------------------------------------------------ + + +def _partial_files(folder: Path) -> list[Path]: + """Temporary files left in a folder.""" + return sorted(folder.glob(".*.partial")) + + +def test_saving_an_output_writes_its_tables_and_stamp(tmp_path): + path = tmp_path / "data" / "output.h5" + stamp = _stamp(_current_year_stamp()) + + save_output(_output_dataset(path, age=41), stamp) + + assert output_stamp_problem(path, _current_year_stamp()) is None + assert read_output_stamp(path) == stamp + assert _age(path) == 41 + assert _partial_files(path.parent) == [] + + +def test_saving_an_output_with_no_filepath_is_refused(): + dataset = PolicyEngineUKDataset( + name="memory", + description="memory", + year=YEAR, + is_output_dataset=True, + data=_tables(), + data_year=DATA_YEAR, + ) + + with pytest.raises(ValueError, match="no filepath"): + save_output(dataset, _stamp()) + + +@pytest.mark.parametrize("stop", ["in the tables", "at the stamp", "at the move"]) +def test_a_save_that_fails_leaves_the_output_already_there_whole(tmp_path, stop): + path = tmp_path / "output.h5" + save_output(_output_dataset(path, age=41), _stamp(_current_year_stamp())) + before = path.read_bytes() + + def stopped(*args, **kwargs): + raise RuntimeError("stopped part-way") + + with pytest.MonkeyPatch.context() as patcher: + if stop == "in the tables": + put = pd.HDFStore.put + + def put_until_household(self, key, value, *args, **kwargs): + if key == "household": + raise RuntimeError("stopped part-way") + return put(self, key, value, *args, **kwargs) + + patcher.setattr(pd.HDFStore, "put", put_until_household) + elif stop == "at the stamp": + patcher.setattr(shared_stamp, "write_json_dataset", stopped) + else: + patcher.setattr(shared_stamp.os, "replace", stopped) + with pytest.raises(RuntimeError, match="stopped part-way"): + save_output(_output_dataset(path, age=99), _stamp(None)) + + assert path.read_bytes() == before + assert output_stamp_problem(path, _current_year_stamp()) is None + assert _age(path) == 41 + assert _partial_files(tmp_path) == [] + + +def test_a_first_save_that_fails_leaves_no_output(tmp_path, monkeypatch): + path = tmp_path / "output.h5" + + def stopped(*args, **kwargs): + raise RuntimeError("stopped part-way") + + monkeypatch.setattr(shared_stamp, "write_json_dataset", stopped) + with pytest.raises(RuntimeError, match="stopped part-way"): + save_output(_output_dataset(path), _stamp()) + + assert not path.exists() + assert _partial_files(tmp_path) == [] + + +def test_two_saves_to_one_path_never_mix_ones_tables_with_the_others_stamp( + tmp_path, monkeypatch +): + """Another process saves a whole output to the path between this save's + tables and its stamp. Whichever save's file is at the path, its tables + and its stamp are that save's.""" + path = tmp_path / "output.h5" + one = _stamp(_year_stamp(source={"uri": f"{SOURCE}@1"})) + other = _stamp(_year_stamp(source={"uri": f"{SOURCE}@2"})) + publish = uk_datasets.publish_year_file + seen = [] + + def publish_after_the_other_save(temporary, target, name, stamp): + if not seen: + seen.append("other") + # The other save runs to completion before this one stamps. + save_output(_output_dataset(path, age=99), other) + assert (read_output_stamp(path), _age(path)) == (other, 99) + return publish(temporary, target, name, stamp) + + monkeypatch.setattr(uk_datasets, "publish_year_file", publish_after_the_other_save) + save_output(_output_dataset(path, age=41), one) + + assert seen == ["other"] + assert (read_output_stamp(path), _age(path)) == (one, 41) + assert _partial_files(tmp_path) == [] + + +def test_an_output_is_saved_where_its_filepath_says(tmp_path): + dataset = _output_dataset(tmp_path / "first.h5") + dataset.filepath = str(tmp_path / "moved" / "second.h5") + + save_output(dataset, _stamp()) + + assert not (tmp_path / "first.h5").exists() + assert output_stamp_problem(tmp_path / "moved" / "second.h5", None) is None + + +# --- One file per load -------------------------------------------------- + + +def _two_cuts(directory: Path): + """A year file at ``directory/year.h5`` and another cut of it beside it, + with other tables (the first person's age) and another stamp.""" + old = _year_stamp(source={"uri": f"{SOURCE}@1", "sha256": "a" * 64}) + new = _year_stamp(source={"uri": f"{SOURCE}@2", "sha256": "b" * 64}) + path = _write_year_file(directory / "year.h5", old, age=70) + replacement = _write_year_file(directory / "next.h5", new, age=99) + return path, replacement, old, new + + +class _DuringTheRead: + """Stand in for ``pd.HDFStore``, calling ``before`` as the tables are + opened and ``after`` as they are closed, as another process acting at + those moments would: on the first open only, or on every open.""" + + def __init__(self, before=None, after=None, every_time=False): + self.before, self.after, self.every_time = before, after, every_time + self.store = pd.HDFStore + self.opened = 0 + + def __call__(self, *args, **kwargs): + acting = self.opened == 0 or self.every_time + self.opened += 1 + outer = self + + class Store: + def __init__(self): + if outer.before and acting: + outer.before() + self.store = outer.store(*args, **kwargs) + + def __enter__(self): + return self.store.__enter__() + + def __exit__(self, *exc): + result = self.store.__exit__(*exc) + if outer.after and acting: + outer.after() + return result + + return Store() + + +@pytest.mark.parametrize("when", ["before", "after"]) +def test_a_year_file_replaced_while_it_is_read_is_read_again( + tmp_path, monkeypatch, when +): + """The dataset keeps the stamp of the cut its tables came from.""" + path, replacement, _, new = _two_cuts(tmp_path) + stores = _DuringTheRead(**{when: lambda: os.replace(replacement, path)}) + + monkeypatch.setattr(pd, "HDFStore", stores) + dataset = _load(path) + + assert stores.opened == 2 + assert dataset.metadata[YEAR_FILE_STAMP_KEY] == new + assert pd.DataFrame(dataset.data.person)["age"].iloc[0] == 99 + assert pd.DataFrame(dataset.data_year_data.person)["age"].iloc[0] == 99 + + +def test_a_year_file_replaced_and_put_back_while_it_is_read_is_read_again( + tmp_path, monkeypatch +): + """Another cut is moved onto the path after the stamp is read, its tables + are read, and a copy of the first cut is moved back before the read ends. + The stamp at the path is the same before and after, yet the tables read + are the other cut's: the file put back is another file, so the read is + done again.""" + path, replacement, old, _ = _two_cuts(tmp_path) + put_back = tmp_path / "back.h5" + shutil.copy(path, put_back) + stores = _DuringTheRead( + before=lambda: os.replace(replacement, path), + after=lambda: os.replace(put_back, path), + ) + + monkeypatch.setattr(pd, "HDFStore", stores) + dataset = _load(path) + + assert stores.opened == 2 + assert dataset.metadata[YEAR_FILE_STAMP_KEY] == old + assert pd.DataFrame(dataset.data.person)["age"].iloc[0] == 70 + + +def test_a_year_file_modified_while_it_is_read_is_read_again(tmp_path, monkeypatch): + """A writer that rewrites the file in place, as ``save()`` does.""" + path = _write_year_file(tmp_path / "year.h5", _current_year_stamp(), age=70) + write = PolicyEngineUKDataset._write + store = pd.HDFStore + + def rewrite(): + # With the real store, so this writer's open is not counted. + with pytest.MonkeyPatch.context() as patcher: + patcher.setattr(pd, "HDFStore", store) + write(_year_dataset(path, age=99), path) + status = os.stat(path) + os.utime(path, ns=(status.st_atime_ns, status.st_mtime_ns + 10**9)) + + stores = _DuringTheRead(after=rewrite) + monkeypatch.setattr(pd, "HDFStore", stores) + dataset = _load(path) + + assert stores.opened == 2 + assert pd.DataFrame(dataset.data.person)["age"].iloc[0] == 99 + # Rewriting the tables wrote no stamp. + assert YEAR_FILE_STAMP_KEY not in dataset.metadata + + +def test_a_year_file_replaced_on_every_read_is_refused(tmp_path, monkeypatch): + path = _write_year_file(tmp_path / "year.h5", _current_year_stamp()) + spare = tmp_path / "spare.h5" + + def replace(): + shutil.copy(path, spare) + os.replace(spare, path) + + stores = _DuringTheRead(after=replace, every_time=True) + monkeypatch.setattr(pd, "HDFStore", stores) + with pytest.raises( + RuntimeError, match="was replaced or modified while it was read, 3 times" + ): + _load(path) + assert stores.opened == 3 + + +def test_an_unchanged_year_file_is_read_once(tmp_path, monkeypatch): + path = _write_year_file(tmp_path / "year.h5", _current_year_stamp()) + stores = _DuringTheRead() + + monkeypatch.setattr(pd, "HDFStore", stores) + dataset = _load(path) + + assert stores.opened == 1 + assert dataset.metadata[YEAR_FILE_STAMP_KEY] == _current_year_stamp() + + +def test_read_one_file_returns_what_was_read_of_one_unchanged_file(tmp_path): + path = tmp_path / "file" + path.write_text("first") + reads = [] + + def read(name): + reads.append(name) + return Path(name).read_text() + + assert read_one_file(path, read) == "first" + assert reads == [str(path)] + + +def test_read_one_file_reads_again_after_a_replacement_and_gives_up(tmp_path): + path = tmp_path / "file" + path.write_text("first") + reads = [] + + def replacing(until): + def read(name): + reads.append(Path(name).read_text()) + if len(reads) <= until: + spare = tmp_path / "spare" + spare.write_text(f"replacement {len(reads)}") + os.replace(spare, name) + return reads[-1] + + return read + + assert read_one_file(path, replacing(1)) == "replacement 1" + assert reads == ["first", "replacement 1"] + + reads.clear() + with pytest.raises(RuntimeError, match="2 times"): + read_one_file(path, replacing(5), attempts=2) + assert len(reads) == 2 + + +def test_read_one_file_reads_again_after_the_file_is_modified_in_place(tmp_path): + path = tmp_path / "file" + path.write_text("first") + reads = [] + + def appending(name): + reads.append(Path(name).read_text()) + if len(reads) == 1: + with open(name, "a") as file: + file.write(" and more") + return reads[-1] + + assert read_one_file(path, appending) == "first and more" + assert reads == ["first", "first and more"] + + # Another size, at the same time. + reads.clear() + + def appending_unseen(name): + reads.append(Path(name).read_text()) + if len(reads) == 1: + status = os.stat(name) + with open(name, "a") as file: + file.write("!") + os.utime(name, ns=(status.st_atime_ns, status.st_mtime_ns)) + return reads[-1] + + assert read_one_file(path, appending_unseen) == "first and more!" + assert len(reads) == 2 + path.write_text("first and more") + + # The same size, at another time. + reads.clear() + + def touching(name): + reads.append(Path(name).read_text()) + if len(reads) == 1: + status = os.stat(name) + os.utime(name, ns=(status.st_atime_ns, status.st_mtime_ns + 10**9)) + return reads[-1] + + assert read_one_file(path, touching) == "first and more" + assert len(reads) == 2 + + +def test_read_one_file_finds_no_file_once_it_is_removed_meanwhile(tmp_path): + path = tmp_path / "file" + path.write_text("first") + + def removing(name): + os.unlink(name) + return "read" + + # The read is not trusted; the second attempt finds no file to read. + with pytest.raises(FileNotFoundError): + read_one_file(path, removing) + + +def test_read_one_file_raises_what_the_read_raises_and_what_opening_does(tmp_path): + path = tmp_path / "file" + path.write_text("first") + + def failing(name): + raise ValueError("refused") + + with pytest.raises(ValueError, match="refused"): + read_one_file(path, failing) + with pytest.raises(FileNotFoundError): + read_one_file(tmp_path / "missing", failing) + with pytest.raises(IsADirectoryError): + read_one_file(tmp_path, failing) diff --git a/tests/test_uk_output_stamp_integration.py b/tests/test_uk_output_stamp_integration.py new file mode 100644 index 00000000..2291b65b --- /dev/null +++ b/tests/test_uk_output_stamp_integration.py @@ -0,0 +1,1040 @@ +"""A saved UK output is never reused after its year file is cut again. + +These tests run, save, load and ensure UK simulations over a real +policyengine-uk engine, on the seven-person 2024 dataset of +``test_uk_year_file_data_year.py``. Each replays one way a saved output could +be reused although it holds results calculated from other data, or with +other package versions: + +- the bundle re-pins a logical dataset name to another revision and sha256, + and ``ensure_datasets`` cuts the year files again; +- a local source is rewritten in place, and the year files are cut again; +- policyengine-uk or policyengine-core is upgraded, before and after the + year files are cut again; +- a year file is run with other versions than cut it, and those that cut it + are installed again; +- the simulation's dataset is another year's file in the same folder; +- the output was saved by an earlier release, before the stamp existed; +- another process saves another output to the path while this one is loaded. + +In each case ``Simulation.load()`` refuses the output, naming what changed, +and ``Simulation.ensure()`` calculates it again from the simulation's dataset +and then reuses it. ``test_uk_output_stamp.py`` states the check's invariants +without an engine. + +Invariant exercised over generated outputs and datasets, with a stand-in for +the engine's run: + +- **Ensure agrees with load.** ``Simulation.ensure()`` runs the simulation + exactly when its output is missing or ``Simulation.load()`` would refuse + it, and the output it saves is current exactly when the simulation's + dataset keeps no year-file stamp or a well-formed one. +""" + +from __future__ import annotations + +import os +import re +import shutil +import tempfile +import uuid +from pathlib import Path +from unittest.mock import patch + +import h5py +import pandas as pd +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +pytest.importorskip("policyengine_uk") + +import policyengine as pe # noqa: E402 +from policyengine.core import Simulation # noqa: E402 +from policyengine.core.scoping_strategy import RowFilterStrategy # noqa: E402 +from policyengine.core.simulation import _cache # noqa: E402 +from policyengine.provenance.dataset_materialization import ( # noqa: E402 + describe_dataset_source_identity, + requested_dataset_source_identity, +) +from policyengine.tax_benefit_models.common.model_version import ( # noqa: E402 + output_dataset_filepath, +) +from policyengine.tax_benefit_models.uk import datasets as uk_datasets # noqa: E402 +from policyengine.tax_benefit_models.uk.datasets import ( # noqa: E402 + DATA_YEAR_KEY, + MALFORMED_OUTPUT_STAMP, + OUTPUT_STAMP_H5_DATASET, + OUTPUT_STAMP_KEY, + OUTPUT_STAMP_REVISION, + UNREADABLE_YEAR_FILE_STAMP, + UNSTAMPED_OUTPUT, + YEAR_FILE_STAMP_KEY, + YEAR_FILE_STAMP_REVISION, + PolicyEngineUKDataset, + ensure_datasets, + input_year_file_stamp, + installed_output_packages, + installed_package_versions, + load_datasets, + output_stamp, + output_stamp_problem, + read_output_stamp, + read_year_file_stamp, + write_output_stamp, + year_file_stamp, +) +from policyengine.utils.hashing import sha256_file # noqa: E402 +from tests.test_uk_year_file_data_year import ( # noqa: E402 + DATA_YEAR, + EXTRA_VARIABLES, + _assert_matches_direct, + _direct, + _tables, + _with_weights, + _write_source, +) +from tests.test_uk_year_file_stamp_integration import ( # noqa: E402 + BUNDLE_NAME, + BUNDLE_URI, + _assert_cut_again, + _assert_reused, + _ensure, + _new_source, + _pinned, + _replacement_tables, + _stored, +) + +YEAR = 2026 +KEY = f"tiny_uk_2024_{YEAR}" +#: The engine's run, kept before any test counts its calls. +RUN = type(pe.uk.model).run + + +@pytest.fixture(autouse=True) +def _no_outputs_kept_in_memory(): + """``ensure()`` keeps each output in memory under its simulation's id. + + These tests reuse an id to reach the output saved under it, as another + process would. + """ + _cache.clear() + yield + _cache.clear() + + +def _run_id() -> str: + return f"uk-output-stamp-{uuid.uuid4().hex}" + + +def _simulation(dataset, run_id: str, **kwargs) -> Simulation: + return Simulation( + id=run_id, + dataset=dataset, + tax_benefit_model_version=pe.uk.model, + extra_variables=EXTRA_VARIABLES, + **kwargs, + ) + + +def _ensured(dataset, run_id: str, run=RUN, **kwargs) -> tuple[Simulation, int]: + """Ensure a simulation's output with none kept in memory. + + Returns the simulation and how many times it was run. + """ + _cache.clear() + simulation = _simulation(dataset, run_id, **kwargs) + with patch.object( + type(pe.uk.model), "run", autospec=True, side_effect=run + ) as counted: + simulation.ensure() + return simulation, counted.call_count + + +def _refusal(dataset, run_id: str) -> str | None: + """Why ``load()`` refuses the saved output, or ``None`` when it loads.""" + simulation = _simulation(dataset, run_id) + try: + simulation.load() + except ValueError as error: + return str(error) + return None + + +def _out_of_date(path, problem: str) -> str: + return ( + f"Saved UK simulation output {path} is out of date: {problem}. Run the " + "simulation again (Simulation.ensure() does so)." + ) + + +def _frames(simulation: Simulation) -> dict[str, pd.DataFrame]: + return { + name: pd.DataFrame(frame) + for name, frame in simulation.output_dataset.data.entity_data.items() + } + + +def _assert_same_output(first: Simulation, second: Simulation) -> None: + for name, frame in _frames(first).items(): + pd.testing.assert_frame_equal( + frame.reset_index(drop=True), + _frames(second)[name].reset_index(drop=True), + check_dtype=False, + check_categorical=False, + ) + + +def _upgrade(patcher, package: str, version: str = "999.0.0") -> None: + """Make ``package`` report ``version`` as installed, to year files and to + outputs alike, as an upgrade would. The engine's code stays the same.""" + cut_with = installed_package_versions() + calculated_with = installed_output_packages() + patcher.setattr( + uk_datasets, + "installed_package_versions", + lambda: {**cut_with, package: version}, + ) + patcher.setattr( + uk_datasets, + "installed_output_packages", + lambda: {**calculated_with, package: version}, + ) + + +def _edit(source: str) -> None: + """Rewrite a local source in place, with a modification time of its own + whatever the file system's resolution.""" + before = requested_dataset_source_identity("uk", source) + _write_source(Path(source), _replacement_tables()) + os.utime(source, ns=(before["mtime_ns"] + 10**9, before["mtime_ns"] + 10**9)) + assert requested_dataset_source_identity("uk", source) != before + + +def _partial_files(folder: Path) -> list[Path]: + return sorted(Path(folder).glob(".*.partial")) + + +# --- What a run records -------------------------------------------------- + + +def test_year_files_keep_their_stamp_in_memory_as_on_disk(tmp_path): + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + identity = requested_dataset_source_identity("uk", source) + + created = _ensure(source, data_folder, years=(DATA_YEAR, YEAR)) + loaded = _assert_reused(source, data_folder, years=(DATA_YEAR, YEAR)) + + for year in (DATA_YEAR, YEAR): + key = f"tiny_uk_2024_{year}" + on_disk = read_year_file_stamp(created[key].filepath) + assert on_disk == year_file_stamp(identity, year) + assert created[key].metadata == {YEAR_FILE_STAMP_KEY: on_disk} + assert loaded[key].metadata == {YEAR_FILE_STAMP_KEY: on_disk} + assert input_year_file_stamp(created[key]) == on_disk + assert input_year_file_stamp(loaded[key]) == on_disk + + +def test_a_saved_output_records_what_it_was_calculated_from_and_with(tmp_path): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + year_file = read_year_file_stamp(dataset.filepath) + run_id = _run_id() + + simulation = _simulation(dataset, run_id) + simulation.run() + + stamp = { + "packages": installed_output_packages(), + "revision": OUTPUT_STAMP_REVISION, + "year_file": year_file, + } + assert set(stamp["packages"]) == {"policyengine-core", "policyengine-uk"} + assert None not in stamp["packages"].values() + assert simulation.output_dataset.metadata == {OUTPUT_STAMP_KEY: stamp} + path = output_dataset_filepath(simulation) + assert not path.exists() + simulation.save() + assert read_output_stamp(path) == stamp + with h5py.File(path, "r") as file: + assert {DATA_YEAR_KEY, OUTPUT_STAMP_H5_DATASET, "person"} <= set(file) + assert output_stamp_problem(path, year_file) is None + assert _partial_files(path.parent) == [] + + reloaded = _simulation(dataset, run_id) + reloaded.load() + assert reloaded.output_dataset.metadata == {OUTPUT_STAMP_KEY: stamp} + assert reloaded.output_dataset.data_year == DATA_YEAR + _assert_same_output(simulation, reloaded) + + +def test_a_run_records_the_output_revision_it_runs_under(tmp_path, monkeypatch): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + run_id = _run_id() + + with monkeypatch.context() as raised: + raised.setattr(uk_datasets, "OUTPUT_STAMP_REVISION", OUTPUT_STAMP_REVISION + 1) + simulation, runs = _ensured(dataset, run_id) + assert runs == 1 + revision = simulation.output_dataset.metadata[OUTPUT_STAMP_KEY]["revision"] + assert revision == OUTPUT_STAMP_REVISION + 1 + assert _refusal(dataset, run_id) is None + + path = output_dataset_filepath(simulation) + assert _refusal(dataset, run_id) == _out_of_date( + path, + f"it was saved with output format revision {OUTPUT_STAMP_REVISION + 1}, " + "and this version of policyengine.py reads only revision " + f"{OUTPUT_STAMP_REVISION}", + ) + + +def test_saving_writes_the_stamp_the_run_recorded_not_one_made_then(tmp_path): + """The year file is cut again between a run and its save: the output + holds results from the old cut, and is stamped with the old cut.""" + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + dataset = _ensure(source, data_folder)[KEY] + old = read_year_file_stamp(dataset.filepath) + run_id = _run_id() + simulation = _simulation(dataset, run_id) + simulation.run() + + _edit(source) + recut = _assert_cut_again(source, data_folder)[KEY] + new = read_year_file_stamp(recut.filepath) + assert new != old + # The simulation's dataset is reloaded from the new cut before the save. + dataset.load() + assert input_year_file_stamp(dataset) == new + simulation.save() + + path = output_dataset_filepath(simulation) + assert read_output_stamp(path) == output_stamp(old) + assert _refusal(recut, run_id) is not None + assert _ensured(recut, run_id)[1] == 1 + assert read_output_stamp(path) == output_stamp(new) + + +def test_a_loaded_output_saved_again_keeps_its_stamp(tmp_path): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + run_id = _run_id() + first, runs = _ensured(dataset, run_id) + assert runs == 1 + path = output_dataset_filepath(first) + stamp = read_output_stamp(path) + + reloaded = _simulation(dataset, run_id) + reloaded.load() + reloaded.save() + + assert read_output_stamp(path) == stamp + assert _refusal(dataset, run_id) is None + + +def test_an_output_is_saved_where_its_filepath_says(tmp_path): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + simulation = _simulation(dataset, _run_id()) + simulation.run() + moved = tmp_path / "elsewhere" / "output.h5" + simulation.output_dataset.filepath = str(moved) + + simulation.save() + + # The tables and the stamp go to one place. + assert not output_dataset_filepath(simulation).exists() + assert output_stamp_problem(moved, read_year_file_stamp(dataset.filepath)) is None + with pd.HDFStore(moved, mode="r") as store: + assert len(store["person"]) == 7 + + +def test_a_scoped_run_records_the_stamp_of_the_year_file_it_filters(tmp_path): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + + simulation = _simulation( + dataset, + _run_id(), + scoping_strategy=RowFilterStrategy( + variable_name="region", variable_value="SCOTLAND" + ), + ) + simulation.run() + + assert sorted(_frames(simulation)["person"]["person_id"]) == [201, 202] + assert simulation.output_dataset.metadata[OUTPUT_STAMP_KEY] == output_stamp( + read_year_file_stamp(dataset.filepath) + ) + + +def test_a_run_of_data_built_in_memory_records_no_year_file(tmp_path): + data_folder = tmp_path / "data" + data_folder.mkdir() + built = PolicyEngineUKDataset( + name="observed-2026", + description="tables taken as observed in 2026", + # Never written: it places the output beside the year files below. + filepath=str(data_folder / "built.h5"), + year=YEAR, + data=_with_weights(_tables()), + ) + run_id = _run_id() + + simulation, runs = _ensured(built, run_id) + assert runs == 1 + path = output_dataset_filepath(simulation) + assert read_output_stamp(path) == { + "packages": installed_output_packages(), + "revision": OUTPUT_STAMP_REVISION, + "year_file": None, + } + assert _ensured(built, run_id)[1] == 0 + + # The same id over a year file in that folder: the output is not its. + dataset = _ensure(_new_source(tmp_path), data_folder)[KEY] + assert output_dataset_filepath(_simulation(dataset, run_id)) == path + assert _refusal(dataset, run_id) == _out_of_date( + path, + "it was calculated from data that kept no year-file stamp, and the " + "simulation's dataset keeps one", + ) + assert _ensured(dataset, run_id)[1] == 1 + assert _refusal(built, run_id) == _out_of_date( + path, + "it was calculated from a year file, and the simulation's dataset keeps " + "no year-file stamp", + ) + + +def test_an_output_from_data_built_in_memory_is_not_reused_after_an_upgrade( + tmp_path, monkeypatch +): + built = PolicyEngineUKDataset( + name="observed-2026", + description="tables taken as observed in 2026", + filepath=str(tmp_path / "built.h5"), + year=YEAR, + data=_with_weights(_tables()), + ) + run_id = _run_id() + simulation, _ = _ensured(built, run_id) + path = output_dataset_filepath(simulation) + calculated_with = installed_output_packages()["policyengine-uk"] + + _upgrade(monkeypatch, "policyengine-uk") + + assert _refusal(built, run_id) == _out_of_date( + path, + f"it was calculated with policyengine-uk {calculated_with}, and " + "policyengine-uk 999.0.0 is installed", + ) + assert _ensured(built, run_id)[1] == 1 + assert _ensured(built, run_id)[1] == 0 + + +# --- Year files cut again ------------------------------------------------ + + +def test_an_output_is_reused_while_its_year_file_is_current(tmp_path): + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + run_id = _run_id() + + first, runs = _ensured(_ensure(source, data_folder)[KEY], run_id) + assert runs == 1 + + # Another process: the year file is loaded, and the output with it. + second, runs = _ensured(_assert_reused(source, data_folder)[KEY], run_id) + assert runs == 0 + _assert_same_output(first, second) + _assert_matches_direct(second, _direct(source), YEAR) + + +def test_an_output_is_not_reused_after_its_local_source_is_edited(tmp_path): + """The smoke-test workflow: rewrite a local dataset in place.""" + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + run_id = _run_id() + first_dataset = _ensure(source, data_folder)[KEY] + old = requested_dataset_source_identity("uk", source) + first, runs = _ensured(first_dataset, run_id) + assert runs == 1 + path = output_dataset_filepath(first) + + _edit(source) + new = requested_dataset_source_identity("uk", source) + + # Until the year file is cut again, a simulation over it holds the tables + # the output was calculated from. ``load_datasets`` refuses the year file. + year_file = data_folder / f"tiny_uk_2024_year_{YEAR}.h5" + assert _refusal(_stored(year_file), run_id) is None + with pytest.raises(ValueError, match="is out of date"): + load_datasets(datasets=[source], years=[YEAR], data_folder=str(data_folder)) + + second_dataset = _assert_cut_again(source, data_folder)[KEY] + assert _refusal(second_dataset, run_id) == _out_of_date( + path, + "it was calculated from a year file cut from " + f"{describe_dataset_source_identity(old)}, and the simulation's dataset " + f"was cut from {describe_dataset_source_identity(new)}", + ) + # The dataset loaded before still holds the old cut's tables, and the + # output is still what a run of it would give. + assert _refusal(first_dataset, run_id) is None + + second, runs = _ensured(second_dataset, run_id) + assert runs == 1 + assert read_output_stamp(path) == output_stamp(read_year_file_stamp(year_file)) + assert read_output_stamp(path)["year_file"]["source"] == new + # Calculated from the rewritten source, not the one the old output held. + _assert_matches_direct(second, _direct(source), YEAR) + assert list(_frames(second)["person"]["state_pension"]) != list( + _frames(first)["person"]["state_pension"] + ) + + third, runs = _ensured(_assert_reused(source, data_folder)[KEY], run_id) + assert runs == 0 + _assert_same_output(second, third) + + +def test_an_output_is_not_reused_after_the_bundle_re_pins_its_dataset(tmp_path): + """A release of a UK dataset in miniature: the bundle points a logical + name at a new revision and sha256, the year files are cut again under + their old names, and outputs calculated from the old release are not + reused, although they sit beside those year files under the same ids.""" + data_folder = tmp_path / "data" + data_folder.mkdir() + bundled = data_folder / f"{BUNDLE_NAME}.h5" + _write_source(bundled) + old_sha256 = sha256_file(bundled) + key = f"{BUNDLE_NAME}_{YEAR}" + run_id = _run_id() + + first, second, third = _pinned("release-1", old_sha256) + with first, second, third: + ensured = ensure_datasets( + datasets=[BUNDLE_NAME], years=[YEAR], data_folder=str(data_folder) + ) + before, runs = _ensured(ensured[key], run_id) + assert runs == 1 + path = output_dataset_filepath(before) + reloaded = load_datasets( + datasets=[BUNDLE_NAME], years=[YEAR], data_folder=str(data_folder) + ) + assert _ensured(reloaded[key], run_id)[1] == 0 + + # The new release reports other State Pension amounts; its verified copy + # replaces the old one, as a download would. + _write_source(bundled, _replacement_tables()) + new_sha256 = sha256_file(bundled) + assert new_sha256 != old_sha256 + + first, second, third = _pinned("release-2", new_sha256) + with first, second, third: + recut = ensure_datasets( + datasets=[BUNDLE_NAME], years=[YEAR], data_folder=str(data_folder) + ) + assert _refusal(recut[key], run_id) == _out_of_date( + path, + f"it was calculated from a year file cut from {BUNDLE_URI}@release-1 " + f"(sha256 {old_sha256}), and the simulation's dataset was cut from " + f"{BUNDLE_URI}@release-2 (sha256 {new_sha256})", + ) + after, runs = _ensured(recut[key], run_id) + assert runs == 1 + _assert_matches_direct(after, _direct(str(bundled)), YEAR) + reloaded = load_datasets( + datasets=[BUNDLE_NAME], years=[YEAR], data_folder=str(data_folder) + ) + again, runs = _ensured(reloaded[key], run_id) + assert runs == 0 + _assert_same_output(after, again) + assert read_output_stamp(path)["year_file"]["source"] == { + "uri": f"{BUNDLE_URI}@release-2", + "sha256": new_sha256, + } + + +@pytest.mark.parametrize("package", ["policyengine-uk", "policyengine-core"]) +def test_an_output_is_not_reused_after_a_package_upgrade( + tmp_path, monkeypatch, package +): + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + run_id = _run_id() + dataset = _ensure(source, data_folder)[KEY] + first, runs = _ensured(dataset, run_id) + assert runs == 1 + path = output_dataset_filepath(first) + before = installed_output_packages()[package] + + _upgrade(monkeypatch, package) + + # Before the year file is cut again: the versions that calculated the + # output are not the ones installed. + year_file = data_folder / f"tiny_uk_2024_year_{YEAR}.h5" + stale = _out_of_date( + path, + f"it was calculated with {package} {before}, and {package} 999.0.0 is " + "installed", + ) + assert _refusal(dataset, run_id) == stale + assert _refusal(_stored(year_file), run_id) == stale + # A simulation over the old cut, opened directly, is run again once with + # the new versions and then reused: it is what a run of that file gives. + assert _ensured(_stored(year_file), run_id)[1] == 1 + assert _refusal(_stored(year_file), run_id) is None + recorded = read_output_stamp(path) + assert recorded["packages"][package] == "999.0.0" + assert recorded["year_file"]["packages"][package] == before + + # ``ensure_datasets`` cuts the year file again, and the output is not the + # new cut's. + recut = _assert_cut_again(source, data_folder)[KEY] + assert _refusal(recut, run_id) == _out_of_date( + path, + f"it was calculated from a year file cut with {package} {before}, and " + f"the simulation's dataset was cut with {package} 999.0.0", + ) + assert _ensured(recut, run_id)[1] == 1 + assert read_output_stamp(path)["year_file"]["packages"][package] == "999.0.0" + assert _ensured(_assert_reused(source, data_folder)[KEY], run_id)[1] == 0 + + +def test_an_output_is_not_reused_when_the_versions_that_cut_its_year_file_return( + tmp_path, monkeypatch +): + """A year file cut under one policyengine-uk is opened directly and run + under another. With the first installed again, the output's year-file + stamp names the installed versions, but the output was not calculated + with them.""" + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + run_id = _run_id() + _ensure(source, data_folder) + year_file = data_folder / f"tiny_uk_2024_year_{YEAR}.h5" + cut_with = installed_package_versions()["policyengine-uk"] + + with monkeypatch.context() as upgraded: + _upgrade(upgraded, "policyengine-uk") + simulation, runs = _ensured(_stored(year_file), run_id) + assert runs == 1 + path = output_dataset_filepath(simulation) + recorded = read_output_stamp(path) + assert recorded["packages"]["policyengine-uk"] == "999.0.0" + assert recorded["year_file"]["packages"]["policyengine-uk"] == cut_with + assert _refusal(_stored(year_file), run_id) is None + + # The year file is current again, and loads. + dataset = _assert_reused(source, data_folder)[KEY] + stale = _out_of_date( + path, + "it was calculated with policyengine-uk 999.0.0, and policyengine-uk " + f"{cut_with} is installed", + ) + assert _refusal(dataset, run_id) == stale + assert _refusal(_stored(year_file), run_id) == stale + assert _ensured(dataset, run_id)[1] == 1 + assert _ensured(dataset, run_id)[1] == 0 + + +def test_an_output_is_not_reused_for_another_years_file(tmp_path): + source = _new_source(tmp_path) + datasets = _ensure(source, tmp_path / "data", years=(YEAR, 2028)) + run_id = _run_id() + first, runs = _ensured(datasets[KEY], run_id) + assert runs == 1 + path = output_dataset_filepath(first) + + # One id, the same folder, another year's file: the same output path. + later = datasets["tiny_uk_2024_2028"] + assert output_dataset_filepath(_simulation(later, run_id)) == path + assert _refusal(later, run_id) == _out_of_date( + path, + f"it was calculated from the year file for {YEAR}, and the simulation's " + "dataset is the one for 2028", + ) + second, runs = _ensured(later, run_id) + assert runs == 1 + _assert_matches_direct(second, _direct(source), 2028) + + +def test_raising_the_year_file_revision_makes_saved_outputs_out_of_date( + tmp_path, monkeypatch +): + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + run_id = _run_id() + dataset = _ensure(source, data_folder)[KEY] + first, _ = _ensured(dataset, run_id) + path = output_dataset_filepath(first) + + monkeypatch.setattr( + uk_datasets, "YEAR_FILE_STAMP_REVISION", YEAR_FILE_STAMP_REVISION + 1 + ) + + # The dataset still holds the tables the output was calculated from. + assert _refusal(dataset, run_id) is None + # ``ensure_datasets`` cuts the year files again under the new revision. + recut = _assert_cut_again(source, data_folder)[KEY] + assert _refusal(recut, run_id) == _out_of_date( + path, + "it was calculated from a year file of year-file format revision " + f"{YEAR_FILE_STAMP_REVISION}, and the simulation's dataset is of " + f"revision {YEAR_FILE_STAMP_REVISION + 1}", + ) + assert _ensured(recut, run_id)[1] == 1 + assert _ensured(_assert_reused(source, data_folder)[KEY], run_id)[1] == 0 + + +# --- Outputs saved before the stamp, or not completely ------------------- + + +def test_outputs_saved_by_an_earlier_release_are_calculated_again_once(tmp_path): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + run_id = _run_id() + first, _ = _ensured(dataset, run_id) + path = output_dataset_filepath(first) + # As the release before wrote it: tables and data year, no stamp. + with h5py.File(path, "a") as file: + del file[OUTPUT_STAMP_H5_DATASET] + + assert _refusal(dataset, run_id) == _out_of_date(path, UNSTAMPED_OUTPUT) + second, runs = _ensured(dataset, run_id) + assert runs == 1 + _assert_same_output(first, second) + assert _ensured(dataset, run_id)[1] == 0 + + +def test_saving_an_output_that_records_no_stamp_is_refused(tmp_path): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + simulation = _simulation(dataset, _run_id()) + simulation.run() + del simulation.output_dataset.metadata[OUTPUT_STAMP_KEY] + + with pytest.raises(ValueError, match="does not record what it was calculated"): + simulation.save() + # Refused before anything is written. + assert not output_dataset_filepath(simulation).exists() + assert _partial_files(tmp_path / "data") == [] + + +@pytest.mark.parametrize("stop", ["in the tables", "at the stamp"]) +def test_a_save_that_fails_leaves_the_saved_output_whole_and_current( + tmp_path, monkeypatch, stop +): + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + dataset = _ensure(source, data_folder)[KEY] + run_id = _run_id() + simulation, _ = _ensured(dataset, run_id) + path = output_dataset_filepath(simulation) + before = path.read_bytes() + + def stopped(*args, **kwargs): + raise RuntimeError("stopped part-way") + + with monkeypatch.context() as patcher: + if stop == "in the tables": + put = pd.HDFStore.put + + def put_until_household(self, key, value, *args, **kwargs): + if key == "household": + raise RuntimeError("stopped part-way") + return put(self, key, value, *args, **kwargs) + + patcher.setattr(pd.HDFStore, "put", put_until_household) + else: + patcher.setattr(uk_datasets, "publish_year_file", stopped) + with pytest.raises(RuntimeError, match="stopped part-way"): + simulation.save() + + # The save wrote beside the path, so the output there is untouched. + assert path.read_bytes() == before + assert _partial_files(data_folder) == [] + assert _refusal(dataset, run_id) is None + assert _ensured(dataset, run_id)[1] == 0 + + +def test_a_first_save_that_fails_leaves_no_output_and_the_next_ensure_runs( + tmp_path, monkeypatch +): + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + dataset = _ensure(source, data_folder)[KEY] + run_id = _run_id() + simulation = _simulation(dataset, run_id) + simulation.run() + path = output_dataset_filepath(simulation) + + def stopped(*args, **kwargs): + raise RuntimeError("stopped before the stamp") + + with monkeypatch.context() as patcher: + patcher.setattr(uk_datasets, "publish_year_file", stopped) + with pytest.raises(RuntimeError, match="stopped before the stamp"): + simulation.save() + + assert not path.exists() + assert _partial_files(data_folder) == [] + with pytest.raises(FileNotFoundError): + _simulation(dataset, run_id).load() + assert _ensured(dataset, run_id)[1] == 1 + assert _refusal(dataset, run_id) is None + + +def test_a_missing_output_is_calculated_without_a_refusal(tmp_path): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + run_id = _run_id() + + with pytest.raises(FileNotFoundError): + _simulation(dataset, run_id).load() + assert _ensured(dataset, run_id)[1] == 1 + + +# --- Another process saving to the same path ------------------------------ + + +def test_an_output_replaced_while_it_is_loaded_is_checked_again(tmp_path, monkeypatch): + """The output is checked, and then, before its tables are read, another + process moves the output of another cut onto the path. The tables read + are that other output's, so the check is made again, and refuses it.""" + source = _new_source(tmp_path) + data_folder = tmp_path / "data" + run_id = _run_id() + first_dataset = _ensure(source, data_folder)[KEY] + old = requested_dataset_source_identity("uk", source) + first, _ = _ensured(first_dataset, run_id) + path = output_dataset_filepath(first) + ours = tmp_path / "ours.h5" + shutil.copy(path, ours) + + # Another process's output: the same id over the year file cut again. + _edit(source) + new = requested_dataset_source_identity("uk", source) + second_dataset = _assert_cut_again(source, data_folder)[KEY] + second, runs = _ensured(second_dataset, run_id) + assert runs == 1 + theirs = tmp_path / "theirs.h5" + shutil.copy(path, theirs) + os.replace(ours, path) + assert _refusal(first_dataset, run_id) is None + + check = uk_datasets.output_stamp_problem + checks = [] + + def check_then_replace(*args, **kwargs): + problem = check(*args, **kwargs) + checks.append(problem) + if len(checks) == 1: + os.replace(theirs, path) + return problem + + monkeypatch.setattr(uk_datasets, "output_stamp_problem", check_then_replace) + refusal = _refusal(first_dataset, run_id) + + mismatch = ( + "it was calculated from a year file cut from " + f"{describe_dataset_source_identity(new)}, and the simulation's dataset " + f"was cut from {describe_dataset_source_identity(old)}" + ) + assert checks == [None, mismatch] + assert refusal == _out_of_date(path, mismatch) + + +# --- Ensure agrees with load --------------------------------------------- + + +def _stand_in_run(self, simulation): + """What the engine's run records, without the engine. + + A fixed output stamped as ``PolicyEngineUKLatest.run`` stamps one: with + the installed versions and the year-file stamp the simulation's dataset + keeps. + """ + simulation.output_dataset = PolicyEngineUKDataset( + id=simulation.id, + name=simulation.dataset.name, + description=simulation.dataset.description, + filepath=str(output_dataset_filepath(simulation)), + year=simulation.dataset.year, + is_output_dataset=True, + data_year=DATA_YEAR, + data=_with_weights(_tables()), + metadata={ + OUTPUT_STAMP_KEY: output_stamp(input_year_file_stamp(simulation.dataset)) + }, + ) + + +def _dataset_stamps() -> dict[str, object]: + """Year-file stamps a simulation's dataset may keep, by name.""" + current = year_file_stamp({"uri": "hf://fixture/repo/tiny.h5"}, YEAR) + packages = installed_package_versions() + return { + "none": None, + "current": current, + "another source": { + **current, + "source": {"uri": "hf://fixture/repo/tiny.h5@next"}, + }, + "another year": {**current, "year": YEAR + 1}, + "another policyengine-uk": { + **current, + "packages": {**packages, "policyengine-uk": "0.0.1"}, + }, + "another revision": {**current, "revision": YEAR_FILE_STAMP_REVISION + 1}, + "unreadable": UNREADABLE_YEAR_FILE_STAMP, + "malformed": {**current, "year": str(YEAR)}, + } + + +DATASET_STAMPS = st.sampled_from(sorted(_dataset_stamps())) +#: The stamps of :func:`_dataset_stamps` that are none or well formed, and so +#: name the tables a run reads. Whether a year file is the one for the source +#: and year requested, cut with the installed versions, is ``load_datasets``' +#: check, not the output's. +NAMES_ITS_TABLES = ( + "none", + "current", + "another source", + "another year", + "another policyengine-uk", + "another revision", +) +#: What is at the output's path: an output saved by a run over a dataset +#: keeping the named stamp, or one of these. +SAVED = st.one_of( + DATASET_STAMPS, + st.sampled_from( + [ + "missing", + "no stamp", + "no data year", + "another output revision", + "other versions", + "malformed stamp", + "not hdf5", + ] + ), +) + + +def _stand_in_dataset(directory: Path, stamp) -> PolicyEngineUKDataset: + return PolicyEngineUKDataset( + name="stand-in", + description="A dataset keeping a given year-file stamp", + # Never written: it places the output in ``directory``. + filepath=str(directory / "input.h5"), + year=YEAR, + data=_with_weights(_tables()), + metadata={} if stamp is None else {YEAR_FILE_STAMP_KEY: stamp}, + ) + + +def _save_as(path: Path, saved: str, directory: Path, run_id: str) -> None: + """Put at ``path`` what ``saved`` names (see :data:`SAVED`).""" + stamps = _dataset_stamps() + if saved == "missing": + return + if saved == "not hdf5": + path.write_text("not an H5 file") + return + simulation = _simulation( + _stand_in_dataset(directory, stamps.get(saved, stamps["current"])), run_id + ) + _stand_in_run(None, simulation) + if saved == "no data year": + simulation.output_dataset.data_year = None + simulation.save() + recorded = read_output_stamp(path) + if saved == "no stamp" or saved == "no data year": + with h5py.File(path, "a") as file: + del file[OUTPUT_STAMP_H5_DATASET] + elif saved == "another output revision": + write_output_stamp(path, {**recorded, "revision": OUTPUT_STAMP_REVISION + 1}) + elif saved == "other versions": + write_output_stamp( + path, + { + **recorded, + "packages": {**recorded["packages"], "policyengine-uk": "0.0.1"}, + }, + ) + elif saved == "malformed stamp": + write_output_stamp(path, {"revision": OUTPUT_STAMP_REVISION}) + + +@settings(max_examples=80, deadline=None) +@given(saved=SAVED, kept=DATASET_STAMPS) +def test_ensure_runs_exactly_when_load_would_refuse(saved, kept): + with tempfile.TemporaryDirectory() as directory: + _check_ensure_agrees_with_load(Path(directory), saved, kept) + + +def _check_ensure_agrees_with_load(directory: Path, saved: str, kept: str) -> None: + stamps = _dataset_stamps() + dataset_stamp = stamps[kept] + run_id = _run_id() + dataset = _stand_in_dataset(directory, dataset_stamp) + path = output_dataset_filepath(_simulation(dataset, run_id)) + _save_as(path, saved, directory, run_id) + + missing = saved == "missing" + problem = None if missing else output_stamp_problem(path, dataset_stamp) + # The check is the one the engine-free tests state: current exactly when + # the output was saved by a run over the stamp the dataset keeps, and + # that stamp is none or well formed. + assert (problem is None and not missing) == ( + saved == kept and kept in NAMES_ITS_TABLES + ) + + if missing: + with pytest.raises(FileNotFoundError): + _simulation(dataset, run_id).load() + else: + refusal = _refusal(dataset, run_id) + assert refusal == (None if problem is None else _out_of_date(path, problem)) + + simulation, runs = _ensured(dataset, run_id, run=_stand_in_run) + assert runs == (1 if missing or problem is not None else 0) + assert simulation.output_dataset is not None + assert simulation.output_dataset.metadata[OUTPUT_STAMP_KEY] == output_stamp( + dataset_stamp + ) + assert read_output_stamp(path) == output_stamp(dataset_stamp) + + # What ensure leaves is current exactly when the dataset keeps no stamp + # or a well-formed one: a stamp that is unreadable or malformed names no + # tables, so an output calculated from such a year file is calculated + # again every time. + current = kept in NAMES_ITS_TABLES + assert (output_stamp_problem(path, dataset_stamp) is None) == current + assert _ensured(dataset, run_id, run=_stand_in_run)[1] == (0 if current else 1) + + +def test_the_refusal_names_the_output_and_why(tmp_path): + source = _new_source(tmp_path) + dataset = _ensure(source, tmp_path / "data")[KEY] + run_id = _run_id() + first, _ = _ensured(dataset, run_id) + path = output_dataset_filepath(first) + write_output_stamp(path, {"revision": OUTPUT_STAMP_REVISION}) + + with pytest.raises( + ValueError, + match=re.escape( + f"Saved UK simulation output {path} is out of date: " + f"{MALFORMED_OUTPUT_STAMP}. Run the simulation again " + "(Simulation.ensure() does so)." + ), + ): + _simulation(dataset, run_id).load()