Commit b3407087 authored by Dimitrios Gogos's avatar Dimitrios Gogos
Browse files

feat: enforce schema_version validation in CommandEnvelopeV1 and add related tests

parent b8a20797

src/srm/api/databus/schemas.py

+1 −1
Original line number Diff line number Diff line
@@ -17,7 +17,7 @@ class InboundMessage(BaseModel): 




class CommandEnvelopeV1(BaseModel):

    schema_version: str = "1.0"

    schema_version: Literal["1.0"]

    operation_id: UUID

    correlation_id: str

    requested_at: datetime

tests/api/databus/test_schemas.py

+31 −0
Original line number Diff line number Diff line
@@ -6,6 +6,7 @@ import pytest 
from pydantic import ValidationError



from srm.api.databus.schemas import (

    CommandEnvelopeV1,

    DeployPayloadV1,

    DeployTargetV1,

    NetworkCapabilityDeactivatePayloadV1,
@@ -28,6 +29,7 @@ from srm.domain.models.canonical_parameters.parameters import ( 


def _envelope(**overrides: object) -> dict[str, object]:

    envelope: dict[str, object] = {

        "schema_version": "1.0",

        "operation_id": str(uuid4()),

        "correlation_id": "corr-1",

        "requested_at": "2026-07-03T12:00:00+00:00",
@@ -38,6 +40,35 @@ def _envelope(**overrides: object) -> dict[str, object]: 
    return envelope





class TestSchemaVersion:

    """§B.1: schema_version is required. §A: an unsupported version is unanswerable

    and must be dead-lettered, so it must never parse as if it were 1.0."""



    def test_envelope_requires_schema_version(self) -> None:

        payload = _envelope()

        del payload["schema_version"]



        with pytest.raises(ValidationError, match="schema_version"):

            CommandEnvelopeV1.model_validate(payload)



    @pytest.mark.parametrize("version", ["2.0", "banana"])

    def test_envelope_rejects_unsupported_schema_version(self, version: str) -> None:

        with pytest.raises(ValidationError, match="schema_version"):

            CommandEnvelopeV1.model_validate(_envelope(schema_version=version))



    def test_every_command_inherits_the_rule(self) -> None:

        """The rule lives on the envelope, so a v2 producer cannot have its message

        silently processed as v1 on any command.srm.* subject."""

        payload = _envelope(

            schema_version="2.0",

            service_instance_id=str(uuid4()),

            scale={"replicas": 3},

        )



        with pytest.raises(ValidationError, match="schema_version"):

            SrmServiceScaleV1.model_validate(payload)





def test_deploy_v1_parses_minimal_valid_payload() -> None:

    payload = _envelope(

        service_specification_id=str(uuid4()),