Merge pull request #345 from digital-land/referential-check-issue-type

Add issue type to check referential integrity for linked datasets

Author: kena-SL

digital_land/cli.py

@@ -44,6 +44,7 @@
     column_field_dir,
     converted_resource_dir,
     output_log_dir,
+    provision_summary_dir,
 )
 
 
@@ -227,6 +228,7 @@ def dataset_dump_flattened_cmd(ctx, input_path, output_path):
 @dataset_resource_dir
 @converted_resource_dir
 @organisation_path
+@provision_summary_dir
 @collection_dir
 @operational_issue_dir
 @output_log_dir
@@ -250,6 +252,7 @@ def pipeline_command(
     config_path,
     resource,
     output_log_dir,
+    provision_summary_dir,
 ):
     dataset = ctx.obj["DATASET"]
     pipeline = ctx.obj["PIPELINE"]
@@ -279,6 +282,7 @@ def pipeline_command(
         config_path=config_path,
         resource=resource,
         output_log_dir=output_log_dir,
+        provision_summary_dir=provision_summary_dir,
     )
 
 

digital_land/command_arguments.py

@@ -97,3 +97,11 @@ def organisation_path(f):
         type=click.Path(exists=True),
         default="var/cache/organisation.csv",
     )(f)
+
+
+def provision_summary_dir(f):
+    return click.option(
+        "--provision-summary-dir",
+        type=click.Path(exists=True),
+        default="var/cache/provision-summary/",
+    )(f)

digital_land/commands.py

@@ -216,6 +216,7 @@ def pipeline_run(
     resource=None,
     output_log_dir=None,
     converted_path=None,
+    provision_summary_dir="var/cache/provision_summary",
 ):
     # set up paths
     cache_dir = Path(cache_dir)
@@ -330,6 +331,7 @@ def pipeline_run(
             issue_log=issue_log,
             operational_issue_log=operational_issue_log,
             entity_range=[entity_range_min, entity_range_max],
+            provision_summary_dir=provision_summary_dir,
         ),
         SavePhase(
             default_output_path("harmonised", input_path),

digital_land/phase/lookup.py

@@ -1,5 +1,6 @@
 import re
 import logging
+import pandas as pd
 
 from .phase import Phase
 
@@ -33,13 +34,15 @@ def __init__(
         issue_log=None,
         operational_issue_log=None,
         entity_range=[],
+        provision_summary_dir=None,
     ):
         self.lookups = lookups
         self.redirect_lookups = redirect_lookups
         self.issues = issue_log
         self.operational_issues = operational_issue_log
         self.reverse_lookups = self.build_reverse_lookups()
         self.entity_range = entity_range
+        self.provision_summary_dir = provision_summary_dir
 
     def build_reverse_lookups(self):
         reverse_lookups = {}
@@ -163,6 +166,47 @@ def process(self, stream):
                         row[self.entity_field] = self.redirect_entity(
                             row[self.entity_field]
                         )
+
+                linked_datasets = ["article-4-direction", "tree-preservation-order"]
+                if row[self.entity_field]:
+                    for linked_dataset in linked_datasets:
+                        if (
+                            row.get(linked_dataset, "")
+                            or row.get(linked_dataset, "").strip()
+                        ):
+                            get_organisations = pd.read_csv(
+                                self.provision_summary_dir
+                                + "/"
+                                + linked_dataset
+                                + ".csv"
+                            )
+
+                            if (
+                                row.get("organisation", "")
+                                in get_organisations["organisation"].values
+                            ):
+                                reference = row.get(linked_dataset, "")
+                                find_entity = self.lookup(
+                                    prefix=linked_dataset,
+                                    organisation=row.get("organisation", ""),
+                                    reference=reference,
+                                )
+                                # raise issue if the found entity is retired in old-entity.csv
+                                if not find_entity or (
+                                    str(find_entity) in self.redirect_lookups
+                                    and int(
+                                        self.redirect_lookups[str(find_entity)].get(
+                                            "status", 0
+                                        )
+                                    )
+                                    == 410
+                                ):
+                                    self.issues.log_issue(
+                                        linked_dataset,
+                                        "no associated documents found for this area",
+                                        reference,
+                                        line_number=line_number,
+                                    )
             yield block
 
 

tests/unit/phase/test_lookup.py

@@ -1,3 +1,4 @@
+import pandas as pd
 import pytest
 
 from digital_land.phase.lookup import LookupPhase, EntityLookupPhase, PrintLookupPhase
@@ -19,6 +20,22 @@ def get_input_stream():
     ]
 
 
