feat: network capability activation (6988c209) · Commits · OOP / code / service-resource-manager

src/srm/api/databus/dependencies.py

+33 −0

Original line number	Diff line number	Diff line
		@@ -14,7 +14,13 @@ from srm.adapters.database.repos.topology import SqlZoneRepository
		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.adapters.databus.nats_publisher import NatsPublisher
		from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator
		from srm.application.command_handlers.network_capability_activate import (
		NetworkCapabilityActivateCommandCoordinator,
		)
		from srm.application.use_cases.deploy_service import DeployServiceUseCase
		from srm.application.use_cases.network_capability_activate import (
		NetworkCapabilityActivateUseCase,
		)
		from srm.domain.ports.databus.publisher import DataBusPublisher


		@@ -50,3 +56,30 @@ def get_deploy_service_command_coordinator(
		use_case_factory=build_deploy_service_use_case,
		publisher=NatsPublisher(connection_manager),
		)


		def build_network_capability_activate_use_case(
		session: AsyncSession,
		publisher: DataBusPublisher,
		) -> NetworkCapabilityActivateUseCase:
		return NetworkCapabilityActivateUseCase(
		service_specifications=SqlServiceSpecificationRepository(session),
		service_capability_requirements=SqlServiceCapabilityRequirementRepository(session),
		service_orders=SqlServiceOrderRepository(session),
		service_instances=SqlServiceInstanceRepository(session),
		capability_instances=SqlCapabilityInstanceRepository(session),
		zones=SqlZoneRepository(session),
		domains=SqlDomainRepository(session),
		publisher=publisher,
		)


		def get_network_capability_activate_command_coordinator(
		session_maker: async_sessionmaker[AsyncSession],
		connection_manager: NatsConnectionManager,
		) -> NetworkCapabilityActivateCommandCoordinator:
		return NetworkCapabilityActivateCommandCoordinator(
		session_maker=session_maker,
		use_case_factory=build_network_capability_activate_use_case,
		publisher=NatsPublisher(connection_manager),
		)

src/srm/api/databus/nats_subscriber.py

+14 −3

Original line number	Diff line number	Diff line
		@@ -7,7 +7,10 @@ from pydantic import ValidationError
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.api.databus.dependencies import get_deploy_service_command_coordinator
		from srm.api.databus.dependencies import (
		get_deploy_service_command_coordinator,
		get_network_capability_activate_command_coordinator,
		)
		from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage

		logger: structlog.BoundLogger = structlog.get_logger(__name__)
		@@ -28,12 +31,15 @@ async def _noop_callback(envelope: AnswerableCommandV1) -> None:
		COMMAND_HANDLERS: dict[str, MessageHandler] = {
		"command.srm.service.scale": _noop_callback,
		"command.srm.service.terminate": _noop_callback,
		"command.srm.network.capability.activate": _noop_callback,
		"command.srm.network.capability.update": _noop_callback,
		"command.srm.network.capability.deactivate": _noop_callback,
		}

		COMMAND_SUBJECTS = ("command.srm.service.deploy", *COMMAND_HANDLERS.keys())
		COMMAND_SUBJECTS = (
		"command.srm.service.deploy",
		"command.srm.network.capability.activate",
		*COMMAND_HANDLERS.keys(),
		)


		async def subscribe_to_subjects(
		@@ -44,9 +50,14 @@ async def subscribe_to_subjects(
		session_maker,
		connection_manager,
		)
		network_activate_coordinator = get_network_capability_activate_command_coordinator(
		session_maker,
		connection_manager,
		)

		command_handlers = {
		"command.srm.service.deploy": deploy_coordinator.handle,
		"command.srm.network.capability.activate": network_activate_coordinator.handle,
		**COMMAND_HANDLERS,
		}
		subscribers = [

src/srm/application/command_handlers/network_capability_activate.py

0 → 100644

+233 −0

Original line number	Diff line number	Diff line
		from __future__ import annotations

		from collections.abc import Callable
		from datetime import UTC, datetime
		from uuid import UUID

		from pydantic import ValidationError
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

		from srm.api.databus.schemas import SrmNetworkCapabilityActivateV1, ValidatedEnvelopeV1
		from srm.application.use_cases.network_capability_activate import (
		NetworkCapabilityActivateCommand,
		NetworkCapabilityActivateResult,
		NetworkCapabilityActivateUseCase,
		NetworkCapabilityPayloadCommand,
		)
		from srm.domain.ports.databus.events import (
		OperationCompletedStatus,
		OperationStatusState,
		SrmOperationCompletedV1,
		SrmOperationStatusV1,
		)
		from srm.domain.ports.databus.publisher import DataBusPublisher

		NetworkCapabilityActivateUseCaseFactory = Callable[
		[AsyncSession, DataBusPublisher],
		NetworkCapabilityActivateUseCase,
		]


		class NetworkCapabilityActivateCommandCoordinator:
		def __init__(
		self,
		session_maker: async_sessionmaker[AsyncSession],
		use_case_factory: NetworkCapabilityActivateUseCaseFactory,
		publisher: DataBusPublisher,
		) -> None:
		self._session_maker = session_maker
		self._use_case_factory = use_case_factory
		self._publisher = publisher

		async def handle(self, envelope: ValidatedEnvelopeV1) -> None:
		command = await self._parse_command(envelope)
		if command is None:
		return

		accepted = await self._accept(command)
		if accepted.duplicate_order_id is not None:
		await self._publish_accepted(command, accepted.duplicate_order_id)
		return
		if accepted.accepted_order_id is None:
		return

		await self._publish_accepted(command, accepted.accepted_order_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: ValidatedEnvelopeV1,
		) -> NetworkCapabilityActivateCommand \| None:
		try:
		activate_message = SrmNetworkCapabilityActivateV1.model_validate(
		envelope.model_dump(mode="python")
		)
		except ValidationError:
		await self._publish_failed_before_start_for_invalid_command(envelope)
		return None

		return NetworkCapabilityActivateCommand(
		operation_id=activate_message.operation_id,
		correlation_id=activate_message.correlation_id,
		requested_at=activate_message.requested_at,
		app_provider_id=activate_message.app_provider_id,
		federation_partner_ref=activate_message.federation_partner_ref,
		source=activate_message.source,
		service_instance_id=activate_message.service_instance_id,
		service_specification_id=activate_message.service_specification_id,
		zone_id=activate_message.zone_id,
		domain_id=activate_message.domain_id,
		network_capability=NetworkCapabilityPayloadCommand(
		capability_type=activate_message.network_capability.capability_type,
		target=activate_message.network_capability.target,
		profile_ref=activate_message.network_capability.profile_ref,
		parameters=activate_message.network_capability.parameters,
		),
		payload_snapshot=activate_message.model_dump(mode="json"),
		)

		async def _accept(
		self,
		command: NetworkCapabilityActivateCommand,
		) -> NetworkCapabilityActivateResult:
		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: NetworkCapabilityActivateCommand,
		accepted: NetworkCapabilityActivateResult,
		) -> 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: NetworkCapabilityActivateCommand,
		accepted: NetworkCapabilityActivateResult,
		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: NetworkCapabilityActivateCommand,
		service_order_id: UUID,
		) -> None:
		status_event = SrmOperationStatusV1(
		schema_version="1.0",
		operation_id=command.operation_id,
		service_order_id=service_order_id,
		service_instance_id=command.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: ValidatedEnvelopeV1,
		) -> None:
		now = datetime.now(UTC)
		status_event = SrmOperationStatusV1(
		schema_version="1.0",
		operation_id=envelope.operation_id,
		service_order_id=None,
		service_instance_id=None,
		capability="network_capability",
		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 activate command is invalid.",
		"status": 400,
		"detail": "Network capability activate 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 _failed_after_accept_event(
		self,
		command: NetworkCapabilityActivateCommand,
		accepted: NetworkCapabilityActivateResult,
		error: Exception,
		) -> SrmOperationCompletedV1:
		now = datetime.now(UTC)
		problem = {
		"type": "about:blank",
		"title": "Network capability activate failed after acceptance.",
		"status": 500,
		"detail": str(error),
		}
		return SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=command.operation_id,
		status=OperationCompletedStatus.FAILED,
		service_order_id=accepted.accepted_order_id,
		instances=[
		{
		"service_instance_id": str(command.service_instance_id),
		**(
		{"zone_id": str(accepted.placement.zone_id)}
		if accepted.placement is not None
		else {}
		),
		"status": "failed",
		"error": problem,
		}
		],
		correlation_id=command.correlation_id,
		completed_at=now,
		)

