diff --git a/src/srm/adapters/database/repos/catalog.py b/src/srm/adapters/database/repos/catalog.py index 1889b9a0cffa2e99929b808cdaf580cf9c87758c..d5721227f00234b7d3d2195f81b6be491c7c5384 100644 --- a/src/srm/adapters/database/repos/catalog.py +++ b/src/srm/adapters/database/repos/catalog.py @@ -147,6 +147,16 @@ class SqlServiceDeploymentUnitRepository(ServiceDeploymentUnitRepository): row = await self._session.scalar(stmt) return ServiceDeploymentUnitMapper.to_domain(row) if row is not None else None + async def list_by_service_specification_id( + self, + service_specification_id: UUID, + ) -> list[ServiceDeploymentUnit]: + stmt = select(ServiceDeploymentUnitRow).where( + ServiceDeploymentUnitRow.service_specification_id == service_specification_id + ) + rows = (await self._session.scalars(stmt)).all() + return [ServiceDeploymentUnitMapper.to_domain(row) for row in rows] + async def create(self, deployment_unit: ServiceDeploymentUnit) -> ServiceDeploymentUnit: row = ServiceDeploymentUnitMapper.to_row(deployment_unit) self._session.add(row) @@ -173,6 +183,16 @@ class SqlServiceCapabilityRequirementRepository(ServiceCapabilityRequirementRepo row = await self._session.scalar(stmt) return ServiceCapabilityRequirementMapper.to_domain(row) if row is not None else None + async def list_by_service_specification_id( + self, + service_specification_id: UUID, + ) -> list[ServiceCapabilityRequirement]: + stmt = select(ServiceCapabilityRequirementRow).where( + ServiceCapabilityRequirementRow.service_specification_id == service_specification_id + ) + rows = (await self._session.scalars(stmt)).all() + return [ServiceCapabilityRequirementMapper.to_domain(row) for row in rows] + async def create( self, capability_requirement: ServiceCapabilityRequirement, diff --git a/src/srm/adapters/database/repos/runtime_inventory.py b/src/srm/adapters/database/repos/runtime_inventory.py index 50980f3e900eea4899c22e5768971d67d227d7e3..23d3dd4b2f2cfa856862cd5bb5b297e02eb5cf52 100644 --- a/src/srm/adapters/database/repos/runtime_inventory.py +++ b/src/srm/adapters/database/repos/runtime_inventory.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime from uuid import UUID import structlog @@ -19,6 +20,7 @@ from srm.domain.models.runtime_inventory import ( ServiceInstance, ServiceOrder, ) +from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState from srm.domain.ports.database.runtime_inventory import ( CapabilityInstanceRepository, ServiceInstanceRepository, @@ -51,6 +53,11 @@ class SqlServiceOrderRepository(ServiceOrderRepository): row = await self._session.scalar(stmt) return ServiceOrderMapper.to_domain(row) if row is not None else None + async def get_by_operation_id(self, operation_id: UUID) -> ServiceOrder | None: + stmt = select(ServiceOrderRow).where(ServiceOrderRow.operation_id == operation_id) + row = await self._session.scalar(stmt) + return ServiceOrderMapper.to_domain(row) if row is not None else None + async def create(self, service_order: ServiceOrder) -> ServiceOrder: row = ServiceOrderMapper.to_row(service_order) self._session.add(row) @@ -65,6 +72,26 @@ class SqlServiceOrderRepository(ServiceOrderRepository): raise RuntimeError("Created service order could not be reloaded") return saved + async def update_state( + self, + id: UUID, + state: ServiceOrderState, + *, + completed_at: datetime | None = None, + failure_detail: dict[str, object] | None = None, + ) -> ServiceOrder: + row = await self._session.get(ServiceOrderRow, id) + if row is None: + raise RuntimeError("Service order could not be reloaded") + row.state = state + row.completed_at = completed_at + row.failure_detail = failure_detail + await self._session.flush() + saved = await self.get_by_id(id) + if saved is None: + raise RuntimeError("Updated service order could not be reloaded") + return saved + class SqlServiceInstanceRepository(ServiceInstanceRepository): def __init__(self, session: AsyncSession) -> None: @@ -81,6 +108,20 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): row = await self._session.scalar(stmt) return ServiceInstanceMapper.to_domain(row) if row is not None else None + async def list_by_originating_service_order_id( + self, + originating_service_order_id: UUID, + ) -> list[ServiceInstance]: + stmt = ( + select(ServiceInstanceRow) + .options( + selectinload(ServiceInstanceRow.capability_instances), + ) + .where(ServiceInstanceRow.originating_service_order_id == originating_service_order_id) + ) + rows = (await self._session.scalars(stmt)).all() + return [ServiceInstanceMapper.to_domain(row) for row in rows] + async def create(self, service_instance: ServiceInstance) -> ServiceInstance: row = ServiceInstanceMapper.to_row(service_instance) self._session.add(row) @@ -95,6 +136,21 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): raise RuntimeError("Created service instance could not be reloaded") return saved + async def update_state( + self, + id: UUID, + state: ServiceInstanceState, + ) -> ServiceInstance: + row = await self._session.get(ServiceInstanceRow, id) + if row is None: + raise RuntimeError("Service instance could not be reloaded") + row.state = state + await self._session.flush() + saved = await self.get_by_id(id) + if saved is None: + raise RuntimeError("Updated service instance could not be reloaded") + return saved + class SqlCapabilityInstanceRepository(CapabilityInstanceRepository): def __init__(self, session: AsyncSession) -> None: diff --git a/src/srm/adapters/database/repos/topology.py b/src/srm/adapters/database/repos/topology.py index bfb024614471c5e0dbf625fa5b693b5657fad05f..d6e1a10df1ae0ffe4b458e3a31bdca426260f42e 100644 --- a/src/srm/adapters/database/repos/topology.py +++ b/src/srm/adapters/database/repos/topology.py @@ -16,6 +16,7 @@ from srm.adapters.database.mappers import ( ) from srm.adapters.database.sql import CapabilityRow, ControlPathBindingRow, DomainRow, ZoneRow from srm.domain.models.topology import Capability, ControlPathBinding, Domain, Zone +from srm.domain.models.topology.enums import ZoneKind, ZoneState from srm.domain.ports.database.topology import ( CapabilityRepository, ControlPathBindingRepository, @@ -57,6 +58,20 @@ class SqlZoneRepository(ZoneRepository): row = await self._session.scalar(stmt) return ZoneMapper.to_domain(row) if row is not None else None + async def list_active_resource_zones(self) -> list[Zone]: + stmt = ( + select(ZoneRow) + .options( + selectinload(ZoneRow.domains) + .selectinload(DomainRow.capabilities) + .selectinload(CapabilityRow.control_path_bindings) + ) + .where(ZoneRow.kind == ZoneKind.RESOURCE, ZoneRow.state == ZoneState.ACTIVE) + .order_by(ZoneRow.ref, ZoneRow.id) + ) + rows = (await self._session.scalars(stmt)).all() + return [ZoneMapper.to_domain(row) for row in rows] + async def create(self, zone: Zone) -> Zone: row = ZoneMapper.to_row(zone) self._session.add(row) diff --git a/src/srm/api/databus/dependencies.py b/src/srm/api/databus/dependencies.py new file mode 100644 index 0000000000000000000000000000000000000000..e72f9fea3a311d5c611097a0bf9bf4ebfcd2e99c --- /dev/null +++ b/src/srm/api/databus/dependencies.py @@ -0,0 +1,53 @@ +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from srm.adapters.database.repos.catalog import ( + SqlServiceCapabilityRequirementRepository, + SqlServiceDeploymentUnitRepository, + SqlServiceSpecificationRepository, +) +from srm.adapters.database.repos.runtime_inventory import ( + SqlCapabilityInstanceRepository, + SqlServiceInstanceRepository, + SqlServiceOrderRepository, +) +from srm.adapters.database.repos.topology import SqlDomainRepository, 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.use_cases.deploy_service import DeployServiceUseCase +from srm.domain.ports.databus.publisher import DataBusPublisher + + +def get_deploy_service_use_case( + session: AsyncSession, + connection_manager: NatsConnectionManager, +) -> DeployServiceUseCase: + return build_deploy_service_use_case(session, NatsPublisher(connection_manager)) + + +def build_deploy_service_use_case( + session: AsyncSession, + publisher: DataBusPublisher, +) -> DeployServiceUseCase: + return DeployServiceUseCase( + service_specifications=SqlServiceSpecificationRepository(session), + service_deployment_units=SqlServiceDeploymentUnitRepository(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_deploy_service_command_coordinator( + session_maker: async_sessionmaker[AsyncSession], + connection_manager: NatsConnectionManager, +) -> DeployServiceCommandCoordinator: + return DeployServiceCommandCoordinator( + session_maker=session_maker, + use_case_factory=build_deploy_service_use_case, + publisher=NatsPublisher(connection_manager), + ) diff --git a/src/srm/api/databus/nats_subscriber.py b/src/srm/api/databus/nats_subscriber.py index f1d2d705251ecd320b05265cb419df20d06f5966..2a358316967bbde895d839bbe5c8405a4bd3934b 100644 --- a/src/srm/api/databus/nats_subscriber.py +++ b/src/srm/api/databus/nats_subscriber.py @@ -3,34 +3,59 @@ from collections.abc import Awaitable, Callable import structlog from nats.aio.msg import Msg from nats.aio.subscription import Subscription +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.schemas import InboundMessage +from srm.api.databus.dependencies import get_deploy_service_command_coordinator +from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage logger: structlog.BoundLogger = structlog.get_logger(__name__) -COMMAND_SUBJECTS = [ - "command.srm.service.deploy", - "command.srm.service.scale", - "command.srm.service.terminate", - "command.srm.network.capability.activate", - "command.srm.network.capability.update", - "command.srm.network.capability.deactivate", -] +MessageHandler = Callable[[AnswerableCommandV1], Awaitable[None]] -async def _noop_router(message: InboundMessage) -> None: - # TODO: the real router must validate in two stages (interface-contract.md §A, ADR-0032). - logger.info("Message Received", subject=message.subject, data=message.payload) - return None + +async def _noop_callback(envelope: AnswerableCommandV1) -> None: + # TODO: non-deploy command handlers must validate in two stages (interface-contract.md §A). + logger.info( + "Message Received", + operation_id=str(envelope.operation_id), + correlation_id=envelope.correlation_id, + ) + + +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()) async def subscribe_to_subjects( connection_manager: NatsConnectionManager, + session_maker: async_sessionmaker[AsyncSession], ) -> list["NatsSubscriber"]: + deploy_coordinator = get_deploy_service_command_coordinator( + session_maker, + connection_manager, + ) + + command_handlers = { + "command.srm.service.deploy": deploy_coordinator.handle, + **COMMAND_HANDLERS, + } subscribers = [ - NatsSubscriber(connection_manager=connection_manager, subject=subject, router=_noop_router) - for subject in COMMAND_SUBJECTS + NatsSubscriber( + connection_manager=connection_manager, + subject=subject, + callback=callback, + ) + for subject, callback in command_handlers.items() ] for subscriber in subscribers: @@ -44,11 +69,11 @@ class NatsSubscriber: self, connection_manager: NatsConnectionManager, subject: str, - router: Callable[[InboundMessage], Awaitable[None]], + callback: MessageHandler, ) -> None: self._connection_manager = connection_manager self._subject = subject - self._router = router + self._callback = callback self._subscription: Subscription | None = None async def start(self) -> None: @@ -64,6 +89,14 @@ class NatsSubscriber: headers=dict(msg.headers) if msg.headers is not None else {}, ) try: - await self._router(inbound_message) + envelope = AnswerableCommandV1.model_validate_json(inbound_message.payload) + except ValidationError: + # Stage 1 failures are unanswerable: publish no SRM event; + # route to JetStream/DLQ when wired. + logger.exception("databus_stage1_unanswerable", subject=msg.subject) + return + + try: + await self._callback(envelope) except Exception: logger.exception("databus_handler_failed", subject=msg.subject) diff --git a/src/srm/api/databus/schemas.py b/src/srm/api/databus/schemas.py index 6dbff88643760e5ab98f961fc3615cf3619c3be9..8029e3f555bd99a3c260b448351f3974a52a8810 100644 --- a/src/srm/api/databus/schemas.py +++ b/src/srm/api/databus/schemas.py @@ -29,14 +29,20 @@ class CommandEnvelopeV1(CommandSchema): federation_partner_ref: str | None = None source: Literal["nbi_camara", "nbi_tmf", "operator_portal", "federation"] + +class CommandPayloadV1(CommandEnvelopeV1): @model_validator(mode="after") - def validate_federation_context(self) -> "CommandEnvelopeV1": + def validate_federation_context(self) -> "CommandPayloadV1": if self.source == "federation" and self.federation_partner_ref is None: raise ValueError("federation_partner_ref is required when source=federation") return self +class AnswerableCommandV1(CommandEnvelopeV1): + model_config = ConfigDict(extra="allow") + + class PlacementConstraintsV1(CommandSchema): model_config = ConfigDict(extra="allow") @@ -59,7 +65,7 @@ class DeployTargetV1(CommandSchema): return self -class SrmServiceDeployV1(CommandEnvelopeV1): +class SrmServiceDeployV1(CommandPayloadV1): service_specification_id: UUID targets: list[DeployTargetV1] = Field(min_length=1) deploy: DeployPayloadV1 @@ -69,7 +75,7 @@ class ScalePayloadV1(CommandSchema): replicas: int = Field(ge=0) -class SrmServiceScaleV1(CommandEnvelopeV1): +class SrmServiceScaleV1(CommandPayloadV1): service_instance_id: UUID service_specification_id: UUID | None = None scale: ScalePayloadV1 @@ -79,7 +85,7 @@ class TerminatePayloadV1(CommandSchema): grace_period_seconds: int = Field(default=0, ge=0) -class SrmServiceTerminateV1(CommandEnvelopeV1): +class SrmServiceTerminateV1(CommandPayloadV1): service_instance_id: UUID service_specification_id: UUID | None = None terminate: TerminatePayloadV1 @@ -92,7 +98,7 @@ class NetworkCapabilityPayloadV1(CommandSchema): parameters: CapabilityParameters -class SrmNetworkCapabilityActivateV1(CommandEnvelopeV1): +class SrmNetworkCapabilityActivateV1(CommandPayloadV1): service_specification_id: UUID zone_id: UUID | None = None domain_id: UUID | None = None @@ -131,7 +137,7 @@ class NetworkCapabilityUpdatePayloadV1(NetworkCapabilityRealizationRefV1): parameters: CapabilityParameters -class SrmNetworkCapabilityUpdateV1(CommandEnvelopeV1): +class SrmNetworkCapabilityUpdateV1(CommandPayloadV1): service_specification_id: UUID | None = None network_capability: NetworkCapabilityUpdatePayloadV1 @@ -140,6 +146,6 @@ class NetworkCapabilityDeactivatePayloadV1(NetworkCapabilityRealizationRefV1): grace_period_seconds: int = Field(default=0, ge=0) -class SrmNetworkCapabilityDeactivateV1(CommandEnvelopeV1): +class SrmNetworkCapabilityDeactivateV1(CommandPayloadV1): service_specification_id: UUID | None = None network_capability: NetworkCapabilityDeactivatePayloadV1 diff --git a/src/srm/application/command_handlers/__init__.py b/src/srm/application/command_handlers/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..455daf430f38ecd8952596553ddca282a561ff44 --- /dev/null +++ b/src/srm/application/command_handlers/__init__.py @@ -0,0 +1 @@ +"""Application-level command handlers.""" diff --git a/src/srm/application/command_handlers/deploy_service.py b/src/srm/application/command_handlers/deploy_service.py new file mode 100644 index 0000000000000000000000000000000000000000..78c46d6cd9fb96714886668ebf7f8ee09a789705 --- /dev/null +++ b/src/srm/application/command_handlers/deploy_service.py @@ -0,0 +1,236 @@ +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 AnswerableCommandV1, SrmServiceDeployV1 +from srm.application.use_cases.deploy_service import ( + DeployPayloadCommand, + DeployServiceCommand, + DeployServiceResult, + DeployServiceUseCase, + DeployTargetCommand, +) +from srm.domain.ports.databus.events import ( + OperationCompletedStatus, + OperationStatusState, + SrmOperationCompletedV1, + SrmOperationStatusV1, +) +from srm.domain.ports.databus.publisher import DataBusPublisher + +DeployUseCaseFactory = Callable[[AsyncSession, DataBusPublisher], DeployServiceUseCase] + + +class DeployServiceCommandCoordinator: + def __init__( + self, + session_maker: async_sessionmaker[AsyncSession], + use_case_factory: DeployUseCaseFactory, + 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(envelope, accepted.duplicate_order_id) + return + if accepted.accepted_order_id is None: + return + + await self._publish_accepted(envelope, 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: AnswerableCommandV1, + ) -> DeployServiceCommand | None: + try: + deploy_message = SrmServiceDeployV1.model_validate(envelope.model_dump(mode="python")) + except ValidationError: + await self._publish_failed_before_start_for_invalid_command(envelope) + return None + + return DeployServiceCommand( + operation_id=deploy_message.operation_id, + correlation_id=deploy_message.correlation_id, + requested_at=deploy_message.requested_at, + app_provider_id=deploy_message.app_provider_id, + federation_partner_ref=deploy_message.federation_partner_ref, + source=deploy_message.source, + service_specification_id=deploy_message.service_specification_id, + targets=[ + DeployTargetCommand( + app_instance_id=target.app_instance_id, + zone_id=target.zone_id, + domain_id=target.domain_id, + ) + for target in deploy_message.targets + ], + deploy=DeployPayloadCommand( + instance_name=deploy_message.deploy.instance_name, + placement_constraints=( + deploy_message.deploy.placement_constraints.model_dump(mode="python") + if deploy_message.deploy.placement_constraints is not None + else None + ), + ), + payload_snapshot=deploy_message.model_dump(mode="json"), + ) + + async def _accept(self, command: DeployServiceCommand) -> DeployServiceResult: + 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 _fail_after_accept( + self, + command: DeployServiceCommand, + accepted: DeployServiceResult, + 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 _complete( + self, + command: DeployServiceCommand, + accepted: DeployServiceResult, + ) -> 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 _publish_accepted( + self, + envelope: AnswerableCommandV1, + service_order_id: UUID, + ) -> None: + status_event = SrmOperationStatusV1( + schema_version="1.0", + operation_id=envelope.operation_id, + service_order_id=service_order_id, + service_instance_id=None, + capability="deploy", + state=OperationStatusState.ACCEPTED, + correlation_id=envelope.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=None, + capability="deploy", + 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": "Deploy service command is invalid.", + "status": 400, + }, + 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: DeployServiceCommand, + accepted: DeployServiceResult, + error: Exception, + ) -> SrmOperationCompletedV1: + now = datetime.now(UTC) + problem = { + "type": "about:blank", + "title": "Deploy service 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(target_placement.app_instance_id), + "zone_id": str(target_placement.zone_id), + "status": "failed", + "error": problem, + } + for target_placement in accepted.target_placements + ], + error=None if accepted.target_placements else problem, + correlation_id=command.correlation_id, + completed_at=now, + ) diff --git a/src/srm/application/services/__init__.py b/src/srm/application/services/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..ede22003c1f23e05282bc6e09bbb6478e8ee02f4 --- /dev/null +++ b/src/srm/application/services/__init__.py @@ -0,0 +1 @@ +"""Application services shared across use cases.""" diff --git a/src/srm/application/services/capability_placement.py b/src/srm/application/services/capability_placement.py new file mode 100644 index 0000000000000000000000000000000000000000..5beaf08c111b89ef1d8b18fc06eff0eb32edd1fb --- /dev/null +++ b/src/srm/application/services/capability_placement.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +from dataclasses import dataclass +from uuid import UUID + +from srm.domain.models.catalog import RuntimeKind +from srm.domain.models.topology import ( + Capability, + CapabilityKind, + CapabilityState, + ControlPathBinding, + ControlPathBindingState, + Domain, + DomainKind, + DomainState, + Zone, + ZoneState, +) +from srm.domain.ports.database.topology import DomainRepository, ZoneRepository + + +@dataclass(frozen=True, slots=True) +class PlacementPin: + zone_id: UUID | None = None + domain_id: UUID | None = None + + +@dataclass(frozen=True, slots=True) +class CapabilityPlacementRequest: + capability_kind: CapabilityKind + domain_kind: DomainKind + action: str + pin: PlacementPin + runtime_kind: RuntimeKind | None = None + + +@dataclass(frozen=True, slots=True) +class CapabilityPlacement: + zone_id: UUID + domain_id: UUID + capability_id: UUID + control_path_binding_id: UUID + control_path_ref: str + + +class CapabilityPlacementPlanner: + def __init__( + self, + zones: ZoneRepository, + domains: DomainRepository, + ) -> None: + self._zones = zones + self._domains = domains + + async def place(self, request: CapabilityPlacementRequest) -> CapabilityPlacement | None: + if request.pin.domain_id is not None and request.pin.zone_id is None: + return None + + if request.pin.domain_id is not None: + return await self._place_pinned_domain(request) + if request.pin.zone_id is not None: + return await self._place_pinned_zone(request) + return await self._place_unpinned(request) + + async def _place_pinned_domain( + self, + request: CapabilityPlacementRequest, + ) -> CapabilityPlacement | None: + assert request.pin.zone_id is not None + assert request.pin.domain_id is not None + + zone = await self._zones.get_by_id(request.pin.zone_id) + if zone is None or zone.state != ZoneState.ACTIVE: + return None + + domain = await self._domains.get_by_id(request.pin.domain_id) + if domain is None: + return None + if domain.zone_id != zone.id or domain.kind != request.domain_kind: + return None + + return self._select_from_domain(zone.id, domain, request) + + async def _place_pinned_zone( + self, + request: CapabilityPlacementRequest, + ) -> CapabilityPlacement | None: + assert request.pin.zone_id is not None + + zone = await self._zones.get_by_id(request.pin.zone_id) + if zone is None: + return None + + return self.place_in_loaded_zone(zone, request) + + async def _place_unpinned( + self, + request: CapabilityPlacementRequest, + ) -> CapabilityPlacement | None: + zones = await self._zones.list_active_resource_zones() + for zone in zones: + placement = self.place_in_loaded_zone(zone, request) + if placement is not None: + return placement + return None + + def place_in_loaded_zone( + self, + zone: Zone, + request: CapabilityPlacementRequest, + ) -> CapabilityPlacement | None: + if zone.state != ZoneState.ACTIVE: + return None + if request.pin.zone_id is not None and request.pin.zone_id != zone.id: + return None + + domains = [ + domain + for domain in zone.domains + if domain.kind == request.domain_kind + and domain.state == DomainState.ACTIVE + and (request.pin.domain_id is None or domain.id == request.pin.domain_id) + ] + return self._select_from_domains(zone.id, domains, request) + + def _select_from_domains( + self, + zone_id: UUID, + domains: list[Domain], + request: CapabilityPlacementRequest, + ) -> CapabilityPlacement | None: + for domain in sorted(domains, key=lambda item: item.ref): + placement = self._select_from_domain(zone_id, domain, request) + if placement is not None: + return placement + return None + + def _select_from_domain( + self, + zone_id: UUID, + domain: Domain, + request: CapabilityPlacementRequest, + ) -> CapabilityPlacement | None: + if domain.state != DomainState.ACTIVE or domain.kind != request.domain_kind: + return None + + for capability in sorted(domain.capabilities, key=lambda item: item.ref): + binding = self._select_binding(capability, request) + if binding is not None: + return CapabilityPlacement( + zone_id=zone_id, + domain_id=domain.id, + capability_id=capability.id, + control_path_binding_id=binding.id, + control_path_ref=binding.control_path_ref, + ) + return None + + def _select_binding( + self, + capability: Capability, + request: CapabilityPlacementRequest, + ) -> ControlPathBinding | None: + if capability.state != CapabilityState.ACTIVE or capability.kind != request.capability_kind: + return None + + candidates = [ + binding + for binding in capability.control_path_bindings + if binding.state == ControlPathBindingState.ACTIVE + and request.action in binding.supported_actions + and ( + request.runtime_kind is None + or request.runtime_kind in binding.supported_runtime_kinds + ) + ] + candidates.sort(key=lambda item: (not item.is_default, item.priority, item.ref)) + return candidates[0] if candidates else None diff --git a/src/srm/application/use_cases/deploy_service.py b/src/srm/application/use_cases/deploy_service.py new file mode 100644 index 0000000000000000000000000000000000000000..615cb2a09bfb5fdd8caa3f7db65b5c7f90aecb62 --- /dev/null +++ b/src/srm/application/use_cases/deploy_service.py @@ -0,0 +1,628 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any, Literal +from uuid import UUID + +import structlog + +from srm.application.services.capability_placement import ( + CapabilityPlacement, + CapabilityPlacementPlanner, + CapabilityPlacementRequest, + PlacementPin, +) +from srm.domain.models.canonical_parameters.result import Result, ResultStatus +from srm.domain.models.catalog import ( + RuntimeKind, + ServiceCapabilityRequirement, + ServiceSpecificationState, +) +from srm.domain.models.runtime_inventory import CapabilityInstance, ServiceInstance, ServiceOrder +from srm.domain.models.runtime_inventory.enums import ( + CapabilityInstanceKind, + CapabilityInstanceState, + ServiceInstanceState, + ServiceOrderState, + ServiceOrderType, +) +from srm.domain.models.topology import CapabilityKind, DomainKind, Zone +from srm.domain.ports.database.catalog import ( + ServiceCapabilityRequirementRepository, + ServiceDeploymentUnitRepository, + ServiceSpecificationRepository, +) +from srm.domain.ports.database.runtime_inventory import ( + CapabilityInstanceRepository, + ServiceInstanceRepository, + ServiceOrderRepository, +) +from srm.domain.ports.database.topology import DomainRepository, ZoneRepository +from srm.domain.ports.databus.events import ( + OperationCompletedStatus, + OperationStatusState, + SrmOperationCompletedV1, + SrmOperationStatusV1, +) +from srm.domain.ports.databus.publisher import DataBusPublisher + + +@dataclass(frozen=True, slots=True) +class DeployTargetCommand: + app_instance_id: UUID + zone_id: UUID | None = None + domain_id: UUID | None = None + + +@dataclass(frozen=True, slots=True) +class DeployPayloadCommand: + instance_name: str | None = None + placement_constraints: dict[str, Any] | None = None + + +@dataclass(frozen=True, slots=True) +class DeployServiceCommand: + operation_id: UUID + correlation_id: str + requested_at: datetime + app_provider_id: str + source: Literal["nbi_camara", "nbi_tmf", "operator_portal", "federation"] + service_specification_id: UUID + targets: list[DeployTargetCommand] + deploy: DeployPayloadCommand + federation_partner_ref: str | None = None + payload_snapshot: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DeployRequirementPlacement: + requirement_id: UUID + placement: CapabilityPlacement + + +@dataclass(frozen=True, slots=True) +class DeployTargetPlacement: + app_instance_id: UUID + zone_id: UUID + requirements: list[DeployRequirementPlacement] + + +@dataclass(frozen=True, slots=True) +class DeployServiceResult: + accepted_order_id: UUID | None = None + duplicate_order_id: UUID | None = None + target_placements: list[DeployTargetPlacement] = field(default_factory=list) + deploy_material: list[tuple[ServiceCapabilityRequirement, RuntimeKind]] = field( + default_factory=list + ) + completed_event: SrmOperationCompletedV1 | None = None + + +class DeployServiceUseCase: + def __init__( + self, + service_specifications: ServiceSpecificationRepository, + service_deployment_units: ServiceDeploymentUnitRepository, + service_capability_requirements: ServiceCapabilityRequirementRepository, + service_orders: ServiceOrderRepository, + service_instances: ServiceInstanceRepository, + capability_instances: CapabilityInstanceRepository, + zones: ZoneRepository, + domains: DomainRepository, + publisher: DataBusPublisher, + placement_planner: CapabilityPlacementPlanner | None = None, + ) -> None: + self._service_specifications = service_specifications + self._service_deployment_units = service_deployment_units + self._service_capability_requirements = service_capability_requirements + self._service_orders = service_orders + self._service_instances = service_instances + self._capability_instances = capability_instances + self._zones = zones + self._domains = domains + self._publisher = publisher + self._placement_planner = placement_planner or CapabilityPlacementPlanner(zones, domains) + + async def accept(self, command: DeployServiceCommand) -> DeployServiceResult: + existing_order = await self._service_orders.get_by_operation_id(command.operation_id) + if existing_order is not None: + return await self._replay_existing_order(command, existing_order) + + service_specification = await self._service_specifications.get_by_id( + command.service_specification_id + ) + if service_specification is None: + await self._publish_failed_before_start(command, service_order_id=None) + return DeployServiceResult() + if service_specification.state != ServiceSpecificationState.ACTIVE: + await self._publish_failed_before_start(command, service_order_id=None) + return DeployServiceResult() + deploy_material = await self._resolve_deploy_requirements(command.service_specification_id) + if deploy_material is None: + await self._publish_failed_before_start(command, service_order_id=None) + return DeployServiceResult() + + placements = await self._place_targets(command.targets, deploy_material) + if placements is None: + await self._publish_failed_before_start(command, service_order_id=None) + return DeployServiceResult() + + service_order = await self._service_orders.create( + ServiceOrder( + operation_id=command.operation_id, + correlation_id=command.correlation_id, + order_type=ServiceOrderType.DEPLOY_SERVICE, + service_specification_id=command.service_specification_id, + target_service_instance_id=None, + state=ServiceOrderState.ACCEPTED, + payload_snapshot=command.payload_snapshot, + app_provider_id=command.app_provider_id, + federation_partner_ref=command.federation_partner_ref, + ) + ) + service_instances: list[ServiceInstance] = [] + for target, target_placement in zip(command.targets, placements, strict=True): + service_instance = ServiceInstance( + id=target.app_instance_id, + service_specification_id=command.service_specification_id, + originating_service_order_id=service_order.id, + zone_id=target_placement.zone_id, + ref=command.deploy.instance_name or str(target.app_instance_id), + state=ServiceInstanceState.CREATING, + app_provider_id=command.app_provider_id, + federation_partner_ref=command.federation_partner_ref, + ) + await self._service_instances.create(service_instance) + service_instances.append(service_instance) + + structlog.get_logger(__name__).info( + "service_deploy_accepted", + operation_id=str(command.operation_id), + service_specification_id=str(command.service_specification_id), + target_count=len(command.targets), + ) + return DeployServiceResult( + accepted_order_id=service_order.id, + target_placements=placements, + deploy_material=deploy_material, + ) + + async def _replay_existing_order( + self, + command: DeployServiceCommand, + service_order: ServiceOrder, + ) -> DeployServiceResult: + if service_order.state == ServiceOrderState.COMPLETED: + return DeployServiceResult( + duplicate_order_id=service_order.id, + completed_event=await self._completed_event_from_order( + command, + service_order, + status=OperationCompletedStatus.COMPLETED, + ), + ) + if service_order.state == ServiceOrderState.FAILED: + return DeployServiceResult( + duplicate_order_id=service_order.id, + completed_event=await self._completed_event_from_order( + command, + service_order, + status=OperationCompletedStatus.FAILED, + ), + ) + + return DeployServiceResult(duplicate_order_id=service_order.id) + + async def _completed_event_from_order( + self, + command: DeployServiceCommand, + service_order: ServiceOrder, + *, + status: OperationCompletedStatus, + ) -> SrmOperationCompletedV1: + service_instances = await self._service_instances.list_by_originating_service_order_id( + service_order.id + ) + instance_status = "completed" if status == OperationCompletedStatus.COMPLETED else "failed" + problem = service_order.failure_detail or { + "type": "about:blank", + "title": "Deploy service failed.", + "status": 500, + } + + return SrmOperationCompletedV1( + schema_version="1.0", + operation_id=command.operation_id, + status=status, + service_order_id=service_order.id, + instances=[ + { + "service_instance_id": str(service_instance.id), + "zone_id": str(service_instance.zone_id), + "status": instance_status, + **({"error": problem} if status == OperationCompletedStatus.FAILED else {}), + } + for service_instance in service_instances + ], + error=( + problem + if status == OperationCompletedStatus.FAILED and not service_instances + else None + ), + correlation_id=command.correlation_id, + completed_at=service_order.completed_at or datetime.now(UTC), + ) + + async def complete( + self, + command: DeployServiceCommand, + accepted: DeployServiceResult, + ) -> SrmOperationCompletedV1 | None: + if accepted.accepted_order_id is None: + return None + + service_order = await self._service_orders.get_by_id(accepted.accepted_order_id) + if service_order is None: + raise RuntimeError("Accepted deploy service order could not be reloaded") + + for service_instance, target_placement in zip( + [ + ServiceInstance( + id=target_placement.app_instance_id, + service_specification_id=command.service_specification_id, + originating_service_order_id=service_order.id, + zone_id=target_placement.zone_id, + ref=command.deploy.instance_name or str(target_placement.app_instance_id), + state=ServiceInstanceState.CREATING, + app_provider_id=command.app_provider_id, + federation_partner_ref=command.federation_partner_ref, + ) + for target_placement in accepted.target_placements + ], + accepted.target_placements, + strict=True, + ): + await self._create_deploy_capability_instances( + command, + service_order, + service_instance, + target_placement, + accepted.deploy_material, + ) + await self._service_instances.update_state( + service_instance.id, + ServiceInstanceState.ACTIVE, + ) + + completed_at = datetime.now(UTC) + await self._service_orders.update_state( + service_order.id, + ServiceOrderState.COMPLETED, + completed_at=completed_at, + ) + completed_event = SrmOperationCompletedV1( + schema_version="1.0", + operation_id=command.operation_id, + status=OperationCompletedStatus.COMPLETED, + service_order_id=service_order.id, + instances=[ + { + "service_instance_id": str(target_placement.app_instance_id), + "zone_id": str(target_placement.zone_id), + "status": "completed", + } + for target_placement in accepted.target_placements + ], + correlation_id=command.correlation_id, + completed_at=completed_at, + ) + + structlog.get_logger(__name__).info( + "service_deploy_use_case_invoked", + operation_id=str(command.operation_id), + service_specification_id=str(command.service_specification_id), + target_count=len(command.targets), + ) + return completed_event + + async def fail_after_accept( + self, + command: DeployServiceCommand, + accepted: DeployServiceResult, + error: Exception, + ) -> SrmOperationCompletedV1: + if accepted.accepted_order_id is None: + raise RuntimeError("Cannot fail deploy after accept without a service order") + + service_order = await self._service_orders.get_by_id(accepted.accepted_order_id) + if service_order is None: + raise RuntimeError("Accepted deploy service order could not be reloaded") + + completed_at = datetime.now(UTC) + problem = { + "type": "about:blank", + "title": "Deploy service failed after acceptance.", + "status": 500, + "detail": str(error), + } + for target_placement in accepted.target_placements: + await self._service_instances.update_state( + target_placement.app_instance_id, + ServiceInstanceState.FAILED, + ) + await self._service_orders.update_state( + service_order.id, + ServiceOrderState.FAILED, + completed_at=completed_at, + failure_detail=problem, + ) + + return SrmOperationCompletedV1( + schema_version="1.0", + operation_id=command.operation_id, + status=OperationCompletedStatus.FAILED, + service_order_id=service_order.id, + instances=[ + { + "service_instance_id": str(target_placement.app_instance_id), + "zone_id": str(target_placement.zone_id), + "status": "failed", + "error": problem, + } + for target_placement in accepted.target_placements + ], + error=None if accepted.target_placements else problem, + correlation_id=command.correlation_id, + completed_at=completed_at, + ) + + async def _create_deploy_capability_instances( + self, + command: DeployServiceCommand, + service_order: ServiceOrder, + service_instance: ServiceInstance, + target_placement: DeployTargetPlacement, + deploy_material: list[tuple[ServiceCapabilityRequirement, RuntimeKind]], + ) -> None: + # TODO: TF-SDK CapabilityExecutionAdapter.activate_capability(...) call using + # capability_kind="deploy_workload" and placement.control_path_ref. + requirements_by_id = {requirement.id: requirement for requirement, _ in deploy_material} + for requirement_placement in target_placement.requirements: + requirement = requirements_by_id[requirement_placement.requirement_id] + placement = requirement_placement.placement + await self._capability_instances.create( + CapabilityInstance( + capability_id=placement.capability_id, + service_instance_id=service_instance.id, + originating_service_order_id=service_order.id, + service_capability_requirement_id=requirement.id, + control_path_binding_id=placement.control_path_binding_id, + control_path_ref_snapshot=placement.control_path_ref, + ref=requirement.ref, + kind=CapabilityInstanceKind.ACTIVATION, + state=CapabilityInstanceState.ACTIVE, + external_id=None, + external_ref=None, + parameters_snapshot=requirement.parameters, + result_summary=Result(status=ResultStatus.ACTIVE), + app_provider_id=command.app_provider_id, + federation_partner_ref=command.federation_partner_ref, + ) + ) + + async def _resolve_deploy_requirements( + self, + service_specification_id: UUID, + ) -> list[tuple[ServiceCapabilityRequirement, RuntimeKind]] | None: + deployment_units = await self._service_deployment_units.list_by_service_specification_id( + service_specification_id + ) + if not deployment_units: + return None + + unit_ids = {unit.id for unit in deployment_units} + units_by_id = {unit.id: unit for unit in deployment_units} + capability_requirements = ( + await self._service_capability_requirements.list_by_service_specification_id( + service_specification_id + ) + ) + deploy_requirements = [ + requirement + for requirement in capability_requirements + if requirement.is_required + and requirement.capability_kind == CapabilityKind.DEPLOY_WORKLOAD + ] + if not deploy_requirements: + return None + + material: list[tuple[ServiceCapabilityRequirement, RuntimeKind]] = [] + for requirement in deploy_requirements: + if requirement.domain_kind not in (None, DomainKind.COMPUTE): + return None + if requirement.deployment_unit_id is not None: + if requirement.deployment_unit_id not in unit_ids: + return None + runtime_kind = units_by_id[requirement.deployment_unit_id].runtime_kind + else: + runtime_kind = deployment_units[0].runtime_kind + material.append((requirement, runtime_kind)) + + return material + + async def _place_targets( + self, + targets: list[DeployTargetCommand], + deploy_material: list[tuple[ServiceCapabilityRequirement, RuntimeKind]], + ) -> list[DeployTargetPlacement] | None: + candidate_zones: list[Zone] | None = None + if len(targets) > 1: + candidate_zones = await self._zones.list_active_resource_zones() + if len(candidate_zones) < len(targets): + return None + + placements: list[DeployTargetPlacement] = [] + placed_zone_ids: set[UUID] = set() + for target in targets: + if target.zone_id in placed_zone_ids: + return None + + target_placement = await self._place_target( + target, + deploy_material, + excluded_zone_ids=placed_zone_ids, + candidate_zones=candidate_zones, + ) + if target_placement is None: + return None + if target_placement.zone_id in placed_zone_ids: + return None + + placements.append(target_placement) + placed_zone_ids.add(target_placement.zone_id) + return placements + + async def _place_target( + self, + target: DeployTargetCommand, + deploy_material: list[tuple[ServiceCapabilityRequirement, RuntimeKind]], + excluded_zone_ids: set[UUID], + candidate_zones: list[Zone] | None, + ) -> DeployTargetPlacement | None: + if target.zone_id is not None: + if target.domain_id is None: + zone = await self._zones.get_by_id(target.zone_id) + if zone is None: + return None + return self._place_target_in_loaded_zone(target, deploy_material, zone) + return await self._place_target_in_pin(target, deploy_material) + + if target.domain_id is not None: + return None + + zones = ( + candidate_zones + if candidate_zones is not None + else await self._zones.list_active_resource_zones() + ) + for zone in zones: + if zone.id in excluded_zone_ids: + continue + placement = self._place_target_in_loaded_zone(target, deploy_material, zone) + if placement is not None: + return placement + return None + + async def _place_target_in_pin( + self, + target: DeployTargetCommand, + deploy_material: list[tuple[ServiceCapabilityRequirement, RuntimeKind]], + ) -> DeployTargetPlacement | None: + requirement_placements: list[DeployRequirementPlacement] = [] + for requirement, runtime_kind in deploy_material: + placement = await self._placement_planner.place( + self._placement_request( + requirement, + runtime_kind, + PlacementPin(zone_id=target.zone_id, domain_id=target.domain_id), + ) + ) + if placement is None: + return None + requirement_placements.append( + DeployRequirementPlacement( + requirement_id=requirement.id, + placement=placement, + ) + ) + + return DeployTargetPlacement( + app_instance_id=target.app_instance_id, + zone_id=requirement_placements[0].placement.zone_id, + requirements=requirement_placements, + ) + + def _place_target_in_loaded_zone( + self, + target: DeployTargetCommand, + deploy_material: list[tuple[ServiceCapabilityRequirement, RuntimeKind]], + zone: Zone, + ) -> DeployTargetPlacement | None: + requirement_placements: list[DeployRequirementPlacement] = [] + for requirement, runtime_kind in deploy_material: + placement = self._placement_planner.place_in_loaded_zone( + zone, + self._placement_request( + requirement, + runtime_kind, + PlacementPin(zone_id=zone.id), + ), + ) + if placement is None: + return None + requirement_placements.append( + DeployRequirementPlacement( + requirement_id=requirement.id, + placement=placement, + ) + ) + + return DeployTargetPlacement( + app_instance_id=target.app_instance_id, + zone_id=zone.id, + requirements=requirement_placements, + ) + + def _placement_request( + self, + requirement: ServiceCapabilityRequirement, + runtime_kind: RuntimeKind, + pin: PlacementPin, + ) -> CapabilityPlacementRequest: + return CapabilityPlacementRequest( + capability_kind=requirement.capability_kind, + domain_kind=requirement.domain_kind or DomainKind.COMPUTE, + action="deploy", + runtime_kind=runtime_kind, + pin=pin, + ) + + async def _publish_failed_before_start( + self, + command: DeployServiceCommand, + *, + service_order_id: UUID | None, + ) -> None: + now = datetime.now(UTC) + status_event = SrmOperationStatusV1( + schema_version="1.0", + operation_id=command.operation_id, + service_order_id=service_order_id, + service_instance_id=None, + capability="deploy", + state=OperationStatusState.FAILED_BEFORE_START, + correlation_id=command.correlation_id, + emitted_at=now, + ) + completed_event = SrmOperationCompletedV1( + schema_version="1.0", + operation_id=command.operation_id, + status=OperationCompletedStatus.FAILED, + service_order_id=service_order_id, + error={ + "type": "about:blank", + "title": "Deploy service failed before start.", + "status": 400, + }, + correlation_id=command.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"), + ) diff --git a/src/srm/domain/ports/database/catalog.py b/src/srm/domain/ports/database/catalog.py index bdfb3fcb107f65fac467c9d521b403edfccd4f50..c1bf0228d9b6880ae9b39523a8077876cad68371 100644 --- a/src/srm/domain/ports/database/catalog.py +++ b/src/srm/domain/ports/database/catalog.py @@ -29,6 +29,13 @@ class ServiceDeploymentUnitRepository(ABC): async def get_by_id(self, id: UUID) -> ServiceDeploymentUnit | None: pass + @abstractmethod + async def list_by_service_specification_id( + self, + service_specification_id: UUID, + ) -> list[ServiceDeploymentUnit]: + pass + @abstractmethod async def create(self, deployment_unit: ServiceDeploymentUnit) -> ServiceDeploymentUnit: pass @@ -39,6 +46,13 @@ class ServiceCapabilityRequirementRepository(ABC): async def get_by_id(self, id: UUID) -> ServiceCapabilityRequirement | None: pass + @abstractmethod + async def list_by_service_specification_id( + self, + service_specification_id: UUID, + ) -> list[ServiceCapabilityRequirement]: + pass + @abstractmethod async def create( self, diff --git a/src/srm/domain/ports/database/runtime_inventory.py b/src/srm/domain/ports/database/runtime_inventory.py index bdf08340c8ca55e540a5318b4f4452fc73a8da4d..3d8ad8394f8490742e89e2f8580b93c7b5f4af5a 100644 --- a/src/srm/domain/ports/database/runtime_inventory.py +++ b/src/srm/domain/ports/database/runtime_inventory.py @@ -1,6 +1,7 @@ """Runtime inventory repository ports.""" from abc import ABC, abstractmethod +from datetime import datetime from uuid import UUID from srm.domain.models.runtime_inventory import ( @@ -8,6 +9,7 @@ from srm.domain.models.runtime_inventory import ( ServiceInstance, ServiceOrder, ) +from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState class ServiceOrderRepository(ABC): @@ -15,20 +17,50 @@ class ServiceOrderRepository(ABC): async def get_by_id(self, id: UUID) -> ServiceOrder | None: pass + @abstractmethod + async def get_by_operation_id(self, operation_id: UUID) -> ServiceOrder | None: + pass + @abstractmethod async def create(self, service_order: ServiceOrder) -> ServiceOrder: pass + @abstractmethod + async def update_state( + self, + id: UUID, + state: ServiceOrderState, + *, + completed_at: datetime | None = None, + failure_detail: dict[str, object] | None = None, + ) -> ServiceOrder: + pass + class ServiceInstanceRepository(ABC): @abstractmethod async def get_by_id(self, id: UUID) -> ServiceInstance | None: pass + @abstractmethod + async def list_by_originating_service_order_id( + self, + originating_service_order_id: UUID, + ) -> list[ServiceInstance]: + pass + @abstractmethod async def create(self, service_instance: ServiceInstance) -> ServiceInstance: pass + @abstractmethod + async def update_state( + self, + id: UUID, + state: ServiceInstanceState, + ) -> ServiceInstance: + pass + class CapabilityInstanceRepository(ABC): @abstractmethod diff --git a/src/srm/domain/ports/database/topology.py b/src/srm/domain/ports/database/topology.py index a208cff74b647ec3fd26557feb4583cad2de56ca..af8fa61d83fed8d18f2ae082f930d0726baacca8 100644 --- a/src/srm/domain/ports/database/topology.py +++ b/src/srm/domain/ports/database/topology.py @@ -16,6 +16,10 @@ class ZoneRepository(ABC): async def get_by_id(self, id: UUID) -> Zone | None: pass + @abstractmethod + async def list_active_resource_zones(self) -> list[Zone]: + pass + @abstractmethod async def create(self, zone: Zone) -> Zone: pass diff --git a/src/srm/domain/ports/databus/events.py b/src/srm/domain/ports/databus/events.py new file mode 100644 index 0000000000000000000000000000000000000000..27ea1813c6e7383d3aa4e31da8290a55ea905ab2 --- /dev/null +++ b/src/srm/domain/ports/databus/events.py @@ -0,0 +1,48 @@ +from datetime import datetime +from enum import StrEnum +from typing import Any, Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + + +class EventSchema(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class OperationStatusState(StrEnum): + ACCEPTED = "accepted" + FAILED_BEFORE_START = "failed_before_start" + COMPLETED = "completed" + FAILED = "failed" + IN_PROGRESS = "in_progress" + + +class OperationCompletedStatus(StrEnum): + COMPLETED = "completed" + PARTIALLY_COMPLETED = "partially_completed" + FAILED = "failed" + + +class SrmOperationStatusV1(EventSchema): + schema_version: Literal["1.0"] + operation_id: UUID + service_order_id: UUID | None = None + service_instance_id: UUID | None = None + capability: str | None = None + state: OperationStatusState + metadata: dict[str, Any] = Field(default_factory=dict) + correlation_id: str + emitted_at: datetime + + +class SrmOperationCompletedV1(EventSchema): + schema_version: Literal["1.0"] + operation_id: UUID + status: OperationCompletedStatus + service_order_id: UUID | None = None + instances: list[dict[str, Any]] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + error: dict[str, Any] | None = None + correlation_id: str + completed_at: datetime diff --git a/src/srm/main.py b/src/srm/main.py index 6c7334a5f41df3616c112eb210e7933ba132a012..e6d0884550eee6d9eef8c0055f5647e342678b03 100644 --- a/src/srm/main.py +++ b/src/srm/main.py @@ -65,7 +65,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: raise try: - databus_subscribers: list[NatsSubscriber] = await subscribe_to_subjects(databus_manager) + databus_subscribers: list[NatsSubscriber] = await subscribe_to_subjects( + connection_manager=databus_manager, + session_maker=session_maker, + ) except Exception as e: logger.error("Databus subscription failed!", error=str(e)) await databus_manager.close() diff --git a/tests/api/databus/test_dependencies.py b/tests/api/databus/test_dependencies.py new file mode 100644 index 0000000000000000000000000000000000000000..7b9908a50fa751bba5d54e536c431b232b766079 --- /dev/null +++ b/tests/api/databus/test_dependencies.py @@ -0,0 +1,46 @@ +from unittest.mock import MagicMock, patch + +from srm.api.databus.dependencies import get_deploy_service_use_case + + +def test_get_deploy_service_use_case_wires_required_repositories() -> None: + session = MagicMock() + connection_manager = MagicMock() + + with ( + patch("srm.api.databus.dependencies.SqlServiceSpecificationRepository") as specs, + patch("srm.api.databus.dependencies.SqlServiceDeploymentUnitRepository") as units, + 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.NatsPublisher") as publisher, + patch("srm.api.databus.dependencies.DeployServiceUseCase") as use_case, + ): + result = get_deploy_service_use_case(session, connection_manager) + + specs.assert_called_once_with(session) + units.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) + publisher.assert_called_once_with(connection_manager) + use_case.assert_called_once_with( + service_specifications=specs.return_value, + service_deployment_units=units.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.return_value, + ) + assert result is use_case.return_value diff --git a/tests/api/databus/test_nats_subscriber.py b/tests/api/databus/test_nats_subscriber.py index 65a97ef29f97a32bb19003aa89764b083cddaede..e4b7c6b9934c52f04fe9f53c7142e54426805fe2 100644 --- a/tests/api/databus/test_nats_subscriber.py +++ b/tests/api/databus/test_nats_subscriber.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest import structlog.testing @@ -12,7 +12,7 @@ from srm.api.databus.nats_subscriber import ( NatsSubscriber, subscribe_to_subjects, ) -from srm.api.databus.schemas import InboundMessage +from srm.api.databus.schemas import AnswerableCommandV1 @dataclass @@ -22,6 +22,27 @@ class FakeMsg: headers: dict[str, str] | None = field(default=None) +class FakeSessionContext: + def __init__(self) -> None: + self.session = AsyncMock() + + async def __aenter__(self) -> AsyncMock: + return self.session + + async def __aexit__(self, *_args: object) -> None: + return None + + +class FakeSessionMaker: + def __init__(self) -> None: + self.contexts: list[FakeSessionContext] = [] + + def __call__(self) -> FakeSessionContext: + context = FakeSessionContext() + self.contexts.append(context) + return context + + @pytest.fixture def connection_manager() -> MagicMock: manager = MagicMock(spec=NatsConnectionManager) @@ -32,13 +53,13 @@ def connection_manager() -> MagicMock: def make_subscriber( connection_manager: MagicMock, subject: str = "command.srm.service.deploy" ) -> tuple[NatsSubscriber, AsyncMock]: - router = AsyncMock() + callback = AsyncMock() subscriber = NatsSubscriber( connection_manager=connection_manager, subject=subject, - router=router, + callback=callback, ) - return subscriber, router + return subscriber, callback async def test_start_subscribes_via_connection_manager_client( @@ -55,65 +76,144 @@ async def test_start_subscribes_via_connection_manager_client( assert subscriber._subscription is not None -async def test_handle_message_builds_inbound_message_and_invokes_router( +async def test_handle_message_builds_inbound_message_and_invokes_callback( connection_manager: MagicMock, ) -> None: - subscriber, router = make_subscriber(connection_manager) + subscriber, callback = make_subscriber(connection_manager) msg = FakeMsg( subject="command.srm.service.deploy", - data=b'{"operation_id": "abc-123"}', + data=( + b'{"schema_version": "1.0",' + b' "operation_id": "11111111-2222-3333-4444-555555555555",' + b' "correlation_id": "corr-1",' + b' "requested_at": "2026-06-19T12:00:00Z",' + b' "app_provider_id": "VideoAppsCo",' + b' "source": "nbi_camara"}' + ), headers={"x-correlation-id": "corr-1"}, ) await subscriber._handle_message(msg) # type: ignore[arg-type] - router.assert_awaited_once() - assert router.await_args is not None - (inbound,) = router.await_args.args - assert isinstance(inbound, InboundMessage) - assert inbound.subject == "command.srm.service.deploy" - assert inbound.payload == b'{"operation_id": "abc-123"}' - assert inbound.headers == {"x-correlation-id": "corr-1"} + callback.assert_awaited_once() + assert callback.await_args is not None + (envelope,) = callback.await_args.args + assert isinstance(envelope, AnswerableCommandV1) + assert envelope.correlation_id == "corr-1" + assert envelope.model_extra == {} + + +async def test_handle_message_preserves_command_payload_fields_on_validated_envelope( + connection_manager: MagicMock, +) -> None: + subscriber, callback = make_subscriber(connection_manager) + msg = FakeMsg( + subject="command.srm.service.deploy", + data=b""" + { + "schema_version": "1.0", + "operation_id": "11111111-2222-3333-4444-555555555555", + "correlation_id": "corr-1", + "requested_at": "2026-06-19T12:00:00Z", + "app_provider_id": "VideoAppsCo", + "source": "nbi_camara", + "service_specification_id": "d4e5f6a7-0000-4000-8000-000000000001", + "targets": [ + { + "app_instance_id": "9a3f1c22-0000-4000-8000-000000000002" + } + ], + "deploy": {} + } + """, + ) + + await subscriber._handle_message(msg) # type: ignore[arg-type] + + callback.assert_awaited_once() + assert callback.await_args is not None + (envelope,) = callback.await_args.args + assert envelope.model_extra is not None + assert envelope.model_extra["service_specification_id"] == ( + "d4e5f6a7-0000-4000-8000-000000000001" + ) + assert envelope.model_extra["targets"] == [ + {"app_instance_id": "9a3f1c22-0000-4000-8000-000000000002"} + ] + assert envelope.model_extra["deploy"] == {} async def test_handle_message_defaults_headers_to_empty_dict_when_none( connection_manager: MagicMock, ) -> None: - subscriber, router = make_subscriber(connection_manager) - msg = FakeMsg(subject="command.srm.service.deploy", data=b"{}", headers=None) + subscriber, callback = make_subscriber(connection_manager) + msg = FakeMsg( + subject="command.srm.service.deploy", + data=( + b'{"schema_version": "1.0",' + b' "operation_id": "11111111-2222-3333-4444-555555555555",' + b' "correlation_id": "corr-1",' + b' "requested_at": "2026-06-19T12:00:00Z",' + b' "app_provider_id": "VideoAppsCo",' + b' "source": "nbi_camara"}' + ), + headers=None, + ) await subscriber._handle_message(msg) # type: ignore[arg-type] - assert router.await_args is not None - (inbound,) = router.await_args.args - assert inbound.headers == {} + assert callback.await_args is not None + (envelope,) = callback.await_args.args + assert envelope.correlation_id == "corr-1" async def test_handle_message_passes_malformed_payload_through_unparsed( connection_manager: MagicMock, ) -> None: - subscriber, router = make_subscriber(connection_manager) + subscriber, callback = make_subscriber(connection_manager) msg = FakeMsg(subject="command.srm.service.deploy", data=b'{"operation_id": ') + with structlog.testing.capture_logs() as logs: + await subscriber._handle_message(msg) # type: ignore[arg-type] + + callback.assert_not_awaited() + assert [(entry["event"], entry["subject"], entry["log_level"]) for entry in logs] == [ + ("databus_stage1_unanswerable", "command.srm.service.deploy", "error") + ] + + +async def test_handle_message_logs_stage1_unanswerable_when_envelope_is_invalid( + connection_manager: MagicMock, +) -> None: + subscriber, callback = make_subscriber(connection_manager) + msg = FakeMsg(subject="command.srm.service.deploy", data=b'{"operation_id": "abc-123"}') + await subscriber._handle_message(msg) # type: ignore[arg-type] - router.assert_awaited_once() - assert router.await_args is not None - (inbound,) = router.await_args.args - assert inbound.payload == b'{"operation_id": ' + callback.assert_not_awaited() -async def test_handle_message_logs_dropped_command_when_router_fails( +async def test_handle_message_logs_dropped_command_when_callback_fails( connection_manager: MagicMock, ) -> None: - subscriber, router = make_subscriber(connection_manager) - router.side_effect = RuntimeError("handler blew up") - msg = FakeMsg(subject="command.srm.service.deploy", data=b"{}") + subscriber, callback = make_subscriber(connection_manager) + callback.side_effect = RuntimeError("handler blew up") + msg = FakeMsg( + subject="command.srm.service.deploy", + data=( + b'{"schema_version": "1.0",' + b' "operation_id": "11111111-2222-3333-4444-555555555555",' + b' "correlation_id": "corr-1",' + b' "requested_at": "2026-06-19T12:00:00Z",' + b' "app_provider_id": "VideoAppsCo",' + b' "source": "nbi_camara"}' + ), + ) with structlog.testing.capture_logs() as logs: await subscriber._handle_message(msg) # type: ignore[arg-type] - router.assert_awaited_once() + callback.assert_awaited_once() assert [(entry["event"], entry["subject"], entry["log_level"]) for entry in logs] == [ ("databus_handler_failed", "command.srm.service.deploy", "error") ] @@ -133,10 +233,52 @@ def test_command_subjects_matches_the_contract() -> None: assert sorted(COMMAND_SUBJECTS) == sorted(EXPECTED_COMMAND_SUBJECTS) +async def test_deploy_subscription_callback_delegates_to_command_coordinator( + connection_manager: MagicMock, +) -> None: + envelope = AnswerableCommandV1.model_validate( + { + "schema_version": "1.0", + "operation_id": "11111111-2222-3333-4444-555555555555", + "correlation_id": "corr-appinst-001", + "requested_at": "2026-06-19T12:00:00Z", + "app_provider_id": "VideoAppsCo", + "source": "nbi_camara", + "service_specification_id": "d4e5f6a7-0000-4000-8000-000000000001", + "targets": [ + { + "app_instance_id": "9a3f1c22-0000-4000-8000-000000000002", + "zone_id": "642f6105-7015-4af1-a4d1-e1ecb8437abc", + } + ], + "deploy": {}, + } + ) + session_maker = FakeSessionMaker() + with patch( + "srm.api.databus.nats_subscriber.get_deploy_service_command_coordinator" + ) as coordinator_factory: + coordinator = MagicMock() + coordinator.handle = AsyncMock() + coordinator_factory.return_value = coordinator + + subscribers = await subscribe_to_subjects(connection_manager, session_maker) # type: ignore[arg-type] + deploy_subscriber = next( + subscriber + for subscriber in subscribers + if subscriber._subject == "command.srm.service.deploy" + ) + await deploy_subscriber._callback(envelope) + + coordinator_factory.assert_called_once_with(session_maker, connection_manager) + coordinator.handle.assert_awaited_once_with(envelope) + assert session_maker.contexts == [] + + async def test_subscribe_to_subjects_registers_all_command_subjects( connection_manager: MagicMock, ) -> None: - subscribers = await subscribe_to_subjects(connection_manager) + subscribers = await subscribe_to_subjects(connection_manager, FakeSessionMaker()) # type: ignore[arg-type] assert sorted(sub._subject for sub in subscribers) == sorted(EXPECTED_COMMAND_SUBJECTS) assert connection_manager.client.subscribe.await_count == len(EXPECTED_COMMAND_SUBJECTS) diff --git a/tests/api/databus/test_schemas.py b/tests/api/databus/test_schemas.py index dac5f7d5dbdc606145d32265043e1308284400c0..6529dbd040cd17049a47164375b84438b376c404 100644 --- a/tests/api/databus/test_schemas.py +++ b/tests/api/databus/test_schemas.py @@ -82,15 +82,11 @@ class TestSource: class TestFederationContext: - def test_federation_source_requires_partner_ref(self) -> None: - with pytest.raises(ValidationError, match="federation_partner_ref is required"): - CommandEnvelopeV1.model_validate(_envelope(source="federation")) + def test_envelope_accepts_federation_without_partner_ref_as_answerable(self) -> None: + command = CommandEnvelopeV1.model_validate(_envelope(source="federation")) - def test_federation_source_rejects_explicit_null_partner_ref(self) -> None: - with pytest.raises(ValidationError, match="federation_partner_ref is required"): - CommandEnvelopeV1.model_validate( - _envelope(source="federation", federation_partner_ref=None) - ) + assert command.source == "federation" + assert command.federation_partner_ref is None def test_federation_source_accepts_partner_ref(self) -> None: command = CommandEnvelopeV1.model_validate( @@ -102,7 +98,7 @@ class TestFederationContext: def test_non_federation_source_may_omit_partner_ref(self) -> None: assert CommandEnvelopeV1.model_validate(_envelope()).federation_partner_ref is None - def test_every_command_inherits_the_rule(self) -> None: + def test_deploy_command_requires_federation_partner_ref(self) -> None: payload = _envelope( source="federation", service_specification_id=str(uuid4()), diff --git a/tests/application/test_deploy_service_command_coordinator.py b/tests/application/test_deploy_service_command_coordinator.py new file mode 100644 index 0000000000000000000000000000000000000000..7a7468119401740a7588d3546878daec3713ce28 --- /dev/null +++ b/tests/application/test_deploy_service_command_coordinator.py @@ -0,0 +1,250 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock +from uuid import UUID + +from srm.api.databus.schemas import AnswerableCommandV1 +from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator +from srm.application.use_cases.deploy_service import DeployServiceResult +from srm.domain.ports.databus.events import ( + OperationCompletedStatus, + SrmOperationCompletedV1, +) + + +@dataclass +class FakeSessionContext: + session: AsyncMock + + async def __aenter__(self) -> AsyncMock: + return self.session + + async def __aexit__(self, *_args: object) -> None: + return None + + +class FakeSessionMaker: + def __init__(self) -> None: + self.contexts: list[FakeSessionContext] = [] + + def __call__(self) -> FakeSessionContext: + context = FakeSessionContext(session=AsyncMock()) + self.contexts.append(context) + return context + + +def _valid_envelope(*, targets: list[dict[str, str]] | None = None) -> AnswerableCommandV1: + return AnswerableCommandV1.model_validate( + { + "schema_version": "1.0", + "operation_id": "11111111-2222-3333-4444-555555555555", + "correlation_id": "corr-appinst-001", + "requested_at": "2026-06-19T12:00:00Z", + "app_provider_id": "VideoAppsCo", + "source": "nbi_camara", + "service_specification_id": "d4e5f6a7-0000-4000-8000-000000000001", + "targets": targets + if targets is not None + else [ + { + "app_instance_id": "9a3f1c22-0000-4000-8000-000000000002", + "zone_id": "642f6105-7015-4af1-a4d1-e1ecb8437abc", + } + ], + "deploy": {"instance_name": "video_analytics_berlin_01"}, + } + ) + + +async def test_handle_publishes_failed_before_start_for_stage2_schema_failure() -> None: + session_maker = FakeSessionMaker() + use_case_factory = MagicMock() + publisher = AsyncMock() + coordinator = DeployServiceCommandCoordinator( + session_maker=session_maker, # type: ignore[arg-type] + use_case_factory=use_case_factory, + publisher=publisher, + ) + + await coordinator.handle(_valid_envelope(targets=[])) + + assert session_maker.contexts == [] + use_case_factory.assert_not_called() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["state"] == "failed_before_start" + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["status"] == "failed" + + +async def test_handle_commits_accept_before_publishing_accepted_then_completes() -> None: + session_maker = FakeSessionMaker() + accepted_order_id = UUID("aaaaaaaa-0000-4000-8000-000000000001") + completed_event = SrmOperationCompletedV1( + schema_version="1.0", + operation_id=UUID("11111111-2222-3333-4444-555555555555"), + status=OperationCompletedStatus.COMPLETED, + service_order_id=accepted_order_id, + instances=[ + { + "service_instance_id": "9a3f1c22-0000-4000-8000-000000000002", + "zone_id": "642f6105-7015-4af1-a4d1-e1ecb8437abc", + "status": "completed", + } + ], + correlation_id="corr-appinst-001", + completed_at=datetime.fromisoformat("2026-06-19T12:01:00+00:00"), + ) + accept_use_case = AsyncMock() + accept_use_case.accept.return_value = DeployServiceResult(accepted_order_id=accepted_order_id) + complete_use_case = AsyncMock() + complete_use_case.complete.return_value = completed_event + use_case_factory = MagicMock(side_effect=[accept_use_case, complete_use_case]) + publisher = AsyncMock() + coordinator = DeployServiceCommandCoordinator( + session_maker=session_maker, # type: ignore[arg-type] + use_case_factory=use_case_factory, + publisher=publisher, + ) + + await coordinator.handle(_valid_envelope()) + + assert len(session_maker.contexts) == 2 + session_maker.contexts[0].session.commit.assert_awaited_once() + session_maker.contexts[0].session.rollback.assert_not_awaited() + session_maker.contexts[1].session.commit.assert_awaited_once() + session_maker.contexts[1].session.rollback.assert_not_awaited() + accept_use_case.accept.assert_awaited_once() + complete_use_case.complete.assert_awaited_once() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["state"] == "accepted" + assert payload["service_order_id"] == str(accepted_order_id) + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["status"] == "completed" + assert payload["service_order_id"] == str(accepted_order_id) + + +async def test_handle_publishes_failed_completed_when_completion_raises_after_accept() -> None: + session_maker = FakeSessionMaker() + accepted_order_id = UUID("aaaaaaaa-0000-4000-8000-000000000001") + accepted = DeployServiceResult( + accepted_order_id=accepted_order_id, + target_placements=[], + ) + failed_event = SrmOperationCompletedV1( + schema_version="1.0", + operation_id=UUID("11111111-2222-3333-4444-555555555555"), + status=OperationCompletedStatus.FAILED, + service_order_id=accepted_order_id, + error={ + "type": "about:blank", + "title": "Deploy service failed after acceptance.", + "status": 500, + "detail": "adapter failed", + }, + correlation_id="corr-appinst-001", + completed_at=datetime.fromisoformat("2026-06-19T12:01:00+00:00"), + ) + accept_use_case = AsyncMock() + accept_use_case.accept.return_value = accepted + complete_use_case = AsyncMock() + complete_use_case.complete.side_effect = RuntimeError("adapter failed") + fail_use_case = AsyncMock() + fail_use_case.fail_after_accept.return_value = failed_event + use_case_factory = MagicMock(side_effect=[accept_use_case, complete_use_case, fail_use_case]) + publisher = AsyncMock() + coordinator = DeployServiceCommandCoordinator( + session_maker=session_maker, # type: ignore[arg-type] + use_case_factory=use_case_factory, + publisher=publisher, + ) + + await coordinator.handle(_valid_envelope()) + + assert len(session_maker.contexts) == 3 + session_maker.contexts[0].session.commit.assert_awaited_once() + session_maker.contexts[1].session.rollback.assert_awaited_once() + session_maker.contexts[2].session.commit.assert_awaited_once() + complete_use_case.complete.assert_awaited_once() + fail_use_case.fail_after_accept.assert_awaited_once() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["state"] == "accepted" + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["status"] == "failed" + assert payload["service_order_id"] == str(accepted_order_id) + + +async def test_handle_republishes_accepted_for_non_terminal_duplicate() -> None: + session_maker = FakeSessionMaker() + duplicate_order_id = UUID("aaaaaaaa-0000-4000-8000-000000000001") + accept_use_case = AsyncMock() + accept_use_case.accept.return_value = DeployServiceResult(duplicate_order_id=duplicate_order_id) + use_case_factory = MagicMock(return_value=accept_use_case) + publisher = AsyncMock() + coordinator = DeployServiceCommandCoordinator( + session_maker=session_maker, # type: ignore[arg-type] + use_case_factory=use_case_factory, + publisher=publisher, + ) + + await coordinator.handle(_valid_envelope()) + + assert len(session_maker.contexts) == 1 + accept_use_case.accept.assert_awaited_once() + publisher.publish.assert_awaited_once() + subject, payload = publisher.publish.await_args.args + assert subject == "event.srm.operation.status" + assert payload["state"] == "accepted" + assert payload["service_order_id"] == str(duplicate_order_id) + + +async def test_handle_republishes_terminal_event_for_terminal_duplicate() -> None: + session_maker = FakeSessionMaker() + duplicate_order_id = UUID("aaaaaaaa-0000-4000-8000-000000000001") + completed_event = SrmOperationCompletedV1( + schema_version="1.0", + operation_id=UUID("11111111-2222-3333-4444-555555555555"), + status=OperationCompletedStatus.COMPLETED, + service_order_id=duplicate_order_id, + instances=[ + { + "service_instance_id": "9a3f1c22-0000-4000-8000-000000000002", + "zone_id": "642f6105-7015-4af1-a4d1-e1ecb8437abc", + "status": "completed", + } + ], + correlation_id="corr-appinst-001", + completed_at=datetime.fromisoformat("2026-06-19T12:01:00+00:00"), + ) + accept_use_case = AsyncMock() + accept_use_case.accept.return_value = DeployServiceResult( + duplicate_order_id=duplicate_order_id, + completed_event=completed_event, + ) + use_case_factory = MagicMock(return_value=accept_use_case) + publisher = AsyncMock() + coordinator = DeployServiceCommandCoordinator( + session_maker=session_maker, # type: ignore[arg-type] + use_case_factory=use_case_factory, + publisher=publisher, + ) + + await coordinator.handle(_valid_envelope()) + + assert len(session_maker.contexts) == 1 + accept_use_case.accept.assert_awaited_once() + publisher.publish.assert_awaited_once() + subject, payload = publisher.publish.await_args.args + assert subject == "event.srm.operation.completed" + assert payload["status"] == "completed" + assert payload["service_order_id"] == str(duplicate_order_id) diff --git a/tests/conftest.py b/tests/conftest.py index dbc6a8676ff8e683e90c69c249b819c59a2779f7..24645ee742462a0f72aac98c0875181b8eceffe6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -52,6 +52,22 @@ def _set_test_settings_env( monkeypatch.setenv("NATS_SETTINGS__URL", nats_url) +async def _truncate_db(postgres_container: PostgresContainer) -> None: + engine = create_async_engine( + _as_asyncpg_url(postgres_container.get_connection_url()), + echo=False, + ) + table_names = list(get_metadata().tables) + + async with engine.begin() as conn: + await conn.run_sync(get_metadata().create_all) + if table_names: + joined = ", ".join(table_names) + await conn.execute(text(f"TRUNCATE TABLE {joined} RESTART IDENTITY CASCADE")) + + await engine.dispose() + + @pytest.fixture(scope="session") def postgres_container() -> Iterator[PostgresContainer]: try: @@ -87,22 +103,12 @@ def app_with_db( @pytest_asyncio.fixture async def clean_db(postgres_container: PostgresContainer) -> AsyncIterator[None]: - engine = create_async_engine( - _as_asyncpg_url(postgres_container.get_connection_url()), - echo=False, - ) - table_names = list(get_metadata().tables) - - async with engine.begin() as conn: - await conn.run_sync(get_metadata().create_all) - if table_names: - joined = ", ".join(table_names) - await conn.execute(text(f"TRUNCATE TABLE {joined} RESTART IDENTITY CASCADE")) + await _truncate_db(postgres_container) try: yield finally: - await engine.dispose() + await _truncate_db(postgres_container) @pytest.fixture diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 87e37d983252c593e40e9004a1e8eded99bf8807..60fc132f6c8947cd5b2c1892f63dac535fa9959e 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -37,7 +37,7 @@ async def db_engine(postgres_container: object) -> AsyncIterator[AsyncEngine]: @pytest_asyncio.fixture -async def db_session(db_engine: AsyncEngine) -> AsyncIterator[AsyncSession]: +async def db_session(clean_db: None, db_engine: AsyncEngine) -> AsyncIterator[AsyncSession]: async with db_engine.connect() as connection: transaction = await connection.begin() session_factory = async_sessionmaker(bind=connection, expire_on_commit=False) diff --git a/tests/integration/test_databus.py b/tests/integration/test_databus.py index 82569c00d51f5f41e64513649a1aaee89b92d628..7d49ce6cea4e892ccececcdb4d11ecd60abfe905 100644 --- a/tests/integration/test_databus.py +++ b/tests/integration/test_databus.py @@ -1,6 +1,7 @@ import asyncio import json from collections.abc import AsyncIterator, Generator +from unittest.mock import AsyncMock import nats import pytest @@ -11,7 +12,7 @@ from testcontainers.nats import NatsContainer from srm.adapters.databus.nats_connection_manager import NatsConnectionManager from srm.adapters.databus.nats_publisher import NatsPublisher from srm.api.databus.nats_subscriber import NatsSubscriber, subscribe_to_subjects -from srm.api.databus.schemas import InboundMessage +from srm.api.databus.schemas import AnswerableCommandV1 from srm.config import NatsSettings from tests.api.databus.test_nats_subscriber import EXPECTED_COMMAND_SUBJECTS @@ -79,40 +80,59 @@ async def test_publisher_publishes_json_message_to_subject( assert received == [{"operation_id": "abc-123"}] -async def test_subscriber_invokes_router_when_message_arrives( +async def test_subscriber_invokes_callback_when_message_arrives( connection_manager: NatsConnectionManager, raw_client: Client, ) -> None: - received: list[InboundMessage] = [] + received: list[AnswerableCommandV1] = [] ready = asyncio.Event() - async def router(message: InboundMessage) -> None: - received.append(message) + async def callback(envelope: AnswerableCommandV1) -> None: + received.append(envelope) ready.set() subscriber = NatsSubscriber( connection_manager=connection_manager, subject="command.srm.service.deploy", - router=router, + callback=callback, ) await subscriber.start() await connection_manager.client.flush() await raw_client.publish( "command.srm.service.deploy", - json.dumps({"operation_id": "abc-123"}).encode("utf-8"), + json.dumps( + { + "schema_version": "1.0", + "operation_id": "11111111-2222-3333-4444-555555555555", + "correlation_id": "corr-1", + "requested_at": "2026-06-19T12:00:00Z", + "app_provider_id": "VideoAppsCo", + "source": "nbi_camara", + } + ).encode("utf-8"), ) await asyncio.wait_for(ready.wait(), timeout=2.0) assert len(received) == 1 - assert received[0].subject == "command.srm.service.deploy" - assert json.loads(received[0].payload) == {"operation_id": "abc-123"} + assert received[0].correlation_id == "corr-1" async def test_subscribe_to_subjects_registers_all_command_subjects( connection_manager: NatsConnectionManager, ) -> None: - subscribers = await subscribe_to_subjects(connection_manager) + class FakeSessionContext: + async def __aenter__(self) -> AsyncMock: + return AsyncMock() + + async def __aexit__(self, *_args: object) -> None: + return None + + class FakeSessionMaker: + def __call__(self) -> FakeSessionContext: + return FakeSessionContext() + + subscribers = await subscribe_to_subjects(connection_manager, FakeSessionMaker()) # type: ignore[arg-type] assert sorted(sub._subject for sub in subscribers) == sorted(EXPECTED_COMMAND_SUBJECTS) for sub in subscribers: diff --git a/tests/integration/test_databus_deploy_e2e.py b/tests/integration/test_databus_deploy_e2e.py new file mode 100644 index 0000000000000000000000000000000000000000..c35445f030795ae4fc5057c9eb36906d0230be2d --- /dev/null +++ b/tests/integration/test_databus_deploy_e2e.py @@ -0,0 +1,367 @@ +from __future__ import annotations + +import asyncio +import json +from collections.abc import AsyncIterator +from datetime import datetime, timezone +from typing import Any, cast +from uuid import UUID + +import nats +import pytest_asyncio +from asgi_lifespan import LifespanManager +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from nats.aio.client import Client +from nats.aio.msg import Msg +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from testcontainers.nats import NatsContainer +from testcontainers.postgres import PostgresContainer + +from srm.adapters.database.repos.runtime_inventory import ( + SqlCapabilityInstanceRepository, + SqlServiceInstanceRepository, + SqlServiceOrderRepository, +) +from srm.adapters.database.repos.topology import SqlZoneRepository +from srm.domain.models.canonical_parameters.compute import ( + ComputeRequirements, + ComputeResources, + TopologyConstraints, +) +from srm.domain.models.catalog import RuntimeKind +from srm.domain.models.runtime_inventory import ( + CapabilityInstance, + ServiceInstance, + ServiceOrder, +) +from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState +from srm.domain.models.topology import Capability, ControlPathBinding, Domain, Zone +from srm.domain.models.topology.enums import ( + CapabilityKind, + CapabilityState, + ControlPathBindingState, + DomainKind, + DomainState, + ZoneKind, + ZoneState, +) +from tests.conftest import _as_asyncpg_url + + +def _now() -> datetime: + return datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc) + + +def _compute_requirements() -> ComputeRequirements: + return ComputeRequirements( + compute=ComputeResources(cpu_millicores=500, memory_mb=1024), + topology=TopologyConstraints(min_nodes=1), + ) + + +def _service_specification_payload( + specification_id: UUID, +) -> dict[str, object]: + return { + "service_specification": { + "id": str(specification_id), + "app_provider_id": "VideoAppsCo", + "ref": "video-analytics", + "name": "Video Analytics", + "version": "1.0.0", + }, + "service_deployment_units": [ + { + "ref": "main", + "name": "Main", + "runtime_kind": RuntimeKind.HELM, + "artifact_ref": "oci://example/video-analytics", + "resource_requirements": _compute_requirements().model_dump(mode="json"), + } + ], + "service_capability_requirements": [ + { + "ref": "deploy-workload", + "deployment_unit_ref": "main", + "capability_kind": CapabilityKind.DEPLOY_WORKLOAD, + "domain_kind": DomainKind.COMPUTE, + "is_required": True, + } + ], + } + + +def _zone(zone_id: UUID, domain_id: UUID, capability_id: UUID) -> Zone: + now = _now() + return Zone( + id=zone_id, + platform_ref="edge-platform", + ref=f"zone-{zone_id}", + name="Athens Zone", + kind=ZoneKind.RESOURCE, + state=ZoneState.ACTIVE, + domains=[ + Domain( + id=domain_id, + zone_id=zone_id, + ref="compute", + name="Compute", + kind=DomainKind.COMPUTE, + state=DomainState.ACTIVE, + capabilities=[ + Capability( + id=capability_id, + domain_id=domain_id, + ref="deploy-workload", + name="Deploy Workload", + kind=CapabilityKind.DEPLOY_WORKLOAD, + state=CapabilityState.ACTIVE, + control_path_bindings=[ + ControlPathBinding( + capability_id=capability_id, + ref="primary", + control_path_ref=f"k8s.{zone_id}.compute", + supported_runtime_kinds=[RuntimeKind.HELM], + supported_actions=["deploy"], + priority=100, + is_default=True, + state=ControlPathBindingState.ACTIVE, + created_at=now, + updated_at=now, + ) + ], + created_at=now, + updated_at=now, + ) + ], + created_at=now, + updated_at=now, + ) + ], + created_at=now, + updated_at=now, + ) + + +async def _create_service_specification( + app: FastAPI, + *, + specification_id: UUID, +) -> None: + async with AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + ) as client: + response = await client.post( + "/internal/catalog/service-specifications", + json=_service_specification_payload(specification_id), + ) + + assert response.status_code == 201 + + +async def _seed_topology( + postgres_container: PostgresContainer, + *, + zone_id: UUID, + domain_id: UUID, + capability_id: UUID, +) -> None: + engine = create_async_engine(_as_asyncpg_url(postgres_container.get_connection_url())) + session_maker = async_sessionmaker(engine, expire_on_commit=False) + async with session_maker() as session: + await SqlZoneRepository(session).create(_zone(zone_id, domain_id, capability_id)) + await session.commit() + await engine.dispose() + + +async def _load_deploy_records( + postgres_container: PostgresContainer, + *, + operation_id: UUID, + app_instance_id: UUID, +) -> tuple[ServiceOrder, ServiceInstance, CapabilityInstance]: + engine = create_async_engine(_as_asyncpg_url(postgres_container.get_connection_url())) + session_maker = async_sessionmaker(engine, expire_on_commit=False) + async with session_maker() as session: + order = await SqlServiceOrderRepository(session).get_by_operation_id(operation_id) + instance = await SqlServiceInstanceRepository(session).get_by_id(app_instance_id) + assert instance is not None + assert instance.capabilities + capability = await SqlCapabilityInstanceRepository(session).get_by_id( + instance.capabilities[0].id + ) + await engine.dispose() + + assert order is not None + assert capability is not None + return order, instance, capability + + +async def _collect_deploy_events(client: Client) -> tuple[list[dict[str, object]], asyncio.Event]: + events: list[dict[str, object]] = [] + completed = asyncio.Event() + + async def handler(msg: Msg) -> None: + events.append( + { + "subject": msg.subject, + "payload": json.loads(msg.data), + } + ) + if msg.subject == "event.srm.operation.completed": + completed.set() + + await client.subscribe("event.srm.operation.status", cb=handler) + await client.subscribe("event.srm.operation.completed", cb=handler) + await client.flush() + return events, completed + + +@pytest_asyncio.fixture +async def raw_nats_client(nats_container: NatsContainer) -> AsyncIterator[Client]: + client = await nats.connect(nats_container.nats_uri()) + try: + yield client + finally: + await client.drain() + + +async def test_databus_deploy_federation_without_partner_ref_returns_failed_before_start( + app_with_db: FastAPI, + raw_nats_client: Client, + clean_db: None, +) -> None: + operation_id = UUID("11111111-2222-3333-4444-555555555555") + events, completed = await _collect_deploy_events(raw_nats_client) + + async with LifespanManager(app_with_db): + await app_with_db.state.databus_connection_manager.client.flush() + await raw_nats_client.flush() + await raw_nats_client.publish( + "command.srm.service.deploy", + json.dumps( + { + "schema_version": "1.0", + "operation_id": str(operation_id), + "correlation_id": "corr-federation-missing-partner", + "requested_at": "2026-06-19T12:00:00Z", + "app_provider_id": "VideoAppsCo", + "source": "federation", + "federation_partner_ref": None, + "service_specification_id": "d4e5f6a7-0000-4000-8000-000000000001", + "targets": [ + { + "app_instance_id": "9a3f1c22-0000-4000-8000-000000000002", + "zone_id": "642f6105-7015-4af1-a4d1-e1ecb8437abc", + } + ], + "deploy": {"instance_name": "video_analytics_athens_01"}, + } + ).encode("utf-8"), + ) + await asyncio.wait_for(completed.wait(), timeout=5.0) + + assert [event["subject"] for event in events] == [ + "event.srm.operation.status", + "event.srm.operation.completed", + ] + status_payload = cast(dict[str, Any], events[0]["payload"]) + completed_payload = cast(dict[str, Any], events[1]["payload"]) + assert status_payload["operation_id"] == str(operation_id) + assert status_payload["correlation_id"] == "corr-federation-missing-partner" + assert status_payload["state"] == "failed_before_start" + assert status_payload["service_order_id"] is None + assert completed_payload["operation_id"] == str(operation_id) + assert completed_payload["correlation_id"] == "corr-federation-missing-partner" + assert completed_payload["status"] == "failed" + assert completed_payload["service_order_id"] is None + assert completed_payload["error"] == { + "type": "about:blank", + "title": "Deploy service command is invalid.", + "status": 400, + } + + +async def test_databus_deploy_federation_with_partner_ref_completes_end_to_end( + app_with_db: FastAPI, + postgres_container: PostgresContainer, + raw_nats_client: Client, + clean_db: None, +) -> None: + operation_id = UUID("22222222-2222-3333-4444-555555555555") + app_instance_id = UUID("9a3f1c22-0000-4000-8000-000000000102") + specification_id = UUID("d4e5f6a7-0000-4000-8000-000000000101") + zone_id = UUID("642f6105-7015-4af1-a4d1-e1ecb8437abd") + domain_id = UUID("bbbbbbbb-0000-4000-8000-000000000111") + capability_id = UUID("cccccccc-0000-4000-8000-000000000112") + await _seed_topology( + postgres_container, + zone_id=zone_id, + domain_id=domain_id, + capability_id=capability_id, + ) + events, completed = await _collect_deploy_events(raw_nats_client) + + async with LifespanManager(app_with_db): + await _create_service_specification( + app_with_db, + specification_id=specification_id, + ) + await app_with_db.state.databus_connection_manager.client.flush() + await raw_nats_client.flush() + await raw_nats_client.publish( + "command.srm.service.deploy", + json.dumps( + { + "schema_version": "1.0", + "operation_id": str(operation_id), + "correlation_id": "corr-federation-valid", + "requested_at": "2026-06-19T12:00:00Z", + "app_provider_id": "VideoAppsCo", + "source": "federation", + "federation_partner_ref": "ptr-OperatorB", + "service_specification_id": str(specification_id), + "targets": [ + { + "app_instance_id": str(app_instance_id), + "zone_id": str(zone_id), + } + ], + "deploy": {"instance_name": "video_analytics_federated_01"}, + } + ).encode("utf-8"), + ) + await asyncio.wait_for(completed.wait(), timeout=5.0) + + assert [event["subject"] for event in events] == [ + "event.srm.operation.status", + "event.srm.operation.completed", + ] + status_payload = cast(dict[str, Any], events[0]["payload"]) + completed_payload = cast(dict[str, Any], events[1]["payload"]) + assert status_payload["operation_id"] == str(operation_id) + assert status_payload["correlation_id"] == "corr-federation-valid" + assert status_payload["state"] == "accepted" + assert completed_payload["operation_id"] == str(operation_id) + assert completed_payload["correlation_id"] == "corr-federation-valid" + assert completed_payload["status"] == "completed" + assert completed_payload["instances"] == [ + { + "service_instance_id": str(app_instance_id), + "zone_id": str(zone_id), + "status": "completed", + } + ] + + order, instance, capability = await _load_deploy_records( + postgres_container, + operation_id=operation_id, + app_instance_id=app_instance_id, + ) + assert order.state == ServiceOrderState.COMPLETED + assert order.federation_partner_ref == "ptr-OperatorB" + assert instance.state == ServiceInstanceState.ACTIVE + assert instance.federation_partner_ref == "ptr-OperatorB" + assert capability.federation_partner_ref == "ptr-OperatorB" diff --git a/tests/integration/test_repositories.py b/tests/integration/test_repositories.py index b9b68cef5508f36206ea29b781cc4251410c4ae9..40658e64e1f442a17bc19a82958ae944989c4a95 100644 --- a/tests/integration/test_repositories.py +++ b/tests/integration/test_repositories.py @@ -98,25 +98,31 @@ def _compute_requirements() -> ComputeRequirements: ) -def _zone() -> Zone: +def _zone( + *, + ref: str = "zone-1", + kind: ZoneKind = ZoneKind.RESOURCE, + state: ZoneState = ZoneState.ACTIVE, + domains: list[Domain] | None = None, +) -> Zone: return Zone( id=uuid4(), platform_ref="edge-platform", - ref="zone-1", + ref=ref, name="Zone", - kind=ZoneKind.RESOURCE, - state=ZoneState.ACTIVE, - domains=[], + kind=kind, + state=state, + domains=domains or [], created_at=_now(), updated_at=_now(), ) -def _domain(zone_id: UUID) -> Domain: +def _domain(zone_id: UUID, *, ref: str = "compute") -> Domain: return Domain( id=uuid4(), zone_id=zone_id, - ref="compute", + ref=ref, name="Domain", kind=DomainKind.COMPUTE, state=DomainState.ACTIVE, @@ -126,11 +132,11 @@ def _domain(zone_id: UUID) -> Domain: ) -def _capability(domain_id: UUID) -> Capability: +def _capability(domain_id: UUID, *, ref: str = "deploy") -> Capability: return Capability( id=uuid4(), domain_id=domain_id, - ref="deploy", + ref=ref, name="Capability", kind=CapabilityKind.DEPLOY_WORKLOAD, state=CapabilityState.ACTIVE, @@ -196,22 +202,31 @@ def _service_specification_with_children() -> ServiceSpecification: return spec -def _service_instance(spec_id: UUID, orig_so_id: UUID, zone_id: UUID) -> ServiceInstance: +def _service_instance( + spec_id: UUID, + orig_so_id: UUID, + zone_id: UUID, + *, + ref: str = "prod", + app_provider_id: str | None = None, + federation_partner_ref: str | None = None, +) -> ServiceInstance: return ServiceInstance( id=uuid4(), service_specification_id=spec_id, originating_service_order_id=orig_so_id, zone_id=zone_id, - ref="prod", + ref=ref, state=ServiceInstanceState.CREATING, - app_provider_id=str(uuid4()), + app_provider_id=app_provider_id or str(uuid4()), + federation_partner_ref=federation_partner_ref, capabilities=[], created_at=_now(), updated_at=_now(), ) -def _service_order(zone_id: UUID) -> ServiceOrder: +def _service_order(zone_id: UUID, *, federation_partner_ref: str | None = None) -> ServiceOrder: return ServiceOrder( id=uuid4(), operation_id=uuid4(), @@ -219,6 +234,7 @@ def _service_order(zone_id: UUID) -> ServiceOrder: state=ServiceOrderState.ACCEPTED, payload_snapshot={"target_pins": {"zone_id": str(zone_id)}}, app_provider_id=str(uuid4()), + federation_partner_ref=federation_partner_ref, created_at=_now(), updated_at=_now(), ) @@ -231,19 +247,25 @@ def _service_order_with_type(zone_id: UUID, order_type: ServiceOrderType) -> Ser def _capability_instance( - service_instance_id: UUID, capability_id: UUID, originating_service_order_id: UUID + service_instance_id: UUID, + capability_id: UUID, + originating_service_order_id: UUID, + *, + ref: str = "deploy", + federation_partner_ref: str | None = None, ) -> CapabilityInstance: return CapabilityInstance( id=uuid4(), capability_id=capability_id, service_instance_id=service_instance_id, originating_service_order_id=originating_service_order_id, - ref="deploy", + ref=ref, kind=CapabilityInstanceKind.ACTIVATION, state=CapabilityInstanceState.CREATING, parameters_snapshot=_parameters(), result_summary=Result(status=ResultStatus.ACTIVE), app_provider_id=str(uuid4()), + federation_partner_ref=federation_partner_ref, created_at=_now(), updated_at=_now(), ) @@ -428,6 +450,282 @@ async def test_service_order_create_accepts_deactivate_capability( assert reloaded.order_type == ServiceOrderType.DEACTIVATE_CAPABILITY +async def test_service_order_get_by_operation_id(db_session: AsyncSession) -> None: + zone = await SqlZoneRepository(db_session).create(_zone()) + repo = SqlServiceOrderRepository(db_session) + saved = await repo.create(_service_order(zone.id)) + + reloaded = await repo.get_by_operation_id(saved.operation_id) + + assert reloaded == saved + + +async def test_service_deployment_unit_list_by_service_specification_id( + db_session: AsyncSession, +) -> None: + spec = await SqlServiceSpecificationRepository(db_session).create(_service_specification()) + other_spec = await SqlServiceSpecificationRepository(db_session).create( + _service_specification() + ) + repo = SqlServiceDeploymentUnitRepository(db_session) + main = await repo.create( + ServiceDeploymentUnit( + id=uuid4(), + service_specification_id=spec.id, + ref="main", + name="Main", + runtime_kind=RuntimeKind.HELM, + artifact_ref="oci://example/main", + resource_requirements=_compute_requirements(), + created_at=_now(), + updated_at=_now(), + ) + ) + worker = await repo.create( + ServiceDeploymentUnit( + id=uuid4(), + service_specification_id=spec.id, + ref="worker", + name="Worker", + runtime_kind=RuntimeKind.HELM, + artifact_ref="oci://example/worker", + resource_requirements=_compute_requirements(), + created_at=_now(), + updated_at=_now(), + ) + ) + await repo.create( + ServiceDeploymentUnit( + id=uuid4(), + service_specification_id=other_spec.id, + ref="other", + name="Other", + runtime_kind=RuntimeKind.HELM, + artifact_ref="oci://example/other", + resource_requirements=_compute_requirements(), + created_at=_now(), + updated_at=_now(), + ) + ) + + reloaded = await repo.list_by_service_specification_id(spec.id) + + assert {unit.id for unit in reloaded} == {main.id, worker.id} + + +async def test_service_capability_requirement_list_by_service_specification_id( + db_session: AsyncSession, +) -> None: + spec = await SqlServiceSpecificationRepository(db_session).create(_service_specification()) + other_spec = await SqlServiceSpecificationRepository(db_session).create( + _service_specification() + ) + deployment_unit = await SqlServiceDeploymentUnitRepository(db_session).create( + ServiceDeploymentUnit( + id=uuid4(), + service_specification_id=spec.id, + ref="main", + name="Main", + runtime_kind=RuntimeKind.HELM, + artifact_ref="oci://example/main", + resource_requirements=_compute_requirements(), + created_at=_now(), + updated_at=_now(), + ) + ) + repo = SqlServiceCapabilityRequirementRepository(db_session) + deploy = await repo.create( + ServiceCapabilityRequirement( + id=uuid4(), + service_specification_id=spec.id, + deployment_unit_id=deployment_unit.id, + ref="deploy-workload", + capability_kind=CapabilityKind.DEPLOY_WORKLOAD, + domain_kind=DomainKind.COMPUTE, + parameters=_parameters(), + created_at=_now(), + updated_at=_now(), + ) + ) + scale = await repo.create( + ServiceCapabilityRequirement( + id=uuid4(), + service_specification_id=spec.id, + deployment_unit_id=deployment_unit.id, + ref="scale-workload", + capability_kind=CapabilityKind.SCALE_WORKLOAD, + domain_kind=DomainKind.COMPUTE, + parameters=_parameters(), + created_at=_now(), + updated_at=_now(), + ) + ) + await repo.create( + ServiceCapabilityRequirement( + id=uuid4(), + service_specification_id=other_spec.id, + ref="other", + capability_kind=CapabilityKind.DEPLOY_WORKLOAD, + domain_kind=DomainKind.COMPUTE, + parameters=_parameters(), + created_at=_now(), + updated_at=_now(), + ) + ) + + reloaded = await repo.list_by_service_specification_id(spec.id) + + assert {requirement.id for requirement in reloaded} == {deploy.id, scale.id} + assert {requirement.deployment_unit_id for requirement in reloaded} == {deployment_unit.id} + + +async def test_service_order_update_state_persists_terminal_fields( + db_session: AsyncSession, +) -> None: + zone = await SqlZoneRepository(db_session).create(_zone()) + repo = SqlServiceOrderRepository(db_session) + saved = await repo.create(_service_order(zone.id)) + completed_at = _now() + failure_detail: dict[str, object] = {"reason": "placement_failed"} + + updated = await repo.update_state( + saved.id, + ServiceOrderState.FAILED, + completed_at=completed_at, + failure_detail=failure_detail, + ) + reloaded = await repo.get_by_id(saved.id) + + assert updated.state == ServiceOrderState.FAILED + assert updated.completed_at == completed_at + assert updated.failure_detail == failure_detail + assert reloaded == updated + + +async def test_service_order_create_round_trips_federation_partner_ref( + db_session: AsyncSession, +) -> None: + zone = await SqlZoneRepository(db_session).create(_zone()) + repo = SqlServiceOrderRepository(db_session) + + saved = await repo.create(_service_order(zone.id, federation_partner_ref="ptr-OperatorB")) + reloaded = await repo.get_by_id(saved.id) + + assert reloaded is not None + assert reloaded.federation_partner_ref == "ptr-OperatorB" + + +async def test_service_instance_list_by_originating_service_order_id( + db_session: AsyncSession, +) -> None: + spec = await SqlServiceSpecificationRepository(db_session).create(_service_specification()) + zone = await SqlZoneRepository(db_session).create(_zone()) + order = await SqlServiceOrderRepository(db_session).create(_service_order(zone.id)) + other_order = await SqlServiceOrderRepository(db_session).create(_service_order(zone.id)) + repo = SqlServiceInstanceRepository(db_session) + app_provider_id = str(uuid4()) + prod_a = await repo.create( + _service_instance( + spec.id, + order.id, + zone.id, + ref="prod-a", + app_provider_id=app_provider_id, + ) + ) + prod_b = await repo.create( + _service_instance( + spec.id, + order.id, + zone.id, + ref="prod-b", + app_provider_id=app_provider_id, + ) + ) + await repo.create( + _service_instance( + spec.id, + other_order.id, + zone.id, + ref="prod-c", + app_provider_id=app_provider_id, + ) + ) + + reloaded = await repo.list_by_originating_service_order_id(order.id) + + assert {instance.id for instance in reloaded} == {prod_a.id, prod_b.id} + + +async def test_service_instance_update_state_persists( + db_session: AsyncSession, +) -> None: + spec = await SqlServiceSpecificationRepository(db_session).create(_service_specification()) + zone = await SqlZoneRepository(db_session).create(_zone()) + order = await SqlServiceOrderRepository(db_session).create(_service_order(zone.id)) + repo = SqlServiceInstanceRepository(db_session) + saved = await repo.create(_service_instance(spec.id, order.id, zone.id)) + + updated = await repo.update_state(saved.id, ServiceInstanceState.ACTIVE) + reloaded = await repo.get_by_id(saved.id) + + assert updated.state == ServiceInstanceState.ACTIVE + assert reloaded == updated + + +async def test_capability_instance_create_round_trips_federation_partner_ref_and_links( + db_session: AsyncSession, +) -> None: + spec = await SqlServiceSpecificationRepository(db_session).create(_service_specification()) + zone = await SqlZoneRepository(db_session).create(_zone()) + order = await SqlServiceOrderRepository(db_session).create(_service_order(zone.id)) + instance = await SqlServiceInstanceRepository(db_session).create( + _service_instance(spec.id, order.id, zone.id) + ) + domain = await SqlDomainRepository(db_session).create(_domain(zone.id)) + capability = await SqlCapabilityRepository(db_session).create(_capability(domain.id)) + repo = SqlCapabilityInstanceRepository(db_session) + + saved = await repo.create( + _capability_instance( + instance.id, + capability.id, + order.id, + federation_partner_ref="ptr-OperatorB", + ) + ) + reloaded = await repo.get_by_id(saved.id) + + assert reloaded is not None + assert reloaded.federation_partner_ref == "ptr-OperatorB" + assert reloaded.service_instance_id == instance.id + assert reloaded.capability_id == capability.id + assert reloaded.originating_service_order_id == order.id + + +async def test_zone_list_active_resource_zones_filters_and_loads_nested_graph( + db_session: AsyncSession, +) -> None: + active = _zone(ref="zone-active") + domain = _domain(active.id) + capability = _capability(domain.id) + binding = _control_path_binding(capability.id) + capability.control_path_bindings = [binding] + domain.capabilities = [capability] + active.domains = [domain] + repo = SqlZoneRepository(db_session) + saved_active = await repo.create(active) + await repo.create(_zone(ref="zone-authority", kind=ZoneKind.AUTHORITY)) + await repo.create(_zone(ref="zone-offline", state=ZoneState.OFFLINE)) + + reloaded = await repo.list_active_resource_zones() + + assert [zone.id for zone in reloaded] == [saved_active.id] + assert reloaded[0].domains[0].id == domain.id + assert reloaded[0].domains[0].capabilities[0].id == capability.id + assert reloaded[0].domains[0].capabilities[0].control_path_bindings[0].id == binding.id + + class TestChildRepositoriesGetById: """Deployment units and capability requirements are written via the aggregate root, so only their get_by_id is exercised directly.""" diff --git a/tests/unit/fakes.py b/tests/unit/fakes.py index 3354b9ce170493fa3cf2f6207f34ca01c8a132cf..b0048607b58f18af02e95de4a5c4afb9d9d329ab 100644 --- a/tests/unit/fakes.py +++ b/tests/unit/fakes.py @@ -14,6 +14,7 @@ from srm.adapters.errors import DuplicateEntryError, DuplicateServiceSpecificati from srm.domain.models.canonical_parameters.result import Result, ResultStatus from srm.domain.models.catalog import RuntimeKind, ServiceSpecification from srm.domain.models.runtime_inventory import CapabilityInstanceState, ServiceInstance +from srm.domain.models.runtime_inventory.enums import ServiceInstanceState from srm.domain.models.topology import ( CapabilityKind, CapabilityState, @@ -89,6 +90,13 @@ class InMemoryZoneRepository(ZoneRepository): found = self.rows.get(id) return found.model_copy(deep=True) if found is not None else None + async def list_active_resource_zones(self) -> list[Zone]: + return [ + zone.model_copy(deep=True) + for zone in self.rows.values() + if zone.kind == ZoneKind.RESOURCE and zone.state == ZoneState.ACTIVE + ] + async def create(self, zone: Zone) -> Zone: if zone.id in self.rows: raise DuplicateEntryError() @@ -105,6 +113,16 @@ class InMemoryServiceInstanceRepository(ServiceInstanceRepository): found = self.rows.get(id) return found.model_copy(deep=True) if found is not None else None + async def list_by_originating_service_order_id( + self, + originating_service_order_id: UUID, + ) -> list[ServiceInstance]: + return [ + row.model_copy(deep=True) + for row in self.rows.values() + if row.originating_service_order_id == originating_service_order_id + ] + async def create(self, service_instance: ServiceInstance) -> ServiceInstance: if service_instance.id in self.rows: raise DuplicateServiceInstanceError() @@ -124,6 +142,18 @@ class InMemoryServiceInstanceRepository(ServiceInstanceRepository): self.rows[stored.id] = stored return stored.model_copy(deep=True) + async def update_state( + self, + id: UUID, + state: ServiceInstanceState, + ) -> ServiceInstance: + found = self.rows.get(id) + if found is None: + raise KeyError(id) + updated = found.model_copy(update={"state": state}, deep=True) + self.rows[id] = updated + return updated.model_copy(deep=True) + class FakeTransformationFunctionAdapter(TransformationFunctionExecutionPort): def get_zones(self, kind: ZoneKind | None = None) -> list[TransformationFunctionZone]: diff --git a/tests/unit/test_databus_events.py b/tests/unit/test_databus_events.py new file mode 100644 index 0000000000000000000000000000000000000000..cd4e9b31d02e25f76a0e709f42ec364c2c485002 --- /dev/null +++ b/tests/unit/test_databus_events.py @@ -0,0 +1,61 @@ +from datetime import UTC, datetime +from uuid import uuid4 + +from srm.domain.ports.databus.events import ( + OperationCompletedStatus, + OperationStatusState, + SrmOperationCompletedV1, + SrmOperationStatusV1, +) + + +def test_operation_status_event_dumps_contract_json() -> None: + operation_id = uuid4() + service_order_id = uuid4() + + event = SrmOperationStatusV1( + schema_version="1.0", + operation_id=operation_id, + service_order_id=service_order_id, + state=OperationStatusState.FAILED_BEFORE_START, + capability="deploy", + correlation_id="corr-1", + emitted_at=datetime(2026, 6, 19, 12, 0, tzinfo=UTC), + ) + + assert event.model_dump(mode="json") == { + "schema_version": "1.0", + "operation_id": str(operation_id), + "service_order_id": str(service_order_id), + "service_instance_id": None, + "capability": "deploy", + "state": "failed_before_start", + "metadata": {}, + "correlation_id": "corr-1", + "emitted_at": "2026-06-19T12:00:00Z", + } + + +def test_operation_completed_event_dumps_failed_contract_json() -> None: + operation_id = uuid4() + + event = SrmOperationCompletedV1( + schema_version="1.0", + operation_id=operation_id, + status=OperationCompletedStatus.FAILED, + error={"type": "about:blank", "title": "Failed.", "status": 400}, + correlation_id="corr-1", + completed_at=datetime(2026, 6, 19, 12, 0, tzinfo=UTC), + ) + + assert event.model_dump(mode="json") == { + "schema_version": "1.0", + "operation_id": str(operation_id), + "status": "failed", + "service_order_id": None, + "instances": [], + "metadata": {}, + "error": {"type": "about:blank", "title": "Failed.", "status": 400}, + "correlation_id": "corr-1", + "completed_at": "2026-06-19T12:00:00Z", + } diff --git a/tests/unit/test_deploy_service_use_case.py b/tests/unit/test_deploy_service_use_case.py new file mode 100644 index 0000000000000000000000000000000000000000..7d0068a1e7d68f2c482768428c537345c941b07d --- /dev/null +++ b/tests/unit/test_deploy_service_use_case.py @@ -0,0 +1,1429 @@ +from datetime import datetime +from typing import Literal +from unittest.mock import ANY, AsyncMock +from uuid import UUID, uuid4 + +from srm.application.use_cases.deploy_service import ( + DeployPayloadCommand, + DeployServiceCommand, + DeployServiceUseCase, + DeployTargetCommand, +) +from srm.domain.models.canonical_parameters.compute import ( + ComputeRequirements, + ComputeResources, + TopologyConstraints, +) +from srm.domain.models.canonical_parameters.parameters import ( + CapabilityParameters, + CapabilityTarget, + Parameters, + SourceSpecification, + SourceSpecificationFamily, +) +from srm.domain.models.catalog import ( + RuntimeKind, + ServiceCapabilityRequirement, + ServiceDeploymentUnit, + ServiceSpecification, + ServiceSpecificationState, +) +from srm.domain.models.runtime_inventory import ServiceInstance, ServiceOrder +from srm.domain.models.runtime_inventory.enums import ( + ServiceInstanceState, + ServiceOrderState, + ServiceOrderType, +) +from srm.domain.models.topology import ( + Capability, + CapabilityKind, + CapabilityState, + ControlPathBinding, + ControlPathBindingState, + Domain, + DomainKind, + DomainState, + Zone, + ZoneKind, + ZoneState, +) + + +def _command( + operation_id: UUID, + *, + targets: list[DeployTargetCommand] | None = None, + source: Literal["nbi_camara", "nbi_tmf", "operator_portal", "federation"] = "nbi_camara", + federation_partner_ref: str | None = None, +) -> DeployServiceCommand: + return DeployServiceCommand( + operation_id=operation_id, + correlation_id="corr-1", + requested_at=datetime.fromisoformat("2026-06-19T12:00:00+00:00"), + app_provider_id="VideoAppsCo", + source=source, + service_specification_id=uuid4(), + targets=targets or [DeployTargetCommand(app_instance_id=uuid4(), zone_id=uuid4())], + deploy=DeployPayloadCommand(instance_name="deployment_1"), + federation_partner_ref=federation_partner_ref, + payload_snapshot={"operation_id": str(operation_id)}, + ) + + +def _service_order( + operation_id: UUID, + *, + state: ServiceOrderState = ServiceOrderState.ACCEPTED, + failure_detail: dict[str, object] | None = None, + service_specification_id: UUID | None = None, +) -> ServiceOrder: + return ServiceOrder( + id=uuid4(), + operation_id=operation_id, + correlation_id="corr-original", + order_type=ServiceOrderType.DEPLOY_SERVICE, + service_specification_id=service_specification_id or uuid4(), + state=state, + payload_snapshot={}, + failure_detail=failure_detail, + app_provider_id="VideoAppsCo", + completed_at=( + datetime.fromisoformat("2026-06-19T12:01:00+00:00") + if state in (ServiceOrderState.COMPLETED, ServiceOrderState.FAILED) + else None + ), + ) + + +def _service_specification( + *, + id: UUID, + state: ServiceSpecificationState, +) -> ServiceSpecification: + return ServiceSpecification( + id=id, + app_provider_id="VideoAppsCo", + ref="video-analytics", + name="Video Analytics", + version="1.0.0", + state=state, + ) + + +def _deployment_unit( + *, + id: UUID, + service_specification_id: UUID, + runtime_kind: RuntimeKind = RuntimeKind.HELM, +) -> ServiceDeploymentUnit: + return ServiceDeploymentUnit( + id=id, + service_specification_id=service_specification_id, + ref="main-runtime", + name="Main Runtime", + runtime_kind=runtime_kind, + artifact_ref="oci://example/video-analytics", + resource_requirements=ComputeRequirements( + compute=ComputeResources(cpu_millicores=500, memory_mb=1024), + topology=TopologyConstraints(min_nodes=1), + ), + ) + + +def _parameters() -> Parameters: + return Parameters( + target=CapabilityTarget(), + parameters=CapabilityParameters(), + source_spec=SourceSpecification( + family=SourceSpecificationFamily.INTERNAL, + api="service.deploy", + version="1.0.0", + ), + ) + + +def _deploy_requirement( + *, + service_specification_id: UUID, + deployment_unit_id: UUID, + domain_kind: DomainKind | None = DomainKind.COMPUTE, + ref: str = "deploy-workload", +) -> ServiceCapabilityRequirement: + return ServiceCapabilityRequirement( + service_specification_id=service_specification_id, + deployment_unit_id=deployment_unit_id, + ref=ref, + capability_kind=CapabilityKind.DEPLOY_WORKLOAD, + domain_kind=domain_kind, + parameters=_parameters(), + ) + + +def _control_path_binding( + capability_id: UUID, + *, + runtime_kinds: list[RuntimeKind] | None = None, +) -> ControlPathBinding: + return ControlPathBinding( + id=uuid4(), + capability_id=capability_id, + ref="deploy-binding", + control_path_ref="k8s.zone-a.compute", + supported_runtime_kinds=runtime_kinds or [RuntimeKind.HELM], + supported_actions=["deploy", "scale", "terminate"], + priority=100, + is_default=True, + state=ControlPathBindingState.ACTIVE, + ) + + +def _capability( + domain_id: UUID, + *, + runtime_kinds: list[RuntimeKind] | None = None, +) -> Capability: + capability_id = uuid4() + return Capability( + id=capability_id, + domain_id=domain_id, + ref="deploy-workload", + name="Deploy Workload", + kind=CapabilityKind.DEPLOY_WORKLOAD, + state=CapabilityState.ACTIVE, + control_path_bindings=[_control_path_binding(capability_id, runtime_kinds=runtime_kinds)], + ) + + +def _domain( + id: UUID, + *, + zone_id: UUID, + kind: DomainKind = DomainKind.COMPUTE, + runtime_kinds: list[RuntimeKind] | None = None, +) -> Domain: + return Domain( + id=id, + zone_id=zone_id, + ref="compute", + name="Compute", + kind=kind, + state=DomainState.ACTIVE, + capabilities=[_capability(id, runtime_kinds=runtime_kinds)] + if kind == DomainKind.COMPUTE + else [], + ) + + +def _zone(id: UUID, *, domains: list[Domain] | None = None) -> Zone: + return Zone( + id=id, + ref="zone-a", + name="Zone A", + kind=ZoneKind.RESOURCE, + state=ZoneState.ACTIVE, + domains=domains or [_domain(uuid4(), zone_id=id)], + ) + + +def _service_instance( + *, + service_specification_id: UUID, + service_order_id: UUID, + zone_id: UUID, + state: ServiceInstanceState = ServiceInstanceState.ACTIVE, +) -> ServiceInstance: + return ServiceInstance( + id=uuid4(), + service_specification_id=service_specification_id, + originating_service_order_id=service_order_id, + zone_id=zone_id, + ref="deployment_1", + state=state, + app_provider_id="VideoAppsCo", + ) + + +def _use_case( + *, + service_orders: AsyncMock, + service_instances: AsyncMock | None = None, + capability_instances: AsyncMock | None = None, + service_specifications: AsyncMock | None = None, + service_deployment_units: AsyncMock | None = None, + service_capability_requirements: AsyncMock | None = None, + zones: AsyncMock | None = None, + domains: AsyncMock | None = None, + publisher: AsyncMock, +) -> DeployServiceUseCase: + deployment_units = service_deployment_units + capability_requirements = service_capability_requirements + if deployment_units is None or capability_requirements is None: + unit_id = uuid4() + deployment_units = deployment_units or AsyncMock() + capability_requirements = capability_requirements or AsyncMock() + deployment_units.list_by_service_specification_id.return_value = [ + _deployment_unit(id=unit_id, service_specification_id=uuid4()) + ] + capability_requirements.list_by_service_specification_id.return_value = [ + _deploy_requirement(service_specification_id=uuid4(), deployment_unit_id=unit_id) + ] + zone_repo = zones or AsyncMock() + domain_repo = domains or AsyncMock() + if zones is None: + zone_id = uuid4() + domain = _domain(uuid4(), zone_id=zone_id) + zone_repo.get_by_id.return_value = _zone(zone_id, domains=[domain]) + zone_repo.list_active_resource_zones.return_value = [_zone(zone_id, domains=[domain])] + if domains is None: + domain_repo.get_by_id.return_value = None + + return DeployServiceUseCase( + service_specifications=service_specifications or AsyncMock(), + service_deployment_units=deployment_units, + service_capability_requirements=capability_requirements, + service_orders=service_orders, + service_instances=service_instances or AsyncMock(), + capability_instances=capability_instances or AsyncMock(), + zones=zone_repo, + domains=domain_repo, + publisher=publisher, + ) + + +async def test_accept_replays_accepted_when_operation_order_exists() -> None: + operation_id = uuid4() + existing_order = _service_order(operation_id) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = existing_order + publisher = AsyncMock() + + result = await _use_case(service_orders=service_orders, publisher=publisher).accept( + _command(operation_id) + ) + + service_orders.get_by_operation_id.assert_awaited_once_with(operation_id) + assert result.accepted_order_id is None + assert result.duplicate_order_id == existing_order.id + assert result.completed_event is None + publisher.publish.assert_not_awaited() + + +async def test_accept_replays_completed_event_when_completed_operation_order_exists() -> None: + operation_id = uuid4() + service_specification_id = uuid4() + existing_order = _service_order( + operation_id, + state=ServiceOrderState.COMPLETED, + service_specification_id=service_specification_id, + ) + zone_id = uuid4() + service_instances = AsyncMock() + service_instances.list_by_originating_service_order_id.return_value = [ + _service_instance( + service_specification_id=service_specification_id, + service_order_id=existing_order.id, + zone_id=zone_id, + ) + ] + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = existing_order + publisher = AsyncMock() + + result = await _use_case( + service_orders=service_orders, + service_instances=service_instances, + publisher=publisher, + ).accept(_command(operation_id)) + + service_instances.list_by_originating_service_order_id.assert_awaited_once_with( + existing_order.id + ) + assert result.duplicate_order_id == existing_order.id + assert result.completed_event is not None + assert result.completed_event.status == "completed" + assert result.completed_event.service_order_id == existing_order.id + assert result.completed_event.instances == [ + { + "service_instance_id": str( + service_instances.list_by_originating_service_order_id.return_value[0].id + ), + "zone_id": str(zone_id), + "status": "completed", + } + ] + publisher.publish.assert_not_awaited() + + +async def test_accept_replays_failed_event_when_failed_operation_order_exists() -> None: + operation_id = uuid4() + service_specification_id = uuid4() + failure_detail = { + "type": "about:blank", + "title": "Deploy service failed after acceptance.", + "status": 500, + } + existing_order = _service_order( + operation_id, + state=ServiceOrderState.FAILED, + failure_detail=failure_detail, + service_specification_id=service_specification_id, + ) + zone_id = uuid4() + service_instances = AsyncMock() + service_instances.list_by_originating_service_order_id.return_value = [ + _service_instance( + service_specification_id=service_specification_id, + service_order_id=existing_order.id, + zone_id=zone_id, + state=ServiceInstanceState.FAILED, + ) + ] + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = existing_order + publisher = AsyncMock() + + result = await _use_case( + service_orders=service_orders, + service_instances=service_instances, + publisher=publisher, + ).accept(_command(operation_id)) + + assert result.duplicate_order_id == existing_order.id + assert result.completed_event is not None + assert result.completed_event.status == "failed" + assert result.completed_event.error is None + assert result.completed_event.instances == [ + { + "service_instance_id": str( + service_instances.list_by_originating_service_order_id.return_value[0].id + ), + "zone_id": str(zone_id), + "status": "failed", + "error": failure_detail, + } + ] + publisher.publish.assert_not_awaited() + + +async def test_execute_publishes_failed_before_start_when_service_specification_is_not_active() -> ( + None +): + operation_id = uuid4() + command = _command(operation_id) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.RETIRED, + ) + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + publisher=publisher, + ).accept(command) + + service_orders.get_by_operation_id.assert_awaited_once_with(operation_id) + service_specifications.get_by_id.assert_awaited_once_with(command.service_specification_id) + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + assert payload["service_order_id"] is None + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["operation_id"] == str(operation_id) + assert payload["status"] == "failed" + assert payload["service_order_id"] is None + + +async def test_execute_publishes_failed_before_start_when_service_specification_is_missing() -> ( + None +): + operation_id = uuid4() + command = _command(operation_id) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = None + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + publisher=publisher, + ).accept(command) + + service_orders.get_by_operation_id.assert_awaited_once_with(operation_id) + service_specifications.get_by_id.assert_awaited_once_with(command.service_specification_id) + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["schema_version"] == "1.0" + assert payload["operation_id"] == str(operation_id) + assert payload["service_order_id"] is None + assert payload["service_instance_id"] is None + assert payload["capability"] == "deploy" + assert payload["state"] == "failed_before_start" + assert payload["metadata"] == {} + assert payload["correlation_id"] == "corr-1" + assert datetime.fromisoformat(payload["emitted_at"]).tzinfo is not None + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["schema_version"] == "1.0" + assert payload["operation_id"] == str(operation_id) + assert payload["status"] == "failed" + assert payload["service_order_id"] is None + assert payload["instances"] == [] + assert payload["metadata"] == {} + assert payload["error"] == { + "type": "about:blank", + "title": "Deploy service failed before start.", + "status": 400, + } + assert payload["correlation_id"] == "corr-1" + assert datetime.fromisoformat(payload["completed_at"]).tzinfo is not None + + +async def test_execute_publishes_failed_before_start_when_catalog_has_no_deployment_units() -> None: + operation_id = uuid4() + command = _command(operation_id) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + service_deployment_units = AsyncMock() + service_deployment_units.list_by_service_specification_id.return_value = [] + service_capability_requirements = AsyncMock() + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + service_deployment_units=service_deployment_units, + service_capability_requirements=service_capability_requirements, + publisher=publisher, + ).accept(command) + + service_deployment_units.list_by_service_specification_id.assert_awaited_once_with( + command.service_specification_id + ) + service_capability_requirements.list_by_service_specification_id.assert_not_awaited() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + + +async def test_execute_fails_when_catalog_has_no_deploy_workload_requirement() -> None: + operation_id = uuid4() + command = _command(operation_id) + unit_id = uuid4() + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + service_deployment_units = AsyncMock() + service_deployment_units.list_by_service_specification_id.return_value = [ + _deployment_unit(id=unit_id, service_specification_id=command.service_specification_id) + ] + service_capability_requirements = AsyncMock() + service_capability_requirements.list_by_service_specification_id.return_value = [] + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + service_deployment_units=service_deployment_units, + service_capability_requirements=service_capability_requirements, + publisher=publisher, + ).accept(command) + + service_capability_requirements.list_by_service_specification_id.assert_awaited_once_with( + command.service_specification_id + ) + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + + +async def test_execute_publishes_failed_before_start_when_deploy_requirement_is_not_compute() -> ( + None +): + operation_id = uuid4() + command = _command(operation_id) + unit_id = uuid4() + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + service_deployment_units = AsyncMock() + service_deployment_units.list_by_service_specification_id.return_value = [ + _deployment_unit(id=unit_id, service_specification_id=command.service_specification_id) + ] + service_capability_requirements = AsyncMock() + service_capability_requirements.list_by_service_specification_id.return_value = [ + _deploy_requirement( + service_specification_id=command.service_specification_id, + deployment_unit_id=unit_id, + domain_kind=DomainKind.NETWORK, + ) + ] + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + service_deployment_units=service_deployment_units, + service_capability_requirements=service_capability_requirements, + publisher=publisher, + ).accept(command) + + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + + +async def test_execute_publishes_failed_before_start_when_target_zone_pin_is_missing() -> None: + operation_id = uuid4() + zone_id = uuid4() + command = _command( + operation_id, + targets=[DeployTargetCommand(app_instance_id=uuid4(), zone_id=zone_id)], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.get_by_id.return_value = None + domains = AsyncMock() + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + zones=zones, + domains=domains, + publisher=publisher, + ).accept(command) + + zones.get_by_id.assert_awaited_once_with(zone_id) + domains.get_by_id.assert_not_awaited() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + assert payload["service_order_id"] is None + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["operation_id"] == str(operation_id) + assert payload["status"] == "failed" + assert payload["service_order_id"] is None + + +async def test_execute_publishes_failed_before_start_when_target_domain_pin_is_missing() -> None: + operation_id = uuid4() + zone_id = uuid4() + domain_id = uuid4() + command = _command( + operation_id, + targets=[ + DeployTargetCommand( + app_instance_id=uuid4(), + zone_id=zone_id, + domain_id=domain_id, + ) + ], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.get_by_id.return_value = _zone(zone_id) + domains = AsyncMock() + domains.get_by_id.return_value = None + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + zones=zones, + domains=domains, + publisher=publisher, + ).accept(command) + + zones.get_by_id.assert_awaited_once_with(zone_id) + domains.get_by_id.assert_awaited_once_with(domain_id) + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + assert payload["service_order_id"] is None + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["operation_id"] == str(operation_id) + assert payload["status"] == "failed" + assert payload["service_order_id"] is None + + +async def test_execute_publishes_failed_before_start_when_target_domain_pin_has_no_zone() -> None: + operation_id = uuid4() + command = _command( + operation_id, + targets=[DeployTargetCommand(app_instance_id=uuid4(), domain_id=uuid4())], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + domains = AsyncMock() + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + zones=zones, + domains=domains, + publisher=publisher, + ).accept(command) + + zones.get_by_id.assert_not_awaited() + domains.get_by_id.assert_not_awaited() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + assert payload["service_order_id"] is None + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["operation_id"] == str(operation_id) + assert payload["status"] == "failed" + assert payload["service_order_id"] is None + + +async def test_execute_publishes_failed_before_start_when_target_domain_is_outside_zone() -> None: + operation_id = uuid4() + zone_id = uuid4() + domain_id = uuid4() + command = _command( + operation_id, + targets=[ + DeployTargetCommand( + app_instance_id=uuid4(), + zone_id=zone_id, + domain_id=domain_id, + ) + ], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.get_by_id.return_value = _zone(zone_id) + domains = AsyncMock() + domains.get_by_id.return_value = _domain(domain_id, zone_id=uuid4()) + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_specifications=service_specifications, + zones=zones, + domains=domains, + publisher=publisher, + ).accept(command) + + zones.get_by_id.assert_awaited_once_with(zone_id) + domains.get_by_id.assert_awaited_once_with(domain_id) + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + assert payload["service_order_id"] is None + subject, payload = publisher.publish.await_args_list[1].args + assert subject == "event.srm.operation.completed" + assert payload["operation_id"] == str(operation_id) + assert payload["status"] == "failed" + assert payload["service_order_id"] is None + + +async def test_execute_publishes_failed_before_start_when_targets_repeat_zone_pin() -> None: + operation_id = uuid4() + zone_id = uuid4() + command = _command( + operation_id, + targets=[ + DeployTargetCommand(app_instance_id=uuid4(), zone_id=zone_id), + DeployTargetCommand(app_instance_id=uuid4(), zone_id=zone_id), + ], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.get_by_id.return_value = _zone(zone_id) + zones.list_active_resource_zones.return_value = [_zone(zone_id), _zone(uuid4())] + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_instances=service_instances, + service_specifications=service_specifications, + zones=zones, + publisher=publisher, + ).accept(command) + + zones.list_active_resource_zones.assert_awaited_once() + zones.get_by_id.assert_awaited_once_with(zone_id) + service_orders.create.assert_not_awaited() + service_instances.create.assert_not_awaited() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + assert payload["service_order_id"] is None + + +async def test_execute_publishes_failed_before_start_when_unpinned_target_has_no_candidate() -> ( + None +): + operation_id = uuid4() + command = _command( + operation_id, + targets=[DeployTargetCommand(app_instance_id=uuid4())], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.list_active_resource_zones.return_value = [] + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_instances=service_instances, + service_specifications=service_specifications, + zones=zones, + publisher=publisher, + ).accept(command) + + zones.list_active_resource_zones.assert_awaited_once() + service_orders.create.assert_not_awaited() + service_instances.create.assert_not_awaited() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + assert payload["service_order_id"] is None + + +async def test_execute_publishes_failed_before_start_when_targets_exceed_zones() -> None: + operation_id = uuid4() + zone_id = uuid4() + command = _command( + operation_id, + targets=[ + DeployTargetCommand(app_instance_id=uuid4()), + DeployTargetCommand(app_instance_id=uuid4()), + ], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.list_active_resource_zones.return_value = [_zone(zone_id)] + publisher = AsyncMock() + + await _use_case( + service_orders=service_orders, + service_instances=service_instances, + service_specifications=service_specifications, + zones=zones, + publisher=publisher, + ).accept(command) + + zones.list_active_resource_zones.assert_awaited_once() + service_orders.create.assert_not_awaited() + service_instances.create.assert_not_awaited() + assert publisher.publish.await_count == 2 + subject, payload = publisher.publish.await_args_list[0].args + assert subject == "event.srm.operation.status" + assert payload["operation_id"] == str(operation_id) + assert payload["state"] == "failed_before_start" + assert payload["service_order_id"] is None + + +async def test_execute_places_unpinned_target_on_candidate_zone() -> None: + operation_id = uuid4() + zone_id = uuid4() + domain = _domain(uuid4(), zone_id=zone_id) + command = _command( + operation_id, + targets=[DeployTargetCommand(app_instance_id=uuid4())], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + accepted_order = _service_order(operation_id) + service_orders.create.return_value = accepted_order + service_instances = AsyncMock() + capability_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.list_active_resource_zones.return_value = [_zone(zone_id, domains=[domain])] + publisher = AsyncMock() + + use_case = _use_case( + service_orders=service_orders, + service_instances=service_instances, + capability_instances=capability_instances, + service_specifications=service_specifications, + zones=zones, + publisher=publisher, + ) + result = await use_case.accept(command) + + zones.list_active_resource_zones.assert_awaited_once() + service_orders.create.assert_awaited_once() + service_instances.create.assert_awaited_once() + (created_instance,) = service_instances.create.await_args.args + assert created_instance.zone_id == zone_id + assert result.accepted_order_id == accepted_order.id + assert result.completed_event is None + capability_instances.create.assert_not_awaited() + service_instances.update_state.assert_not_awaited() + service_orders.update_state.assert_not_awaited() + + service_orders.get_by_id.return_value = accepted_order + completed_event = await use_case.complete(command, result) + + capability_instances.create.assert_awaited_once() + (created_capability,) = capability_instances.create.await_args.args + assert created_capability.service_instance_id == command.targets[0].app_instance_id + assert created_capability.originating_service_order_id == accepted_order.id + assert created_capability.control_path_binding_id is not None + assert created_capability.control_path_ref_snapshot == "k8s.zone-a.compute" + assert created_capability.ref == "deploy-workload" + assert created_capability.kind == "activation" + assert created_capability.state == "active" + assert created_capability.result_summary.status == "active" + service_instances.update_state.assert_awaited_once_with( + command.targets[0].app_instance_id, + ServiceInstanceState.ACTIVE, + ) + service_orders.update_state.assert_awaited_once_with( + accepted_order.id, + ServiceOrderState.COMPLETED, + completed_at=ANY, + ) + assert completed_event is not None + assert completed_event.status == "completed" + assert completed_event.instances == [ + { + "service_instance_id": str(command.targets[0].app_instance_id), + "zone_id": str(zone_id), + "status": "completed", + } + ] + publisher.publish.assert_not_awaited() + + +async def test_execute_places_unpinned_targets_on_distinct_candidate_zones() -> None: + operation_id = uuid4() + first_zone_id = uuid4() + second_zone_id = uuid4() + command = _command( + operation_id, + targets=[ + DeployTargetCommand(app_instance_id=uuid4()), + DeployTargetCommand(app_instance_id=uuid4()), + ], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + accepted_order = _service_order(operation_id) + service_orders.create.return_value = accepted_order + service_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.list_active_resource_zones.return_value = [ + _zone(first_zone_id), + _zone(second_zone_id), + ] + publisher = AsyncMock() + + result = await _use_case( + service_orders=service_orders, + service_instances=service_instances, + service_specifications=service_specifications, + zones=zones, + publisher=publisher, + ).accept(command) + + zones.list_active_resource_zones.assert_awaited_once() + assert service_instances.create.await_count == 2 + assert result.accepted_order_id == accepted_order.id + assert [placement.zone_id for placement in result.target_placements] == [ + first_zone_id, + second_zone_id, + ] + publisher.publish.assert_not_awaited() + + +async def test_execute_places_unpinned_target_in_zone_satisfying_all_requirements() -> None: + operation_id = uuid4() + first_zone_id = uuid4() + second_zone_id = uuid4() + helm_unit_id = uuid4() + container_unit_id = uuid4() + command = _command( + operation_id, + targets=[DeployTargetCommand(app_instance_id=uuid4())], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + accepted_order = _service_order(operation_id) + service_orders.create.return_value = accepted_order + service_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + service_deployment_units = AsyncMock() + service_deployment_units.list_by_service_specification_id.return_value = [ + _deployment_unit( + id=helm_unit_id, + service_specification_id=command.service_specification_id, + runtime_kind=RuntimeKind.HELM, + ), + _deployment_unit( + id=container_unit_id, + service_specification_id=command.service_specification_id, + runtime_kind=RuntimeKind.CONTAINER, + ), + ] + service_capability_requirements = AsyncMock() + service_capability_requirements.list_by_service_specification_id.return_value = [ + _deploy_requirement( + service_specification_id=command.service_specification_id, + deployment_unit_id=helm_unit_id, + ref="deploy-helm", + ), + _deploy_requirement( + service_specification_id=command.service_specification_id, + deployment_unit_id=container_unit_id, + ref="deploy-container", + ), + ] + zones = AsyncMock() + zones.list_active_resource_zones.return_value = [ + _zone( + first_zone_id, + domains=[ + _domain( + uuid4(), + zone_id=first_zone_id, + runtime_kinds=[RuntimeKind.HELM], + ) + ], + ), + _zone( + second_zone_id, + domains=[ + _domain( + uuid4(), + zone_id=second_zone_id, + runtime_kinds=[RuntimeKind.HELM, RuntimeKind.CONTAINER], + ) + ], + ), + ] + capability_instances = AsyncMock() + publisher = AsyncMock() + + use_case = _use_case( + service_orders=service_orders, + service_instances=service_instances, + capability_instances=capability_instances, + service_specifications=service_specifications, + service_deployment_units=service_deployment_units, + service_capability_requirements=service_capability_requirements, + zones=zones, + publisher=publisher, + ) + result = await use_case.accept(command) + + service_instances.create.assert_awaited_once() + (created_instance,) = service_instances.create.await_args.args + assert created_instance.zone_id == second_zone_id + assert result.accepted_order_id == accepted_order.id + assert len(result.target_placements) == 1 + assert result.target_placements[0].zone_id == second_zone_id + assert len(result.target_placements[0].requirements) == 2 + assert result.completed_event is None + capability_instances.create.assert_not_awaited() + service_instances.update_state.assert_not_awaited() + service_orders.update_state.assert_not_awaited() + + service_orders.get_by_id.return_value = accepted_order + completed_event = await use_case.complete(command, result) + + assert capability_instances.create.await_count == 2 + assert service_instances.update_state.await_count == 1 + service_orders.update_state.assert_awaited_once_with( + accepted_order.id, + ServiceOrderState.COMPLETED, + completed_at=ANY, + ) + assert completed_event is not None + assert completed_event.instances == [ + { + "service_instance_id": str(command.targets[0].app_instance_id), + "zone_id": str(second_zone_id), + "status": "completed", + } + ] + publisher.publish.assert_not_awaited() + + +async def test_execute_loads_pinned_zone_once_for_multiple_requirements() -> None: + operation_id = uuid4() + zone_id = uuid4() + helm_unit_id = uuid4() + container_unit_id = uuid4() + command = _command( + operation_id, + targets=[DeployTargetCommand(app_instance_id=uuid4(), zone_id=zone_id)], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + service_orders.create.return_value = _service_order(operation_id) + service_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + service_deployment_units = AsyncMock() + service_deployment_units.list_by_service_specification_id.return_value = [ + _deployment_unit( + id=helm_unit_id, + service_specification_id=command.service_specification_id, + runtime_kind=RuntimeKind.HELM, + ), + _deployment_unit( + id=container_unit_id, + service_specification_id=command.service_specification_id, + runtime_kind=RuntimeKind.CONTAINER, + ), + ] + service_capability_requirements = AsyncMock() + service_capability_requirements.list_by_service_specification_id.return_value = [ + _deploy_requirement( + service_specification_id=command.service_specification_id, + deployment_unit_id=helm_unit_id, + ref="deploy-helm", + ), + _deploy_requirement( + service_specification_id=command.service_specification_id, + deployment_unit_id=container_unit_id, + ref="deploy-container", + ), + ] + zones = AsyncMock() + zones.get_by_id.return_value = _zone( + zone_id, + domains=[ + _domain( + uuid4(), + zone_id=zone_id, + runtime_kinds=[RuntimeKind.HELM, RuntimeKind.CONTAINER], + ) + ], + ) + publisher = AsyncMock() + + use_case = _use_case( + service_orders=service_orders, + service_instances=service_instances, + service_specifications=service_specifications, + service_deployment_units=service_deployment_units, + service_capability_requirements=service_capability_requirements, + zones=zones, + publisher=publisher, + ) + result = await use_case.accept(command) + + zones.get_by_id.assert_awaited_once_with(zone_id) + service_instances.create.assert_awaited_once() + assert result.accepted_order_id is not None + assert len(result.target_placements) == 1 + assert result.target_placements[0].zone_id == zone_id + assert len(result.target_placements[0].requirements) == 2 + publisher.publish.assert_not_awaited() + + +async def test_execute_accepts_valid_target_zone_and_domain_pins() -> None: + operation_id = uuid4() + zone_id = uuid4() + domain_id = uuid4() + command = _command( + operation_id, + targets=[ + DeployTargetCommand( + app_instance_id=uuid4(), + zone_id=zone_id, + domain_id=domain_id, + ) + ], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + accepted_order = _service_order(operation_id) + service_orders.create.return_value = accepted_order + service_instances = AsyncMock() + capability_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.get_by_id.return_value = _zone(zone_id) + domains = AsyncMock() + domains.get_by_id.return_value = _domain(domain_id, zone_id=zone_id) + publisher = AsyncMock() + + use_case = _use_case( + service_orders=service_orders, + service_instances=service_instances, + capability_instances=capability_instances, + service_specifications=service_specifications, + zones=zones, + domains=domains, + publisher=publisher, + ) + result = await use_case.accept(command) + + zones.get_by_id.assert_awaited_once_with(zone_id) + domains.get_by_id.assert_awaited_once_with(domain_id) + service_orders.create.assert_awaited_once() + (created_order,) = service_orders.create.await_args.args + assert created_order.operation_id == operation_id + assert created_order.correlation_id == "corr-1" + assert created_order.order_type == ServiceOrderType.DEPLOY_SERVICE + assert created_order.service_specification_id == command.service_specification_id + assert created_order.target_service_instance_id is None + assert created_order.state == ServiceOrderState.ACCEPTED + assert created_order.payload_snapshot == command.payload_snapshot + assert created_order.app_provider_id == "VideoAppsCo" + assert service_instances.create.await_count == 1 + (created_instance,) = service_instances.create.await_args.args + assert created_instance.id == command.targets[0].app_instance_id + assert created_instance.service_specification_id == command.service_specification_id + assert created_instance.originating_service_order_id == accepted_order.id + assert created_instance.zone_id == zone_id + assert created_instance.ref == "deployment_1" + assert created_instance.app_provider_id == "VideoAppsCo" + assert result.accepted_order_id == accepted_order.id + assert result.completed_event is None + capability_instances.create.assert_not_awaited() + service_instances.update_state.assert_not_awaited() + service_orders.update_state.assert_not_awaited() + + service_orders.get_by_id.return_value = accepted_order + completed_event = await use_case.complete(command, result) + + capability_instances.create.assert_awaited_once() + service_instances.update_state.assert_awaited_once_with( + command.targets[0].app_instance_id, + ServiceInstanceState.ACTIVE, + ) + service_orders.update_state.assert_awaited_once_with( + accepted_order.id, + ServiceOrderState.COMPLETED, + completed_at=ANY, + ) + assert completed_event is not None + assert completed_event.service_order_id == accepted_order.id + assert completed_event.status == "completed" + publisher.publish.assert_not_awaited() + + +async def test_execute_preserves_federation_partner_ref_on_created_records() -> None: + operation_id = uuid4() + zone_id = uuid4() + command = _command( + operation_id, + source="federation", + federation_partner_ref="ptr-OperatorB", + targets=[DeployTargetCommand(app_instance_id=uuid4(), zone_id=zone_id)], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + accepted_order = _service_order(operation_id) + service_orders.create.return_value = accepted_order + service_instances = AsyncMock() + capability_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.get_by_id.return_value = _zone(zone_id) + publisher = AsyncMock() + + use_case = _use_case( + service_orders=service_orders, + service_instances=service_instances, + capability_instances=capability_instances, + service_specifications=service_specifications, + zones=zones, + publisher=publisher, + ) + result = await use_case.accept(command) + + service_orders.create.assert_awaited_once() + (created_order,) = service_orders.create.await_args.args + assert created_order.federation_partner_ref == "ptr-OperatorB" + service_instances.create.assert_awaited_once() + (created_instance,) = service_instances.create.await_args.args + assert created_instance.federation_partner_ref == "ptr-OperatorB" + + service_orders.get_by_id.return_value = accepted_order + await use_case.complete(command, result) + + capability_instances.create.assert_awaited_once() + (created_capability,) = capability_instances.create.await_args.args + assert created_capability.federation_partner_ref == "ptr-OperatorB" + publisher.publish.assert_not_awaited() + + +async def test_fail_after_accept_marks_instances_and_order_failed() -> None: + operation_id = uuid4() + zone_id = uuid4() + command = _command( + operation_id, + targets=[DeployTargetCommand(app_instance_id=uuid4(), zone_id=zone_id)], + ) + service_orders = AsyncMock() + service_orders.get_by_operation_id.return_value = None + accepted_order = _service_order(operation_id) + service_orders.create.return_value = accepted_order + service_orders.get_by_id.return_value = accepted_order + service_instances = AsyncMock() + service_specifications = AsyncMock() + service_specifications.get_by_id.return_value = _service_specification( + id=command.service_specification_id, + state=ServiceSpecificationState.ACTIVE, + ) + zones = AsyncMock() + zones.get_by_id.return_value = _zone(zone_id) + publisher = AsyncMock() + use_case = _use_case( + service_orders=service_orders, + service_instances=service_instances, + service_specifications=service_specifications, + zones=zones, + publisher=publisher, + ) + accepted = await use_case.accept(command) + + completed_event = await use_case.fail_after_accept( + command, + accepted, + RuntimeError("adapter failed"), + ) + + service_instances.update_state.assert_awaited_once_with( + command.targets[0].app_instance_id, + ServiceInstanceState.FAILED, + ) + service_orders.update_state.assert_awaited_once_with( + accepted_order.id, + ServiceOrderState.FAILED, + completed_at=ANY, + failure_detail={ + "type": "about:blank", + "title": "Deploy service failed after acceptance.", + "status": 500, + "detail": "adapter failed", + }, + ) + assert completed_event.status == "failed" + assert completed_event.service_order_id == accepted_order.id + assert completed_event.instances == [ + { + "service_instance_id": str(command.targets[0].app_instance_id), + "zone_id": str(zone_id), + "status": "failed", + "error": { + "type": "about:blank", + "title": "Deploy service failed after acceptance.", + "status": 500, + "detail": "adapter failed", + }, + } + ]