feat: network capability deactivate command (48c501a1) · Commits · OOP / code / service-resource-manager

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