Add unit tests for core

Author: TheByronHimes

openapi.yaml

@@ -9,10 +9,12 @@ components:
           type: string
         storage_alias:
           description: S3 storage alias to use for uploads
+          minLength: 1
           title: Storage Alias
           type: string
         title:
           description: Short meaningful name for the box
+          minLength: 1
           title: Title
           type: string
       required:

src/uos/adapters/inbound/fastapi_/configure.py

@@ -25,7 +25,7 @@
 from uos.adapters.inbound.fastapi_.routes import router
 from uos.config import Config
 
-config = Config()
+config = Config()  # type: ignore
 
 
 def get_openapi_schema(api) -> dict[str, Any]:

src/uos/adapters/inbound/fastapi_/routes.py

@@ -112,7 +112,9 @@ async def create_research_data_upload_box(
 
     try:
         box_id = await upload_service.create_research_data_upload_box(
-            request=request,
+            title=request.title,
+            description=request.description,
+            storage_alias=request.storage_alias,
             user_id=UUID(auth_context.id),
         )
         return box_id

src/uos/core/models.py

@@ -140,9 +140,13 @@ class ViewFileBoxWorkOrder(BaseWorkOrderToken):
 class CreateUploadBoxRequest(BaseModel):
     """Request model for creating a new research data upload box."""
 
-    title: str = Field(..., description="Short meaningful name for the box")
+    title: str = Field(
+        ..., description="Short meaningful name for the box", min_length=1
+    )
     description: str = Field(..., description="Describes the upload box in more detail")
-    storage_alias: str = Field(..., description="S3 storage alias to use for uploads")
+    storage_alias: str = Field(
+        ..., description="S3 storage alias to use for uploads", min_length=1
+    )
 
 
 class CreateUploadBoxResponse(BaseModel):

src/uos/core/orchestrator.py

@@ -25,7 +25,6 @@
 from pydantic import UUID4
 
 from uos.core.models import (
-    CreateUploadBoxRequest,
     FileUploadBox,
     GrantAccessRequest,
     ResearchDataUploadBox,
@@ -51,16 +50,18 @@ def __init__(
         box_dao: BoxDao,
         audit_repository: AuditRepositoryPort,
         ucs_client: UCSClientPort,
-        claims_client: AccessClientPort,
+        access_client: AccessClientPort,
     ):
         self._box_dao = box_dao
         self._audit_repository = audit_repository
         self._ucs_client = ucs_client
-        self._access_client = claims_client
+        self._access_client = access_client
 
     async def create_research_data_upload_box(
         self,
-        request: CreateUploadBoxRequest,
+        title: str,
+        description: str,
+        storage_alias: str,
         user_id: UUID4,
     ) -> UUID4:
         """Create a new research data upload box.
@@ -78,18 +79,18 @@ async def create_research_data_upload_box(
         """
         # Create FileUploadBox in UCS
         file_upload_box_id = await self._ucs_client.create_file_upload_box(
-            storage_alias=request.storage_alias
+            storage_alias=storage_alias
         )
 
         # Create ResearchDataUploadBox
         box = ResearchDataUploadBox(
             state=ResearchDataUploadBoxState.OPEN,
-            title=request.title,
-            description=request.description,
+            title=title,
+            description=description,
             last_changed=now_utc_ms_prec(),
             changed_by=user_id,
             file_upload_box_id=file_upload_box_id,
-            storage_alias=request.storage_alias,
+            storage_alias=storage_alias,
         )
 
         # Store in repository & create audit record
@@ -157,6 +158,7 @@ async def grant_upload_access(
         Raises:
             BoxNotFoundError: If the box doesn't exist.
         """
+        # TODO: Test this method after adding audit record
         # Verify the upload box exists
         await self._box_dao.get_by_id(request.box_id)
 
@@ -166,7 +168,7 @@ async def grant_upload_access(
             iva_id=request.iva_id,
             box_id=request.box_id,
         )
-        # TODO: Create audit record?
+        # TODO: Create audit record
 
     async def get_upload_box_files(
         self,
@@ -184,7 +186,10 @@ async def get_upload_box_files(
             UCSCallError: if there's a problem querying the UCS.
         """
         # Verify access
-        upload_box = await self._box_dao.get_by_id(box_id)
+        try:
+            upload_box = await self._box_dao.get_by_id(box_id)
+        except ResourceNotFoundError as err:
+            raise self.BoxNotFoundError(box_id=box_id) from err
 
         is_data_steward = "data_steward" in auth_context.roles
         user_id = UUID(auth_context.id)
@@ -224,7 +229,7 @@ async def upsert_file_upload_box(self, file_upload_box: FileUploadBox) -> None:
 
             # Conditionally update data
             if updated_model.model_dump() != research_data_upload_box.model_dump():
-                await self._box_dao.update(research_data_upload_box)
+                await self._box_dao.update(updated_model)
         except NoHitsFoundError:
             # This might happen during initial creation - ignore
             log.info(

src/uos/inject.py

@@ -102,7 +102,7 @@ async def prepare_core(
         yield UploadOrchestrator(
             box_dao=box_dao,
             audit_repository=audit_repository,
-            claims_client=claims_client,
+            access_client=claims_client,
             ucs_client=ucs_client,
         )
 

src/uos/ports/inbound/orchestrator.py

@@ -22,7 +22,6 @@
 from pydantic import UUID4
 
 from uos.core.models import (
-    CreateUploadBoxRequest,
     FileUploadBox,
     GrantAccessRequest,
     ResearchDataUploadBox,
@@ -46,7 +45,9 @@ def __init__(self, *, box_id: UUID4):
     @abstractmethod
     async def create_research_data_upload_box(
         self,
-        request: CreateUploadBoxRequest,
+        title: str,
+        description: str,
+        storage_alias: str,
         user_id: UUID4,
     ) -> UUID4:
         """Create a new research data upload box.

tests/fixtures/in_mem_dao.py

@@ -0,0 +1,131 @@
+# Copyright 2021 - 2025 Universität Tübingen, DKFZ, EMBL, and Universität zu Köln
+# for the German Human Genome-Phenome Archive (GHGA)
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A dummy DAO"""
+
+from collections.abc import AsyncIterator, Mapping
+from copy import deepcopy
+from typing import Any, TypeVar
+from unittest.mock import AsyncMock, Mock
+
+from hexkit.custom_types import ID
+from hexkit.protocols.dao import (
+    MultipleHitsFoundError,
+    NoHitsFoundError,
+    ResourceAlreadyExistsError,
+    ResourceNotFoundError,
+)
+from pydantic import BaseModel
+
+from uos.core.models import ResearchDataUploadBox
+
+DTO = TypeVar("DTO", bound=BaseModel)
+
+
+class BaseInMemDao[DTO: BaseModel]:
+    """Base class for dummy DAOs with proper typing and in-memory storage"""
+
+    _id_field: str
+    publish_pending = AsyncMock()
+    republish = AsyncMock()
+    with_transaction = Mock()
+
+    def __init__(self) -> None:
+        self.resources: list[DTO] = []
+
+    @property
+    def latest(self) -> DTO:
+        """Return the most recently inserted resource"""
+        return deepcopy(self.resources[-1])
+
+    async def get_by_id(self, id_: ID) -> DTO:
+        """Get the resource via ID."""
+        for resource in self.resources:
+            if id_ == getattr(resource, self._id_field):
+                return deepcopy(resource)
+        raise ResourceNotFoundError(id_=id_)
+
+    async def find_one(self, *, mapping: Mapping[str, Any]) -> DTO:
+        """Find the resource that matches the specified mapping."""
+        hits = self.find_all(mapping=mapping)
+        try:
+            dto = await hits.__anext__()
+        except StopAsyncIteration as error:
+            raise NoHitsFoundError(mapping=mapping) from error
+
+        try:
+            _ = await hits.__anext__()
+        except StopAsyncIteration:
+            # This is expected:
+            return dto
+
+        raise MultipleHitsFoundError(mapping=mapping)
+
+    async def find_all(self, *, mapping: Mapping[str, Any]) -> AsyncIterator[DTO]:
+        """Find all resources that match the specified mapping."""
+        for resource in self.resources:
+            if all([getattr(resource, k) == v for k, v in mapping.items()]):
+                yield deepcopy(resource)
+
+    async def insert(self, dto: DTO) -> None:
+        """Insert a resource"""
+        dto_id = getattr(dto, self._id_field)
+        for resource in self.resources:
+            if getattr(resource, self._id_field) == dto_id:
+                raise ResourceAlreadyExistsError(id_=dto_id)
+        self.resources.append(deepcopy(dto))
+
+    async def update(self, dto: DTO) -> None:
+        """Update a resource"""
+        for i, resource in enumerate(self.resources):
+            if getattr(resource, self._id_field) == getattr(dto, self._id_field):
+                self.resources[i] = deepcopy(dto)
+                break
+        else:
+            raise ResourceNotFoundError(id_=getattr(dto, self._id_field))
+
+    async def delete(self, id_: ID) -> None:
+        """Delete a resource by ID"""
+        for i, resource in enumerate(self.resources):
+            if getattr(resource, self._id_field) == id_:
+                del self.resources[i]
+                break
+        else:
+            raise ResourceNotFoundError(id_=id_)
+
+    async def upsert(self, dto: DTO) -> None:
+        """Upsert a resource"""
+        for i, resource in enumerate(self.resources):
+            if getattr(resource, self._id_field) == getattr(dto, self._id_field):
+                self.resources[i] = deepcopy(dto)
+                break
+        else:
+            self.resources.append(deepcopy(dto))
+
+
+def get_dao[DTO: BaseModel](
+    *, dto_model: type[DTO], id_field: str
+) -> type[BaseInMemDao[DTO]]:
+    """Produce a dummy DAO for the given DTO model and id field"""
+
+    class DummyDao(BaseInMemDao[DTO]):
+        """Dummy dao that stores data in memory"""
+
+        _id_field: str = id_field
+
+    return DummyDao
+
+
+InMemBoxDao = get_dao(dto_model=ResearchDataUploadBox, id_field="id")