From 350d43fba9143e48dc0f04f7a901b03888416f4b Mon Sep 17 00:00:00 2001 From: Sian Teesdale <43341988+sianteesdale@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:52:02 +0100 Subject: [PATCH] Update deleted entities expectation --- .../expectations/checkpoints/dataset.py | 12 +- .../expectations/operations/dataset.py | 116 ++++++++----- .../expectations/checkpoints/test_dataset.py | 14 +- .../expectations/operations/test_dataset.py | 161 ++++++++++++++---- 4 files changed, 225 insertions(+), 78 deletions(-) diff --git a/digital_land/expectations/checkpoints/dataset.py b/digital_land/expectations/checkpoints/dataset.py index 47aeb6c43..774353f21 100644 --- a/digital_land/expectations/checkpoints/dataset.py +++ b/digital_land/expectations/checkpoints/dataset.py @@ -18,6 +18,7 @@ duplicate_geometry_check, duplicate_name_check, fetch_active_resources_for_dataset, + fetch_all_resources_for_dataset, name_is_a_code_check, name_is_a_placeholder_check, ) @@ -169,16 +170,23 @@ def run(self, prefetch_resources=False): ) resources_cache = None + all_resources_cache = None if prefetch_resources: resources_cache = fetch_active_resources_for_dataset(self.dataset) + all_resources_cache = fetch_all_resources_for_dataset(self.dataset) logger.warning("[expectations] Prefetched resources cache") for i, expectation in enumerate(self.expectations): if resources_cache is not None: operation = expectation["operation"] - # only pass resources_cache to operations that explicitly accept it - if "resources_cache" in inspect.signature(operation).parameters: + operation_parameters = inspect.signature(operation).parameters + # only pass the caches to operations that explicitly accept them + if "resources_cache" in operation_parameters: expectation["parameters"]["resources_cache"] = resources_cache + if "all_resources_cache" in operation_parameters: + expectation["parameters"][ + "all_resources_cache" + ] = all_resources_cache passed, message, details = self.run_expectation(expectation) diff --git a/digital_land/expectations/operations/dataset.py b/digital_land/expectations/operations/dataset.py index f03982f93..502bdfac1 100644 --- a/digital_land/expectations/operations/dataset.py +++ b/digital_land/expectations/operations/dataset.py @@ -135,18 +135,22 @@ def count_lpa_boundary( return result, message, details -# Previously named count_deleted_entities() function -def fetch_active_resources_for_dataset(dataset_name): - params = urllib.parse.urlencode( - { - "sql": f"""select o.entity as organisation_entity, rhe.resource - from reporting_historic_endpoints rhe - join organisation o on rhe.organisation=o.organisation - where pipeline == '{dataset_name}' and (resource_end_date == "" or resource_end_date is null) - group by o.entity, rhe.endpoint""", - "_size": "max", - } - ) +def _fetch_resources_for_dataset(dataset_name, active_only): + if active_only: + # one active resource per endpoint + sql = f"""select o.entity as organisation_entity, rhe.resource + from reporting_historic_endpoints rhe + join organisation o on rhe.organisation=o.organisation + where pipeline == '{dataset_name}' and (resource_end_date == "" or resource_end_date is null) + group by o.entity, rhe.endpoint""" + else: + # every resource each organisation has ever supplied, including ended ones + sql = f"""select o.entity as organisation_entity, rhe.resource + from reporting_historic_endpoints rhe + join organisation o on rhe.organisation=o.organisation + where pipeline == '{dataset_name}' and rhe.resource != "" + group by o.entity, rhe.resource""" + params = urllib.parse.urlencode({"sql": sql, "_size": "max"}) base_url = f"https://datasette.planning.data.gov.uk/digital-land.csv?{params}" max_retries = 60 @@ -154,7 +158,7 @@ def fetch_active_resources_for_dataset(dataset_name): try: df = pd.read_csv(base_url) logging.warning( - f"[expectations] Fetched {len(df)} active resources for '{dataset_name}' from datasette" + f"[expectations] Fetched {len(df)} {'active' if active_only else 'historic'} resources for '{dataset_name}' from datasette" ) cache = {} for _, row in df.iterrows(): @@ -171,46 +175,74 @@ def fetch_active_resources_for_dataset(dataset_name): ) +# Previously named count_deleted_entities() function +def fetch_active_resources_for_dataset(dataset_name): + return _fetch_resources_for_dataset(dataset_name, active_only=True) + + +def fetch_all_resources_for_dataset(dataset_name): + return _fetch_resources_for_dataset(dataset_name, active_only=False) + + +def _fetch_organisation_resources(db_name, organisation_entity, active_only): + end_date_filter = 'and resource_end_date == ""' if active_only else "" + params = urllib.parse.urlencode( + { + "sql": f"""select * from reporting_historic_endpoints rhe join organisation o on rhe.organisation=o.organisation + where pipeline == '{db_name}' and o.entity='{organisation_entity}' {end_date_filter} group by {'endpoint' if active_only else 'resource'}""", + "_size": "max", + } + ) + base_url = f"https://datasette.planning.data.gov.uk/digital-land.csv?{params}" + + # Can have an issue getting data from datasette. If this occurs then wait a minute and retry + max_retries = 60 # Retry for an hour + for attempt in range(max_retries): + try: + get_resource = pd.read_csv(base_url) + break + except urllib.error.HTTPError as e: + logging.warning( + f"HTTP error fetching datasette for organisation {organisation_entity}, " + f"attempt {attempt + 1}/{max_retries}: {e}. Retrying in 60s..." + ) + time.sleep(60) + else: + raise Exception("Failed to fetch datasette after multiple attempts") + + return get_resource["resource"].to_list() + + def count_deleted_entities( conn, expected: int, organisation_entity: int = None, resources_cache: dict = None, + all_resources_cache: dict = None, ): + """ + Counts entities the organisation has supplied in the past which are missing + from its active resources. Entities assigned to the organisation but only + ever supplied by another organisation (e.g. MHCLG seeded data) are ignored. + """ # get database name to identify dataset db_path = conn.execute("PRAGMA database_list").fetchall()[0][2] db_name = os.path.splitext(os.path.basename(db_path))[0] - # check if entity data has been fetched from datasette and stored in the cache + # check if resources have been fetched from datasette and stored in the cache if resources_cache is not None: resource_list = resources_cache.get(organisation_entity, []) else: - # get dataset specific active resource list if nothing is found it - params = urllib.parse.urlencode( - { - "sql": f"""select * from reporting_historic_endpoints rhe join organisation o on rhe.organisation=o.organisation - where pipeline == '{db_name}' and o.entity='{organisation_entity}' and resource_end_date == "" group by endpoint""", - "_size": "max", - } + resource_list = _fetch_organisation_resources( + db_name, organisation_entity, active_only=True ) - base_url = f"https://datasette.planning.data.gov.uk/digital-land.csv?{params}" - - # Can have an issue getting data from datasette. If this occurs then wait a minute and retry - max_retries = 60 # Retry for an hour - for attempt in range(max_retries): - try: - get_resource = pd.read_csv(base_url) - break - except urllib.error.HTTPError as e: - logging.warning( - f"HTTP error fetching datasette for organisation {organisation_entity}, " - f"attempt {attempt + 1}/{max_retries}: {e}. Retrying in 60s..." - ) - time.sleep(60) - else: - raise Exception("Failed to fetch datasette after multiple attempts") - resource_list = get_resource["resource"].to_list() + if all_resources_cache is not None: + all_resource_list = all_resources_cache.get(organisation_entity, []) + else: + all_resource_list = _fetch_organisation_resources( + db_name, organisation_entity, active_only=False + ) # use resource list to get current entities query = f"""select f.entity @@ -221,9 +253,13 @@ def count_deleted_entities( rows = conn.execute(query).fetchall() get_active_entities = [row[0] for row in rows] - # get entities from entity table to compare against resource entities + # get the organisation's entities which it has supplied itself at some point, + # using every resource it has ever provided, to compare against resource entities query = f""" - select entity from entity where organisation_entity = '{organisation_entity}'; + select distinct e.entity + from entity e join fact f on f.entity=e.entity join fact_resource fe on fe.fact=f.fact + where e.organisation_entity = '{organisation_entity}' + and fe.resource in ({','.join(f"'{x}'" for x in all_resource_list)}); """ rows = conn.execute(query).fetchall() get_entities = [row[0] for row in rows] diff --git a/tests/integration/expectations/checkpoints/test_dataset.py b/tests/integration/expectations/checkpoints/test_dataset.py index e3add5fbb..98144f11d 100644 --- a/tests/integration/expectations/checkpoints/test_dataset.py +++ b/tests/integration/expectations/checkpoints/test_dataset.py @@ -113,6 +113,10 @@ def test_run_with_prefetch_fetches_once_and_injects_cache( "digital_land.expectations.checkpoints.dataset.fetch_active_resources_for_dataset", return_value={101: ["resource_a"]}, ) + mock_fetch_all = mocker.patch( + "digital_land.expectations.checkpoints.dataset.fetch_all_resources_for_dataset", + return_value={101: ["resource_a", "resource_ended"]}, + ) mock_function = mocker.create_autospec( count_deleted_entities, return_value=(True, "passed", {}) ) @@ -144,10 +148,14 @@ def test_run_with_prefetch_fetches_once_and_injects_cache( # datasette fetched exactly once despite two organisations mock_fetch.assert_called_once_with("test-dataset") + mock_fetch_all.assert_called_once_with("test-dataset") - # cache was injected into every operation call + # caches were injected into every operation call for call in mock_function.call_args_list: assert call.kwargs["resources_cache"] == {101: ["resource_a"]} + assert call.kwargs["all_resources_cache"] == { + 101: ["resource_a", "resource_ended"] + } assert len(checkpoint.log.entries) == 2 @@ -164,6 +172,10 @@ def test_run_with_prefetch_does_not_inject_cache_into_incompatible_operations( "digital_land.expectations.checkpoints.dataset.fetch_active_resources_for_dataset", return_value={101: ["resource_a"]}, ) + mocker.patch( + "digital_land.expectations.checkpoints.dataset.fetch_all_resources_for_dataset", + return_value={101: ["resource_a"]}, + ) def operation_without_cache(conn, expected: int): return True, "passed", {} diff --git a/tests/integration/expectations/operations/test_dataset.py b/tests/integration/expectations/operations/test_dataset.py index d78ece32c..c951683a0 100644 --- a/tests/integration/expectations/operations/test_dataset.py +++ b/tests/integration/expectations/operations/test_dataset.py @@ -13,6 +13,7 @@ duplicate_geometry_check, duplicate_name_check, fetch_active_resources_for_dataset, + fetch_all_resources_for_dataset, name_is_a_code_check, name_is_a_placeholder_check, ) @@ -138,13 +139,29 @@ def test_fetch_active_resources_for_dataset(mocker): assert result == {101: ["resource_a", "resource_b"], 102: ["resource_c"]} +def test_fetch_all_resources_for_dataset_includes_ended_resources(mocker): + mock_df = pd.DataFrame( + { + "organisation_entity": [101, 101, 102], + "resource": ["resource_a", "resource_ended", "resource_c"], + } + ) + mock_read_csv = mocker.patch("pandas.read_csv", return_value=mock_df) + + result = fetch_all_resources_for_dataset("test-dataset") + + assert result == {101: ["resource_a", "resource_ended"], 102: ["resource_c"]} + assert "resource_end_date" not in mock_read_csv.call_args.args[0] + + def test_count_deleted_entities(dataset_path, mocker): """Entity with no facts in the active resource is reported as deleted.""" organisation_entity = 109 expected = 0 # load data into sqlite for entity, fact_resource and fact table - # entity 1001 has facts in the active resource; entity 1002 does not + # entity 1001 has facts in the active resource; entity 1002 only has facts + # in an ended resource previously supplied by the same organisation test_entity_data = pd.DataFrame.from_dict( { "entity": ["1001", "1002"], @@ -156,24 +173,29 @@ def test_count_deleted_entities(dataset_path, mocker): test_fact_resource_data = pd.DataFrame.from_dict( { - "fact": ["036d2b946bd41", "16bf38800aafd"], - "resource": ["2f7d900dd48fd02", "2f7d900dd48fd02"], - "entry_number": ["1", "1"], + "fact": ["036d2b946bd41", "16bf38800aafd", "fact-old"], + "resource": ["2f7d900dd48fd02", "2f7d900dd48fd02", "res-ended"], + "entry_number": ["1", "1", "1"], }, ) test_fact_data = pd.DataFrame.from_dict( { - "fact": ["036d2b946bd41", "16bf38800aafd"], - "entity": ["1001", "1001"], - "field": ["name", "reference"], - "value": ["abc", "ref1"], + "fact": ["036d2b946bd41", "16bf38800aafd", "fact-old"], + "entity": ["1001", "1001", "1002"], + "field": ["name", "reference", "name"], + "value": ["abc", "ref1", "test2"], } ) - # mock `pandas.read_csv` to return the mock DataFrame - mock_df = pd.DataFrame({"resource": ["2f7d900dd48fd02"]}) - mocker.patch("pandas.read_csv", return_value=mock_df) + # mock `pandas.read_csv` to return the active then all resources + mocker.patch( + "pandas.read_csv", + side_effect=[ + pd.DataFrame({"resource": ["2f7d900dd48fd02"]}), + pd.DataFrame({"resource": ["2f7d900dd48fd02", "res-ended"]}), + ], + ) with spatialite.connect(dataset_path) as conn: # load data into required tables @@ -254,7 +276,8 @@ def test_count_deleted_entities_org_isolation(dataset_path, mocker): Facts for a second organisation's entity in the same resource must not cause that entity to be counted as active for the queried organisation. - Setup: org 109 has entities 1001 (active) and 1002 (deleted). Org 200 + Setup: org 109 has entities 1001 (active) and 1002 (deleted, last seen + in an ended resource). Org 200 has entity 2001 with a fact in the same resource. Only 1002 should be reported as deleted for org 109; entity 2001 must be excluded entirely. """ @@ -271,24 +294,29 @@ def test_count_deleted_entities_org_isolation(dataset_path, mocker): test_fact_resource_data = pd.DataFrame.from_dict( { - "fact": ["fact-a", "fact-x"], - "resource": ["res-active", "res-active"], - "entry_number": ["1", "2"], + "fact": ["fact-a", "fact-x", "fact-old"], + "resource": ["res-active", "res-active", "res-ended"], + "entry_number": ["1", "2", "1"], } ) # fact-a belongs to org 109's entity 1001; fact-x belongs to org 200's entity 2001 test_fact_data = pd.DataFrame.from_dict( { - "fact": ["fact-a", "fact-x"], - "entity": ["1001", "2001"], - "field": ["name", "name"], - "value": ["org109 tree A", "org200 tree"], + "fact": ["fact-a", "fact-x", "fact-old"], + "entity": ["1001", "2001", "1002"], + "field": ["name", "name", "name"], + "value": ["org109 tree A", "org200 tree", "org109 tree B"], } ) - mock_df = pd.DataFrame({"resource": ["res-active"]}) - mocker.patch("pandas.read_csv", return_value=mock_df) + mocker.patch( + "pandas.read_csv", + side_effect=[ + pd.DataFrame({"resource": ["res-active"]}), + pd.DataFrame({"resource": ["res-active", "res-ended"]}), + ], + ) with spatialite.connect(dataset_path) as conn: test_entity_data.to_sql("entity", conn, if_exists="replace", index=False) @@ -312,6 +340,66 @@ def test_count_deleted_entities_org_isolation(dataset_path, mocker): assert "2001" not in details["entities"] +def test_count_deleted_entities_ignores_entities_supplied_by_another_organisation( + dataset_path, mocker +): + """ + Entities assigned to an organisation but only ever supplied by another + organisation (e.g. MHCLG seeded plan-timetable data for an LPA) were never + in the organisation's own resources, so must not be reported as deleted. + + Setup: org 109 has entity 1001 (active), 1002 (deleted, last seen in an + ended org 109 resource) and 1003 (only supplied in another organisation's + resource). Only 1002 should be reported as deleted. + """ + organisation_entity = 109 + + test_entity_data = pd.DataFrame.from_dict( + { + "entity": ["1001", "1002", "1003"], + "name": ["test1", "test2", "test3"], + "organisation_entity": [109, 109, 109], + "reference": ["ref1", "ref2", "ref3"], + } + ) + + test_fact_resource_data = pd.DataFrame.from_dict( + { + "fact": ["fact-a", "fact-b", "fact-c"], + "resource": ["res-active", "res-ended", "res-other-org"], + "entry_number": ["1", "1", "1"], + } + ) + + test_fact_data = pd.DataFrame.from_dict( + { + "fact": ["fact-a", "fact-b", "fact-c"], + "entity": ["1001", "1002", "1003"], + "field": ["name", "name", "name"], + "value": ["test1", "test2", "test3"], + } + ) + + with spatialite.connect(dataset_path) as conn: + test_entity_data.to_sql("entity", conn, if_exists="replace", index=False) + test_fact_resource_data.to_sql( + "fact_resource", conn, if_exists="replace", index=False + ) + test_fact_data.to_sql("fact", conn, if_exists="replace", index=False) + + passed, _, details = count_deleted_entities( + conn, + expected=0, + organisation_entity=organisation_entity, + resources_cache={109: ["res-active"]}, + all_resources_cache={109: ["res-active", "res-ended"]}, + ) + + assert not passed + assert details["actual"] == 1 + assert details["entities"] == ["1002"] + + def test_count_deleted_entities_uses_cache_instead_of_http(dataset_path, mocker): organisation_entity = 109 expected = 0 @@ -326,22 +414,23 @@ def test_count_deleted_entities_uses_cache_instead_of_http(dataset_path, mocker) ) test_fact_resource_data = pd.DataFrame.from_dict( { - "fact": ["036d2b946bd41", "16bf38800aafd"], - "resource": ["2f7d900dd48fd02", "2f7d900dd48fd02"], - "entry_number": ["1", "1"], + "fact": ["036d2b946bd41", "16bf38800aafd", "fact-old"], + "resource": ["2f7d900dd48fd02", "2f7d900dd48fd02", "res-ended"], + "entry_number": ["1", "1", "1"], } ) test_fact_data = pd.DataFrame.from_dict( { - "fact": ["036d2b946bd41", "16bf38800aafd"], - "entity": ["1001", "1001"], - "field": ["name", "reference"], - "value": ["abc", "ref1"], + "fact": ["036d2b946bd41", "16bf38800aafd", "fact-old"], + "entity": ["1001", "1001", "1002"], + "field": ["name", "reference", "name"], + "value": ["abc", "ref1", "test2"], } ) mock_read_csv = mocker.patch("pandas.read_csv") resources_cache = {109: ["2f7d900dd48fd02"]} + all_resources_cache = {109: ["2f7d900dd48fd02", "res-ended"]} with spatialite.connect(dataset_path) as conn: test_entity_data.to_sql("entity", conn, if_exists="replace", index=False) @@ -355,6 +444,7 @@ def test_count_deleted_entities_uses_cache_instead_of_http(dataset_path, mocker) expected=expected, organisation_entity=organisation_entity, resources_cache=resources_cache, + all_resources_cache=all_resources_cache, ) mock_read_csv.assert_not_called() @@ -1200,17 +1290,17 @@ def test_count_deleted_entities_failures_carry_organisation_as_a_string( ) test_fact_resource_data = pd.DataFrame.from_dict( { - "fact": ["036d2b946bd41", "16bf38800aafd"], - "resource": ["2f7d900dd48fd02", "2f7d900dd48fd02"], - "entry_number": ["1", "1"], + "fact": ["036d2b946bd41", "16bf38800aafd", "fact-old"], + "resource": ["2f7d900dd48fd02", "2f7d900dd48fd02", "res-ended"], + "entry_number": ["1", "1", "1"], } ) test_fact_data = pd.DataFrame.from_dict( { - "fact": ["036d2b946bd41", "16bf38800aafd"], - "entity": ["1001", "1001"], - "field": ["name", "reference"], - "value": ["abc", "ref1"], + "fact": ["036d2b946bd41", "16bf38800aafd", "fact-old"], + "entity": ["1001", "1001", "1002"], + "field": ["name", "reference", "name"], + "value": ["abc", "ref1", "test2"], } ) mock_read_csv = mocker.patch("pandas.read_csv") @@ -1226,6 +1316,7 @@ def test_count_deleted_entities_failures_carry_organisation_as_a_string( expected=0, organisation_entity=organisation_entity, resources_cache={109: ["2f7d900dd48fd02"]}, + all_resources_cache={109: ["2f7d900dd48fd02", "res-ended"]}, ) mock_read_csv.assert_not_called()