diff --git a/digital_land/expectations/checkpoints/dataset.py b/digital_land/expectations/checkpoints/dataset.py index 11dca14e..592f6d4c 100644 --- a/digital_land/expectations/checkpoints/dataset.py +++ b/digital_land/expectations/checkpoints/dataset.py @@ -1,6 +1,7 @@ import inspect import json import logging +import pyarrow.parquet as pq import spatialite from datetime import datetime from pathlib import Path @@ -25,6 +26,7 @@ def __init__(self, dataset, file_path, organisations: Organisation): self.dataset_path = Path(file_path) self.organisations = organisations self.log = ExpectationLog(dataset=dataset) + self.skipped_rule_names = [] def operation_factory(self, operation_string: str): """ @@ -115,11 +117,13 @@ def load(self, rules): expectations need parsing """ self.expectations = [] + self.skipped_rule_names = [] today = datetime.now().strftime("%A").lower() for rule in rules: schedule = rule.get("schedule", "") # skip rule if today does not match scheduled day of the week. if schedule and schedule.lower() != today: + self.skipped_rule_names.append(rule["name"]) continue if rule["organisations"]: rule_orgs = self.get_rule_orgs(rule) @@ -208,4 +212,14 @@ def save(self, output_dir: Path): save the outputs as a file, the file is named based the the dataset and stored in the provided directory """ + if self.skipped_rule_names: + existing_path = ( + Path(output_dir) / self.log.pq_partition / (self.dataset + ".parquet") + ) + if existing_path.exists(): + previous = pq.ParquetFile(existing_path).read().to_pylist() + for row in previous: + if row.get("name") in self.skipped_rule_names: + self.log.entries.append(row) + self.log.save_parquet(output_dir) diff --git a/tests/integration/expectations/checkpoints/test_dataset.py b/tests/integration/expectations/checkpoints/test_dataset.py index a8798b50..e7716835 100644 --- a/tests/integration/expectations/checkpoints/test_dataset.py +++ b/tests/integration/expectations/checkpoints/test_dataset.py @@ -171,6 +171,7 @@ def test_load_skips_rule_not_scheduled_today(self, test_organisations, mocker): checkpoint.load(rules) assert len(checkpoint.expectations) == 0 + assert "a saturday only rule" in checkpoint.skipped_rule_names def test_load_includes_rule_scheduled_for_today(self, test_organisations, mocker): """test loading includes rules scheduled for today, shoulld be moved to unit""" @@ -199,6 +200,96 @@ def test_load_includes_rule_scheduled_for_today(self, test_organisations, mocker assert len(checkpoint.expectations) == 1 + def test_save_carries_forward_skipped_rule_results( + self, tmp_path, test_organisations + ): + import pyarrow.parquet as pq + from digital_land.expectations.log import ExpectationLog + + dataset = "test-dataset" + output_path = tmp_path / "expectations" + + # write a previous parquet with a result for the skipped rule + partition_dir = output_path / f"dataset={dataset}" + partition_dir.mkdir(parents=True) + previous_log = ExpectationLog(dataset=dataset) + previous_log.add( + { + "organisation": "", + "name": "Check no duplicate geometries", + "operation": "duplicate_geometry_check", + "passed": False, + "message": "2 duplicates found", + "details": "", + "description": "", + "severity": "notice", + "responsibility": "internal", + "parameters": '{"spatial_field": "geometry"}', + } + ) + previous_log.save_parquet(output_path) + + checkpoint = DatasetCheckpoint( + dataset=dataset, file_path="", organisations=test_organisations + ) + checkpoint.skipped_rule_names = ["Check no duplicate geometries"] + checkpoint.log.add( + { + "organisation": "", + "name": "A fresh check", + "passed": True, + "message": "passed", + "details": "", + "description": "", + "severity": "notice", + "responsibility": "internal", + "parameters": "{}", + "operation": "test", + } + ) + + checkpoint.save(str(output_path)) + + result = pq.ParquetFile(partition_dir / f"{dataset}.parquet").read().to_pylist() + names = [row["name"] for row in result] + assert "Check no duplicate geometries" in names # carried forward + assert "A fresh check" in names # fresh result preserved + + def test_save_skipped_rule_no_existing_parquet(self, tmp_path, test_organisations): + import pyarrow.parquet as pq + + dataset = "test-dataset" + output_path = tmp_path / "expectations" + + checkpoint = DatasetCheckpoint( + dataset=dataset, file_path="", organisations=test_organisations + ) + checkpoint.skipped_rule_names = ["some skipped rule"] + checkpoint.log.add( + { + "organisation": "", + "name": "A fresh check", + "passed": True, + "message": "passed", + "details": "", + "description": "", + "severity": "notice", + "responsibility": "internal", + "parameters": "{}", + "operation": "test", + } + ) + + checkpoint.save(str(output_path)) + + result = ( + pq.ParquetFile(output_path / f"dataset={dataset}" / f"{dataset}.parquet") + .read() + .to_pylist() + ) + assert len(result) == 1 + assert result[0]["name"] == "A fresh check" + def test_run_success( self, tmp_path, sqlite3_with_entity_tables_path, mocker, test_organisations ):