feat: network capability deactivate command

Paris Stentoumisrequested to merge
refactor/network-capability-deactivate into refactor/tf-sdk-wiring

Summary

  • Implement command.srm.network.capability.deactivate end-to-end: replace the NATS no-op with a real command coordinator and use case.
  • Resolve the target realization by service_instance_id or external_ref (exactly one live, active match for the caller's app_provider_id); create a DEACTIVATE_CAPABILITY order with target_service_instance_id set to the session holder.
  • Call adapter.deactivate_capability, treat already-gone backends as success (ResourceNotFound / runtime-state probe), terminate the capability instance, and terminate the holder when it was the last live child.
  • Mirror activate/terminate patterns for accept → status → complete / fail_after_accept, including terminate-style replay of COMPLETED/FAILED orders.

Test plan

  • Unit: accept happy paths (service_instance_id / external_ref), zero/many matches, unsupported & sync-query kinds, non-active / missing refs, replay
  • Unit: complete success, already-gone paths, non-terminated adapter result, last-live-child holder teardown, fail_after_accept
  • Application: coordinator invalid payload → failed_before_start; accept → status + completed; replay; fail path
  • API: dependencies wiring + NATS subject bound to deactivate coordinator (not no-op)
  • pytest for the above + ruff/mypy on touched modules

Merge request reports

src/srm/api/databus/dependencies.py

+41 −0
Changes for src/srm/api/databus/dependencies.py: 41 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -23,6 +23,9 @@ from srm.application.command_handlers.federated_deploy_outcome import ( 
from srm.application.command_handlers.network_capability_activate import (

    NetworkCapabilityActivateCommandCoordinator,

)

from srm.application.command_handlers.network_capability_deactivate import (

    NetworkCapabilityDeactivateCommandCoordinator,

)

from srm.application.command_handlers.terminate_service import TerminateServiceCommandCoordinator

from srm.application.services.capability_placement import CapabilityPlacementPlanner

from srm.application.services.device_targeted_control_path import (
@@ -32,6 +35,9 @@ from srm.application.use_cases.deploy_service import DeployServiceUseCase 
from srm.application.use_cases.network_capability_activate import (

    NetworkCapabilityActivateUseCase,

)

from srm.application.use_cases.network_capability_deactivate import (

    NetworkCapabilityDeactivateUseCase,

)

from srm.application.use_cases.terminate_service import TerminateServiceUseCase

from srm.domain.ports.databus.publisher import DataBusPublisher

from srm.domain.ports.transformation_functions import ControlPathRegistry
@@ -158,6 +164,41 @@ def get_terminate_service_command_coordinator( 
    )





def build_network_capability_deactivate_use_case(

    session: AsyncSession,

    publisher: DataBusPublisher,

    control_paths: ControlPathRegistry,

) -> NetworkCapabilityDeactivateUseCase:

    return NetworkCapabilityDeactivateUseCase(

        service_orders=SqlServiceOrderRepository(session),

        service_instances=SqlServiceInstanceRepository(session),

        capability_instances=SqlCapabilityInstanceRepository(session),

        capabilities=SqlCapabilityRepository(session),

        domains=SqlDomainRepository(session),

        zones=SqlZoneRepository(session),

        publisher=publisher,

        control_paths=control_paths,

    )





def get_network_capability_deactivate_command_coordinator(

    session_maker: async_sessionmaker[AsyncSession],

    publisher: DataBusPublisher,

    control_paths: ControlPathRegistry,

) -> NetworkCapabilityDeactivateCommandCoordinator:

    def use_case_factory(

        session: AsyncSession,

        publisher: DataBusPublisher,

    ) -> NetworkCapabilityDeactivateUseCase:

        return build_network_capability_deactivate_use_case(session, publisher, control_paths)



    return NetworkCapabilityDeactivateCommandCoordinator(

        session_maker=session_maker,

        use_case_factory=use_case_factory,

        publisher=publisher,

    )





def get_federated_deploy_outcome_coordinator(

    session_maker: async_sessionmaker[AsyncSession],

    publisher: DataBusPublisher,

src/srm/api/databus/nats_subscriber.py

+8 −1
Changes for src/srm/api/databus/nats_subscriber.py: 8 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -12,6 +12,7 @@ from srm.api.databus.dependencies import ( 
    get_deploy_service_command_coordinator,

    get_federated_deploy_outcome_coordinator,

    get_network_capability_activate_command_coordinator,

    get_network_capability_deactivate_command_coordinator,

    get_terminate_service_command_coordinator,

)

from srm.api.databus.schemas import AnswerableCommandV1, AnswerableEventV1, InboundMessage
@@ -37,12 +38,12 @@ async def _noop_callback(envelope: AnswerableCommandV1) -> None: 
COMMAND_HANDLERS: dict[str, MessageHandler] = {

    "command.srm.service.scale": _noop_callback,

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

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

}



COMMAND_SUBJECTS = (

    "command.srm.service.deploy",

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

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

    "command.srm.service.terminate",

    *COMMAND_HANDLERS.keys(),

)
@@ -72,6 +73,11 @@ async def subscribe_to_subjects( 
        publisher=publisher,

        control_paths=control_paths,

    )

    network_deactivate_coordinator = get_network_capability_deactivate_command_coordinator(

        session_maker,

        publisher=publisher,

        control_paths=control_paths,

    )

    federated_outcome_coordinator = get_federated_deploy_outcome_coordinator(

        session_maker, publisher=publisher

    )
@@ -79,6 +85,7 @@ async def subscribe_to_subjects( 
    command_handlers = {

        "command.srm.service.deploy": deploy_coordinator.handle,

        "command.srm.network.capability.activate": network_activate_coordinator.handle,

        "command.srm.network.capability.deactivate": network_deactivate_coordinator.handle,

        "command.srm.service.terminate": terminate_coordinator.handle,

        **COMMAND_HANDLERS,

    }

src/srm/application/command_handlers/network_capability_deactivate.py

0 → 100644
+283 −0
Changes for src/srm/application/command_handlers/network_capability_deactivate.py: 283 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from __future__ import annotations



from collections.abc import Callable

from datetime import UTC, datetime

from typing import Any

from uuid import UUID



from pydantic import ValidationError

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker



from srm.api.databus.schemas import AnswerableCommandV1, SrmNetworkCapabilityDeactivateV1

from srm.application.use_cases.network_capability_deactivate import (

    NetworkCapabilityDeactivateCommand,

    NetworkCapabilityDeactivatePayloadCommand,

    NetworkCapabilityDeactivateResult,

    NetworkCapabilityDeactivateUseCase,

)

from srm.domain.ports.databus.events import (

    OperationCompletedStatus,

    OperationStatusState,

    SrmOperationCompletedV1,

    SrmOperationStatusV1,

)

from srm.domain.ports.databus.publisher import DataBusPublisher



NetworkCapabilityDeactivateUseCaseFactory = Callable[

    [AsyncSession, DataBusPublisher],

    NetworkCapabilityDeactivateUseCase,

]





class NetworkCapabilityDeactivateCommandCoordinator:

    def __init__(

        self,

        session_maker: async_sessionmaker[AsyncSession],

        use_case_factory: NetworkCapabilityDeactivateUseCaseFactory,

        publisher: DataBusPublisher,

    ) -> None:

        self._session_maker = session_maker

        self._use_case_factory = use_case_factory

        self._publisher = publisher



    async def handle(self, envelope: AnswerableCommandV1) -> None:

        command = await self._parse_command(envelope)

        if command is None:

            return



        accepted = await self._accept(command)

        if accepted.completed_event is not None:

            await self._publisher.publish(

                "event.srm.operation.completed",

                accepted.completed_event.model_dump(mode="json"),

            )

            return

        if accepted.duplicate_order_id is not None:

            await self._publish_accepted(

                command,

                accepted.duplicate_order_id,

                service_instance_id=accepted.service_instance_id,

            )

            return

        if accepted.accepted_order_id is None:

            return



        await self._publish_accepted(

            command,

            accepted.accepted_order_id,

            service_instance_id=accepted.service_instance_id,

        )



        try:

            completed_event = await self._complete(command, accepted)

        except Exception as exc:

            completed_event = await self._fail_after_accept(command, accepted, exc)

        if completed_event is not None:

            await self._publisher.publish(

                "event.srm.operation.completed",

                completed_event.model_dump(mode="json"),

            )



    async def _parse_command(

        self,

        envelope: AnswerableCommandV1,

    ) -> NetworkCapabilityDeactivateCommand | None:

        try:

            deactivate_message = SrmNetworkCapabilityDeactivateV1.model_validate(

                envelope.model_dump(mode="python")

            )

        except ValidationError:

            await self._publish_failed_before_start_for_invalid_command(envelope)

            return None



        return NetworkCapabilityDeactivateCommand(

            operation_id=deactivate_message.operation_id,

            correlation_id=deactivate_message.correlation_id,

            requested_at=deactivate_message.requested_at,

            app_provider_id=deactivate_message.app_provider_id,

            federation_partner_ref=deactivate_message.federation_partner_ref,

            source=deactivate_message.source,

            service_specification_id=deactivate_message.service_specification_id,

            network_capability=NetworkCapabilityDeactivatePayloadCommand(

                capability_type=deactivate_message.network_capability.capability_type,

                service_instance_id=deactivate_message.network_capability.service_instance_id,

                external_ref=deactivate_message.network_capability.external_ref,

                grace_period_seconds=deactivate_message.network_capability.grace_period_seconds,

            ),

            payload_snapshot=deactivate_message.model_dump(mode="json"),

        )



    async def _accept(

        self,

        command: NetworkCapabilityDeactivateCommand,

    ) -> NetworkCapabilityDeactivateResult:

        async with self._session_maker() as session:

            try:

                use_case = self._use_case_factory(session, self._publisher)

                accepted = await use_case.accept(command)

                await session.commit()

                return accepted

            except Exception:

                await session.rollback()

                raise



    async def _complete(

        self,

        command: NetworkCapabilityDeactivateCommand,

        accepted: NetworkCapabilityDeactivateResult,

    ) -> SrmOperationCompletedV1 | None:

        async with self._session_maker() as session:

            try:

                use_case = self._use_case_factory(session, self._publisher)

                completed_event = await use_case.complete(command, accepted)

                await session.commit()

                return completed_event

            except Exception:

                await session.rollback()

                raise



    async def _fail_after_accept(

        self,

        command: NetworkCapabilityDeactivateCommand,

        accepted: NetworkCapabilityDeactivateResult,

        error: Exception,

    ) -> SrmOperationCompletedV1:

        async with self._session_maker() as session:

            try:

                use_case = self._use_case_factory(session, self._publisher)

                completed_event = await use_case.fail_after_accept(command, accepted, error)

                await session.commit()

                return completed_event

            except Exception:

                await session.rollback()

                return self._failed_after_accept_event(command, accepted, error)



    async def _publish_accepted(

        self,

        command: NetworkCapabilityDeactivateCommand,

        service_order_id: UUID,

        *,

        service_instance_id: UUID | None = None,

    ) -> None:

        status_event = SrmOperationStatusV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            service_order_id=service_order_id,

            service_instance_id=(

                service_instance_id or command.network_capability.service_instance_id

            ),

            capability=command.network_capability.capability_type,

            state=OperationStatusState.ACCEPTED,

            correlation_id=command.correlation_id,

            emitted_at=datetime.now(UTC),

        )

        await self._publisher.publish(

            "event.srm.operation.status",

            status_event.model_dump(mode="json"),

        )



    async def _publish_failed_before_start_for_invalid_command(

        self,

        envelope: AnswerableCommandV1,

    ) -> None:

        now = datetime.now(UTC)

        status_event = SrmOperationStatusV1(

            schema_version="1.0",

            operation_id=envelope.operation_id,

            service_order_id=None,

            service_instance_id=self._service_instance_id_from_envelope(envelope),

            capability=self._capability_from_envelope(envelope),

            state=OperationStatusState.FAILED_BEFORE_START,

            correlation_id=envelope.correlation_id,

            emitted_at=now,

        )

        completed_event = SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=envelope.operation_id,

            status=OperationCompletedStatus.FAILED,

            service_order_id=None,

            error={

                "type": "about:blank",

                "title": "Network capability deactivate command is invalid.",

                "status": 400,

                "detail": "Network capability deactivate command payload is invalid.",

            },

            correlation_id=envelope.correlation_id,

            completed_at=now,

        )

        await self._publisher.publish(

            "event.srm.operation.status",

            status_event.model_dump(mode="json"),

        )

        await self._publisher.publish(

            "event.srm.operation.completed",

            completed_event.model_dump(mode="json"),

        )



    def _capability_from_envelope(self, envelope: AnswerableCommandV1) -> str | None:

        if envelope.model_extra is None:

            return None



        network_capability = envelope.model_extra.get("network_capability")

        if not isinstance(network_capability, dict):

            return None



        capability_type = network_capability.get("capability_type")

        if isinstance(capability_type, str) and capability_type:

            return capability_type

        return None



    def _service_instance_id_from_envelope(self, envelope: AnswerableCommandV1) -> UUID | None:

        if envelope.model_extra is None:

            return None



        network_capability = envelope.model_extra.get("network_capability")

        if not isinstance(network_capability, dict):

            return None



        service_instance_id = network_capability.get("service_instance_id")

        if isinstance(service_instance_id, UUID):

            return service_instance_id

        if isinstance(service_instance_id, str):

            try:

                return UUID(service_instance_id)

            except ValueError:

                return None

        return None



    def _failed_after_accept_event(

        self,

        command: NetworkCapabilityDeactivateCommand,

        accepted: NetworkCapabilityDeactivateResult,

        error: Exception,

    ) -> SrmOperationCompletedV1:

        now = datetime.now(UTC)

        problem = {

            "type": "about:blank",

            "title": "Network capability deactivate failed after acceptance.",

            "status": 500,

            "detail": str(error),

        }

        service_instance_id = (

            accepted.service_instance_id or command.network_capability.service_instance_id

        )

        instance_entry: dict[str, Any] = {

            **(

                {"service_instance_id": str(service_instance_id)}

                if service_instance_id is not None

                else {}

            ),

            "status": "failed",

            "error": problem,

        }

        if accepted.zone_id is not None:

            instance_entry["zone_id"] = str(accepted.zone_id)

        return SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            status=OperationCompletedStatus.FAILED,

            service_order_id=accepted.accepted_order_id,

            instances=[instance_entry],

            correlation_id=command.correlation_id,

            completed_at=now,

        )

src/srm/application/use_cases/network_capability_deactivate.py

0 → 100644
+600 −0

File added.

Preview size limit exceeded, changes collapsed.

tests/api/databus/test_dependencies.py

+38 −0
Changes for tests/api/databus/test_dependencies.py: 38 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -3,6 +3,7 @@ from unittest.mock import MagicMock, patch 
from srm.api.databus.dependencies import (

    build_deploy_service_use_case,

    build_network_capability_activate_use_case,

    build_network_capability_deactivate_use_case,

    build_terminate_service_use_case,

)
@@ -128,3 +129,40 @@ def test_build_network_capability_activate_use_case_wires_required_repositories( 
        control_paths=control_paths,

    )

    assert result is use_case.return_value





def test_build_network_capability_deactivate_use_case_wires_required_repositories() -> None:

    session = MagicMock()

    publisher = MagicMock()

    control_paths = MagicMock()



    with (

        patch("srm.api.databus.dependencies.SqlServiceOrderRepository") as orders,

        patch("srm.api.databus.dependencies.SqlServiceInstanceRepository") as instances,

        patch(

            "srm.api.databus.dependencies.SqlCapabilityInstanceRepository"

        ) as capability_instances,

        patch("srm.api.databus.dependencies.SqlCapabilityRepository") as capabilities,

        patch("srm.api.databus.dependencies.SqlDomainRepository") as domains,

        patch("srm.api.databus.dependencies.SqlZoneRepository") as zones,

        patch("srm.api.databus.dependencies.NetworkCapabilityDeactivateUseCase") as use_case,

    ):

        result = build_network_capability_deactivate_use_case(session, publisher, control_paths)



    orders.assert_called_once_with(session)

    instances.assert_called_once_with(session)

    capability_instances.assert_called_once_with(session)

    capabilities.assert_called_once_with(session)

    domains.assert_called_once_with(session)

    zones.assert_called_once_with(session)

    use_case.assert_called_once_with(

        service_orders=orders.return_value,

        service_instances=instances.return_value,

        capability_instances=capability_instances.return_value,

        capabilities=capabilities.return_value,

        domains=domains.return_value,

        zones=zones.return_value,

        publisher=publisher,

        control_paths=control_paths,

    )

    assert result is use_case.return_value