suggested way of network capability parameters validation and passing (bb2a8692) · Commits · OOP / code / service-resource-manager

src/srm/application/services/network_capability_activation_specs.py

0 → 100644

+265 −0

Changes for src/srm/application/services/network_capability_activation_specs.py: 265 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from collections.abc import Mapping
		from dataclasses import dataclass
		from datetime import datetime
		from typing import Any, Protocol

		from pydantic import Field, ValidationError

		from srm.domain.models.canonical_parameters.parameters import (
		ApplicationReferenceTarget,
		ApplicationServerTarget,
		CapabilityParameters,
		CapabilityTarget,
		DeviceTarget,
		Parameters,
		SourceSpecification,
		SourceSpecificationFamily,
		TrafficFilters,
		)
		from srm.domain.models.common import DomainModel
		from srm.domain.models.topology import CapabilityKind


		class CapabilityActivationValidationError(ValueError):
		pass


		@dataclass(frozen=True, slots=True)
		class CapabilityActivationContext:
		source: str
		requested_at: datetime
		api_version: str = "1.0.0"


		@dataclass(frozen=True, slots=True)
		class NormalizedCapabilityActivation:
		parameters: Parameters


		class NetworkCapabilityPayloadLike(Protocol):
		@property
		def capability_type(self) -> str: ...

		@property
		def target(self) -> CapabilityTarget: ...

		@property
		def profile_ref(self) -> str \| None: ...

		@property
		def parameters(self) -> CapabilityParameters: ...


		class CapabilityActivationSpec(Protocol):
		capability_kind: CapabilityKind

		def validate_and_normalize(
		self,
		payload: NetworkCapabilityPayloadLike,
		context: CapabilityActivationContext,
		) -> NormalizedCapabilityActivation: ...


		class QodSessionFields(DomainModel):
		qos_profile: str
		duration_seconds: int \| None = Field(default=None, gt=0)
		device: DeviceTarget \| None = None
		application_server: ApplicationServerTarget \| None = None


		class TrafficInfluenceFields(DomainModel):
		duration_seconds: int \| None = Field(default=None, gt=0)
		traffic_filters: TrafficFilters
		application_reference: ApplicationReferenceTarget \| None = None
		application_server: ApplicationServerTarget \| None = None
		notification_url: str \| None = None
		notification_auth_token: str \| None = None


		class QodSessionActivationSpec:
		capability_kind = CapabilityKind.QOD_SESSION

		def validate_and_normalize(
		self,
		payload: NetworkCapabilityPayloadLike,
		context: CapabilityActivationContext,
		) -> NormalizedCapabilityActivation:
		raw_fields = _raw_fields(payload)
		try:
		fields = QodSessionFields.model_validate(
		{
		"qos_profile": _first_present(
		payload.profile_ref,
		payload.parameters.profile_ref,
		raw_fields.get("qos_profile"),
		raw_fields.get("qosProfile"),
		),
		"duration_seconds": _first_present(
		payload.parameters.duration_seconds,
		raw_fields.get("duration_seconds"),
		raw_fields.get("duration"),
		),
		"device": _first_present(payload.target.device, raw_fields.get("device")),
		"application_server": _first_present(
		payload.target.application_server,
		raw_fields.get("application_server"),
		raw_fields.get("applicationServer"),
		),
		}
		)
		except ValidationError as exc:
		raise CapabilityActivationValidationError(
		_validation_error_detail("qod_session", exc)
		) from exc

		if fields.device is None and fields.application_server is None:
		raise CapabilityActivationValidationError(
		"qod_session requires device or application_server target"
		)

		return NormalizedCapabilityActivation(
		parameters=Parameters(
		target=CapabilityTarget(
		device=fields.device,
		application_server=fields.application_server,
		),
		parameters=CapabilityParameters(
		profile_ref=fields.qos_profile,
		duration_seconds=fields.duration_seconds,
		),
		source_spec=SourceSpecification(
		family=_source_family(context.source),
		api="quality-on-demand",
		version=context.api_version,
		),
		)
		)


		class TrafficInfluenceActivationSpec:
		capability_kind = CapabilityKind.TRAFFIC_INFLUENCE

		def validate_and_normalize(
		self,
		payload: NetworkCapabilityPayloadLike,
		context: CapabilityActivationContext,
		) -> NormalizedCapabilityActivation:
		raw_fields = _raw_fields(payload)
		try:
		fields = TrafficInfluenceFields.model_validate(
		{
		"duration_seconds": _first_present(
		payload.parameters.duration_seconds,
		raw_fields.get("duration_seconds"),
		raw_fields.get("duration"),
		),
		"traffic_filters": _first_present(
		payload.parameters.traffic_filters,
		raw_fields.get("traffic_filters"),
		raw_fields.get("trafficFilters"),
		),
		"application_reference": _first_present(
		payload.target.application_reference,
		raw_fields.get("application_reference"),
		raw_fields.get("applicationReference"),
		),
		"application_server": _first_present(
		payload.target.application_server,
		raw_fields.get("application_server"),
		raw_fields.get("applicationServer"),
		),
		"notification_url": _first_present(
		raw_fields.get("notification_url"),
		raw_fields.get("notificationUrl"),
		),
		"notification_auth_token": _first_present(
		raw_fields.get("notification_auth_token"),
		raw_fields.get("notificationAuthToken"),
		),
		}
		)
		except ValidationError as exc:
		raise CapabilityActivationValidationError(
		_validation_error_detail("traffic_influence", exc)
		) from exc

		return NormalizedCapabilityActivation(
		parameters=Parameters(
		target=CapabilityTarget(
		application_reference=fields.application_reference,
		application_server=fields.application_server,
		),
		parameters=CapabilityParameters(
		duration_seconds=fields.duration_seconds,
		traffic_filters=fields.traffic_filters,
		extra={
		key: value
		for key, value in {
		"notification_url": fields.notification_url,
		"notification_auth_token": fields.notification_auth_token,
		}.items()
		if value is not None
		},
		),
		source_spec=SourceSpecification(
		family=_source_family(context.source),
		api="traffic-influence",
		version=context.api_version,
		),
		)
		)


		CAPABILITY_ACTIVATION_SPECS: Mapping[CapabilityKind, CapabilityActivationSpec] = {
		CapabilityKind.QOD_SESSION: QodSessionActivationSpec(),
		CapabilityKind.TRAFFIC_INFLUENCE: TrafficInfluenceActivationSpec(),
		}


		def normalize_network_capability_activation(
		payload: NetworkCapabilityPayloadLike,
		context: CapabilityActivationContext,
		) -> Parameters:
		try:
		capability_kind = CapabilityKind(payload.capability_type)
		except ValueError as exc:
		raise CapabilityActivationValidationError(
		"Requested network capability type is not supported."
		) from exc

		spec = CAPABILITY_ACTIVATION_SPECS.get(capability_kind)
		if spec is None:
		raise CapabilityActivationValidationError(
		"Requested network capability type is not supported."
		)

		return spec.validate_and_normalize(payload, context).parameters


		def _raw_fields(payload: NetworkCapabilityPayloadLike) -> dict[str, Any]:
		return dict(payload.parameters.extra)


		def _first_present(*values: object) -> object:
		for value in values:
		if value is not None:
		return value
		return None


		def _validation_error_detail(capability_type: str, error: ValidationError) -> str:
		first_error = error.errors()[0]
		field = ".".join(str(part) for part in first_error["loc"])
		return f"{capability_type} invalid field {field}: {first_error['msg']}"


		def _source_family(source: str) -> SourceSpecificationFamily:
		if source == "nbi_camara":
		return SourceSpecificationFamily.CAMARA
		if source == "nbi_tmf":
		return SourceSpecificationFamily.TMF
		if source == "federation":
		return SourceSpecificationFamily.GSMA_OPG
		return SourceSpecificationFamily.INTERNAL

