Example: steward-validator — a custom governance rule

"""Steward validator: fails any contract missing a data-steward identifier."""

from __future__ import annotations

from typing import Any, List, Mapping

from fluid_sdk import ContractHelper, Finding, PluginMetadata, Validator


class StewardValidator(Validator):
    """Every contract must carry labels['principal.steward.id']."""

    name = "steward-required"

    @classmethod
    def get_plugin_info(cls) -> PluginMetadata:
        return PluginMetadata(
            name=cls.name,
            role=cls.role,
            display_name="Steward Required Validator",
            description="Every contract must declare a data steward identifier.",
            version="0.1.0",
            author="my-org platform team",
            tags=["governance", "compliance"],
        )

    def plan(self, contract: Mapping[str, Any]) -> List[dict]:
        c = ContractHelper(contract)
        findings: List[Finding] = []

        labels = c.labels  # the contract's top-level `labels:` map
        steward_id = labels.get("principal.steward.id")
        steward_email = labels.get("principal.steward.email")

        if not steward_id:
            findings.append(Finding(
                severity="error",
                code="STEWARD_ID_MISSING",
                message=f"Contract {c.id!r} is missing the required label 'principal.steward.id'.",
                path='labels["principal.steward.id"]',
                remediation="Add labels['principal.steward.id'] with the identifier of the data steward.",
            ))

        if steward_id and not steward_email:
            findings.append(Finding(
                severity="warn",
                code="STEWARD_EMAIL_MISSING",
                message=f"Contract {c.id!r} declares a steward id but no email.",
                path='labels["principal.steward.email"]',
                remediation="Add labels['principal.steward.email'] with the steward's email.",
            ))

        if steward_email and not steward_email.endswith("@my-org.example.com"):
            findings.append(Finding(
                severity="error",
                code="STEWARD_EMAIL_DOMAIN",
                message=f"Steward email {steward_email!r} must be on the @my-org.example.com domain.",
                path='labels["principal.steward.email"]',
                remediation="Use the steward's official my-org email address.",
            ))

        return [f.to_action() for f in findings]

from fluid_sdk.testing import ValidatorTestHarness, LOCAL_CONTRACT
from steward_validator.validator import StewardValidator

CONTRACT_NO_STEWARD = {"id": "p1", "labels": {}}
CONTRACT_GOOD_STEWARD = {"id": "p2", "labels": {
    "principal.steward.id": "emp-12345",
    "principal.steward.email": "alice@my-org.example.com"}}
CONTRACT_STEWARD_NO_EMAIL = {"id": "p3", "labels": {"principal.steward.id": "emp-12345"}}
CONTRACT_WRONG_DOMAIN = {"id": "p4", "labels": {
    "principal.steward.id": "emp-12345",
    "principal.steward.email": "alice@example.com"}}


class TestStewardValidator(ValidatorTestHarness):
    plugin_class = StewardValidator
    sample_contracts = [LOCAL_CONTRACT, CONTRACT_GOOD_STEWARD]

    def _codes(self, contract):
        actions = self.get_plugin().plan(contract)
        return {a["params"]["code"] for a in actions if a["op"] == "emit_finding"}

    def test_missing_steward_id_is_error(self):
        assert "STEWARD_ID_MISSING" in self._codes(CONTRACT_NO_STEWARD)

    def test_steward_id_present_no_email_is_warning(self):
        assert self._codes(CONTRACT_STEWARD_NO_EMAIL) == {"STEWARD_EMAIL_MISSING"}

    def test_wrong_email_domain_is_error(self):
        assert self._codes(CONTRACT_WRONG_DOMAIN) == {"STEWARD_EMAIL_DOMAIN"}

    def test_fully_specified_contract_is_clean(self):
        assert self._codes(CONTRACT_GOOD_STEWARD) == set()