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):
    """Fails the contract validation if a steward identifier is missing."""

    name = "steward-required"

    @classmethod
    def get_plugin_info(cls) -> PluginMetadata:
        return PluginMetadata(
            name=cls.name,
            role=cls.role,
            display_name="Steward Required Validator",
            description=(
                "Enforces that every contract declares "
                "metadata.labels['principal.steward.id']."
            ),
            version="0.1.0",
            author="FLUID SDK Examples",
            tags=["governance", "compliance"],
        )

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

        labels = c.metadata.get("labels") or {}
        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 "
                        f"'principal.steward.id'."
                    ),
                    path='metadata.labels["principal.steward.id"]',
                    remediation=(
                        "Add metadata.labels['principal.steward.id'] with the "
                        "employee/user 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 — "
                        "operations notifications will go nowhere."
                    ),
                    path='metadata.labels["principal.steward.email"]',
                    remediation=(
                        "Add metadata.labels['principal.steward.email'] with the "
                        "team / steward email."
                    ),
                )
            )

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

# tests/test_validator.py (excerpts)
# ValidatorTestHarness (SDK 0.10.0) runs the 13 generic invariants plus
# validator-specific conformance. Subclass it directly for any Validator.

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

    def test_missing_steward_id_is_error(self):
        plugin = self._instantiate()
        actions = plugin.plan(CONTRACT_WITHOUT_STEWARD)
        findings = [a for a in actions if a["op"] == "emit_finding"]
        assert any(f["params"]["severity"] == "error" and
                   f["params"]["code"] == "STEWARD_ID_MISSING"
                   for f in findings)

    def test_missing_steward_email_is_warning(self):
        plugin = self._instantiate()
        actions = plugin.plan(CONTRACT_WITH_ID_NO_EMAIL)
        findings = [a for a in actions if a["op"] == "emit_finding"]
        assert any(f["params"]["severity"] == "warn" and
                   f["params"]["code"] == "STEWARD_EMAIL_MISSING"
                   for f in findings)

    def test_fully_specified_contract_passes_clean(self):
        plugin = self._instantiate()
        actions = plugin.plan(CONTRACT_WITH_STEWARD_AND_EMAIL)
        findings = [a for a in actions if a["op"] == "emit_finding"]
        assert findings == []