+@pytest.fixture
+def get_input_stream_with_linked_field():
+    return [
+        {
+            "row": {
+                "prefix": "article-4-direction-area",
+                "reference": "1",
+                "organisation": "local-authority:ABC",
+                "article-4-direction": "a4d2",
+            },
+            "entry-number": 1,
+            "line-number": 2,
+        }
+    ]
+
+
 @pytest.fixture
 def get_lookup():
     return {",dataset,1,test": "1"}
@@ -123,6 +140,88 @@ def test_process_empty_prefix(self, get_lookup):
 
         assert output[0]["row"]["entity"] == "10"
 
+    def test_no_associated_documents_issue(
+        self, get_input_stream_with_linked_field, mocker
+    ):
+        input_stream = get_input_stream_with_linked_field
+
+        lookups = {
+            ",article-4-direction,a4d1,local-authorityabc": "1",
+            ",article-4-direction-area,1,local-authorityabc": "2",
+        }
+        issues = IssueLog()
+
+        phase = LookupPhase(
+            lookups=lookups,
+            issue_log=issues,
+            provision_summary_dir="var/cache/provision-summary/",
+        )
+        phase.entity_field = "entity"
+        mock_df = pd.DataFrame({"organisation": ["local-authority:ABC"]})
+        mocker.patch("pandas.read_csv", return_value=mock_df)
+        output = [block for block in phase.process(input_stream)]
+
+        assert output[0]["row"]["entity"] == "2"
+        assert (
+            issues.rows[0]["issue-type"]
+            == "no associated documents found for this area"
+        )
+        assert issues.rows[0]["value"] == "a4d2"
+
+    def test_no_associated_documents_issue_for_missing_dataset(
+        self, get_input_stream_with_linked_field, mocker
+    ):
+        input_stream = get_input_stream_with_linked_field
+
+        lookups = {
+            ",article-4-direction,a4d1,local-authorityabc": "1",
+            ",article-4-direction-area,1,local-authorityabc": "2",
+        }
+        issues = IssueLog()
+
+        phase = LookupPhase(
+            lookups=lookups,
+            issue_log=issues,
+            provision_summary_dir="var/cache/provision-summary/",
+        )
+        phase.entity_field = "entity"
+        mock_df = pd.DataFrame({"organisation": ["local-authority:XYZ"]})
+        mocker.patch("pandas.read_csv", return_value=mock_df)
+        output = [block for block in phase.process(input_stream)]
+
+        assert output[0]["row"]["entity"] == "2"
+        assert len(issues.rows) == 0
+
+    def test_no_associated_documents_issue_for_retired_entity(
+        self, get_input_stream_with_linked_field, mocker
+    ):
+        input_stream = get_input_stream_with_linked_field
+
+        lookups = {
+            ",article-4-direction,a4d2,local-authorityabc": "1",
+            ",article-4-direction-area,1,local-authorityabc": "2",
+        }
+        issues = IssueLog()
+        redirect_lookups = {"1": {"entity": "", "status": "410"}}
+
+        phase = LookupPhase(
+            lookups=lookups,
+            redirect_lookups=redirect_lookups,
+            issue_log=issues,
+            provision_summary_dir="var/cache/provision-summary/",
+        )
+        phase.entity_field = "entity"
+        mock_df = pd.DataFrame({"organisation": ["local-authority:ABC"]})
+        mocker.patch("pandas.read_csv", return_value=mock_df)
+        output = [block for block in phase.process(input_stream)]
+
+        assert output[0]["row"]["entity"] == "2"
+        assert (
+            issues.rows[0]["issue-type"]
+            == "no associated documents found for this area"
+        )
+        assert issues.rows[0]["value"] == "a4d2"
+
 
 class TestPrintLookupPhase:
     def test_process_does_not_produce_new_lookup(self, get_input_stream, get_lookup):