feat: add check_overdue_plan_events expectation

digital_land/expectations/checkpoints/dataset.py

@@ -13,6 +13,7 @@
 from ..log import ExpectationLog
 from ..operations.dataset import (
     check_fields_required_after_plan_event,
+    check_overdue_plan_events,
     count_lpa_boundary,
     count_deleted_entities,
     duplicate_geometry_check,
@@ -43,6 +44,7 @@ def operation_factory(self, operation_string: str):
         """
         operation_map = {
             "check_fields_required_after_plan_event": check_fields_required_after_plan_event,
+            "check_overdue_plan_events": check_overdue_plan_events,
             "count_lpa_boundary": count_lpa_boundary,
             "count_deleted_entities": count_deleted_entities,
             "duplicate_geometry_check": duplicate_geometry_check,

digital_land/expectations/operations/dataset.py

@@ -8,7 +8,7 @@
 import os
 import time
 from collections import defaultdict
-from datetime import datetime
+from datetime import datetime, timedelta
 
 
 # # TODO is there a way to represent this in a generalised count or not
@@ -936,3 +936,85 @@ def check_fields_required_after_plan_event(
     }
 
     return result, message, details
+
+
+def check_overdue_plan_events(
+    conn,
+    plan_events: list,
+    earliest_event_date: str,
+    grace_days: int = 7,
+    today: str = None,
+):
+    """
+    Checks plan-timetable rows have an actual-date once their event-date has passed.
+
+    A blank actual-date is expected while an event is in the future, so a row is only
+    flagged once its event-date is more than grace_days old. This runs against the
+    assembled dataset so a row becomes overdue without its resource having to change.
+
+    Each organisation's row is checked separately, including for joint plans, as each
+    authority provides its own endpoint and is responsible for keeping it up to date.
+
+    args:
+        conn: connection to the dataset being checked, created by the checkpoint class
+        plan_events: the plan-events which must be given an actual-date
+        earliest_event_date: ignore rows with an event-date before this, as YYYY-MM-DD
+        grace_days: how many days after the event-date to allow for an update
+        today: date to compare against as YYYY-MM-DD, defaults to the current date
+    """
+    if not plan_events:
+        raise ValueError("At least one plan event must be provided to check")
+
+    # a single string in config would otherwise iterate as characters and match
+    # nothing, so treat it as one plan event
+    if isinstance(plan_events, str):
+        plan_events = [plan_events]
+
+    today = today or datetime.now().strftime("%Y-%m-%d")
+    overdue_before = (
+        datetime.strptime(today, "%Y-%m-%d") - timedelta(days=grace_days)
+    ).strftime("%Y-%m-%d")
+
+    placeholders = ", ".join("?" for _ in plan_events)
+    query = f"""
+        select
+            organisation_entity,
+            reference,
+            json_extract(json, '$.plan') as plan,
+            json_extract(json, '$."plan-event"') as plan_event,
+            json_extract(json, '$."event-date"') as event_date
+        from entity
+        where coalesce(end_date, '') = ''
+            and coalesce(json_extract(json, '$."actual-date"'), '') = ''
+            and json_extract(json, '$."plan-event"') in ({placeholders})
+            and json_extract(json, '$."event-date"') >= ?
+            and json_extract(json, '$."event-date"') < ?
+        order by reference, organisation_entity
+    """
+    rows = conn.execute(query, (*plan_events, earliest_event_date, overdue_before))
+
+    failures = [
+        {
+            "organisation_entity": organisation_entity,
+            "reference": reference,
+            "plan": plan,
+            "plan-event": plan_event,
+            "event-date": event_date,
+        }
+        for organisation_entity, reference, plan, plan_event, event_date in rows
+    ]
+
+    result = len(failures) == 0
+    message = (
+        f"{len(failures)} plan events have no actual-date "
+        f"more than {grace_days} days after their event-date"
+    )
+    details = {
+        "failures": failures,
+        "field": "actual-date",
+        "actual": len(failures),
+        "expected": 0,
+        "overdue_before": overdue_before,
+    }
+
+    return result, message, details

tests/integration/expectations/operations/test_dataset.py

@@ -9,6 +9,7 @@
 from digital_land.expectations.operations.dataset import (
     check_columns,
     check_fields_required_after_plan_event,
+    check_overdue_plan_events,
     count_lpa_boundary,
     count_deleted_entities,
     duplicate_geometry_check,
@@ -1047,6 +1048,201 @@ def test_check_fields_required_after_plan_event_null_value(
     ]
 
 
+def _insert_timetable_rows(dataset_path, rows):
+    """insert plan-timetable entities, numbering them in the order given"""
+    with spatialite.connect(dataset_path) as con:
+        for entity, (reference, organisation_entity, end_date, fields) in enumerate(
+            rows, start=1
+        ):
+            con.execute(
+                "INSERT INTO entity"
+                " (entity, reference, organisation_entity, end_date, json)"
+                " VALUES (?, ?, ?, ?, ?)",
+                (
+                    entity,
+                    reference,
+                    organisation_entity,
+                    end_date,
+                    json.dumps({"plan": "plan-1", **fields}),
+                ),
+            )
+
+
+def _timetable_fields(event_date=None, actual_date=None, plan_event="adopted"):
+    """build an entity's json fields, leaving out any date which isn't given"""
+    fields = {"plan-event": plan_event}
+    if event_date is not None:
+        fields["event-date"] = event_date
+    if actual_date is not None:
+        fields["actual-date"] = actual_date
+    return fields
+
+
+def test_check_overdue_plan_events(dataset_path):
+    """only rows in scope with no actual-date past the grace period are flagged"""
+    _insert_timetable_rows(
+        dataset_path,
+        [
+            # event-date has passed the grace period, should be flagged
+            ("overdue", "1", None, _timetable_fields("2026-06-01")),
+            # an explicit blank actual-date is treated the same as no actual-date
+            ("blank", "1", None, _timetable_fields("2026-06-01", "")),
+            # actual-date provided, should not be flagged
+            ("updated", "1", None, _timetable_fields("2026-06-01", "2026-06-02")),
+            # event-date in the future, should not be flagged
+            ("future", "1", None, _timetable_fields("2026-12-01")),
+            # no event-date, nothing to be overdue against
+            ("undated", "1", None, _timetable_fields()),
+            # event-date before the earliest date, should not be flagged
+            ("too-early", "1", None, _timetable_fields("2026-03-01")),
+            # first day in range, should be flagged
+            ("earliest", "1", None, _timetable_fields("2026-03-02")),
+            # a plan event which is not being checked, should not be flagged
+            (
+                "other-event",
+                "1",
+                None,
+                _timetable_fields("2026-06-01", plan_event="withdrawn"),
+            ),
+            # removed from the timetable, should not be flagged
+            ("ended", "1", "2026-07-01", _timetable_fields("2026-06-01")),
+        ],
+    )
+
+    with spatialite.connect(dataset_path) as con:
+        passed, message, details = check_overdue_plan_events(
+            conn=con,
+            plan_events=["adopted"],
+            earliest_event_date="2026-03-02",
+            today="2026-11-06",
+        )
+
+    assert not passed, message
+    assert details["field"] == "actual-date"
+    assert details["actual"] == 3
+    assert [failure["reference"] for failure in details["failures"]] == [
+        "blank",
+        "earliest",
+        "overdue",
+    ]
+    assert details["failures"][2] == {
+        "organisation_entity": "1",
+        "reference": "overdue",
+        "plan": "plan-1",
+        "plan-event": "adopted",
+        "event-date": "2026-06-01",
+    }
+
+
+@pytest.mark.parametrize(
+    "event_date,grace_days,is_overdue",
+    [
+        ("2026-10-29", 7, True),
+        # exactly on the grace period, the LPA still has until the end of the day
+        ("2026-10-30", 7, False),
+        ("2026-11-05", 7, False),
+        ("2026-11-05", 0, True),
+        ("2026-11-06", 0, False),
+    ],
+)
+def test_check_overdue_plan_events_grace_period(
+    dataset_path, event_date, grace_days, is_overdue
+):
+    _insert_timetable_rows(
+        dataset_path, [("event", "1", None, _timetable_fields(event_date))]
+    )
+
+    with spatialite.connect(dataset_path) as con:
+        passed, _, details = check_overdue_plan_events(
+            conn=con,
+            plan_events=["adopted"],
+            earliest_event_date="2026-03-02",
+            grace_days=grace_days,
+            today="2026-11-06",
+        )
+
+    assert passed is not is_overdue
+    assert details["actual"] == int(is_overdue)
+
+
+def test_check_overdue_plan_events_joint_plan(dataset_path):
+    """each authority must update its own row, even if another already has"""
+    _insert_timetable_rows(
+        dataset_path,
+        [
+            ("joint", "1", None, _timetable_fields("2026-06-01", "2026-06-01")),
+            ("joint", "2", None, _timetable_fields("2026-06-01")),
+            ("joint", "3", None, _timetable_fields("2026-06-01")),
+        ],
+    )
+
+    with spatialite.connect(dataset_path) as con:
+        _, _, details = check_overdue_plan_events(
+            conn=con,
+            plan_events=["adopted"],
+            earliest_event_date="2026-03-02",
+            today="2026-11-06",
+        )
+
+    assert [failure["organisation_entity"] for failure in details["failures"]] == [
+        "2",
+        "3",
+    ]
+
+
+def test_check_overdue_plan_events_passes_with_no_overdue_rows(dataset_path):
+    _insert_timetable_rows(
+        dataset_path,
+        [("updated", "1", None, _timetable_fields("2026-06-01", "2026-06-02"))],
+    )
+
+    with spatialite.connect(dataset_path) as con:
+        passed, _, details = check_overdue_plan_events(
+            conn=con,
+            plan_events=["adopted"],
+            earliest_event_date="2026-03-02",
+            today="2026-11-06",
+        )
+
+    assert passed
+    assert details["failures"] == []
+
+
+def test_check_overdue_plan_events_requires_plan_events(dataset_path):
+    with spatialite.connect(dataset_path) as con:
+        with pytest.raises(ValueError):
+            check_overdue_plan_events(
+                conn=con, plan_events=[], earliest_event_date="2026-03-02"
+            )
+
+
+def test_check_overdue_plan_events_rejects_empty_string(dataset_path):
+    """an empty string must raise, not become a plan event which matches nothing"""
+    with spatialite.connect(dataset_path) as con:
+        with pytest.raises(ValueError):
+            check_overdue_plan_events(
+                conn=con, plan_events="", earliest_event_date="2026-03-02"
+            )
+
+
+def test_check_overdue_plan_events_single_string(dataset_path):
+    """a single plan event given as a string is checked, not split into characters"""
+    _insert_timetable_rows(
+        dataset_path, [("overdue", "1", None, _timetable_fields("2026-06-01"))]
+    )
+
+    with spatialite.connect(dataset_path) as con:
+        passed, _, details = check_overdue_plan_events(
+            conn=con,
+            plan_events="adopted",
+            earliest_event_date="2026-03-02",
+            today="2026-11-06",
+        )
+
+    assert not passed
+    assert details["actual"] == 1
+
+
 def test_name_is_a_code_check(dataset_path):
     entities = [
         # bare codes, should be flagged