src/srm/application/use_cases/network_capability_activate.py

0 → 100644

+420 −0

File added.

Preview size limit exceeded, changes collapsed.

tests/api/databus/test_dependencies.py

+42 −1

Original line number	Diff line number	Diff line
		from unittest.mock import MagicMock, patch

		from srm.api.databus.dependencies import get_deploy_service_use_case
		from srm.api.databus.dependencies import (
		build_network_capability_activate_use_case,
		get_deploy_service_use_case,
		)


		def test_get_deploy_service_use_case_wires_required_repositories() -> None:
		@@ -41,3 +44,41 @@ def test_get_deploy_service_use_case_wires_required_repositories() -> None:
		publisher=publisher.return_value,
		)
		assert result is use_case.return_value


		def test_build_network_capability_activate_use_case_wires_required_repositories() -> None:
		session = MagicMock()
		publisher = MagicMock()

		with (
		patch("srm.api.databus.dependencies.SqlServiceSpecificationRepository") as specs,
		patch(
		"srm.api.databus.dependencies.SqlServiceCapabilityRequirementRepository"
		) as requirements,
		patch("srm.api.databus.dependencies.SqlServiceOrderRepository") as orders,
		patch("srm.api.databus.dependencies.SqlServiceInstanceRepository") as instances,
		patch("srm.api.databus.dependencies.SqlCapabilityInstanceRepository") as capabilities,
		patch("srm.api.databus.dependencies.SqlZoneRepository") as zones,
		patch("srm.api.databus.dependencies.SqlDomainRepository") as domains,
		patch("srm.api.databus.dependencies.NetworkCapabilityActivateUseCase") as use_case,
		):
		result = build_network_capability_activate_use_case(session, publisher)

		specs.assert_called_once_with(session)
		requirements.assert_called_once_with(session)
		orders.assert_called_once_with(session)
		instances.assert_called_once_with(session)
		capabilities.assert_called_once_with(session)
		zones.assert_called_once_with(session)
		domains.assert_called_once_with(session)
		use_case.assert_called_once_with(
		service_specifications=specs.return_value,
		service_capability_requirements=requirements.return_value,
		service_orders=orders.return_value,
		service_instances=instances.return_value,
		capability_instances=capabilities.return_value,
		zones=zones.return_value,
		domains=domains.return_value,
		publisher=publisher,
		)
		assert result is use_case.return_value