diff --git a/docs/CHANGELOG-SSA.md b/docs/CHANGELOG-SSA.md index 83d7412..138156f 100644 --- a/docs/CHANGELOG-SSA.md +++ b/docs/CHANGELOG-SSA.md @@ -3,6 +3,31 @@ Scoped changelog for the Site Safety Audit pipeline and its quality guards. Newest at top. +## 2026-05-15 + +### Added + +- **Full drift checker.** Two new checks beyond the minimal v1 CSV + header check: + - Every `PH_*` placeholder constant in `ssa_pipeline.py` must appear + as literal text in the SSA report template `.docx` (concatenated + `` text across all `word/*.xml` parts, so tokens split across + formatting boundaries are still caught). + - Every `STAGING_HEADERS` entry must have a corresponding + `_STAGING_COL_WIDTHS` key. Subset semantics, not equality — + `_STAGING_COL_WIDTHS` legitimately carries forward-defined + gap-4/5/6 columns (phase, activity_ref, hrcw, swms_required, etc.) + that aren't yet in `STAGING_HEADERS`. + +### Changed + +- **Placeholder tokens centralised.** `build_ssa_report_docx` no longer + carries inline `{{SITE_ADDRESS}}` / `{{NARRATIVE_SUMMARY}}` / + `{{AUDIT_DATE}}` / `{{PREPARED_BY}}` strings in its `body_replacements` + and `footer_replacements` dicts. They're now module-level + `PH_*` constants plus an `SSA_REPORT_PLACEHOLDERS` tuple, which the + drift checker walks. Pure refactor — no behaviour change. + ## 2026-05-14 ### Added diff --git a/pims/services/ssa_pipeline.py b/pims/services/ssa_pipeline.py index d114e5e..787b7f3 100644 --- a/pims/services/ssa_pipeline.py +++ b/pims/services/ssa_pipeline.py @@ -77,6 +77,22 @@ def flag(self, reason: str) -> None: _REQUIRED_HEADER = ("timestamp", "observation", "filename") _TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}$") +# Placeholder tokens substituted by build_ssa_report_docx. Centralised here so +# the drift checker (tools/check_ssa_drift.py) can verify each token exists as +# literal text in the SSA report template. Body and footer dicts both use the +# same set of tokens; only the source value differs (long-form date in body +# vs short-form date in footer for AUDIT_DATE). +PH_SITE_ADDRESS = "{{SITE_ADDRESS}}" +PH_NARRATIVE_SUMMARY = "{{NARRATIVE_SUMMARY}}" +PH_AUDIT_DATE = "{{AUDIT_DATE}}" +PH_PREPARED_BY = "{{PREPARED_BY}}" +SSA_REPORT_PLACEHOLDERS: tuple[str, ...] = ( + PH_SITE_ADDRESS, + PH_NARRATIVE_SUMMARY, + PH_AUDIT_DATE, + PH_PREPARED_BY, +) + def _decode_csv_bytes(raw: bytes) -> str: """Detect UTF-8 / UTF-8-SIG / CP1252 and return a unicode string. @@ -2681,16 +2697,16 @@ def build_ssa_report_docx( audit_date_ordinal = _to_ordinal_date(audit_date_ddmmyyyy) audit_date_short = _to_short_date(audit_date_ddmmyyyy) body_replacements = { - "{{SITE_ADDRESS}}": full_site, - "{{NARRATIVE_SUMMARY}}": combined_narrative, - "{{AUDIT_DATE}}": audit_date_ordinal or audit_date_ddmmyyyy or "", - "{{PREPARED_BY}}": prepared_by or "", + PH_SITE_ADDRESS: full_site, + PH_NARRATIVE_SUMMARY: combined_narrative, + PH_AUDIT_DATE: audit_date_ordinal or audit_date_ddmmyyyy or "", + PH_PREPARED_BY: prepared_by or "", } footer_replacements = { - "{{SITE_ADDRESS}}": full_site, - "{{NARRATIVE_SUMMARY}}": combined_narrative, - "{{AUDIT_DATE}}": audit_date_short or audit_date_ddmmyyyy or "", - "{{PREPARED_BY}}": prepared_by or "", + PH_SITE_ADDRESS: full_site, + PH_NARRATIVE_SUMMARY: combined_narrative, + PH_AUDIT_DATE: audit_date_short or audit_date_ddmmyyyy or "", + PH_PREPARED_BY: prepared_by or "", } _replace_tokens_in_part(doc.part, body_replacements) # Split the combined narrative into the canonical two-paragraph diff --git a/tests/test_ssa_drift_checker.py b/tests/test_ssa_drift_checker.py index d491156..2c999a5 100644 --- a/tests/test_ssa_drift_checker.py +++ b/tests/test_ssa_drift_checker.py @@ -5,28 +5,80 @@ from pathlib import Path -def test_drift_checker_clean_state_returns_zero(capsys, monkeypatch): - """Current main passes — no drift between paired constants.""" +def _import_checker(): + """Import (or re-import) the drift checker module from tools/.""" sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools")) import check_ssa_drift # noqa: E402 - rc = check_ssa_drift.main() + return check_ssa_drift + + +def test_drift_checker_clean_state_returns_zero(capsys): + """Current main passes — no drift across any of the wired checks.""" + drift = _import_checker() + rc = drift.main() out = capsys.readouterr().out - assert rc == 0 + assert rc == 0, out assert "[X]" in out assert "no drift" in out.lower() def test_drift_checker_flags_csv_header_mismatch(capsys, monkeypatch): """Forcing the preflight constant to diverge from ssa_pipeline's - must trip the drift check.""" - sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools")) - import check_ssa_drift # noqa: E402 + must trip the CSV header check.""" + drift = _import_checker() monkeypatch.setattr( - check_ssa_drift, "_EXPECTED_CSV_HEADERS", + drift, "_EXPECTED_CSV_HEADERS", ("timestamp", "observation", "DIVERGED"), ) - rc = check_ssa_drift.main() + rc = drift.main() out = capsys.readouterr().out assert rc == 1 assert "DRIFT" in out assert "DIVERGED" in out + + +def test_drift_checker_flags_missing_placeholder_in_template( + capsys, monkeypatch, +): + """Adding a fake PH_* that doesn't exist in the template must trip + the placeholder check.""" + drift = _import_checker() + monkeypatch.setattr( + drift, "SSA_REPORT_PLACEHOLDERS", + ("{{SITE_ADDRESS}}", "{{NEVER_IN_TEMPLATE}}"), + ) + rc = drift.main() + out = capsys.readouterr().out + assert rc == 1 + assert "DRIFT" in out + assert "NEVER_IN_TEMPLATE" in out + + +def test_drift_checker_flags_staging_header_without_width( + capsys, monkeypatch, +): + """A STAGING_HEADERS entry with no corresponding width entry must + trip the staging-width check.""" + drift = _import_checker() + monkeypatch.setattr( + drift, "STAGING_HEADERS", + drift.STAGING_HEADERS + ("phantom_column",), + ) + rc = drift.main() + out = capsys.readouterr().out + assert rc == 1 + assert "DRIFT" in out + assert "phantom_column" in out + + +def test_drift_checker_tolerates_extra_width_entries(capsys): + """_STAGING_COL_WIDTHS legitimately carries forward-defined gap-4/5/6 + columns that aren't yet in STAGING_HEADERS. The subset semantics must + NOT flag this as drift.""" + drift = _import_checker() + rc = drift.main() + out = capsys.readouterr().out + # Today's main has gap-4/5/6 width entries (phase, hrcw, swms_*) + # absent from STAGING_HEADERS. If this test fails, the check + # tightened to equality and is now too strict. + assert rc == 0, out diff --git a/tools/check_ssa_drift.py b/tools/check_ssa_drift.py index 4fa5a6b..68367d2 100644 --- a/tools/check_ssa_drift.py +++ b/tools/check_ssa_drift.py @@ -4,15 +4,27 @@ python tools/check_ssa_drift.py -Drift is the silent divergence between paired constants in different -files. Today this checker covers one case — the Evidence_Master.csv -header tuple, which is duplicated in: - - - pims/services/ssa_pipeline.py ``_REQUIRED_HEADER`` - - pims/services/ssa_quality/preflight.py ``_EXPECTED_CSV_HEADERS`` - -Both must stay aligned; the preflight check would silently start -accepting CSVs the pipeline can't parse if they drift. +Drift is the silent divergence between paired facts in different files. +A constant in one module, a literal in a template, a header tuple in a +docs file — they all encode the same truth, and if one moves while the +others don't, the build keeps working but produces wrong output. Word's +recovery prompt tolerates it; downstream consumers do not. + +Checks today: + + 1. ``Evidence_Master.csv`` header tuple — duplicated in + ``ssa_pipeline._REQUIRED_HEADER`` and + ``preflight._EXPECTED_CSV_HEADERS``. Must stay aligned or preflight + silently accepts CSVs the pipeline can't parse. + 2. SSA report template placeholders — every ``PH_*`` constant in + ``ssa_pipeline.py`` must appear as literal text somewhere in the + report template ``.docx``. Catches "renamed PH_FOO in code but + forgot the template" (or vice versa). + 3. Staging xlsx header widths — every entry in + ``STAGING_HEADERS`` must have a corresponding entry in + ``_STAGING_COL_WIDTHS``. Subset (not equality): extra width entries + are tolerated (gap-4/5/6 forward-defined columns); missing entries + are not. Future checks can be added by appending to ``_CHECKS`` — each is a ``(label, callable)`` returning ``None`` on pass or a string detail on @@ -20,19 +32,34 @@ """ from __future__ import annotations +import re import sys +import zipfile from pathlib import Path +from typing import Callable, Optional # Add repo root to sys.path so this script can be run from anywhere. _REPO_ROOT = Path(__file__).resolve().parent.parent if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) -from pims.services.ssa_pipeline import _REQUIRED_HEADER +from pims.services.ssa_pipeline import ( + _REQUIRED_HEADER, + _STAGING_COL_WIDTHS, + SSA_REPORT_PLACEHOLDERS, + SSA_REPORT_TEMPLATE, + STAGING_HEADERS, +) from pims.services.ssa_quality.preflight import _EXPECTED_CSV_HEADERS -def _check_csv_headers() -> str | None: +# Matches ``text`` runs so we can extract Word's visible +# text content from word/*.xml parts (placeholders may be split across +# multiple runs when formatting boundaries land mid-token). +_W_T_RE = re.compile(r"]*>([^<]*)") + + +def _check_csv_headers() -> Optional[str]: if _REQUIRED_HEADER == _EXPECTED_CSV_HEADERS: return None return ( @@ -41,9 +68,49 @@ def _check_csv_headers() -> str | None: ) -_CHECKS: tuple[tuple[str, callable], ...] = ( +def _check_template_placeholders() -> Optional[str]: + """Every PH_* must appear as literal text in the report template.""" + if not SSA_REPORT_TEMPLATE.is_file(): + return f"template missing: {SSA_REPORT_TEMPLATE}" + # Concatenate text across every word/*.xml part — covers + # body, headers, footers, cover-page text boxes. Tag-strip via + # findall then join, which collapses split-across-runs tokens. + parts: list[str] = [] + with zipfile.ZipFile(SSA_REPORT_TEMPLATE) as z: + for name in z.namelist(): + if name.startswith("word/") and name.endswith(".xml"): + xml = z.read(name).decode("utf-8", errors="ignore") + parts.extend(_W_T_RE.findall(xml)) + haystack = "".join(parts) + missing = [ph for ph in SSA_REPORT_PLACEHOLDERS if ph not in haystack] + if missing: + return f"placeholder(s) not found in template: {missing}" + return None + + +def _check_staging_header_widths() -> Optional[str]: + """Every STAGING_HEADERS entry must have an _STAGING_COL_WIDTHS key. + + Subset check, not equality — _STAGING_COL_WIDTHS legitimately + carries forward-defined gap-4/5/6 columns (phase, activity_ref, + hrcw, swms_required, etc.) that aren't yet in STAGING_HEADERS. + """ + missing = [h for h in STAGING_HEADERS if h not in _STAGING_COL_WIDTHS] + if missing: + return ( + f"STAGING_HEADERS entries missing _STAGING_COL_WIDTHS entry: " + f"{missing}" + ) + return None + + +_CHECKS: tuple[tuple[str, Callable[[], Optional[str]]], ...] = ( ("Evidence_Master.csv header tuple matches across modules", _check_csv_headers), + ("SSA report template carries every PH_* placeholder", + _check_template_placeholders), + ("Every STAGING_HEADERS entry has a width definition", + _check_staging_header_widths), )