Commit 6061d7f9 authored by Dimitrios Gogos's avatar Dimitrios Gogos
Browse files

feat: add validation for federation context in CommandEnvelopeV1 and enhance related tests

parent b3407087

src/srm/api/databus/schemas.py

+7 −0
Original line number Diff line number Diff line
@@ -25,6 +25,13 @@ class CommandEnvelopeV1(BaseModel): 
    federation_partner_ref: str | None = None

    source: Literal["nbi_camara", "nbi_tmf", "operator_portal", "federation"]



    @model_validator(mode="after")

    def validate_federation_context(self) -> "CommandEnvelopeV1":

        if self.source == "federation" and self.federation_partner_ref is None:

            raise ValueError("federation_partner_ref is required when source=federation")



        return self





class PlacementConstraintsV1(BaseModel):

    model_config = {"extra": "allow"}

tests/api/databus/test_nats_subscriber.py

+38 −4
Original line number Diff line number Diff line
@@ -4,6 +4,7 @@ from dataclasses import dataclass, field 
from unittest.mock import AsyncMock, MagicMock



import pytest

import structlog.testing



from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

from srm.api.databus.nats_subscriber import (
@@ -88,14 +89,47 @@ async def test_handle_message_defaults_headers_to_empty_dict_when_none( 
    assert inbound.headers == {}





async def test_handle_message_swallows_router_exceptions(connection_manager: MagicMock) -> None:

async def test_handle_message_passes_malformed_payload_through_unparsed(

    connection_manager: MagicMock,

) -> None:

    subscriber, router = make_subscriber(connection_manager)

    msg = FakeMsg(subject="command.srm.service.deploy", data=b'{"operation_id": ')



    await subscriber._handle_message(msg)  # type: ignore[arg-type]



    router.assert_awaited_once()

    assert router.await_args is not None

    (inbound,) = router.await_args.args

    assert inbound.payload == b'{"operation_id": '





async def test_handle_message_logs_dropped_command_when_router_fails(

    connection_manager: MagicMock,

) -> None:

    subscriber, router = make_subscriber(connection_manager)

    router.side_effect = RuntimeError("handler blew up")

    msg = FakeMsg(subject="command.srm.service.deploy", data=b"{}")



    with structlog.testing.capture_logs() as logs:

        await subscriber._handle_message(msg)  # type: ignore[arg-type]



    router.assert_awaited_once()

    assert [(entry["event"], entry["subject"], entry["log_level"]) for entry in logs] == [

        ("databus_command_dropped", "command.srm.service.deploy", "error")

    ]



EXPECTED_COMMAND_SUBJECTS = [

    "command.srm.service.deploy",

    "command.srm.service.scale",

    "command.srm.service.terminate",

    "command.srm.network.capability.activate",

    "command.srm.network.capability.update",

    "command.srm.network.capability.deactivate",

]





def test_command_subjects_matches_the_contract() -> None:

    assert sorted(COMMAND_SUBJECTS) == sorted(EXPECTED_COMMAND_SUBJECTS)





async def test_subscribe_to_subjects_registers_all_command_subjects(
@@ -103,8 +137,8 @@ async def test_subscribe_to_subjects_registers_all_command_subjects( 
) -> None:

    subscribers = await subscribe_to_subjects(connection_manager)



    assert sorted(sub._subject for sub in subscribers) == sorted(COMMAND_SUBJECTS)

    assert connection_manager.client.subscribe.await_count == len(COMMAND_SUBJECTS)

    assert sorted(sub._subject for sub in subscribers) == sorted(EXPECTED_COMMAND_SUBJECTS)

    assert connection_manager.client.subscribe.await_count == len(EXPECTED_COMMAND_SUBJECTS)





class TestCommandDeliveryDurability:

tests/api/databus/test_schemas.py

+44 −2
Original line number Diff line number Diff line
@@ -69,6 +69,50 @@ class TestSchemaVersion: 
            SrmServiceScaleV1.model_validate(payload)





class TestSource:

    @pytest.mark.parametrize("source", ["nbi_camara", "nbi_tmf", "operator_portal"])

    def test_envelope_accepts_each_non_federation_source(self, source: str) -> None:

        assert CommandEnvelopeV1.model_validate(_envelope(source=source)).source == source



    @pytest.mark.parametrize("source", ["nbi_rest", "FEDERATION", ""])

    def test_envelope_rejects_unknown_source(self, source: str) -> None:

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

            CommandEnvelopeV1.model_validate(_envelope(source=source))





class TestFederationContext:

    def test_federation_source_requires_partner_ref(self) -> None:

        with pytest.raises(ValidationError, match="federation_partner_ref is required"):

            CommandEnvelopeV1.model_validate(_envelope(source="federation"))



    def test_federation_source_rejects_explicit_null_partner_ref(self) -> None:

        with pytest.raises(ValidationError, match="federation_partner_ref is required"):

            CommandEnvelopeV1.model_validate(

                _envelope(source="federation", federation_partner_ref=None)

            )



    def test_federation_source_accepts_partner_ref(self) -> None:

        command = CommandEnvelopeV1.model_validate(

            _envelope(source="federation", federation_partner_ref="ptr-OperatorB")

        )



        assert command.federation_partner_ref == "ptr-OperatorB"



    def test_non_federation_source_may_omit_partner_ref(self) -> None:

        assert CommandEnvelopeV1.model_validate(_envelope()).federation_partner_ref is None



    def test_every_command_inherits_the_rule(self) -> None:

        payload = _envelope(

            source="federation",

            service_specification_id=str(uuid4()),

            targets=[{"app_instance_id": str(uuid4())}],

            deploy={},

        )



        with pytest.raises(ValidationError, match="federation_partner_ref is required"):

            SrmServiceDeployV1.model_validate(payload)





def test_deploy_v1_parses_minimal_valid_payload() -> None:

    payload = _envelope(

        service_specification_id=str(uuid4()),
@@ -109,7 +153,6 @@ def test_deploy_target_accepts_zone_and_domain_pin() -> None: 




class TestPinShape:



    def test_deploy_target_rejects_domain_pin_without_zone(self) -> None:

        with pytest.raises(ValidationError, match="domain_id requires zone_id"):

            DeployTargetV1(app_instance_id=uuid4(), domain_id=uuid4())
@@ -255,7 +298,6 @@ def test_network_capability_payload_requires_capability_type() -> None: 




class TestNetworkCapabilityRealizationRef:



    def test_accepts_external_ref_alone(self) -> None:

        ref = NetworkCapabilityUpdatePayloadV1(

            capability_type="qod_session",

tests/api/test_app.py

+70 −0
Original line number Diff line number Diff line 
from collections.abc import Iterator

from unittest.mock import AsyncMock, MagicMock, patch



import pytest

from fastapi import FastAPI



from srm.config import get_settings

from srm.main import lifespan





def test_create_app_uses_settings(app: FastAPI) -> None:
@@ -8,3 +13,68 @@ def test_create_app_uses_settings(app: FastAPI) -> None: 
    assert app.title == settings.app_name

    assert app.version == settings.app_version

    assert app.description == settings.app_description





@pytest.fixture

def stub_database() -> Iterator[AsyncMock]:

    """Lifespan builds the DB engine before touching NATS; stubbing it keeps these tests

    focused on the DataBus stage and off Docker."""

    engine = AsyncMock()

    with (

        patch(

            "srm.main.build_engine_and_session_maker",

            new_callable=AsyncMock,

            return_value=(engine, MagicMock()),

        ),

        patch("srm.main.schema_initialization", new_callable=AsyncMock),

    ):

        yield engine





class TestLifespanDatabusInit:

    async def test_startup_fails_when_databus_connection_fails(

        self, app: FastAPI, stub_database: AsyncMock

    ) -> None:

        with patch(

            "srm.main.init_databus_manager",

            new_callable=AsyncMock,

            side_effect=OSError("nats unavailable"),

        ):

            with pytest.raises(OSError, match="nats unavailable"):

                async with lifespan(app):

                    pass



    async def test_startup_fails_when_subject_subscription_fails(

        self, app: FastAPI, stub_database: AsyncMock

    ) -> None:

        with (

            patch("srm.main.init_databus_manager", new_callable=AsyncMock),

            patch(

                "srm.main.subscribe_to_subjects",

                new_callable=AsyncMock,

                side_effect=OSError("subscription refused"),

            ),

        ):

            with pytest.raises(OSError, match="subscription refused"):

                async with lifespan(app):

                    pass



    async def test_successful_startup_exposes_databus_state(

        self, app: FastAPI, stub_database: AsyncMock

    ) -> None:

        manager, subscribers = AsyncMock(), [MagicMock(), MagicMock()]



        with (

            patch("srm.main.init_databus_manager", new_callable=AsyncMock, return_value=manager),

            patch(

                "srm.main.subscribe_to_subjects",

                new_callable=AsyncMock,

                return_value=subscribers,

            ),

        ):

            async with lifespan(app):

                assert app.state.databus_connection_manager is manager

                assert app.state.databus_subscribers == subscribers



        manager.close.assert_awaited_once()

        assert app.state.databus_subscribers == []

tests/integration/test_databus.py

+4 −6
Original line number Diff line number Diff line
@@ -10,14 +10,12 @@ from testcontainers.nats import NatsContainer 


from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

from srm.adapters.databus.nats_publisher import NatsPublisher

from srm.api.databus.nats_subscriber import (

    COMMAND_SUBJECTS,

    NatsSubscriber,

    subscribe_to_subjects,

)

from srm.api.databus.nats_subscriber import NatsSubscriber, subscribe_to_subjects

from srm.api.databus.schemas import InboundMessage

from srm.config import NatsSettings



from tests.api.databus.test_nats_subscriber import EXPECTED_COMMAND_SUBJECTS





@pytest.fixture(scope="session")

def nats_url() -> Generator[str, None, None]:
@@ -117,6 +115,6 @@ async def test_subscribe_to_subjects_registers_all_command_subjects( 
) -> None:

    subscribers = await subscribe_to_subjects(connection_manager)



    assert sorted(sub._subject for sub in subscribers) == sorted(COMMAND_SUBJECTS)

    assert sorted(sub._subject for sub in subscribers) == sorted(EXPECTED_COMMAND_SUBJECTS)

    for sub in subscribers:

        assert sub._subscription is not None