Post conversion expectations

digital_land/expectations/checkpoints/converted_resource.py

@@ -5,9 +5,82 @@
 # a checkpoint represents the moment in the process where we tell it the
 # type of data it is validating and where the data is
 # the primary different between checkpoints is how it loads expectations (i.e. where that are loaded from)
+from pathlib import Path
+import csv
 from .base import BaseCheckpoint
 
 
-class CovertedResourceCheckpoint(BaseCheckpoint):
-    def load():
-        pass
+class ConvertedResourceCheckpoint(BaseCheckpoint):
+    def __init__(self, data_path):
+        super().__init__("converted_resource", data_path)
+        self.csv_path = Path(data_path)
+
+    def load(self):
+        self.expectations = [
+            {
+                "function": self.check_for_duplicate_references,
+                "name": "Check for Duplicate References",
+                "severity": "error",
+                "responsibility": "system",
+            },
+            {
+                "function": self.validate_references,
+                "name": "Validate References",
+                "severity": "error",
+                "responsibility": "system",
+            },
+        ]
+
+    def check_for_duplicate_references(self):
+        duplicates = {}
+        issues = []
+
+        with self.csv_path.open(newline="") as csvfile:
+            reader = csv.DictReader(csvfile)
+            for row_number, row in enumerate(reader, start=1):
+                ref = row.get("reference")
+                if ref in duplicates:
+                    duplicates[ref].append(row_number)
+                else:
+                    duplicates[ref] = [row_number]
+
+        for ref, rows in duplicates.items():
+            if len(rows) > 1:
+                issues.append(
+                    {
+                        "scope": "duplicate_reference",
+                        "message": f"Duplicate reference '{ref}' found on rows: {', '.join(map(str, rows))}",
+                        "rows": rows,
+                        "reference": ref,
+                        "dataset": "dataset",
+                        "field_name": "reference",
+                        "row_id": str(rows[0]),
+                        "organisation": "organisation",
+                    }
+                )
+
+        return True, "Checked for duplicate references.", issues
+
+    def validate_references(self):
+        issues = []
+
+        with self.csv_path.open(newline="") as csvfile:
+            reader = csv.DictReader(csvfile)
+            for row_number, row in enumerate(reader, start=1):
+                ref = row.get("reference")
+
+                if not ref:  # This will be True for both None and empty strings
+                    issues.append(
+                        {
+                            "scope": "invalid_reference",
+                            "message": f"Reference is missing on row {row_number}.",
+                            "row": row_number,
+                            "reference": ref,
+                            "dataset": "dataset",
+                            "field_name": "reference",
+                            "row_id": str(row_number),
+                            "organisation": "organisation",
+                        }
+                    )
+
+        return len(issues) == 0, "Checked for unpopulated references.", issues

digital_land/expectations/issue.py

@@ -19,6 +19,8 @@ def issue_factory(scope):
         "row-group": RowGroupIssue,
         "row": RowIssue,
         "value": ValueIssue,
+        "duplicate_reference": DuplicateReferenceIssue,
+        "invalid_reference": InvalidReferenceIssue,
     }
     if scope in SCOPE_MAP:
         return SCOPE_MAP[scope]
@@ -129,3 +131,29 @@ def __post_init__(self):
         issue_scope = "value"
         if self.scope != issue_scope:
             raise ValueError(f"scope must be '{issue_scope}'.")
+
+
+@dataclass
+class DuplicateReferenceIssue(Issue):
+    dataset: str
+    field_name: str = field(metadata=config(field_name="field_name"))
+    rows: list = field(metadata=config(field_name="rows"))
+    organisation: str
+
+    def __post_init__(self):
+        issue_scope = "duplicate_reference"
+        if self.scope != issue_scope:
+            raise ValueError(f"scope must be '{issue_scope}'.")
+
+
+@dataclass
+class InvalidReferenceIssue(Issue):
+    dataset: str
+    field_name: str = field(metadata=config(field_name="field_name"))
+    row_id: str = field(metadata=config(field_name="row_id"))
+    organisation: str
+
+    def __post_init__(self):
+        issue_scope = "invalid_reference"
+        if self.scope != issue_scope:
+            raise ValueError(f"scope must be '{issue_scope}'.")

digital_land/phase/convert.py

@@ -11,6 +11,9 @@
 import pandas as pd
 from .load import Stream
 from .phase import Phase
