BDRSPS-811 Add mutual exclusive check

abis_mapping/plugins/__init__.py

@@ -7,6 +7,7 @@
 from . import empty
 from . import list
 from . import logical_or
+from . import mutual_exclusion
 from . import mutual_inclusion
 from . import sites_geometry
 from . import tabular

abis_mapping/plugins/mutual_exclusion.py

@@ -0,0 +1,47 @@
+"""Provides extra frictionless mutual exclusion checks for the package"""
+
+
+# Third-Party
+import frictionless
+import frictionless.errors
+import attrs
+
+# Typing
+from typing import Iterator
+
+
+@attrs.define(kw_only=True, repr=False)
+class MutuallyExclusive(frictionless.Check):
+    """Checks that mutually exclusive columns are not provided together."""
+
+    # Check Attributes
+    type = "mutually-exclusive"
+    Errors = [frictionless.errors.RowConstraintError]
+
+    # Attributes specific to this check
+    field_names: list[str]
+
+    def validate_row(self, row: frictionless.Row) -> Iterator[frictionless.Error]:
+        """Called to validate the given row (on every row).
+
+        Args:
+            row (frictionless.Row): The row to check the mutual exclusivity of.
+
+        Yields:
+            frictionless.Error: For when the mutual exclusion is violated.
+        """
+        # Retrieve Field Names for cells with values
+        fields_provided = [f for f in self.field_names if row[f] not in (None, "")]
+
+        # Check Mutual Exclusivity
+        # If 2 or more of the mutually exclusive fields are provided, that's an error.
+        if len(fields_provided) >= 2:
+            # Yield Error
+            yield frictionless.errors.RowConstraintError.from_row(
+                row=row,
+                note=(
+                    f"The columns {self.field_names} are mutually exclusive and must "
+                    f"not be provided together "
+                    f"(columns {fields_provided} were provided together)"
+                ),
+            )

tests/plugins/test_mutual_exclusion.py

@@ -0,0 +1,70 @@
+"""Provides Unit Tests for the `abis_mapping.plugins.mutual_exclusion` module"""
+
+
+# Third-Party
+import frictionless
+
+# Local
+from abis_mapping import plugins
+
+
+def test_checks_mutually_exclusive_valid() -> None:
+    """Tests the MutuallyExclusive Checker"""
+    # Construct Fake Resource
+    resource = frictionless.Resource(
+        source=[
+            # valid: neither field provided
+            {"C0": "R1", "C1": "", "C2": None},
+            # valid: only one field provided
+            {"C0": "R4", "C1": "", "C2": 1},
+            {"C0": "R5", "C1": "", "C2": 0},
+            {"C0": "R6", "C1": "A", "C2": None},
+        ],
+    )
+
+    # Validate
+    report: frictionless.Report = resource.validate(
+        checklist=frictionless.Checklist(
+            checks=[
+                plugins.mutual_exclusion.MutuallyExclusive(
+                    field_names=["C1", "C2"],
+                ),
+            ],
+        ),
+    )
+
+    # Check
+    assert report.valid
+
+
+def test_checks_mutually_exclusive_not_valid() -> None:
+    """Tests the MutuallyExclusive Checker"""
+    # Construct Fake Resource
+    resource = frictionless.Resource(
+        source=[
+            # not valid: both fields provided
+            {"C0": "R1", "C1": "A", "C2": 1},
+            {"C0": "R2", "C1": "A", "C2": 0},
+        ],
+    )
+
+    # Validate
+    report: frictionless.Report = resource.validate(
+        checklist=frictionless.Checklist(
+            checks=[
+                plugins.mutual_exclusion.MutuallyExclusive(
+                    field_names=["C1", "C2"],
+                ),
+            ],
+        ),
+    )
+
+    # Check
+    assert not report.valid
+    assert len(report.tasks[0].errors) == 2
+    for error in report.tasks[0].errors:
+        assert error.type == 'row-constraint'
+        assert error.note == (
+            "The columns ['C1', 'C2'] are mutually exclusive and must not be "
+            "provided together (columns ['C1', 'C2'] were provided together)"
+        )