feat: tf-sdk wiring for network capability activation, logging, tests (1d5c06c4) · Commits · OOP / code / service-resource-manager

src/srm/application/use_cases/network_capability_activate.py

+83 −15

Changes for src/srm/application/use_cases/network_capability_activate.py: 83 added lines, 15 removed lines.
Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@ from typing import Any, Literal
		from uuid import UUID

		import structlog
		from anyio import to_thread

		from srm.application.services.capability_placement import (
		CapabilityPlacement,
		@@ -59,10 +60,34 @@ from srm.domain.ports.databus.events import (
		SrmOperationStatusV1,
		)
		from srm.domain.ports.databus.publisher import DataBusPublisher
		from srm.domain.ports.transformation_functions.registry import ControlPathRegistry
		from srm.domain.ports.transformation_functions import (
		CapabilityLifecycleRequest,
		ControlPathRegistry,
		DomainExecutionFailed,
		ResolvedCapabilityTarget,
		)

		logger: structlog.BoundLogger = structlog.get_logger(__name__)

		ACTIVATE_ACTION = "activate"


		def _resolved_target(
		placement: CapabilityPlacement,
		*,
		capability_kind: CapabilityKind,
		) -> ResolvedCapabilityTarget:
		return ResolvedCapabilityTarget(
		zone_ref=placement.zone_ref,
		zone_kind=placement.zone_kind,
		domain_ref=placement.domain_ref,
		domain_kind=placement.domain_kind,
		capability_ref=placement.capability_ref,
		capability_kind=capability_kind,
		control_path_ref=placement.control_path_ref,
		action=ACTIVATE_ACTION,
		)


		@dataclass(frozen=True, slots=True)
		class NetworkCapabilityPayloadCommand:
		@@ -125,9 +150,6 @@ _UNRESOLVED: dict[TargetResolutionFailure, tuple[NetworkCapabilityActivateFailur
		}


		ACTIVATE_ACTION = "activate"


		class NetworkCapabilityActivateUseCase:
		def __init__(
		self,
		@@ -297,25 +319,71 @@ class NetworkCapabilityActivateUseCase:
		if service_order is None:
		raise RuntimeError("Accepted network capability order could not be reloaded")

		# TODO: replace this placeholder success with a TF-SDK
		# CapabilityExecutionAdapter.activate_capability(...) call using
		# capability_kind=command.network_capability.capability_type and
		# accepted.placement.control_path_ref.
		capability_kind = self._capability_kind(command)
		if capability_kind is None:
		raise RuntimeError("Accepted network capability activate has unsupported kind")

		placement = accepted.placement
		adapter = self._control_paths.get_adapter(placement.control_path_ref)
		logger.info(
		"network_capability_adapter_activate",
		operation_id=str(command.operation_id),
		correlation_id=command.correlation_id,
		control_path_ref=placement.control_path_ref,
		adapter=type(adapter).__name__,
		capability_kind=capability_kind,
		)
		result = await to_thread.run_sync(
		adapter.activate_capability,
		CapabilityLifecycleRequest(
		operation_id=str(command.operation_id),
		tenant_id=command.app_provider_id,
		capability_kind=capability_kind,
		target=_resolved_target(placement, capability_kind=capability_kind),
		parameters=accepted.parameters_snapshot,
		),
		)
		logger.info(
		"network_capability_adapter_activate_result",
		operation_id=str(command.operation_id),
		correlation_id=command.correlation_id,
		control_path_ref=placement.control_path_ref,
		adapter=type(adapter).__name__,
		state=result.state,
		external_id=result.external_id,
		external_ref=result.external_ref,
		details=result.details,
		)
		if result.state is not CapabilityInstanceState.ACTIVE:
		raise DomainExecutionFailed(
		error_code="domain_execution_failed",
		message=(
		"network capability activate did not become active on "
		f"{placement.control_path_ref}: {result.state}"
		),
		details=result.details,
		)

		await self._capability_instances.create(
		CapabilityInstance(
		capability_id=accepted.placement.capability_id,
		capability_id=placement.capability_id,
		service_instance_id=command.service_instance_id,
		originating_service_order_id=service_order.id,
		service_capability_requirement_id=accepted.requirement.id,
		control_path_binding_id=accepted.placement.control_path_binding_id,
		control_path_ref_snapshot=accepted.placement.control_path_ref,
		control_path_binding_id=placement.control_path_binding_id,
		control_path_ref_snapshot=placement.control_path_ref,
		ref=accepted.requirement.ref,
		kind=self._capability_instance_kind(command),
		state=CapabilityInstanceState.ACTIVE,
		external_id=None,
		external_ref=None,
		state=result.state,
		external_id=result.external_id,
		external_ref=result.external_ref,
		parameters_snapshot=accepted.parameters_snapshot,
		result_summary=Result(status=ResultStatus.ACTIVE),
		result_summary=result.result_summary
		or Result(
		status=ResultStatus.ACTIVE,
		endpoints=result.endpoints,
		backend=result.details,
		),
		app_provider_id=command.app_provider_id,
		federation_partner_ref=command.federation_partner_ref,
		)

tests/unit/test_network_capability_activate_use_case.py

+112 −3

Changes for tests/unit/test_network_capability_activate_use_case.py: 112 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,6 +3,9 @@ from datetime import datetime
		from unittest.mock import AsyncMock
		from uuid import UUID, uuid4

		import pytest

		from srm.adapters.transformation_functions.registry import ConfiguredControlPathRegistry
		from srm.application.services.capability_placement import (
		CapabilityPlacement,
		CapabilityPlacementPlanner,
		@@ -53,11 +56,15 @@ from srm.domain.models.topology import (
		ZoneState,
		)
		from srm.domain.ports.transformation_functions import (
		CapabilityLifecycleRequest,
		CapabilityLifecycleResult,
		ControlPathRegistry,
		DomainExecutionFailed,
		DomainTimeoutError,
		TargetServeability,
		TargetServeabilityRequest,
		TransformationFunctionExecutionPort,
		UnknownControlPathRef,
		)
		from tests.unit.fakes import FakeTransformationFunctionAdapter, InMemoryZoneRepository

		@@ -246,8 +253,12 @@ def _use_case(
		zones: AsyncMock \| None = None,
		publisher: AsyncMock \| None = None,
		resolver: AsyncMock \| None = None,
		control_paths: AsyncMock \| None = None,
		control_paths: ConfiguredControlPathRegistry \| None = None,
		adapter: FakeTransformationFunctionAdapter \| None = None,
		) -> NetworkCapabilityActivateUseCase:
		registry = control_paths or ConfiguredControlPathRegistry(
		{"nef.zone-a": adapter or FakeTransformationFunctionAdapter()}
		)
		return NetworkCapabilityActivateUseCase(
		service_specifications=service_specifications or AsyncMock(),
		service_capability_requirements=service_capability_requirements or AsyncMock(),
		@@ -257,10 +268,25 @@ def _use_case(
		zones=zones or AsyncMock(),
		publisher=publisher or AsyncMock(),
		resolver=resolver or AsyncMock(),
		control_paths=control_paths or AsyncMock(),
		control_paths=registry,
		)


		class _RecordingAdapter(FakeTransformationFunctionAdapter):
		def __init__(self) -> None:
		self.requests: list[CapabilityLifecycleRequest] = []

		def activate_capability(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:
		self.requests.append(request)
		return super().activate_capability(request)


		class _FailedAdapter(FakeTransformationFunctionAdapter):
		def activate_capability(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:
		result = super().activate_capability(request)
		return result.model_copy(update={"state": CapabilityInstanceState.FAILED})


		async def test_accept_validates_network_capability_activate_command() -> None:
		command = _command(uuid4())
		zone_id = uuid4()
		@@ -631,7 +657,7 @@ async def test_complete_creates_capability_instance_and_completed_event() -> Non
		assert created_capability.control_path_ref_snapshot == placement.control_path_ref
		assert created_capability.kind == CapabilityInstanceKind.ACTIVATION
		assert created_capability.state == CapabilityInstanceState.ACTIVE
		assert created_capability.external_id is None
		assert created_capability.external_id == "backend-1"
		assert created_capability.external_ref is None
		service_instances.update_state.assert_awaited_once_with(
		command.service_instance_id,
		@@ -644,6 +670,43 @@ async def test_complete_creates_capability_instance_and_completed_event() -> Non
		)


		async def test_complete_activates_network_capability_through_control_path_adapter() -> None:
		command = _command(uuid4())
		accepted_order = _service_order(command.operation_id)
		requirement = _network_requirement(command.service_specification_id)
		placement = _placement()
		parameters_snapshot = _parameters()
		accepted = NetworkCapabilityActivateResult(
		accepted_order_id=accepted_order.id,
		requirement=requirement,
		placement=placement,
		parameters_snapshot=parameters_snapshot,
		)
		service_orders = AsyncMock()
		service_orders.get_by_id.return_value = accepted_order
		capability_instances = AsyncMock()
		adapter = _RecordingAdapter()

		await _use_case(
		service_orders=service_orders,
		capability_instances=capability_instances,
		adapter=adapter,
		).complete(command, accepted)

		assert len(adapter.requests) == 1
		request = adapter.requests[0]
		assert request.operation_id == str(command.operation_id)
		assert request.tenant_id == "VideoAppsCo"
		assert request.capability_kind is CapabilityKind.QOD_SESSION
		assert request.target.control_path_ref == "nef.zone-a"
		assert request.target.action == "activate"
		assert request.target.runtime_kind is None
		assert request.deployment_unit is None
		assert request.parameters == parameters_snapshot
		(created_capability,) = capability_instances.create.await_args.args
		assert created_capability.external_id == "backend-1"


		async def test_complete_uses_subscription_kind_for_traffic_influence() -> None:
		command = _command(uuid4(), capability_type="traffic_influence")
		accepted_order = _service_order(command.operation_id)
		@@ -670,6 +733,52 @@ async def test_complete_uses_subscription_kind_for_traffic_influence() -> None:
		assert created_capability.kind == CapabilityInstanceKind.SUBSCRIPTION


		async def test_complete_raises_when_control_path_has_no_adapter() -> None:
		command = _command(uuid4())
		accepted_order = _service_order(command.operation_id)
		accepted = NetworkCapabilityActivateResult(
		accepted_order_id=accepted_order.id,
		requirement=_network_requirement(command.service_specification_id),
		placement=_placement(),
		parameters_snapshot=_parameters(),
		)
		service_orders = AsyncMock()
		service_orders.get_by_id.return_value = accepted_order
		capability_instances = AsyncMock()

		with pytest.raises(UnknownControlPathRef):
		await _use_case(
		service_orders=service_orders,
		capability_instances=capability_instances,
		control_paths=ConfiguredControlPathRegistry({}),
		).complete(command, accepted)

		capability_instances.create.assert_not_awaited()


		async def test_complete_raises_when_adapter_returns_failed_state() -> None:
		command = _command(uuid4())
		accepted_order = _service_order(command.operation_id)
		accepted = NetworkCapabilityActivateResult(
		accepted_order_id=accepted_order.id,
		requirement=_network_requirement(command.service_specification_id),
		placement=_placement(),
		parameters_snapshot=_parameters(),
		)
		service_orders = AsyncMock()
		service_orders.get_by_id.return_value = accepted_order
		capability_instances = AsyncMock()

		with pytest.raises(DomainExecutionFailed):
		await _use_case(
		service_orders=service_orders,
		capability_instances=capability_instances,
		adapter=_FailedAdapter(),
		).complete(command, accepted)

		capability_instances.create.assert_not_awaited()


		async def test_fail_after_accept_marks_instance_and_order_failed() -> None:
		command = _command(uuid4())
		accepted_order = _service_order(command.operation_id)