+from digital_land.expectations.checkpoints.converted_resource import (
+    ConvertedResourceCheckpoint,
+)
 
 
 def detect_file_encoding(path):
@@ -187,12 +190,41 @@ def _read_text_file(self, input_path, encoding):
 
         if converted_csv_file:
             f.close()
+            self.run_checkpoint(converted_csv_file)
             reader = read_csv(converted_csv_file)
         else:
             reader = f
 
         return reader
 
+    def run_checkpoint(self, path):
+        checkpoint = ConvertedResourceCheckpoint(data_path=path)
+        checkpoint.load()
+        result = checkpoint.run()
+
+        # Check if the result is not None and is iterable (unpackable)
+        if result is not None and isinstance(result, tuple) and len(result) == 2:
+            checkpoint_result, issues = result
+        else:
+            logging.error("Checkpoint did not return the expected result format.")
+            return
+
+        if issues:
+            for issue in issues:
+                log_message = self.format_issue_message(issue)
+
+                if issue["severity"] == "error":
+                    logging.error(log_message)
+                elif issue["severity"] == "warning":
+                    logging.warning(log_message)
+                else:
+                    logging.info(log_message)
+        else:
+            logging.info(f"Checkpoint completed with result: {checkpoint_result}")
+
+    def format_issue_message(self, issue):
+        return f"Checkpoint Issue: {issue['message']} at line {issue.get('line_number', 'N/A')} (Severity: {issue['severity']})"
+
     def _find_zip_file(self, input_file, suffix=".gml"):
         zip_ = zipfile.ZipFile(input_file)
         files = zip_.namelist()

tests/integration/expectations/test_checkpoint.py

@@ -2,8 +2,11 @@
 import os
 import spatialite
 import pandas as pd
-from csv import DictReader
+from csv import DictReader, DictWriter
 from digital_land.expectations.checkpoints.dataset import DatasetCheckpoint
+from digital_land.expectations.checkpoints.converted_resource import (
+    ConvertedResourceCheckpoint,
+)
 
 
 @pytest.fixture
@@ -43,6 +46,22 @@ def sqlite3_with_entity_tables_path(tmp_path):
     return dataset_path
 
 
+@pytest.fixture
+def csv_path(tmp_path):
+    data = [
+        {"reference": "REF-001", "name": "Test 1"},
+        {"reference": "REF-002", "name": "Test 2"},
+        {"reference": "REF-001", "name": "Test 3"},  # Duplicate
+        {"reference": "", "name": "Test 4"},  # Invalid format
+    ]
+    csv_file = tmp_path / "test_data.csv"
+    with csv_file.open(mode="w", newline="") as f:
+        writer = DictWriter(f, fieldnames=["reference", "name"])
+        writer.writeheader()
+        writer.writerows(data)
+    return csv_file
+
+
 def test_run_checkpoint_success(tmp_path, sqlite3_with_entity_tables_path):
     # load data
     test_entity_data = pd.DataFrame.from_dict({"entity": [1], "name": ["test1"]})
@@ -126,3 +145,33 @@ def test_run_checkpoint_failure(tmp_path, sqlite3_with_entity_tables_path):
     assert issues[0]["rows"] == ""
     assert issues[0]["row"] != ""  # Just check it's there
     assert issues[0]["value"] == ""
+
+
+def test_check_for_duplicate_references(csv_path):
+    checkpoint = ConvertedResourceCheckpoint(data_path=csv_path)
+    checkpoint.load()
+
+    success, message, issues = checkpoint.check_for_duplicate_references()
+
+    assert success is True, "The function should successfully identify issues."
+    assert len(issues) == 1, "There should be one issue identified."
+    assert (
+        issues[0]["scope"] == "duplicate_reference"
+    ), "The issue should be identified as a duplicate reference."
+    assert (
+        "REF-001" in issues[0]["message"]
+    ), "REF-001 should be identified as a duplicate."
+
+
+def test_validate_references(csv_path):
+    checkpoint = ConvertedResourceCheckpoint(data_path=csv_path)
+    checkpoint.load()
+
+    success, message, issues = checkpoint.validate_references()
+
+    assert success is False, "The function should fail due to invalid references."
+    assert len(issues) == 1, "There should be one issue identified."
+    assert (
+        issues[0]["scope"] == "invalid_reference"
+    ), "The issue should be identified as an invalid reference."
+    assert "" in issues[0]["message"], " 4th value should be identified as invalid."