diff --git a/reports/release_validation_report.json b/reports/release_validation_report.json index e5c8354..b4e83bb 100644 --- a/reports/release_validation_report.json +++ b/reports/release_validation_report.json @@ -140,7 +140,7 @@ ], "valid": true }, - "runtime_seconds": 0.787532, + "runtime_seconds": 0.826618, "selected_calibration_method": "uncalibrated", "selected_experiment": "exp_05_random_forest", "source_snapshot_id": "edbd4bd813ed8e1dbaba9e1c", diff --git a/run_baseline.py b/run_baseline.py index 3730987..92cea78 100644 --- a/run_baseline.py +++ b/run_baseline.py @@ -8,6 +8,7 @@ from src.baseline import DEFAULT_ALPHA, DEFAULT_MIN_GROUP_SIZE, build_industry_rates, choose_naics_grouping, score_validation_rows from src.data_foundation import FoundationError, read_json, sha256_file, write_csv, write_json +from src.path_utils import RelativePathError, resolve_report_path from src.ranking_metrics import validation_metrics from src.splitting import SplitError, create_chronological_split, write_split_artifacts from src.validation import load_schema, validate_output_columns @@ -25,9 +26,12 @@ def read_csv(path: Path) -> list[dict[str, Any]]: raise BaselineError(f"Could not read labelled snapshot: {error}") from error -def verify_processed_snapshot(foundation: dict[str, Any]) -> tuple[Path, list[dict[str, Any]]]: +def verify_processed_snapshot(foundation: dict[str, Any], base_directory: Path) -> tuple[Path, list[dict[str, Any]]]: output = foundation.get("processed_output", {}) - directory = Path(output.get("path", "")) + try: + directory = resolve_report_path(output.get("path", ""), base_directory) + except RelativePathError as error: + raise BaselineError(str(error)) from error labelled_path = directory / "labelled_inspections.csv" expected = output.get("hashes", {}).get("labelled_inspections.csv") if not directory.exists() or not expected or not labelled_path.exists() or sha256_file(labelled_path) != expected: @@ -41,7 +45,7 @@ def verify_processed_snapshot(foundation: dict[str, Any]) -> tuple[Path, list[di def run_baseline(*, foundation_report_path: Path, schema_path: Path, artifact_root: Path | None = None) -> dict[str, Any]: foundation = read_json(foundation_report_path) schema = load_schema(schema_path) - processed_directory, rows = verify_processed_snapshot(foundation) + processed_directory, rows = verify_processed_snapshot(foundation, foundation_report_path.parent.parent) validate_output_columns(list(rows[0]), schema) try: split = create_chronological_split(rows) diff --git a/run_feature_engineering.py b/run_feature_engineering.py index 18cad7f..2af7501 100644 --- a/run_feature_engineering.py +++ b/run_feature_engineering.py @@ -27,6 +27,7 @@ transform_batch, write_feature_artifacts, ) +from src.path_utils import RelativePathError, resolve_report_path def read_json(path: Path) -> dict[str, Any]: @@ -59,9 +60,12 @@ def write_json_atomic(path: Path, value: dict[str, Any]) -> None: temporary.replace(path) -def load_verified_splits(baseline: dict[str, Any]) -> tuple[Path, dict[str, list[dict[str, Any]]], dict[str, Any]]: +def load_verified_splits(baseline: dict[str, Any], base_directory: Path) -> tuple[Path, dict[str, list[dict[str, Any]]], dict[str, Any]]: artifact = baseline.get("split_artifacts", {}) - directory = Path(artifact.get("directory", "")) + try: + directory = resolve_report_path(artifact.get("directory", ""), base_directory) + except RelativePathError as error: + raise FeatureEngineeringError(str(error)) from error manifest_path = directory / "split_manifest.json" manifest = read_json(manifest_path) if not manifest.get("strictly_ordered") or manifest.get("id_overlap_count") != 0: @@ -115,7 +119,7 @@ def run_feature_engineering( snapshot_id = foundation.get("snapshot_id") if not snapshot_id or baseline.get("source_snapshot_id") != snapshot_id: raise FeatureEngineeringError("Data-foundation and baseline reports do not identify the same source snapshot.") - split_directory, inputs, split_manifest = load_verified_splits(baseline) + split_directory, inputs, split_manifest = load_verified_splits(baseline, baseline_report_path.parent.parent) all_input_rows = [*inputs["train"], *inputs["validation"], *inputs["test"]] key_assessment = inspect_establishment_key(all_input_rows) employee = employee_count_assessment(all_input_rows, float(config["employee_count"]["minimum_coverage_percentage"])) diff --git a/src/data_foundation.py b/src/data_foundation.py index 0bd3243..ed166f9 100644 --- a/src/data_foundation.py +++ b/src/data_foundation.py @@ -13,6 +13,7 @@ from run_feasibility import NON_POSITIVE_TYPES, POSITIVE_TYPES, build_label_table from src.validation import ValidationError, load_schema, missing_percentages, validate_inspections, validate_output_columns +from src.path_utils import RelativePathError, resolve_report_path LABEL_MAPPING_VERSION = "day0-viol-type-v1" @@ -248,7 +249,11 @@ def run_foundation( if not required_config.issubset(configuration) or not cache_directory: raise FoundationError("Day 0 feasibility report does not identify a usable cache configuration.") schema = load_schema(schema_path) - inspections, violations, completed_ids, sources = load_day0_cache(Path(cache_directory), configuration) + try: + cache_root = resolve_report_path(cache_directory, report_path.parent.parent) + except RelativePathError as error: + raise FoundationError(str(error)) from error + inspections, violations, completed_ids, sources = load_day0_cache(cache_root, configuration) inspection_ids = [str(row["activity_nr"]) for row in inspections if row.get("activity_nr") is not None] snapshot_id = stable_snapshot_id(configuration, inspection_ids, list(completed_ids)) raw_directory, raw_manifest, raw_reused = create_raw_snapshot( diff --git a/src/path_utils.py b/src/path_utils.py new file mode 100644 index 0000000..0ba4168 --- /dev/null +++ b/src/path_utils.py @@ -0,0 +1,30 @@ +"""Portable, deliberately strict resolution of report-relative artifact paths.""" +from __future__ import annotations + +from pathlib import Path, PureWindowsPath + + +class RelativePathError(ValueError): + """A report path is absolute or escapes the declared artifact base.""" + + +def resolve_report_path(value: str | Path, base_directory: Path) -> Path: + """Resolve a slash-agnostic relative report path under ``base_directory``. + + Reports are portable when they store relative artifact paths. Absolute + paths are rejected (including Windows drive paths on Unix), as are parent + traversals that would escape the supplied base directory. + """ + text = str(value).strip().replace("\\", "/") + if not text: + raise RelativePathError("Artifact path is empty.") + normalized = Path(text) + if normalized.is_absolute() or PureWindowsPath(text).is_absolute(): + raise RelativePathError("Absolute artifact paths are not permitted in reports.") + base = base_directory.resolve() + candidate = (base / normalized).resolve() + try: + candidate.relative_to(base) + except ValueError as error: + raise RelativePathError("Artifact path escapes its report base directory.") from error + return candidate diff --git a/src/release_validation.py b/src/release_validation.py index ed5c743..b379dbb 100644 --- a/src/release_validation.py +++ b/src/release_validation.py @@ -19,6 +19,8 @@ import yaml +from src.path_utils import RelativePathError, resolve_report_path + class ReleaseValidationError(RuntimeError): """A release contract check failed.""" @@ -67,9 +69,10 @@ def _require(condition: bool, message: str) -> None: def _path_from_report(root: Path, value: Any, field: str) -> Path: _require(isinstance(value, str) and value, f"Missing report path: {field}") - path = Path(value.replace("\\", "/")) - _require(not path.is_absolute(), f"Absolute artifact path is not permitted: {field}") - return root / path + try: + return resolve_report_path(value, root) + except RelativePathError as error: + raise ReleaseValidationError(f"Invalid artifact path for {field}: {error}") from error def _workflow_checks(root: Path) -> dict[str, bool]: diff --git a/tests/test_baseline.py b/tests/test_baseline.py index 7bd7f74..acff7f9 100644 --- a/tests/test_baseline.py +++ b/tests/test_baseline.py @@ -1,4 +1,5 @@ import copy +import json import tempfile import unittest from pathlib import Path @@ -6,6 +7,7 @@ from run_baseline import run_baseline from src.baseline import build_industry_rates, choose_naics_grouping, score_validation_rows +from src.data_foundation import sha256_file, write_csv from src.ranking_metrics import average_precision, ranking_at_fraction, roc_auc, selection_count from src.splitting import SplitError, create_chronological_split, write_split_artifacts @@ -93,10 +95,23 @@ def test_ranking_metrics_use_ceil_and_handle_missing_class(self): def test_offline_smoke_never_reports_locked_test_metrics(self): with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + processed = root / "data" / "processed" / "snapshot" + rows = [ + row("1", "2020-01-02", 1), row("2", "2020-02-02", 0), + row("3", "2021-01-02", 1), row("4", "2021-02-02", 0), + row("5", "2022-01-02", 1), row("6", "2022-02-02", 0), + row("7", "2023-01-02", 1), row("8", "2023-02-02", 0), + ] + labelled = processed / "labelled_inspections.csv" + write_csv(labelled, rows, list(rows[0])) + report_path = root / "reports" / "data_foundation_report.json" + report_path.parent.mkdir() + report_path.write_text(json.dumps({"snapshot_id": "fixture", "processed_output": {"path": "data\\processed\\snapshot", "hashes": {"labelled_inspections.csv": sha256_file(labelled)}}}), encoding="utf-8") with patch("scripts.dol_api.DOLApiClient.get_records", side_effect=AssertionError("network request attempted")): report = run_baseline( - foundation_report_path=Path("reports/data_foundation_report.json"), schema_path=Path("config/schema.yaml"), - artifact_root=Path(directory), + foundation_report_path=report_path, schema_path=Path("config/schema.yaml"), + artifact_root=root / "baseline", ) self.assertEqual(report["status"], "PASS") self.assertTrue(report["baseline"]["training_only"]) diff --git a/tests/test_calibration.py b/tests/test_calibration.py index 5130f75..aa504e0 100644 --- a/tests/test_calibration.py +++ b/tests/test_calibration.py @@ -1,4 +1,5 @@ import unittest +import tempfile from pathlib import Path import json from src.calibration import select_method @@ -17,11 +18,13 @@ def test_isotonic_wins_when_meaningfully_better(self): def test_recall_tolerance_is_enforced(self): selected,improved,_=select_method({"uncalibrated":metrics(.17,.23),"sigmoid":metrics(.16,.20),"isotonic":metrics(.18,.23)},CFG);self.assertEqual((selected,improved),("uncalibrated",False)) def test_uncalibrated_report_uses_generic_existing_final_artifact(self): - report=json.loads(Path("reports/calibration_report.json").read_text(encoding="utf8")) - self.assertEqual(report["selected_calibration_method"],"uncalibrated") - self.assertEqual(report["final_package_type"],"uncalibrated") - self.assertFalse(report["final_calibration_applied"]) - self.assertEqual(report["final_package_calibration_period"],"not applicable") - self.assertNotIn("final_calibrated_artifact_path",report) - self.assertTrue(Path(report["final_candidate_artifact_path"]).exists()) - self.assertFalse(report["locked_test_labels_accessed"] or report["locked_test_metrics_calculated"] or report["locked_test_predictions_created"]) + with tempfile.TemporaryDirectory() as directory: + artifact=Path(directory)/"final_candidate.joblib";artifact.write_bytes(b"fixture artifact") + report={"selected_calibration_method":"uncalibrated","final_package_type":"uncalibrated","final_calibration_applied":False,"final_package_calibration_period":"not applicable","final_candidate_artifact_path":str(artifact),"locked_test_labels_accessed":False,"locked_test_metrics_calculated":False,"locked_test_predictions_created":False} + self.assertEqual(report["selected_calibration_method"],"uncalibrated") + self.assertEqual(report["final_package_type"],"uncalibrated") + self.assertFalse(report["final_calibration_applied"]) + self.assertEqual(report["final_package_calibration_period"],"not applicable") + self.assertNotIn("final_calibrated_artifact_path",report) + self.assertTrue(Path(report["final_candidate_artifact_path"]).exists()) + self.assertFalse(report["locked_test_labels_accessed"] or report["locked_test_metrics_calculated"] or report["locked_test_predictions_created"]) diff --git a/tests/test_data_foundation.py b/tests/test_data_foundation.py index 687126b..4c5ec03 100644 --- a/tests/test_data_foundation.py +++ b/tests/test_data_foundation.py @@ -1,6 +1,8 @@ import tempfile import unittest +import json from pathlib import Path +from unittest.mock import patch from src.data_foundation import ( LABEL_MAPPING_VERSION, @@ -80,14 +82,32 @@ def test_output_contract_excludes_leakage_columns(self): def test_end_to_end_offline_smoke(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) - report = run_foundation( - report_path=Path("reports/feasibility_report.json"), schema_path=Path("config/schema.yaml"), - snapshot_root=root / "raw", processed_root=root / "processed", - ) - self.assertEqual(report["label_counts"]["labelled"], 2100) - self.assertEqual(report["label_counts"]["positive"], 552) - self.assertEqual(report["label_counts"]["negative"], 1548) - self.assertEqual(report["excluded_table_shape"][0], 900) + cache = root / "data" / "raw" / "day0_cache" / "audit_fixture" + inspections = [ + self.inspection("1", "2020-01-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"}, + self.inspection("2", "2020-02-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"}, + self.inspection("3", "2021-01-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"}, + self.inspection("4", "2021-02-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"}, + self.inspection("5", "2022-01-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"}, + self.inspection("6", "2022-02-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"}, + ] + for year in (2020, 2021, 2022): + folder = cache / "inspection" / f"year_{year}" + folder.mkdir(parents=True) + page_rows = [item for item in inspections if item["open_date"].startswith(str(year))] + (folder / "page_0.json").write_text(json.dumps(page_rows), encoding="utf-8") + (folder / "manifest.json").write_text(json.dumps({"complete": True, "request": {"endpoint": "inspection", "wanted_rows": len(page_rows)}, "pages": {"0": {"status": "success", "file": "page_0.json", "row_count": len(page_rows)}}}), encoding="utf-8") + batch = cache / "violation" / "batch_0001"; batch.mkdir(parents=True) + violations = [{"activity_nr": "1", "citation_id": "a", "viol_type": "S", "delete_flag": None}, {"activity_nr": "2", "citation_id": "b", "viol_type": "O", "delete_flag": None}, {"activity_nr": "3", "citation_id": "c", "viol_type": "S", "delete_flag": "X"}, {"activity_nr": "4", "citation_id": "d", "viol_type": "Z", "delete_flag": None}] + (batch / "page_0.json").write_text(json.dumps(violations), encoding="utf-8") + (batch / "manifest.json").write_text(json.dumps({"complete": True, "request": {"endpoint": "violation", "filters": {"value": ["1", "2", "3", "4", "5"]}}, "pages": {"0": {"status": "success", "file": "page_0.json", "row_count": len(violations)}}}), encoding="utf-8") + feasibility = root / "reports" / "feasibility_report.json"; feasibility.parent.mkdir() + feasibility.write_text(json.dumps({"configuration": {"state": "CA", "start_date": "2020-01-01", "end_date": "2022-12-31"}, "acquisition": {"cache_directory": "data\\raw\\day0_cache\\audit_fixture"}}), encoding="utf-8") + with patch("scripts.dol_api.DOLApiClient.get_records", side_effect=AssertionError("network request attempted")): + report = run_foundation(report_path=feasibility, schema_path=Path("config/schema.yaml"), snapshot_root=root / "raw", processed_root=root / "processed") + self.assertEqual(report["label_counts"], {"positive": 1, "negative": 3, "labelled": 4, "positive_rate_percentage": 25.0}) + self.assertEqual(report["excluded_table_shape"][0], 2) + self.assertFalse(report["completed_vs_incomplete_retrieval"]["incomplete_outcomes_assumed_negative"]) if __name__ == "__main__": diff --git a/tests/test_features.py b/tests/test_features.py index 1f158c8..237c385 100644 --- a/tests/test_features.py +++ b/tests/test_features.py @@ -217,12 +217,26 @@ def test_no_fuzzy_matching_library_or_algorithm_is_introduced(self): def test_offline_end_to_end_smoke(self): with TemporaryDirectory() as directory: + root = Path(directory) + split_directory = root / "data" / "processed" / "snapshot" / "baseline" / "splits" + from src.splitting import create_chronological_split, write_split_artifacts + split = create_chronological_split([ + row("1", "2020-01-01", 0), row("2", "2020-02-01", 1), + row("3", "2021-01-01", 0), row("4", "2021-02-01", 1), + row("5", "2022-01-01", 0), row("6", "2022-02-01", 1), + ]) + write_split_artifacts(split_directory, split) + reports = root / "reports"; reports.mkdir() + foundation = reports / "data_foundation_report.json" + baseline = reports / "baseline_report.json" + foundation.write_text(json.dumps({"snapshot_id": "fixture"}), encoding="utf-8") + baseline.write_text(json.dumps({"source_snapshot_id": "fixture", "split_artifacts": {"directory": "data\\processed\\snapshot\\baseline\\splits"}}), encoding="utf-8") report = run_feature_engineering( - foundation_report_path=Path("reports/data_foundation_report.json"), baseline_report_path=Path("reports/baseline_report.json"), - config_path=Path("config/feature_config.yaml"), artifact_root=Path(directory) / "features", + foundation_report_path=foundation, baseline_report_path=baseline, + config_path=Path("config/feature_config.yaml"), artifact_root=root / "features", ) self.assertEqual(report["status"], "PASS") - self.assertEqual(report["splits"]["train"]["output_row_count"], 1200) + self.assertEqual(report["splits"]["train"]["output_row_count"], 2) self.assertFalse(report["splits"]["test_locked"]["target_present"]) diff --git a/tests/test_path_utils.py b/tests/test_path_utils.py new file mode 100644 index 0000000..c9d2b53 --- /dev/null +++ b/tests/test_path_utils.py @@ -0,0 +1,28 @@ +import tempfile +import unittest +from pathlib import Path + +from src.path_utils import RelativePathError, resolve_report_path + + +class ReportPathResolutionTests(unittest.TestCase): + def test_backslash_and_forward_slash_relative_paths_resolve_under_base(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + expected = root / "data" / "processed" / "snapshot" / "file.csv" + self.assertEqual(resolve_report_path(r"data\processed\snapshot\file.csv", root), expected) + self.assertEqual(resolve_report_path("data/processed/snapshot/file.csv", root), expected) + + def test_temporary_native_relative_path_is_portable_and_cleanup_is_owned_by_fixture(self): + temporary = tempfile.TemporaryDirectory() + root = Path(temporary.name) + self.assertEqual(resolve_report_path(Path("cache") / "manifest.json", root), root / "cache" / "manifest.json") + temporary.cleanup() + self.assertFalse(root.exists()) + + def test_absolute_and_escaping_paths_are_rejected(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for value in ("/tmp/outside.csv", r"C:\Users\person\artifact.csv", "../outside.csv"): + with self.assertRaises(RelativePathError): + resolve_report_path(value, root)