src/srm/application/use_cases/network_capability_activate.py

+20 −16

Changes for src/srm/application/use_cases/network_capability_activate.py: 20 added lines, 16 removed lines.
Original line number	Diff line number	Diff line
		@@ -11,12 +11,15 @@ from srm.application.services.capability_placement import (
		CapabilityPlacementRequest,
		PlacementPin,
		)
		from srm.application.services.network_capability_activation_specs import (
		CapabilityActivationContext,
		CapabilityActivationValidationError,
		normalize_network_capability_activation,
		)
		from srm.domain.models.canonical_parameters.parameters import (
		CapabilityParameters,
		CapabilityTarget,
		Parameters,
		SourceSpecification,
		SourceSpecificationFamily,
		)
		from srm.domain.models.canonical_parameters.result import Result, ResultStatus
		from srm.domain.models.catalog import ServiceCapabilityRequirement, ServiceSpecificationState
		@@ -120,6 +123,15 @@ class NetworkCapabilityActivateUseCase:
		)
		return NetworkCapabilityActivateResult()

		try:
		parameters_snapshot = self._parameters_snapshot(command)
		except CapabilityActivationValidationError as exc:
		await self._publish_failed_before_start(
		command,
		detail=str(exc),
		)
		return NetworkCapabilityActivateResult()

		existing_instance = await self._service_instances.get_by_id(command.service_instance_id)
		if existing_instance is not None:
		await self._publish_failed_before_start(
		@@ -200,7 +212,7 @@ class NetworkCapabilityActivateUseCase:
		accepted_order_id=service_order.id,
		requirement=requirement,
		placement=placement,
		parameters_snapshot=self._parameters_snapshot(command),
		parameters_snapshot=parameters_snapshot,
		)

		async def complete(
		@@ -363,19 +375,11 @@ class NetworkCapabilityActivateUseCase:
		return None

		def _parameters_snapshot(self, command: NetworkCapabilityActivateCommand) -> Parameters:
		parameters = command.network_capability.parameters
		if command.network_capability.profile_ref is not None:
		parameters = parameters.model_copy(
		update={"profile_ref": command.network_capability.profile_ref}
		)

		return Parameters(
		target=command.network_capability.target,
		parameters=parameters,
		source_spec=SourceSpecification(
		family=SourceSpecificationFamily.INTERNAL,
		api="network.capability.activate",
		version="1.0.0",
		return normalize_network_capability_activation(
		command.network_capability,
		CapabilityActivationContext(
		source=command.source,
		requested_at=command.requested_at,
		),
		)

tests/unit/test_network_capability_activate_use_case.py

+32 −1

Changes for tests/unit/test_network_capability_activate_use_case.py: 32 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		from dataclasses import replace
		from datetime import datetime
		from unittest.mock import AsyncMock
		from uuid import UUID, uuid4
		@@ -12,6 +13,7 @@ from srm.application.use_cases.network_capability_activate import (
		from srm.domain.models.canonical_parameters.parameters import (
		CapabilityParameters,
		CapabilityTarget,
		DeviceTarget,
		Parameters,
		SourceSpecification,
		SourceSpecificationFamily,
		@@ -64,7 +66,7 @@ def _command(
		domain_id=domain_id,
		network_capability=NetworkCapabilityPayloadCommand(
		capability_type=capability_type,
		target=CapabilityTarget(),
		target=CapabilityTarget(device=DeviceTarget(phone_number="+306900000000")),
		profile_ref="voice",
		parameters=CapabilityParameters(duration_seconds=3600),
		),
		@@ -359,6 +361,35 @@ async def test_accept_fails_before_start_when_requested_capability_type_is_unkno
		assert payload["error"]["detail"] == "Requested network capability type is not supported."


		async def test_accept_fails_before_start_when_capability_payload_is_invalid() -> None:
		valid_command = _command(uuid4())
		command = replace(
		valid_command,
		network_capability=replace(
		valid_command.network_capability,
		target=CapabilityTarget(),
		),
		)
		service_orders = AsyncMock()
		service_orders.get_by_operation_id.return_value = None
		service_instances = AsyncMock()
		publisher = AsyncMock()

		result = await _use_case(
		service_orders=service_orders,
		service_instances=service_instances,
		publisher=publisher,
		).accept(command)

		assert result.accepted_order_id is None
		assert result.placement is None
		assert publisher.publish.await_count == 2
		_subject, payload = publisher.publish.await_args_list[1].args
		assert payload["error"]["detail"] == "qod_session requires device or application_server target"
		service_instances.get_by_id.assert_not_awaited()
		service_orders.create.assert_not_awaited()


		async def test_accept_fails_before_start_when_requirement_is_not_network() -> None:
		command = _command(uuid4())
		service_orders = AsyncMock()

tests/unit/test_network_capability_activation_specs.py

0 → 100644

+108 −0

Changes for tests/unit/test_network_capability_activation_specs.py: 108 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from datetime import UTC, datetime

		import pytest

		from srm.application.services.network_capability_activation_specs import (
		CapabilityActivationContext,
		CapabilityActivationValidationError,
		normalize_network_capability_activation,
		)
		from srm.application.use_cases.network_capability_activate import NetworkCapabilityPayloadCommand
		from srm.domain.models.canonical_parameters.parameters import (
		ApplicationReferenceTarget,
		ApplicationServerTarget,
		CapabilityParameters,
		CapabilityTarget,
		DeviceTarget,
		TrafficEndpoint,
		TrafficFilters,
		)
		from srm.domain.models.common import TransportProtocol


		def _context() -> CapabilityActivationContext:
		return CapabilityActivationContext(
		source="nbi_camara",
		requested_at=datetime(2026, 8, 27, 12, 0, tzinfo=UTC),
		api_version="0.10.0",
		)


		def test_qod_activation_normalizer_maps_api_specific_fields_to_canonical_parameters() -> None:
		payload = NetworkCapabilityPayloadCommand(
		capability_type="qod_session",
		target=CapabilityTarget(
		device=DeviceTarget(phone_number="+306900000000"),
		application_server=ApplicationServerTarget.model_validate({"ipv4": "203.0.113.0/32"}),
		),
		profile_ref="voice",
		parameters=CapabilityParameters(duration_seconds=3600),
		)

		normalized = normalize_network_capability_activation(payload, _context())

		assert normalized.source_spec.api == "quality-on-demand"
		assert normalized.source_spec.version == "0.10.0"
		assert normalized.parameters.profile_ref == "voice"
		assert normalized.parameters.duration_seconds == 3600
		assert normalized.target.device is not None
		assert normalized.target.application_server is not None


		def test_qod_activation_normalizer_can_read_camara_names_from_extra() -> None:
		payload = NetworkCapabilityPayloadCommand(
		capability_type="qod_session",
		target=CapabilityTarget(),
		profile_ref=None,
		parameters=CapabilityParameters(
		extra={
		"qosProfile": "gold",
		"duration": 600,
		"applicationServer": {"ipv4": "203.0.113.10/32"},
		}
		),
		)

		normalized = normalize_network_capability_activation(payload, _context())

		assert normalized.parameters.profile_ref == "gold"
		assert normalized.parameters.duration_seconds == 600
		assert normalized.target.application_server is not None


		def test_qod_activation_normalizer_rejects_missing_target() -> None:
		payload = NetworkCapabilityPayloadCommand(
		capability_type="qod_session",
		target=CapabilityTarget(),
		profile_ref="voice",
		parameters=CapabilityParameters(duration_seconds=3600),
		)

		with pytest.raises(CapabilityActivationValidationError, match="requires device"):
		normalize_network_capability_activation(payload, _context())


		def test_traffic_influence_normalizer_keeps_optional_api_metadata_in_extra() -> None:
		payload = NetworkCapabilityPayloadCommand(
		capability_type="traffic_influence",
		target=CapabilityTarget(
		application_reference=ApplicationReferenceTarget(service_specification_ref="video-app")
		),
		profile_ref=None,
		parameters=CapabilityParameters(
		duration_seconds=1200,
		traffic_filters=TrafficFilters(
		destination=TrafficEndpoint(port=443, protocol=TransportProtocol.TCP)
		),
		extra={"notificationUrl": "https://callback.example/ti"},
		),
		)

		normalized = normalize_network_capability_activation(payload, _context())

		assert normalized.source_spec.api == "traffic-influence"
		assert normalized.target.application_reference is not None
		assert normalized.parameters.traffic_filters is not None
		assert normalized.parameters.extra == {
		"notification_url": "https://callback.example/ti",
		}