feat: use yesterdays expectations if expectation is skipped today

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)

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
     ):