diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index e0dc48da2146d068004c9ca36f2a7ec5b33ade4c..48ee72b06c3f9d4c05c4e2d8d1cc5e607d74df8e 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -49,7 +49,7 @@ test: services: - docker:dind script: - - pytest tests/unit tests/integration tests/api --cov=src/srm --cov-branch --cov-report=term-missing || [ $? -eq 5 ] #bypass no tests found error. + - pytest --cov=src/srm --cov-branch --cov-report=term-missing coverage: '/TOTAL.+ ([0-9]{1,3}(?:\.[0-9]+)?%)/' build-and-push: diff --git a/src/srm/adapters/database/repos/catalog.py b/src/srm/adapters/database/repos/catalog.py index 93d05b587b33c7bf78aa1b74b64d2a9f44122abd..1889b9a0cffa2e99929b808cdaf580cf9c87758c 100644 --- a/src/srm/adapters/database/repos/catalog.py +++ b/src/srm/adapters/database/repos/catalog.py @@ -3,7 +3,7 @@ from __future__ import annotations from uuid import UUID import structlog -from sqlalchemy import select +from sqlalchemy import delete, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -16,14 +16,17 @@ from srm.adapters.database.mappers import ( from srm.adapters.database.sql import ( ServiceCapabilityRequirementRow, ServiceDeploymentUnitRow, + ServiceInstanceRow, ServiceSpecificationRow, ) -from srm.adapters.errors import DuplicateServiceSpecificationError +from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError from srm.domain.models.catalog import ( ServiceCapabilityRequirement, ServiceDeploymentUnit, ServiceSpecification, ) +from srm.domain.models.catalog.enums import ServiceSpecificationState +from srm.domain.models.runtime_inventory.enums import ServiceInstanceState from srm.domain.ports.database.catalog import ( ServiceCapabilityRequirementRepository, ServiceDeploymentUnitRepository, @@ -47,6 +50,13 @@ def _log_integrity_error(exc: IntegrityError, *, entity: str, operation: str) -> ) +def _link_requirement_deployment_unit_rows(row: ServiceSpecificationRow) -> None: + deployment_units_by_id = {unit.id: unit for unit in row.deployment_units} + for requirement in row.capability_requirements: + if requirement.deployment_unit_id in deployment_units_by_id: + requirement.deployment_unit = deployment_units_by_id[requirement.deployment_unit_id] + + class SqlServiceSpecificationRepository(ServiceSpecificationRepository): def __init__(self, session: AsyncSession) -> None: self._session = session @@ -65,6 +75,7 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository): async def create(self, service_specification: ServiceSpecification) -> ServiceSpecification: row = ServiceSpecificationMapper.to_row(service_specification) + _link_requirement_deployment_unit_rows(row) self._session.add(row) try: await self._session.flush() @@ -79,6 +90,53 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository): raise RuntimeError("Created service specification could not be reloaded") return saved + async def delete_or_retire(self, id: UUID) -> bool: + row = await self._session.get(ServiceSpecificationRow, id) + if row is None: + return False + + any_instance_id = await self._session.scalar( + select(ServiceInstanceRow.id) + .where(ServiceInstanceRow.service_specification_id == id) + .limit(1) + ) + if any_instance_id is None: + await self._session.execute( + delete(ServiceCapabilityRequirementRow).where( + ServiceCapabilityRequirementRow.service_specification_id == id + ) + ) + await self._session.execute( + delete(ServiceDeploymentUnitRow).where( + ServiceDeploymentUnitRow.service_specification_id == id + ) + ) + await self._session.execute( + delete(ServiceSpecificationRow).where(ServiceSpecificationRow.id == id) + ) + await self._session.flush() + return True + + live_instance_id = await self._session.scalar( + select(ServiceInstanceRow.id) + .where(ServiceInstanceRow.service_specification_id == id) + .where( + ServiceInstanceRow.state.not_in( + [ + ServiceInstanceState.TERMINATED, + ServiceInstanceState.FAILED, + ] + ) + ) + .limit(1) + ) + if live_instance_id is not None: + raise ServiceSpecificationInUseError() + + row.state = ServiceSpecificationState.RETIRED + await self._session.flush() + return True + class SqlServiceDeploymentUnitRepository(ServiceDeploymentUnitRepository): def __init__(self, session: AsyncSession) -> None: diff --git a/src/srm/adapters/database/sql.py b/src/srm/adapters/database/sql.py index 063ff6771c37dc2ebcf639abc128aced62c3c0bc..e8b80322193770efe3e19ccbb5f7c7cd4d1fb720 100644 --- a/src/srm/adapters/database/sql.py +++ b/src/srm/adapters/database/sql.py @@ -227,6 +227,9 @@ class ServiceCapabilityRequirementRow(AuditedMixin, Base): deployment_unit_id: Mapped[UUID | None] = mapped_column( ForeignKey("service_deployment_unit.id") ) + deployment_unit: Mapped["ServiceDeploymentUnitRow | None"] = relationship( + foreign_keys=[deployment_unit_id] + ) ref: Mapped[str] = mapped_column(String(255), nullable=False) capability_kind: Mapped[CapabilityKind] = mapped_column( _enum_type(CapabilityKind), nullable=False diff --git a/src/srm/adapters/errors.py b/src/srm/adapters/errors.py index cb3e5d1769cc7f4a46f9c16e61f1a39390ea7d8f..31f2d1556ba810c8b2ffd1841c69a5d4cd1c4418 100644 --- a/src/srm/adapters/errors.py +++ b/src/srm/adapters/errors.py @@ -10,3 +10,8 @@ class DuplicateEntryError(Exception): class DuplicateServiceSpecificationError(DuplicateEntryError): entity_name: ClassVar[str] = "Service Specification" + + +class ServiceSpecificationInUseError(Exception): + def __init__(self) -> None: + super().__init__("Service specification has live service instances.") diff --git a/src/srm/api/dependencies.py b/src/srm/api/dependencies.py index 20b8b7d78139d6c967bbda8a614746c104ee5bdb..0f52613bd858ae925a7552b665dd74015a67bff8 100644 --- a/src/srm/api/dependencies.py +++ b/src/srm/api/dependencies.py @@ -3,7 +3,13 @@ from typing import Annotated, AsyncGenerator, cast from fastapi import Depends, Request from sqlalchemy.ext.asyncio import AsyncSession +from srm.adapters.database.repos.catalog import SqlServiceSpecificationRepository from srm.app_state import AppState +from srm.application.use_cases.catalog import ( + CreateServiceSpecificationUseCase, + DeleteServiceSpecificationUseCase, + GetServiceSpecificationUseCase, +) def get_app_state(request: Request) -> AppState: @@ -21,3 +27,35 @@ async def get_session(request: Request) -> AsyncGenerator[AsyncSession, None]: SessionDep = Annotated[AsyncSession, Depends(get_session)] + + +def get_create_service_specification_use_case( + session: SessionDep, +) -> CreateServiceSpecificationUseCase: + return CreateServiceSpecificationUseCase(SqlServiceSpecificationRepository(session)) + + +def get_service_specification_use_case( + session: SessionDep, +) -> GetServiceSpecificationUseCase: + return GetServiceSpecificationUseCase(SqlServiceSpecificationRepository(session)) + + +def get_delete_service_specification_use_case( + session: SessionDep, +) -> DeleteServiceSpecificationUseCase: + return DeleteServiceSpecificationUseCase(SqlServiceSpecificationRepository(session)) + + +CreateServiceSpecificationUseCaseDep = Annotated[ + CreateServiceSpecificationUseCase, + Depends(get_create_service_specification_use_case), +] +GetServiceSpecificationUseCaseDep = Annotated[ + GetServiceSpecificationUseCase, + Depends(get_service_specification_use_case), +] +DeleteServiceSpecificationUseCaseDep = Annotated[ + DeleteServiceSpecificationUseCase, + Depends(get_delete_service_specification_use_case), +] diff --git a/src/srm/api/rest/router.py b/src/srm/api/rest/router.py new file mode 100644 index 0000000000000000000000000000000000000000..b26c30be2e6636e5631c0f6b87bccb351d7bc7b9 --- /dev/null +++ b/src/srm/api/rest/router.py @@ -0,0 +1,238 @@ +from uuid import UUID + +import structlog +from fastapi import APIRouter, HTTPException, status + +from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError +from srm.api.dependencies import ( + CreateServiceSpecificationUseCaseDep, + DeleteServiceSpecificationUseCaseDep, + GetServiceSpecificationUseCaseDep, +) +from srm.api.rest.schemas import ( + CreateServiceSpecificationRequest, + CreateServiceSpecificationResponse, + ErrorResponse, + GetServiceSpecificationResponse, + ServiceCapabilityRequirementResponseSchema, + ServiceDeploymentUnitCreateSchema, + ServiceSpecificationCreateSchema, +) +from srm.domain.models.canonical_parameters.parameters import ( + CapabilityParameters, + CapabilityTarget, + Parameters, + SourceSpecification, + SourceSpecificationFamily, +) +from srm.domain.models.catalog import ( + ServiceCapabilityRequirement, + ServiceDeploymentUnit, + ServiceSpecification, +) + +internal = APIRouter(prefix="/internal", tags=["Internal"]) +logger: structlog.BoundLogger = structlog.get_logger(__name__) + + +def _default_parameters(specification: ServiceSpecification) -> Parameters: + source_api = str(specification.descriptor.get("source_api", "internal")) + return Parameters( + target=CapabilityTarget(), + parameters=CapabilityParameters(), + source_spec=SourceSpecification( + family=SourceSpecificationFamily.INTERNAL, + api=source_api, + version=specification.version, + ), + ) + + +def _to_domain(request: CreateServiceSpecificationRequest) -> ServiceSpecification: + exclude_fields = set() + if request.service_specification.id is None: + exclude_fields.add("id") + + specification = ServiceSpecification.model_validate( + request.service_specification.model_dump(mode="python", exclude=exclude_fields) + ) + default_parameters = _default_parameters(specification) + + specification.deployment_units = [ + ServiceDeploymentUnit.model_validate( + { + **unit.model_dump(mode="python"), + "service_specification_id": specification.id, + } + ) + for unit in request.service_deployment_units + ] + specification.capability_requirements = [ + ServiceCapabilityRequirement.model_validate( + { + **requirement.model_dump(mode="python", exclude={"deployment_unit_ref"}), + "service_specification_id": specification.id, + "deployment_unit_id": None, + "parameters": default_parameters.model_copy(deep=True).model_dump(mode="python"), + } + ) + for requirement in request.service_capability_requirements + ] + + return specification + + +def _requirement_deployment_unit_refs( + request: CreateServiceSpecificationRequest, +) -> dict[str, str]: + return { + requirement.ref: requirement.deployment_unit_ref + for requirement in request.service_capability_requirements + if requirement.deployment_unit_ref is not None + } + + +def _build_response(saved: ServiceSpecification) -> CreateServiceSpecificationResponse: + return CreateServiceSpecificationResponse(id=saved.id, ref=saved.ref) + + +def _build_get_response(saved: ServiceSpecification) -> GetServiceSpecificationResponse: + return GetServiceSpecificationResponse( + service_specification=ServiceSpecificationCreateSchema( + id=saved.id, + app_provider_id=saved.app_provider_id, + ref=saved.ref, + name=saved.name, + version=saved.version, + descriptor=saved.descriptor, + metadata=saved.metadata, + ), + service_deployment_units=[ + ServiceDeploymentUnitCreateSchema( + ref=unit.ref, + name=unit.name, + runtime_kind=unit.runtime_kind, + artifact_ref=unit.artifact_ref, + resource_requirements=unit.resource_requirements, + parameters_schema=unit.parameters_schema, + metadata=unit.metadata, + ) + for unit in saved.deployment_units + ], + service_capability_requirements=[ + ServiceCapabilityRequirementResponseSchema( + ref=requirement.ref, + capability_kind=requirement.capability_kind, + domain_kind=requirement.domain_kind, + is_required=requirement.is_required, + selector=requirement.selector, + policy=requirement.policy, + metadata=requirement.metadata, + ) + for requirement in saved.capability_requirements + ], + ) + + +@internal.post( + "/catalog/service-specifications", + status_code=status.HTTP_201_CREATED, + responses={ + status.HTTP_409_CONFLICT: { + "model": ErrorResponse, + "description": "Service specification already exists.", + } + }, +) +async def create_service_specification( + request: CreateServiceSpecificationRequest, + use_case: CreateServiceSpecificationUseCaseDep, +) -> CreateServiceSpecificationResponse: + logger.info( + "create_service_specification_requested", + service_specification_ref=request.service_specification.ref, + version=request.service_specification.version, + ) + + try: + saved = await use_case.execute( + _to_domain(request), + _requirement_deployment_unit_refs(request), + ) + except DuplicateServiceSpecificationError as exc: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Service specification already exists.", + ) from exc + + logger.info( + "create_service_specification_succeeded", + service_specification_id=str(saved.id), + ) + return _build_response(saved) + + +@internal.get( + "/catalog/service-specifications/{id}", + responses={ + status.HTTP_404_NOT_FOUND: { + "model": ErrorResponse, + "description": "Service specification not found.", + } + }, +) +async def get_service_specification( + id: UUID, + use_case: GetServiceSpecificationUseCaseDep, +) -> GetServiceSpecificationResponse: + logger.info("get_service_specification_requested", service_specification_id=str(id)) + + specification = await use_case.execute(id) + if specification is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Service specification not found.", + ) + + logger.info( + "get_service_specification_succeeded", + service_specification_id=str(specification.id), + ) + return _build_get_response(specification) + + +@internal.delete( + "/catalog/service-specifications/{id}", + status_code=status.HTTP_204_NO_CONTENT, + responses={ + status.HTTP_404_NOT_FOUND: { + "model": ErrorResponse, + "description": "Service specification not found.", + }, + status.HTTP_409_CONFLICT: { + "model": ErrorResponse, + "description": "Service specification has live service instances.", + }, + }, +) +async def delete_service_specification( + id: UUID, + use_case: DeleteServiceSpecificationUseCaseDep, +) -> None: + logger.info("delete_service_specification_requested", service_specification_id=str(id)) + + try: + deleted = await use_case.execute(id) + except ServiceSpecificationInUseError as exc: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Service specification has live service instances.", + ) from exc + + if not deleted: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Service specification not found.", + ) + + logger.info("delete_service_specification_succeeded", service_specification_id=str(id)) diff --git a/src/srm/api/rest/schemas.py b/src/srm/api/rest/schemas.py new file mode 100644 index 0000000000000000000000000000000000000000..12c4882d1d6a373949bb8a84bbda571888cf3746 --- /dev/null +++ b/src/srm/api/rest/schemas.py @@ -0,0 +1,91 @@ +from typing import Any +from uuid import UUID + +from pydantic import Field, model_validator + +from srm.domain.models.canonical_parameters.compute import ComputeRequirements +from srm.domain.models.catalog.enums import RuntimeKind +from srm.domain.models.common import DomainModel +from srm.domain.models.topology.enums import CapabilityKind, DomainKind + + +class ServiceSpecificationCreateSchema(DomainModel): + id: UUID | None = None + app_provider_id: str + ref: str + name: str + version: str + descriptor: dict[str, Any] = Field(default_factory=dict) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class ServiceDeploymentUnitCreateSchema(DomainModel): + ref: str + name: str + runtime_kind: RuntimeKind + artifact_ref: str | None = None + resource_requirements: ComputeRequirements + parameters_schema: dict[str, Any] = Field(default_factory=dict) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class ServiceCapabilityRequirementCreateSchema(DomainModel): + ref: str + deployment_unit_ref: str | None = None + capability_kind: CapabilityKind + domain_kind: DomainKind | None = None + is_required: bool = True + selector: dict[str, Any] = Field(default_factory=dict) + policy: dict[str, Any] = Field(default_factory=dict) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class ServiceCapabilityRequirementResponseSchema(DomainModel): + ref: str + capability_kind: CapabilityKind + domain_kind: DomainKind | None = None + is_required: bool = True + selector: dict[str, Any] = Field(default_factory=dict) + policy: dict[str, Any] = Field(default_factory=dict) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class CreateServiceSpecificationRequest(DomainModel): + service_specification: ServiceSpecificationCreateSchema + service_deployment_units: list[ServiceDeploymentUnitCreateSchema] = Field(default_factory=list) + service_capability_requirements: list[ServiceCapabilityRequirementCreateSchema] = Field( + default_factory=list + ) + + @model_validator(mode="after") + def validate_deployment_unit_refs(self) -> "CreateServiceSpecificationRequest": + deployment_unit_refs = {unit.ref for unit in self.service_deployment_units} + missing_refs = sorted( + { + requirement.deployment_unit_ref + for requirement in self.service_capability_requirements + if requirement.deployment_unit_ref is not None + and requirement.deployment_unit_ref not in deployment_unit_refs + } + ) + if missing_refs: + raise ValueError( + "Unknown deployment unit refs in service_capability_requirements: " + + ", ".join(missing_refs) + ) + return self + + +class CreateServiceSpecificationResponse(DomainModel): + id: UUID + ref: str + + +class GetServiceSpecificationResponse(DomainModel): + service_specification: ServiceSpecificationCreateSchema + service_deployment_units: list[ServiceDeploymentUnitCreateSchema] + service_capability_requirements: list[ServiceCapabilityRequirementResponseSchema] + + +class ErrorResponse(DomainModel): + detail: str diff --git a/src/srm/application/use_cases/catalog.py b/src/srm/application/use_cases/catalog.py new file mode 100644 index 0000000000000000000000000000000000000000..ceae0933f6bf04484f64fb60a73063fd32ea5c02 --- /dev/null +++ b/src/srm/application/use_cases/catalog.py @@ -0,0 +1,38 @@ +from uuid import UUID + +from srm.domain.models.catalog import ServiceSpecification +from srm.domain.ports.database.catalog import ServiceSpecificationRepository + + +class CreateServiceSpecificationUseCase: + def __init__(self, repository: ServiceSpecificationRepository) -> None: + self._repository = repository + + async def execute( + self, + specification: ServiceSpecification, + requirement_deployment_unit_refs: dict[str, str], + ) -> ServiceSpecification: + deployment_unit_ids = {unit.ref: unit.id for unit in specification.deployment_units} + for requirement in specification.capability_requirements: + deployment_unit_ref = requirement_deployment_unit_refs.get(requirement.ref) + if deployment_unit_ref is not None: + requirement.deployment_unit_id = deployment_unit_ids[deployment_unit_ref] + + return await self._repository.create(specification) + + +class GetServiceSpecificationUseCase: + def __init__(self, repository: ServiceSpecificationRepository) -> None: + self._repository = repository + + async def execute(self, id: UUID) -> ServiceSpecification | None: + return await self._repository.get_by_id(id) + + +class DeleteServiceSpecificationUseCase: + def __init__(self, repository: ServiceSpecificationRepository) -> None: + self._repository = repository + + async def execute(self, id: UUID) -> bool: + return await self._repository.delete_or_retire(id) diff --git a/src/srm/domain/ports/database/catalog.py b/src/srm/domain/ports/database/catalog.py index e7e0215c46b05c5f4810c3c137f744e05ebce50e..bdfb3fcb107f65fac467c9d521b403edfccd4f50 100644 --- a/src/srm/domain/ports/database/catalog.py +++ b/src/srm/domain/ports/database/catalog.py @@ -19,6 +19,10 @@ class ServiceSpecificationRepository(ABC): async def create(self, service_specification: ServiceSpecification) -> ServiceSpecification: pass + @abstractmethod + async def delete_or_retire(self, id: UUID) -> bool: + pass + class ServiceDeploymentUnitRepository(ABC): @abstractmethod diff --git a/src/srm/main.py b/src/srm/main.py index 4e4e5f7100619460e03de948b996619b41108411..6c7334a5f41df3616c112eb210e7933ba132a012 100644 --- a/src/srm/main.py +++ b/src/srm/main.py @@ -28,6 +28,7 @@ from srm.adapters.databus.nats_connection_manager import ( from srm.api.databus.nats_subscriber import NatsSubscriber, subscribe_to_subjects from srm.api.health import health_router from srm.api.middlewares.middlewares import register_middlewares +from srm.api.rest.router import internal from srm.config import Settings, get_settings @@ -101,5 +102,6 @@ def create_app() -> FastAPI: # Add routers app.include_router(health_router) + app.include_router(internal) return app diff --git a/tests/api/rest/test_catalog.py b/tests/api/rest/test_catalog.py new file mode 100644 index 0000000000000000000000000000000000000000..aba344a8022c6665ea674ceb9b9ad53e151a6afb --- /dev/null +++ b/tests/api/rest/test_catalog.py @@ -0,0 +1,197 @@ +from typing import Any +from uuid import UUID, uuid4 + +import pytest +from fastapi import HTTPException +from httpx import AsyncClient + +from srm.adapters.errors import ServiceSpecificationInUseError +from srm.api.rest.router import _build_response, _to_domain, delete_service_specification +from srm.api.rest.schemas import CreateServiceSpecificationRequest +from srm.application.use_cases.catalog import DeleteServiceSpecificationUseCase + + +def _request_payload(*, service_specification_id: str | None) -> dict[str, Any]: + service_specification = { + "app_provider_id": "VideoAppsCo", + "ref": "video-analytics", + "name": "Video Analytics", + "version": "1.2.3", + "descriptor": {"source_api": "edge-application-management"}, + } + if service_specification_id is not None: + service_specification["id"] = service_specification_id + + return { + "service_specification": service_specification, + "service_deployment_units": [ + { + "ref": "main", + "name": "Main Unit", + "runtime_kind": "helm", + "artifact_ref": "oci://example/video-analytics", + "resource_requirements": { + "compute": {"cpu_millicores": 500, "memory_mb": 1024}, + "topology": {"min_nodes": 1}, + }, + } + ], + "service_capability_requirements": [ + { + "ref": "compute-capability", + "deployment_unit_ref": "main", + "capability_kind": "deploy_workload", + "domain_kind": "compute", + } + ], + } + + +def test_create_service_specification_generates_missing_id() -> None: + request = CreateServiceSpecificationRequest.model_validate( + _request_payload(service_specification_id=None) + ) + + specification = _to_domain(request) + + assert isinstance(specification.id, UUID) + assert specification.deployment_units[0].service_specification_id == specification.id + assert specification.capability_requirements[0].service_specification_id == specification.id + + +def test_create_service_specification_preserves_supplied_id() -> None: + service_specification_id = uuid4() + request = CreateServiceSpecificationRequest.model_validate( + _request_payload(service_specification_id=str(service_specification_id)) + ) + + specification = _to_domain(request) + + assert specification.id == service_specification_id + + +def test_create_service_specification_response_returns_id_and_ref() -> None: + request = CreateServiceSpecificationRequest.model_validate( + _request_payload(service_specification_id=None) + ) + specification = _to_domain(request) + + response = _build_response(specification) + + assert response.model_dump(mode="json") == { + "id": str(specification.id), + "ref": "video-analytics", + } + + +class _DeleteUseCase(DeleteServiceSpecificationUseCase): + def __init__(self, *, result: bool | None = True, error: Exception | None = None) -> None: + self.result = result + self.error = error + + async def execute(self, id: UUID) -> bool: + if self.error is not None: + raise self.error + assert self.result is not None + return self.result + + +async def test_delete_service_specification_returns_none_on_success() -> None: + await delete_service_specification(uuid4(), _DeleteUseCase(result=True)) + + +async def test_delete_service_specification_returns_404_when_missing() -> None: + with pytest.raises(HTTPException) as exc_info: + await delete_service_specification(uuid4(), _DeleteUseCase(result=False)) + + assert exc_info.value.status_code == 404 + assert exc_info.value.detail == "Service specification not found." + + +async def test_delete_service_specification_returns_409_when_live_instances_exist() -> None: + with pytest.raises(HTTPException) as exc_info: + await delete_service_specification( + uuid4(), + _DeleteUseCase(error=ServiceSpecificationInUseError()), + ) + + assert exc_info.value.status_code == 409 + assert exc_info.value.detail == "Service specification has live service instances." + + +@pytest.mark.integration +async def test_create_then_get_returns_full_specification(client_with_db: AsyncClient) -> None: + payload = _request_payload(service_specification_id=None) + + create_response = await client_with_db.post( + "/internal/catalog/service-specifications", json=payload + ) + + assert create_response.status_code == 201 + body = create_response.json() + assert body["ref"] == "video-analytics" + + get_response = await client_with_db.get( + f"/internal/catalog/service-specifications/{body['id']}" + ) + + assert get_response.status_code == 200 + get_body = get_response.json() + assert get_body["service_specification"]["ref"] == "video-analytics" + assert get_body["service_deployment_units"][0]["ref"] == "main" + assert get_body["service_capability_requirements"][0]["ref"] == "compute-capability" + + +@pytest.mark.integration +async def test_create_duplicate_id_returns_409(client_with_db: AsyncClient) -> None: + payload = _request_payload(service_specification_id=str(uuid4())) + + first = await client_with_db.post("/internal/catalog/service-specifications", json=payload) + assert first.status_code == 201 + + second = await client_with_db.post("/internal/catalog/service-specifications", json=payload) + + assert second.status_code == 409 + + +@pytest.mark.integration +async def test_create_unknown_deployment_unit_ref_returns_422(client_with_db: AsyncClient) -> None: + payload = _request_payload(service_specification_id=None) + payload["service_capability_requirements"][0]["deployment_unit_ref"] = "missing-unit" + + response = await client_with_db.post("/internal/catalog/service-specifications", json=payload) + + assert response.status_code == 422 + + +@pytest.mark.integration +async def test_get_missing_returns_404(client_with_db: AsyncClient) -> None: + response = await client_with_db.get(f"/internal/catalog/service-specifications/{uuid4()}") + + assert response.status_code == 404 + + +@pytest.mark.integration +async def test_delete_removes_specification(client_with_db: AsyncClient) -> None: + payload = _request_payload(service_specification_id=None) + create_response = await client_with_db.post( + "/internal/catalog/service-specifications", json=payload + ) + specification_id = create_response.json()["id"] + + delete_response = await client_with_db.delete( + f"/internal/catalog/service-specifications/{specification_id}" + ) + assert delete_response.status_code == 204 + + get_response = await client_with_db.get( + f"/internal/catalog/service-specifications/{specification_id}" + ) + assert get_response.status_code == 404 + + +@pytest.mark.integration +async def test_delete_missing_returns_404(client_with_db: AsyncClient) -> None: + response = await client_with_db.delete(f"/internal/catalog/service-specifications/{uuid4()}") + + assert response.status_code == 404 diff --git a/tests/application/test_service_specification_create.py b/tests/application/test_service_specification_create.py new file mode 100644 index 0000000000000000000000000000000000000000..2bba6287f26c904056d0e7263845b34e17372d8a --- /dev/null +++ b/tests/application/test_service_specification_create.py @@ -0,0 +1,103 @@ +from uuid import uuid4 + +from srm.application.use_cases.catalog import ( + CreateServiceSpecificationUseCase, + DeleteServiceSpecificationUseCase, +) +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, +) +from srm.domain.models.topology import CapabilityKind, DomainKind +from tests.unit.fakes import InMemoryServiceSpecificationRepository + + +def _compute_requirements() -> ComputeRequirements: + return ComputeRequirements( + compute=ComputeResources(cpu_millicores=500, memory_mb=1024), + topology=TopologyConstraints(min_nodes=1), + ) + + +def _capability_parameters() -> Parameters: + return Parameters( + target=CapabilityTarget(), + parameters=CapabilityParameters(profile_ref="gold"), + source_spec=SourceSpecification( + family=SourceSpecificationFamily.INTERNAL, + api="internal", + version="1.0.0", + ), + ) + + +async def test_create_links_capability_requirement_to_deployment_unit_ref() -> None: + service_specification_id = uuid4() + deployment_unit = ServiceDeploymentUnit( + service_specification_id=service_specification_id, + ref="main", + name="Main Unit", + runtime_kind=RuntimeKind.HELM, + artifact_ref="oci://example/video-analytics", + resource_requirements=_compute_requirements(), + ) + requirement = ServiceCapabilityRequirement( + service_specification_id=service_specification_id, + ref="compute-capability", + capability_kind=CapabilityKind.DEPLOY_WORKLOAD, + domain_kind=DomainKind.COMPUTE, + parameters=_capability_parameters(), + ) + specification = ServiceSpecification( + id=service_specification_id, + app_provider_id=str(uuid4()), + ref="video-analytics", + name="Video Analytics", + version="1.2.3", + deployment_units=[deployment_unit], + capability_requirements=[requirement], + ) + repository = InMemoryServiceSpecificationRepository() + use_case = CreateServiceSpecificationUseCase(repository) + + created = await use_case.execute( + specification, + requirement_deployment_unit_refs={"compute-capability": "main"}, + ) + + assert created.capability_requirements[0].deployment_unit_id == deployment_unit.id + reloaded = await repository.get_by_id(created.id) + assert reloaded is not None + assert reloaded.capability_requirements[0].deployment_unit_id == deployment_unit.id + + +async def test_delete_removes_service_specification_without_instances() -> None: + specification = ServiceSpecification( + app_provider_id=str(uuid4()), + ref="video-analytics", + name="Video Analytics", + version="1.2.3", + ) + repository = InMemoryServiceSpecificationRepository() + created = await repository.create(specification) + use_case = DeleteServiceSpecificationUseCase(repository) + + deleted = await use_case.execute(created.id) + + assert deleted is True + reloaded = await repository.get_by_id(created.id) + assert reloaded is None diff --git a/tests/integration/test_postgres.py b/tests/integration/test_postgres.py index e857216e872dd3ec65a50f8394f62fa4daaa7113..bf2ef163e53bf00acf61469fd1bc3a638f73ebfd 100644 --- a/tests/integration/test_postgres.py +++ b/tests/integration/test_postgres.py @@ -8,8 +8,12 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker from srm.adapters.database.repos.catalog import SqlServiceSpecificationRepository +from srm.adapters.database.repos.runtime_inventory import ( + SqlServiceInstanceRepository, + SqlServiceOrderRepository, +) from srm.adapters.database.repos.topology import SqlZoneRepository -from srm.adapters.errors import DuplicateServiceSpecificationError +from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError from srm.domain.models.canonical_parameters.compute import ( ComputeRequirements, ComputeResources, @@ -32,6 +36,12 @@ from srm.domain.models.catalog import ( ServiceSpecificationState, ) from srm.domain.models.common import TransportProtocol +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 CapabilityKind, DomainKind, Zone from srm.domain.models.topology.enums import ZoneKind, ZoneState @@ -134,6 +144,41 @@ def _zone() -> Zone: ) +def _service_order(*, app_provider_id: str, specification_id: UUID, zone_id: UUID) -> ServiceOrder: + return ServiceOrder( + id=uuid4(), + operation_id=uuid4(), + order_type=ServiceOrderType.DEPLOY_SERVICE, + service_specification_id=specification_id, + state=ServiceOrderState.ACCEPTED, + payload_snapshot={"target_pins": {"zone_id": str(zone_id)}}, + app_provider_id=app_provider_id, + created_at=_timestamp(), + updated_at=_timestamp(), + ) + + +def _service_instance( + *, + app_provider_id: str, + specification_id: UUID, + order_id: UUID, + zone_id: UUID, + state: ServiceInstanceState = ServiceInstanceState.ACTIVE, +) -> ServiceInstance: + return ServiceInstance( + id=uuid4(), + service_specification_id=specification_id, + originating_service_order_id=order_id, + zone_id=zone_id, + ref="prod", + state=state, + app_provider_id=app_provider_id, + created_at=_timestamp(), + updated_at=_timestamp(), + ) + + async def test_service_specification_repo_persists_and_loads_aggregate( db_session: AsyncSession, ) -> None: @@ -161,6 +206,22 @@ async def test_service_specification_repo_persists_and_loads_aggregate( assert reloaded.capability_requirements[0].parameters == _parameters() +async def test_service_specification_repo_persists_linked_deployment_unit_requirement( + db_session: AsyncSession, +) -> None: + repo = SqlServiceSpecificationRepository(db_session) + specification = _service_specification() + deployment_unit_id = specification.deployment_units[0].id + specification.capability_requirements[0].deployment_unit_id = deployment_unit_id + + saved = await repo.create(specification) + + assert saved.capability_requirements[0].deployment_unit_id == deployment_unit_id + reloaded = await repo.get_by_id(saved.id) + assert reloaded is not None + assert reloaded.capability_requirements[0].deployment_unit_id == deployment_unit_id + + async def test_service_specification_duplicate_entry( db_session: AsyncSession, ) -> None: @@ -208,6 +269,81 @@ async def test_service_specification_duplicate_id_is_conflict_without_overwrite( assert reloaded.ref == saved.ref +async def test_service_specification_delete_rejects_live_instances( + db_session: AsyncSession, +) -> None: + specification_repo = SqlServiceSpecificationRepository(db_session) + zone_repo = SqlZoneRepository(db_session) + order_repo = SqlServiceOrderRepository(db_session) + service_instance_repo = SqlServiceInstanceRepository(db_session) + specification = await specification_repo.create(_service_specification()) + zone = await zone_repo.create(_zone()) + order = await order_repo.create( + _service_order( + app_provider_id=specification.app_provider_id, + specification_id=specification.id, + zone_id=zone.id, + ) + ) + await service_instance_repo.create( + _service_instance( + app_provider_id=specification.app_provider_id, + specification_id=specification.id, + order_id=order.id, + zone_id=zone.id, + ) + ) + + with pytest.raises(ServiceSpecificationInUseError): + await specification_repo.delete_or_retire(specification.id) + + +async def test_service_specification_delete_hard_deletes_without_instances( + db_session: AsyncSession, +) -> None: + specification_repo = SqlServiceSpecificationRepository(db_session) + specification = await specification_repo.create(_service_specification()) + + deleted = await specification_repo.delete_or_retire(specification.id) + + assert deleted is True + assert await specification_repo.get_by_id(specification.id) is None + + +async def test_service_specification_delete_retires_when_only_terminal_instances_exist( + db_session: AsyncSession, +) -> None: + specification_repo = SqlServiceSpecificationRepository(db_session) + zone_repo = SqlZoneRepository(db_session) + order_repo = SqlServiceOrderRepository(db_session) + service_instance_repo = SqlServiceInstanceRepository(db_session) + specification = await specification_repo.create(_service_specification()) + zone = await zone_repo.create(_zone()) + order = await order_repo.create( + _service_order( + app_provider_id=specification.app_provider_id, + specification_id=specification.id, + zone_id=zone.id, + ) + ) + await service_instance_repo.create( + _service_instance( + app_provider_id=specification.app_provider_id, + specification_id=specification.id, + order_id=order.id, + zone_id=zone.id, + state=ServiceInstanceState.TERMINATED, + ) + ) + + deleted = await specification_repo.delete_or_retire(specification.id) + + assert deleted is True + reloaded = await specification_repo.get_by_id(specification.id) + assert reloaded is not None + assert reloaded.state == ServiceSpecificationState.RETIRED + + async def test_zone_repo_duplicate_id_is_conflict_without_overwrite( db_engine: AsyncEngine, ) -> None: diff --git a/tests/unit/fakes.py b/tests/unit/fakes.py index c3f7140754056cfa0f318c8061a4b02d7648c0ba..d17942202a514c72e8d21310bcaff3b1a61554d5 100644 --- a/tests/unit/fakes.py +++ b/tests/unit/fakes.py @@ -50,6 +50,13 @@ class InMemoryServiceSpecificationRepository(ServiceSpecificationRepository): self.rows[stored.id] = stored return stored.model_copy(deep=True) + async def delete_or_retire(self, id: UUID) -> bool: + found = self.rows.get(id) + if found is None: + return False + del self.rows[id] + return True + class InMemoryZoneRepository(ZoneRepository): def __init__(self) -> None: diff --git a/tests/unit/test_service_specification_repo.py b/tests/unit/test_service_specification_repo.py index 4f4e6cbcbc08b86be5bc6bf8a53b5f169a96869e..a13eb85488494240e185c7e99811e8b43fe1cfe8 100644 --- a/tests/unit/test_service_specification_repo.py +++ b/tests/unit/test_service_specification_repo.py @@ -13,7 +13,7 @@ from srm.adapters.database.sql import ( ServiceDeploymentUnitRow, ServiceSpecificationRow, ) -from srm.adapters.errors import DuplicateServiceSpecificationError +from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError from srm.domain.models.canonical_parameters.compute import ( ComputeRequirements, ComputeResources, @@ -180,3 +180,63 @@ async def test_create_propagates_check_violation_integrity_errors() -> None: with pytest.raises(IntegrityError): await repo.create(domain) + + +async def test_delete_or_retire_returns_false_when_missing() -> None: + session = AsyncMock(spec=AsyncSession) + session.get.return_value = None + repo = SqlServiceSpecificationRepository(session) + + result = await repo.delete_or_retire(uuid4()) + + assert result is False + session.scalar.assert_not_awaited() + session.execute.assert_not_awaited() + session.flush.assert_not_awaited() + + +async def test_delete_or_retire_hard_deletes_when_no_instances_exist() -> None: + row = _service_specification_row() + session = AsyncMock(spec=AsyncSession) + session.get.return_value = row + session.scalar.return_value = None + repo = SqlServiceSpecificationRepository(session) + + result = await repo.delete_or_retire(row.id) + + assert result is True + session.scalar.assert_awaited_once() + assert session.execute.await_count == 3 + session.flush.assert_awaited_once() + + +async def test_delete_or_retire_sets_state_to_retired_when_only_terminal_instances_exist() -> None: + row = _service_specification_row() + session = AsyncMock(spec=AsyncSession) + session.get.return_value = row + session.scalar.side_effect = [uuid4(), None] + repo = SqlServiceSpecificationRepository(session) + + result = await repo.delete_or_retire(row.id) + + assert result is True + assert row.state == ServiceSpecificationState.RETIRED + assert session.scalar.await_count == 2 + session.execute.assert_not_awaited() + session.flush.assert_awaited_once() + + +async def test_delete_or_retire_raises_when_live_instance_exists() -> None: + row = _service_specification_row() + session = AsyncMock(spec=AsyncSession) + session.get.return_value = row + session.scalar.side_effect = [uuid4(), uuid4()] + repo = SqlServiceSpecificationRepository(session) + + with pytest.raises(ServiceSpecificationInUseError): + await repo.delete_or_retire(row.id) + + assert row.state == ServiceSpecificationState.ACTIVE + assert session.scalar.await_count == 2 + session.execute.assert_not_awaited() + session.flush.assert_not_awaited()