feat: add name_is_a_code expectation

digital_land/expectations/checkpoints/dataset.py

@@ -17,6 +17,7 @@
     count_deleted_entities,
     duplicate_geometry_check,
     fetch_active_resources_for_dataset,
+    name_is_a_code_check,
 )
 
 
@@ -42,6 +43,7 @@ def operation_factory(self, operation_string: str):
             "count_lpa_boundary": count_lpa_boundary,
             "count_deleted_entities": count_deleted_entities,
             "duplicate_geometry_check": duplicate_geometry_check,
+            "name_is_a_code_check": name_is_a_code_check,
         }
         operation = operation_map[operation_string]
         return operation

digital_land/expectations/operations/dataset.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import re
 import sqlite3
 import requests
 import pandas as pd
@@ -464,6 +465,70 @@ def duplicate_geometry_check(conn, spatial_field: str):
     return result, message, details
 
 
+# Flags a name which contains at least one digit, is 20 characters or fewer, and is
+# made up only of letters, digits, dots, slashes and hyphens - a bare reference code
+# rather than a description. The digit lookahead is what stops plain single-word place
+# names such as 'Napsbury' being flagged.
+CODE_LIKE_NAME_RE = re.compile(r"^(?=.*[0-9])[A-Za-z0-9./-]{1,20}$")
+
+
+def name_is_a_code_check(conn):
+    """
+    Checks for entities whose name is just a bare reference code rather than a
+    description, e.g. '59', '0164B2' or '11/802'.
+
+    Failures carry organisation_entity rather than an organisation curie. The dataset
+    package deliberately keeps 'organisation' out of the entity json (package/dataset.py),
+    so organisation_entity is the only provenance available here - resolving it to a
+    curie is the bridge's job.
+
+    Blank names are not flagged here - they are already covered by the existing
+    missing values issue, and the pattern cannot match an empty string anyway.
+
+    args:
+        conn: connection to the dataset being checked, created by the checkpoint class
+    """
+    # length is a cheap superset of the pattern's {1,20}, so this can only ever exclude
+    # rows the pattern would have rejected, and it keeps the regex off most of the table
+    query = """
+        select entity, reference, name, organisation_entity
+        from entity
+        where name is not null
+            and trim(name) != ''
+            and length(name) <= 20
+    """
+    rows = conn.execute(query).fetchall()
+
+    failures = [
+        {
+            "organisation_entity": organisation_entity,
+            "entity": entity,
+            "reference": reference,
+            "name": name,
+        }
+        for entity, reference, name, organisation_entity in rows
+        if CODE_LIKE_NAME_RE.match(name)
+    ]
+
+    failures.sort(
+        key=lambda failure: (
+            failure["organisation_entity"] or "",
+            failure["reference"] or "",
+        )
+    )
+
+    result = len(failures) == 0
+    message = f"{len(failures)} entities have a name which is only a reference code"
+    details = {
+        "failures": failures,
+        "field": "name",
+        "actual": len(failures),
+        "expected": 0,
+    }
+
+    return result, message, details
+
+
 def check_fields_required_after_plan_event(
     conn,
     fields: list,

tests/acceptance/test_run_expectations_on_sqlite.py

@@ -137,6 +137,71 @@ def config_path(tmp_path, specification_dir):
     return config_path
 
 
+@pytest.fixture
+def name_check_dataset_path(tmp_path):
+    """a dataset with one bare code name and one descriptive name"""
+    dataset_path = tmp_path / "name_check.sqlite3"
+    create_entity_table_sql = """
+        CREATE TABLE entity (
+            dataset TEXT,
+            end_date TEXT,
+            entity INTEGER PRIMARY KEY,
+            entry_date TEXT,
+            geojson JSON,
+            geometry TEXT,
+            json JSON,
+            name TEXT,
+            organisation_entity TEXT,
+            point TEXT,
+            prefix TEXT,
+            reference TEXT,
+            start_date TEXT,
+            typology TEXT
+        );
+    """
+    with spatialite.connect(dataset_path) as con:
+        con.execute(create_entity_table_sql)
+        con.executemany(
+            "INSERT INTO entity (entity, reference, name, organisation_entity)"
+            " VALUES (?, ?, ?, ?)",
+            [
+                (1, "CA-59", "59", "122"),
+                (2, "CA-WYM", "Wymondham Conservation Area", "122"),
+            ],
+        )
+
+    return dataset_path
+
+
+@pytest.fixture
+def name_check_config_path(tmp_path, specification_dir):
+    """a configuration enabling name_is_a_code_check, with no organisations and empty
+    parameters, matching how the rule will be written in config"""
+    config_path = tmp_path / "name_check_config.sqlite3"
+    rules = [
+        {
+            "datasets": "test",
+            "organisations": "",
+            "name": "Check no entities have a name which is only a code",
+            "operation": "name_is_a_code_check",
+            "parameters": "{}",
+            "responsibility": "external",
+            "severity": "warning",
+        }
+    ]
+    expect_path = tmp_path / "expect.csv"
+    with open(expect_path, mode="w", newline="") as file:
+        writer = csv.DictWriter(file, fieldnames=rules[0].keys())
+        writer.writeheader()
+        writer.writerows(rules)
+
+    spec = Specification(specification_dir)
+    config = Config(path=config_path, specification=spec)
+    config.create()
+    config.load({"expect": str(tmp_path)})
+    return config_path
+
+
 def test_run_some_expectations(
     tmp_path, organisation_path, dataset_path, config_path, specification_dir, mocker
 ):
@@ -228,3 +293,58 @@ def test_run_some_expectations(
     assert (
         len([result for result in results if result["passed"] == "True"]) == 1
     ), "expected 1 passed expectation"
+
+
+def test_run_name_is_a_code_expectation(
+    tmp_path,
+    organisation_path,
+    name_check_dataset_path,
+    name_check_config_path,
+    specification_dir,
+):
+    dataset = "test"
+    runner = CliRunner()
+
+    result = runner.invoke(
+        expectations_run_dataset_checkpoint,
+        [
+            "--dataset",
+            dataset,
+            "--file-path",
+            str(name_check_dataset_path),
+            "--log-dir",
+            str(tmp_path / "log"),
+            "--configuration-path",
+            str(name_check_config_path),
+            "--organisation-path",
+            str(organisation_path),
+            "--specification-dir",
+            str(specification_dir),
+        ],
+        catch_exceptions=False,
+    )
+
+    assert result.exit_code == 0, result.output
+
+    log_path = tmp_path / f"log/expectation/dataset={dataset}/{dataset}.parquet"
+    assert log_path.exists(), "no logs created"
+
+    conn = duckdb.connect()
+    cursor = conn.execute(f"SELECT * FROM read_parquet('{str(log_path)}')")
+    columns = [desc[0] for desc in cursor.description]
+    results = [dict(zip(columns, row)) for row in cursor.fetchall()]
+
+    assert len(results) == 1, "expected one expectation to have run"
+    log = results[0]
+    assert log["operation"] == "name_is_a_code_check"
+    assert log["passed"] == "False", "the bare code name should have failed the check"
+    assert log["severity"] == "warning"
+    assert log["responsibility"] == "external"
+    assert (
+        log["organisation"] == ""
+    ), "a rule with no organisations logs a blank organisation"
+
+    details = json.loads(log["details"])
+    assert details["actual"] == 1
+    assert [failure["name"] for failure in details["failures"]] == ["59"]
+    assert details["failures"][0]["organisation_entity"] == "122"

tests/integration/expectations/operations/test_dataset.py

@@ -11,6 +11,7 @@
     count_deleted_entities,
     duplicate_geometry_check,
     fetch_active_resources_for_dataset,
+    name_is_a_code_check,
 )
 
 
@@ -722,3 +723,72 @@ def test_check_fields_required_after_plan_event_null_value(
             "actual-date": "2024-12-12",
         }
     ]
