orchestrata_ultimate.cxp.validator

Reference › orchestrata_ultimate › cxp › validator

CXPValidator checks packets against a list of named rules and against JSON schemas (Spec 004 US-007). Each ValidationRule has a name, a check function returning a pair of flag and message, and an error message. The default rules, built in _setup_default_rules, require a valid sender, a valid recipient, a payload that is a dictionary, bytes or packet payload, a sender different from the recipient, a priority from 0 to 10 and a non-empty subject within the length limit. validate runs every rule, collects messages that start with Warning: as warnings and the others as errors, turns an exception inside a rule into an error, and returns a ValidationResult with the packet id. validate_schema registers an ad-hoc schema in a schema registry and returns the same result type. add_rule, remove_rule and register_validator extend or trim the rule set. See headers, exceptions and cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

ValidationRule (dataclass)

Decorators: dataclass, traces(spec='004', story='US-007', reqs=['REQ-004-US-007-018', 'REQ-004-US-007-021']) · Line 43

Individual validation rule.

CXPValidator

Line 51

Validates CXP packets against the CXP specification.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/cxp/validator.py

"""
CXP Validator - Validates CXP packets.

Validates packet structure, fields, and cross-field constraints
against the CXP specification.

Spec: 004-cxp
Author: Mohamed
"""

import logging
import re
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, PacketPayload, ValidationResult
from orchestrata_ultimate.cxp.constants import VOLK_ALIAS_REGEX

logger = logging.getLogger(__name__)

# Volk alias pattern: U: followed by uppercase then lowercase
VOLK_ALIAS_PATTERN = re.compile(r"^U:[A-Z][a-z]$")

# Subject constraints
SUBJECT_MAX_LENGTH = 200

# Priority range
PRIORITY_MIN = 0
PRIORITY_MAX = 10

# Semver pattern
SEMVER_PATTERN = re.compile(r"^\d+\.\d+\.\d+$")


@dataclass
@traces(
    spec="004",
    story="US-007",
    reqs=["REQ-004-US-007-018", "REQ-004-US-007-021"],
)
class ValidationRule:
    """Individual validation rule."""

    name: str
    validate: Callable
    error_message: str