+
+
+def test_name_is_a_code_check(dataset_path):
+    entities = [
+        # bare codes, should be flagged
+        (1, "A4D-59", "59", "600001", "local-authority:EXE"),
+        (2, "CA-3B", "3B", "600002", "local-authority:ARU"),
+        (3, "LBO-802", "11/802", "600002", "local-authority:ARU"),
+        # descriptive names, should not be flagged
+        (4, "CA-WYM", "Wymondham Conservation Area", "600001", "local-authority:EXE"),
+        (5, "LBO-GOODGE", "56, GOODGE STREET", "600002", "local-authority:ARU"),
+        # no digit, should not be flagged - this is what the lookahead is for
+        (6, "CA-NAP", "Napsbury", "600001", "local-authority:EXE"),
+        # over 20 characters, should not be flagged
+        (7, "CA-LONG", "123456789012345678901", "600001", "local-authority:EXE"),
+        # blank, belongs to the missing values issue not this check
+        (8, "CA-BLANK", "", "600001", "local-authority:EXE"),
+    ]
+    with spatialite.connect(dataset_path) as con:
+        for entity, reference, name, organisation_entity, _organisation in entities:
+            con.execute(
+                "INSERT INTO entity (entity, reference, name, organisation_entity)"
+                " VALUES (?, ?, ?, ?)",
+                (entity, reference, name, organisation_entity),
+            )
+
+    with spatialite.connect(dataset_path) as con:
+        passed, message, details = name_is_a_code_check(conn=con)
+
+    assert not passed, message
+    assert details["actual"] == 3
+    assert details["expected"] == 0
+    assert details["field"] == "name"
+    assert details["failures"] == [
+        {
+            "organisation_entity": "600001",
+            "entity": 1,
+            "reference": "A4D-59",
+            "name": "59",
+        },
+        {
+            "organisation_entity": "600002",
+            "entity": 2,
+            "reference": "CA-3B",
+            "name": "3B",
+        },
+        {
+            "organisation_entity": "600002",
+            "entity": 3,
+            "reference": "LBO-802",
+            "name": "11/802",
+        },
+    ]
+
+
+def test_name_is_a_code_check_all_descriptive(dataset_path):
+    with spatialite.connect(dataset_path) as con:
+        con.execute(
+            "INSERT INTO entity (entity, reference, name, organisation_entity)"
+            " VALUES (?, ?, ?, ?)",
+            (1, "CA-WYM", "Wymondham Conservation Area", "600001"),
+        )
+
+    with spatialite.connect(dataset_path) as con:
+        passed, message, details = name_is_a_code_check(conn=con)
+
+    assert passed, message
+    assert details["failures"] == []
+    assert details["actual"] == 0