class CXPValidator:
    """Validates CXP packets against the CXP specification."""

    def __init__(self, config: dict | None = None) -> None:
        self._config = config or {}
        self._rules: list[ValidationRule] = []
        self._setup_default_rules()
        logger.info("CXPValidator initialized")

    @traces(
        spec="004",
        story="US-007",
        reqs=["REQ-004-US-007-003", "REQ-004-US-007-020"],
    )
    def _setup_default_rules(self) -> None:
        self._rules = [
            ValidationRule(
                name="valid_sender",
                validate=lambda p: self._check_sender(p),
                error_message="Invalid or missing sender",
            ),
            ValidationRule(
                name="valid_recipient",
                validate=lambda p: self._check_recipient(p),
                error_message="Invalid or missing recipient",
            ),
            ValidationRule(
                name="valid_payload",
                validate=lambda p: self._check_payload(p),
                error_message="Payload must be a dictionary, bytes, or PacketPayload",
            ),
            ValidationRule(
                name="sender_not_recipient",
                validate=lambda p: self._check_sender_not_recipient(p),
                error_message="Sender and recipient must be different",
            ),
            ValidationRule(
                name="valid_priority",
                validate=lambda p: self._check_priority(p),
                error_message="Priority must be 0-10",
            ),
            ValidationRule(
                name="valid_subject",
                validate=lambda p: self._check_subject(p),
                error_message="Subject too long or empty",
            ),
        ]

    @traces(
        spec="004",
        story="US-007",
        reqs=["REQ-004-US-007-008", "REQ-004-US-007-012"],
    )
    def validate(self, packet: CXPPacket) -> ValidationResult:
        errors = []
        warnings = []

        for rule in self._rules:
            try:
                valid, message = rule.validate(packet)
                if not valid:
                    if message.startswith("Warning:"):
                        warnings.append(message)
                    else:
                        errors.append(message)
            except Exception as e:
                errors.append(f"Validation error: {str(e)}")

        return ValidationResult(
            valid=len(errors) == 0,
            packet_id=packet.message_id,
            errors=errors,
            warnings=warnings,
        )

    @traces(
        spec="004",
        story="US-003",
        reqs=[
            "REQ-004-PERF-007",
            "REQ-004-US-003-004",
            "REQ-004-US-007-009",
            "REQ-004-US-007-010",
            "REQ-004-US-007-011",
        ],
    )
    def validate_schema(
        self, packet: CXPPacket, schema: dict[str, Any]
    ) -> ValidationResult:
        """Validate packet against a JSON schema. Delegates to SchemaRegistry."""
        from orchestrata_ultimate.cxp.schema import SchemaRegistry

        registry = SchemaRegistry()
        schema_name = f"_adhoc_{id(schema)}"
        registry.register_schema(schema_name, schema)

        packet_data = packet.to_dict()
        is_valid, errors = registry.validate_against_schema(packet_data, schema_name)

        return ValidationResult(
            valid=is_valid,
            packet_id=packet.message_id,
            errors=errors,
            warnings=[],
        )

    @traces(
        spec="004",
        story="US-007",
        reqs=["REQ-004-US-007-018", "REQ-004-US-007-019"],
    )
    def register_validator(self, field: str, validator: Callable[..., bool]) -> None:
        """Register a custom field validator. Wraps in add_rule()."""

        def _field_check(packet: CXPPacket) -> tuple[bool, str]:
            value = getattr(packet, field, None)
            if validator(value):
                return True, ""
            return False, f"Custom validation failed for field '{field}'"

        self.add_rule(
            name=f"custom_{field}",
            validate=_field_check,
            error_message=f"Custom validation failed for field '{field}'",
        )

    @traces(spec="004", story="US-001", reqs=["REQ-004-US-001-014"])
    def _check_sender(self, packet: CXPPacket) -> tuple[bool, str]:
        if not packet.from_agent:
            return False, "Missing sender"
        if not VOLK_ALIAS_REGEX.match(packet.from_agent):
            return (
                False,
                f"Invalid sender alias format: {packet.from_agent!r} (expected U:Xx pattern)",
            )
        return True, ""

    @traces(spec="004", story="US-001", reqs=["REQ-004-US-001-015"])
    def _check_recipient(self, packet: CXPPacket) -> tuple[bool, str]:
        if not packet.to_agent:
            return False, "Missing recipient"
        if not VOLK_ALIAS_REGEX.match(packet.to_agent):
            return (
                False,
                f"Invalid recipient alias format: {packet.to_agent!r} (expected U:Xx pattern)",
            )
        return True, ""

    @traces(
        spec="004",
        story="US-001",
        reqs=[
            "REQ-004-US-001-019",
            "REQ-004-US-003-015",
        ],
    )
    def _check_payload(self, packet: CXPPacket) -> tuple[bool, str]:
        if packet.payload is None:
            return True, ""
        if not isinstance(packet.payload, (dict, bytes, PacketPayload)):
            return False, "Payload must be a dictionary, bytes, or PacketPayload"
        return True, ""

    @traces(
        spec="004",
        story="US-007",
        reqs=["REQ-004-PERF-008", "REQ-004-US-007-015"],
    )
    def _check_sender_not_recipient(self, packet: CXPPacket) -> tuple[bool, str]:
        if packet.from_agent == packet.to_agent:
            return False, f"Sender and recipient must be different: {packet.from_agent}"
        return True, ""

    @traces(
        spec="004",
        story="US-007",
        reqs=["REQ-004-US-007-006", "REQ-004-US-009-004"],
    )
    def _check_priority(self, packet: CXPPacket) -> tuple[bool, str]:
        p = packet.priority
        if not isinstance(p, int) or p < PRIORITY_MIN or p > PRIORITY_MAX:
            return False, f"Priority {p} out of range [{PRIORITY_MIN}, {PRIORITY_MAX}]"
        return True, ""

    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-017", "REQ-004-US-007-005"],
    )
    def _check_subject(self, packet: CXPPacket) -> tuple[bool, str]:
        if not packet.subject:
            return False, "Subject must not be empty"
        if len(packet.subject) > SUBJECT_MAX_LENGTH:
            return (
                False,
                f"Subject exceeds {SUBJECT_MAX_LENGTH} chars: {len(packet.subject)}",
            )
        return True, ""

    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-021"])
    def add_rule(self, name: str, validate: Callable, error_message: str) -> None:
        self._rules.append(
            ValidationRule(name=name, validate=validate, error_message=error_message)
        )

    def remove_rule(self, name: str) -> None:
        self._rules = [r for r in self._rules if r.name != name]