brian/visual-semiotic-ai-analysis
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

added infrastructure code #1

Merged
brian merged 4 commits from add-infrastructure into main 2025-09-18 13:23:29 +02:00
Conversation 1 Commits 4 Files Changed 34
+3390 -1
34 changed files with 3390 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 0ba860b892 - Show all commits
.gitea/workflows/code-quality.yml
+43
View File
@@ -0,0 +1,43 @@
name: Code Quality Pipeline
on:
push:
branches:
- main
pull_request:
jobs:
code-quality:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: ${{ inputs.python-version }}
- name: Install uv
run: pip install uv
- name: Install dependencies
env:
UV_LINK_MODE: copy
run: uv sync --all-extras
- name: Type check with mypy
run: uv run mypy .
- name: Ruff lint
run: uv run ruff check .
- name: Ruff format
run: uv run ruff format --check .
- name: Pyupgrade check
run: uv run pyupgrade --py312-plus $(git ls-files '*.py') && git diff --exit-code
- name: Prettier format
run: |
uv run prettier --check .
.gitea/workflows/test.yml
+47
View File
@@ -0,0 +1,47 @@
name: Test Python Package
on:
push:
branches:
- main
pull_request:
jobs:
test:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: ${{ inputs.python-version }}
- name: Install uv
run: pip install uv
- name: Install dependencies
env:
UV_LINK_MODE: copy
run: uv sync --all-extras
- name: Run pytest
env:
PYTHONPATH: .
run: uv run pytest --cov=python_repositories --cov-report=term-missing > coverage.txt
- name: Post coverage summary to PR
env:
API_URL: ${{ vars.API_URL }}
REPO_OWNER: ${{ github.repository_owner }}
REPO_NAME: ${{ github.event.repository.name }}
PR_NUMBER: ${{ github.event.pull_request.number }}
CI_RUNNER_TOKEN: ${{ secrets.CI_RUNNER_TOKEN }}
run: |
COVERAGE=$(cat coverage.txt)
COMMENT_BODY="**Test Coverage Report:**\n\`\`\`\n$COVERAGE\n\`\`\`"
curl -s -X POST "$API_URL/repos/$REPO_OWNER/$REPO_NAME/issues/$PR_NUMBER/comments" \
-H "Authorization: token $CI_RUNNER_TOKEN" \
-H "Content-Type: application/json" \
-d "{\"body\": \"$COMMENT_BODY\"}"
.pre-commit-config.yaml
+40
View File
@@ -0,0 +1,40 @@
repos:
# General repository hygiene hooks
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: check-added-large-files
- id: debug-statements
- id: name-tests-test
- id: check-merge-conflict
# Python linting and formatting with Ruff
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.12.8
hooks:
- id: ruff-check
args: [--fix]
- id: ruff-format
# Static type checking with mypy
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.8.0
hooks:
- id: mypy
# Python syntax modernization with pyupgrade
- repo: https://github.com/asottile/pyupgrade
rev: v3.20.0
hooks:
- id: pyupgrade
args: ["--py311-plus"]
# Formatting for Markdown, JSON, and YAML with Prettier
- repo: https://github.com/pre-commit/mirrors-prettier
rev: v3.1.0
hooks:
- id: prettier
files: "\\.(md|json|yaml|yml)$"
.python-version
+1
View File
@@ -0,0 +1 @@
3.12
README.md
-1
View File
@@ -1,2 +1 @@
# visual-semiotic-ai-analysis
data_store/__init__.py
View File
data_store/dto/__init__.py
+20
View File
@@ -0,0 +1,20 @@
"""
DTO package for visual-semiotic-ai-analysis.
Exposes core data transfer objects and enums for use throughout the project.
"""
from .angle_enum import AngleEnum
from .annotation import Annotation
from .contact_enum import ContactEnum
from .distance_enum import DistanceEnum
from .job import Job
from .point_of_view_enum import PointOfViewEnum
__all__ = [
"AngleEnum",
"Annotation",
"ContactEnum",
"DistanceEnum",
"Job",
"PointOfViewEnum",
]
data_store/dto/angle_enum.py
+11
View File
@@ -0,0 +1,11 @@
"""Definition of AngleEnum DTO."""
from enum import StrEnum
class AngleEnum(StrEnum):
"""Enumeration for Angle values."""
HIGH = "high"
EYE_LEVEL = "eye-level"
LOW = "low"
data_store/dto/annotation.py
+16
View File
@@ -0,0 +1,16 @@
"""Definition of Annotation DTO."""
from .type_checking_base_model import TypeCheckingBaseModel
from .distance_enum import DistanceEnum
from .angle_enum import AngleEnum
from .point_of_view_enum import PointOfViewEnum
from .contact_enum import ContactEnum
class Annotation(TypeCheckingBaseModel):
"""Data Transfer Object representing an Annotation."""
distance: DistanceEnum
angle: AngleEnum
point_of_view: PointOfViewEnum
contact: ContactEnum
data_store/dto/contact_enum.py
+10
View File
@@ -0,0 +1,10 @@
"""Definition of ContactEnum DTO."""
from enum import StrEnum
class ContactEnum(StrEnum):
"""Enumeration for Contact."""
OFFER = "offer"
DEMAND = "demand"
data_store/dto/distance_enum.py
+11
View File
@@ -0,0 +1,11 @@
"""Definition of DistanceEnum DTO."""
from enum import StrEnum
class DistanceEnum(StrEnum):
"""Enumeration for Distance values."""
CLOSE = "close"
MEDIUM = "medium"
LONG = "long"
data_store/dto/job.py
+21
View File
@@ -0,0 +1,21 @@
"""Definition of Job DTO."""
from uuid import uuid4, UUID
from pydantic import Field
from PIL import Image
from .type_checking_base_model import TypeCheckingBaseModel
from .annotation import Annotation
class Job(TypeCheckingBaseModel):
"""Data Transfer Object representing a Job."""
class Config:
"""Pydantic configuration for Job DTO."""
arbitrary_types_allowed = True
job_id: UUID = Field(default_factory=uuid4)
image: Image.Image
annotation: Annotation
data_store/dto/point_of_view_enum.py
+10
View File
@@ -0,0 +1,10 @@
"""Definition of PointOfViewEnum DTO."""
from enum import StrEnum
class PointOfViewEnum(StrEnum):
"""Enumeration for Point of View."""
FRONTAL = "frontal"
OBLIQUE = "oblique"
data_store/dto/type_checking_base_model.py
+12
View File
@@ -0,0 +1,12 @@
"""Definition of class_base class."""
from pydantic import BaseModel, ConfigDict
class TypeCheckingBaseModel(BaseModel):
"""BaseModel with added type checking on input types."""
model_config = ConfigDict(
validate_assignment=True, # ensure that dynamic validations adhere to defined types
frozen=True, # ensure data immutability
)
data_store/interfaces/__init__.py
+14
View File
@@ -0,0 +1,14 @@
"""
Interfaces package for visual-semiotic-ai-analysis.
Exposes repository interfaces for use throughout the project.
"""
from .annotation_repository_interface import AnnotationRepositoryInterface
from .job_repository_interface import JobRepositoryInterface
from .image_repository_interface import ImageRepositoryInterface
__all__ = [
"JobRepositoryInterface",
"ImageRepositoryInterface",
"AnnotationRepositoryInterface",
]
data_store/interfaces/annotation_repository_interface.py
+29
View File
@@ -0,0 +1,29 @@
"""Definition of AnnotationRepositoryInterface."""
from uuid import UUID
from abc import ABC, abstractmethod
from data_store.dto import Annotation
class AnnotationRepositoryInterface(ABC):
"""AnnotationRepository interface."""
@abstractmethod
def get(self, job_id: UUID) -> Annotation | None:
"""Retrieve an Annotation by its associated Job ID."""
raise NotImplementedError()
@abstractmethod
def put(self, annotation: Annotation, job_id: UUID) -> None:
"""Update or create an Annotation."""
raise NotImplementedError()
@abstractmethod
def delete(self, job_id: UUID) -> None:
"""Delete an Annotation by its associated Job ID."""
raise NotImplementedError()
@abstractmethod
def list_all(self) -> list[UUID]:
"""List all Job IDs with associated Annotations in the repository."""
raise NotImplementedError()
data_store/interfaces/image_repository_interface.py
+29
View File
@@ -0,0 +1,29 @@
"""Definition of ImageRepositoryInterface"""
from uuid import UUID
from abc import ABC, abstractmethod
from PIL import Image
class ImageRepositoryInterface(ABC):
"""ImageRepository interface."""
@abstractmethod
def get(self, image_id: UUID) -> Image.Image | None:
"""Retrieve an Image by its ID."""
raise NotImplementedError()
@abstractmethod
def put(self, image: Image.Image, image_id: UUID) -> None:
"""Update or create an Image."""
raise NotImplementedError()
@abstractmethod
def delete(self, image_id: UUID) -> None:
"""Delete an Image by its ID."""
raise NotImplementedError()
@abstractmethod
def list_all(self) -> list[UUID]:
"""List all image IDs in the repository."""
raise NotImplementedError()
data_store/interfaces/job_repository_interface.py
+30
View File
@@ -0,0 +1,30 @@
"""Definition of JobRepositoryInterface."""
from uuid import UUID
from abc import ABC, abstractmethod
from data_store.dto import Job
class JobRepositoryInterface(ABC):
"""JobRepository interface."""
@abstractmethod
def get(self, job_id: UUID) -> Job | None:
"""Retrieve a Job by its ID."""
raise NotImplementedError()
@abstractmethod
def put(self, job: Job) -> None:
"""Update or create a Job."""
raise NotImplementedError()
@abstractmethod
def delete(self, job_id: UUID) -> None:
"""Delete a Job by its ID."""
raise NotImplementedError()
@abstractmethod
def list_all(self) -> list[UUID]:
"""List all Job IDs in the repository."""
raise NotImplementedError()
data_store/repositories/__init__.py
+14
View File
@@ -0,0 +1,14 @@
"""
Repositories package for visual-semiotic-ai-analysis.
Exposes repository implementations for use throughout the project.
"""
from .annotation_repository import AnnotationRepository
from .image_repository import ImageRepository
from .job_repository import JobRepository
__all__ = [
"AnnotationRepository",
"ImageRepository",
"JobRepository",
]
data_store/repositories/annotation_repository.py
+106
View File
@@ -0,0 +1,106 @@
"""Definition of AnnotationRepository."""
from __future__ import annotations
from uuid import UUID
import json
from pathlib import Path
from io import BytesIO
from python_repositories.adapters import MinioAdapter
from data_store.interfaces import AnnotationRepositoryInterface
from data_store.dto import (
Annotation,
DistanceEnum,
AngleEnum,
PointOfViewEnum,
ContactEnum,
)
class AnnotationRepository(MinioAdapter, AnnotationRepositoryInterface):
"""AnnotationRepository implementation using MinIO as the backend."""
encoding: str = "utf-8"
data_format: str = "json"
folder_name: str = "annotations"
@classmethod
def _object_name(cls, job_id: UUID) -> str:
"""Generate the object name for a given job ID."""
# Check input
if not isinstance(job_id, UUID):
raise ValueError("job_id must be a valid UUID.")
# Return object name
return f"{cls.folder_name}/{job_id}.{cls.data_format}"
def __enter__(self) -> AnnotationRepository:
"""Enter the runtime context related to this object."""
super().__enter__()
return self
def get(self, job_id: UUID) -> Annotation | None:
"""Retrieve an Annotation by its job ID."""
# Check input
if not isinstance(job_id, UUID):
raise ValueError("job_id must be a valid UUID.")
# Get object
object_name = self._object_name(job_id)
buffer = self._get(object_name)
if buffer is None:
return None
# Convert buffer to Annotation
try:
data_str = buffer.read().decode(self.encoding)
data = json.loads(data_str)
annotation = Annotation(
distance=DistanceEnum(data["distance"]),
angle=AngleEnum(data["angle"]),
point_of_view=PointOfViewEnum(data["point_of_view"]),
contact=ContactEnum(data["contact"]),
)
except (KeyError, ValueError) as e:
self.logger.error(
"Failed to parse Annotation from data",
object_name=object_name,
error=str(e),
)
return None
return annotation
def put(self, annotation: Annotation, job_id: UUID) -> None:
"""Update or create an Annotation."""
# Check inputs
if not isinstance(annotation, Annotation):
raise ValueError("annotation must be an Annotation instance.")
if not isinstance(job_id, UUID):
raise ValueError("job_id must be a valid UUID.")
# Convert Annotation to json
data_json = annotation.model_dump(mode="json")
data_bytes = json.dumps(data_json).encode(self.encoding)
buffer = BytesIO(data_bytes)
# Put object
object_name = self._object_name(job_id)
self._put(object_name, buffer)
def delete(self, job_id: UUID) -> None:
"""Delete an Annotation by its job ID."""
# Check input
if not isinstance(job_id, UUID):
raise ValueError("job_id must be a valid UUID.")
# Delete object
object_name = self._object_name(job_id)
self._delete(object_name)
def list_all(self) -> list[UUID]:
"""List all job IDs in the repository."""
# List objects in the bucket
objects = self._list_objects(prefix=self.folder_name + "/")
# Stop early if no objects
if len(objects) == 0:
return []
# Remove folder prefix and file extension
object_stems = [Path(obj).stem for obj in objects]
# Convert to list of UUIDs
job_ids = [UUID(name) for name in object_stems]
return job_ids
data_store/repositories/image_repository.py
+82
View File
@@ -0,0 +1,82 @@
"""Definition of ImageRepository class."""
from __future__ import annotations
from pathlib import Path
from io import BytesIO
from uuid import UUID
from PIL import Image
from python_repositories.adapters import MinioAdapter
from data_store.interfaces import ImageRepositoryInterface
class ImageRepository(MinioAdapter, ImageRepositoryInterface):
"""ImageRepository implementation using MinIO as the backend."""
image_format: str = "PNG"
folder_name: str = "images"
@classmethod
def _object_name(cls, image_id: UUID) -> str:
"""Generate the object name for a given image ID."""
# Check input
if not isinstance(image_id, UUID):
raise ValueError("image_id must be a valid UUID.")
# Return object name
return f"{cls.folder_name}/{image_id}.{cls.image_format.lower()}"
def __enter__(self) -> ImageRepository:
"""Enter the runtime context related to this object."""
super().__enter__()
return self
def get(self, image_id: UUID) -> Image.Image | None:
"""Retrieve an Image by its ID."""
# Check input
if not isinstance(image_id, UUID):
raise ValueError("image_id must be a valid UUID.")
# Get object
object_name = self._object_name(image_id)
buffer = self._get(object_name)
if buffer is None:
return None
# Convert bytes back to PIL Image
image: Image.Image = Image.open(buffer, formats=[self.image_format])
return image
def put(self, image: Image.Image, image_id: UUID) -> None:
"""Update or create an Image."""
# Check inputs
if not isinstance(image, Image.Image):
raise ValueError("image must be a PIL Image instance.")
if not isinstance(image_id, UUID):
raise ValueError("image_id must be a valid UUID.")
# Convert image to bytes
buffer = BytesIO()
image.save(buffer, self.image_format)
# Put object
object_name = self._object_name(image_id)
self._put(object_name, buffer)
def delete(self, image_id: UUID) -> None:
"""Delete an Image by its ID."""
# Check input
if not isinstance(image_id, UUID):
raise ValueError("image_id must be a valid UUID.")
# Delete object
object_name = self._object_name(image_id)
self._delete(object_name)
def list_all(self) -> list[UUID]:
"""List all image IDs in the repository."""
# List objects in the bucket
objects = self._list_objects(prefix=self.folder_name + "/")
# Stop early if no objects
if len(objects) == 0:
return []
# Remove folder prefix and file extension
object_stems = [Path(obj).stem for obj in objects]
# Convert to list of UUIDs
image_ids = [UUID(name) for name in object_stems]
return image_ids
data_store/repositories/job_repository.py
+78
View File
@@ -0,0 +1,78 @@
"""Definition of JobRepository class."""
from __future__ import annotations
from uuid import UUID
from python_repositories.adapters import MinioAdapter
from data_store.interfaces import JobRepositoryInterface
from data_store.repositories import AnnotationRepository, ImageRepository
from data_store.dto import Job
class JobRepository(MinioAdapter, JobRepositoryInterface):
"""JobRepository implementation using MinIO as the backend."""
def __enter__(self) -> JobRepository:
"""Enter the runtime context related to this object."""
super().__enter__()
return self
def get(self, job_id: UUID) -> Job | None:
"""Retrieve a Job by its ID."""
# Check input
if not isinstance(job_id, UUID):
raise ValueError("job_id must be a valid UUID.")
# Get image
with ImageRepository() as image_repo:
image = image_repo.get(job_id)
if image is None:
self.logger.warning(f"Image for job ID {job_id} not found.")
return None
# Get annotation
with AnnotationRepository() as annotation_repo:
annotation = annotation_repo.get(job_id)
if annotation is None:
self.logger.warning(f"Annotation for job ID {job_id} not found.")
return None
# Create Job DTO
job = Job(
job_id=job_id,
image=image,
annotation=annotation,
)
return job
def put(self, job: Job) -> None:
"""Update or create a Job."""
# Check input
if not isinstance(job, Job):
raise ValueError("job must be a Job instance.")
# Store image
with ImageRepository() as image_repo:
image_repo.put(job.image, job.job_id)
# Store annotation
with AnnotationRepository() as annotation_repo:
annotation_repo.put(job.annotation, job.job_id)
def delete(self, job_id: UUID) -> None:
"""Delete a Job by its ID."""
# Check input
if not isinstance(job_id, UUID):
raise ValueError("job_id must be a valid UUID.")
# Delete image
with ImageRepository() as image_repo:
image_repo.delete(job_id)
# Delete annotation
with AnnotationRepository() as annotation_repo:
annotation_repo.delete(job_id)
def list_all(self) -> list[UUID]:
"""List all Job IDs."""
# List image IDs
with ImageRepository() as image_repo:
image_ids = set(image_repo.list_all())
# List annotation IDs
with AnnotationRepository() as annotation_repo:
annotation_ids = set(annotation_repo.list_all())
# Find intersection of IDs
job_ids = list(image_ids.intersection(annotation_ids))
return job_ids
pyproject.toml
+84
View File
@@ -0,0 +1,84 @@
[project]
name = "visual-semiotic-ai-analysis"
version = "0.1.0"
description = "Add your description here"
authors = [
{ name = "Brian Bjarke Jensen", email = "[email protected]" }
]
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
classifiers = [
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
]
dependencies = [
"pandas>=2.3.2",
"pillow>=11.3.0",
"pydantic>=2.11.9",
"python-repositories[minio]>=0.2.0",
"python-utils>=0.1.1",
]
[build-system]
requires = ["uv_build"]
build-backend = "uv_build"
[tool.uv.build-backend]
module-name = [
"data_store",
]
module-root = "."
[tool.uv.sources]
python-repositories = { index = "gitea" }
python-utils = { index = "gitea" }
[[tool.uv.index]]
name = "threadripper-proxpi-cache"
url = "http://10.0.0.2:5001/index/"
default = true
[[tool.uv.index]]
name = "gitea"
url = "https://gitea.gt-proj.com/api/packages/brian/pypi/simple/"
explicit = true
[tool.mypy]
python_version = "3.12"
warn_return_any = true # nudge to use stricter types
warn_unused_configs = true # nudge to remove unused configs
disallow_untyped_defs = true # disallow untyped function definitions
disallow_incomplete_defs = true # ensure all parts of a function has type annotation
check_untyped_defs = true # dont skip untyped functions
disallow_untyped_decorators = false # allow untyped decorators to keep code more readable
no_implicit_optional = true # disallow implicit optional types leading to None-mess
warn_redundant_casts = true # nudge to use explicit type casts
warn_unused_ignores = true # nudge to remove unused ignores
warn_no_return = true # catch missing return statements
warn_unreachable = true # catch unreachable code
show_error_codes = true # show error codes in output
explicit_package_bases = true # reduce risk of import confusion
namespace_packages = true # enable namespace packages
[[tool.mypy.overrides]]
module = "testcontainers.*"
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = "minio.*"
ignore_missing_imports = true
[dependency-groups]
dev = [
"mypy>=1.18.1",
"pandas-stubs>=2.3.2.250827",
"pre-commit>=4.3.0",
"pytest>=8.4.2",
"pytest-cov>=7.0.0",
"pyupgrade>=3.20.0",
"ruff>=0.13.0",
"safety>=3.2.11",
"testcontainers>=4.13.0",
]
tests/integration/annotation_repository_interface_test.py
+110
View File
@@ -0,0 +1,110 @@
"""Integration tests for AnnotationRepositoryInterface."""
from uuid import UUID
import pytest
from data_store.interfaces import AnnotationRepositoryInterface
from data_store.dto import Annotation
def test_get_raises_not_implemented_if_not_overwritten(
job_id_list: list[UUID],
) -> None:
"""Test that calling get on a subclass without get implemented raises NotImplementedError."""
class Incomplete(AnnotationRepositoryInterface):
"""A class missing the get method."""
def get(self, job_id: UUID) -> Annotation | None:
return super().get(job_id) # type: ignore
def put(self, annotation: Annotation, job_id: UUID) -> None:
return None
def delete(self, job_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.get(job_id_list[0])
def test_put_raises_not_implemented_if_not_overwritten(
annotation: Annotation,
job_id_list: list[UUID],
) -> None:
"""Test that calling put on a subclass without put implemented raises NotImplementedError."""
class Incomplete(AnnotationRepositoryInterface):
"""A class missing the put method."""
def get(self, job_id: UUID) -> Annotation | None:
return None
def put(self, annotation: Annotation, job_id: UUID) -> None:
return super().put(annotation, job_id) # type: ignore
def delete(self, job_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.put(annotation, job_id_list[0])
def test_delete_raises_not_implemented_if_not_overwritten(
job_id_list: list[UUID],
) -> None:
"""Test that calling delete on a subclass without delete implemented raises NotImplementedError."""
class Incomplete(AnnotationRepositoryInterface):
"""A class missing the delete method."""
def get(self, job_id: UUID) -> Annotation | None:
return None
def put(self, annotation: Annotation, job_id: UUID) -> None:
return None
def delete(self, job_id: UUID) -> None:
return super().delete(job_id) # type: ignore
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.delete(job_id_list[0])
def test_list_all_raises_not_implemented_if_not_overwritten() -> None:
"""Test that calling list_all on a subclass without list_all implemented raises NotImplementedError."""
class Incomplete(AnnotationRepositoryInterface):
"""A class missing the list_all method."""
def get(self, job_id: UUID) -> Annotation | None:
return None
def put(self, annotation: Annotation, job_id: UUID) -> None:
return None
def delete(self, job_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return super().list_all() # type: ignore
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.list_all()
# Allows local debugging by running file as script
if __name__ == "__main__":
pytest.main(["-s", "-v", __file__])
tests/integration/annotation_repository_test.py
+86
View File
@@ -0,0 +1,86 @@
"""Integration tests for AnnotationRepository."""
from uuid import UUID
import pytest
from data_store.repositories.annotation_repository import AnnotationRepository
from data_store.dto import (
Annotation,
)
def test_get_annotation(
annotation_in_minio: tuple[UUID, Annotation],
annotation_repository: AnnotationRepository,
) -> None:
"""Test retrieving an annotation from AnnotationRepository."""
# Arrange
job_id, expected_annotation = annotation_in_minio
# Act
retrieved_annotation = annotation_repository.get(job_id)
# Assert
assert retrieved_annotation is not None
assert isinstance(retrieved_annotation, Annotation)
assert retrieved_annotation == expected_annotation
def test_put_annotation(
annotation_repository: AnnotationRepository,
annotation: Annotation,
job_id_list: list[UUID],
) -> None:
"""Test storing an annotation in AnnotationRepository."""
# Arrange
job_id = job_id_list[0]
assert annotation_repository.get(job_id) is None
# Act
annotation_repository.put(annotation, job_id)
# Assert
received_annotation = annotation_repository.get(job_id)
assert received_annotation is not None
assert isinstance(received_annotation, Annotation)
assert received_annotation == annotation
def test_delete_annotation(
annotation_in_minio: tuple[UUID, Annotation],
annotation_repository: AnnotationRepository,
) -> None:
"""Test deleting an annotation from AnnotationRepository."""
# Arrange
job_id, _ = annotation_in_minio
assert annotation_repository.get(job_id) is not None
# Act
annotation_repository.delete(job_id)
# Assert
assert annotation_repository.get(job_id) is None
def test_list_all_annotations(
annotation_in_minio: tuple[UUID, Annotation],
annotation_repository: AnnotationRepository,
) -> None:
"""Test listing all annotations in AnnotationRepository."""
# Arrange
job_id, _ = annotation_in_minio
# Act
annotation_ids = annotation_repository.list_all()
# Assert
assert isinstance(annotation_ids, list)
assert all(isinstance(i, UUID) for i in annotation_ids)
assert job_id in annotation_ids
def test_list_all_annotations_empty(
annotation_repository: AnnotationRepository,
) -> None:
"""Test listing all annotations in an empty AnnotationRepository."""
# Act
annotation_ids = annotation_repository.list_all()
# Assert
assert isinstance(annotation_ids, list)
assert len(annotation_ids) == 0
# Allows local debugging by running file as script
if __name__ == "__main__":
pytest.main(["-s", "-v", __file__])
tests/integration/conftest.py
+340
View File
@@ -0,0 +1,340 @@
"""Integration test configuration."""
import logging
import os
import shutil
import tempfile
from pathlib import Path
import json
from collections.abc import Generator
from uuid import UUID
from io import BytesIO
import pandas as pd
from PIL import Image
import pytest
import structlog
from minio import Minio, S3Error
from testcontainers.minio import MinioContainer
from python_repositories.adapters import MinioAdapter
from data_store.repositories import (
ImageRepository,
AnnotationRepository,
JobRepository,
)
from data_store.dto import (
Annotation,
DistanceEnum,
AngleEnum,
PointOfViewEnum,
ContactEnum,
Job,
)
# Last Docker image with a useful WebUI
MINIO_DOCKER_IMAGE = "minio/minio:RELEASE.2023-03-20T20-16-18Z"
@pytest.fixture(scope="session", autouse=True)
def configure_logging() -> None:
"""Configure logging for the test session."""
# Configure structlog
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer(),
],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
# Set up basic logging configuration
logging.basicConfig(level=logging.ERROR)
@pytest.fixture(scope="session")
def minio_container() -> Generator[dict[str, str]]:
"""Set up MinIO container and provide credentials for testing."""
# Set variables
access_key = "minioadmin"
secret_key = "minioadmin"
bucket_name = "test-bucket"
# Start container
container = MinioContainer(
image=MINIO_DOCKER_IMAGE,
access_key=access_key,
secret_key=secret_key,
).with_bind_ports( # expose webUI on 9001
container=9001,
host=9001,
)
container.start()
# Build environment variables dictionary
minio_host = container.get_container_host_ip()
minio_port = container.get_exposed_port(9000)
minio_endpoint = f"{minio_host}:{minio_port}"
env_vars = {
"MINIO_ENDPOINT": minio_endpoint,
"MINIO_ACCESS_KEY": access_key,
"MINIO_SECRET_KEY": secret_key,
"MINIO_BUCKET": bucket_name,
}
yield env_vars
# Stop container
container.stop()
@pytest.fixture(scope="session", autouse=True)
def set_environment_variables(
minio_container: dict[str, str],
) -> Generator[dict[str, str]]:
"""Set environment variables needed for tests."""
# Build environment variables dictionary
env_vars = {}
env_vars.update(minio_container)
# Set environment variables
for key, value in env_vars.items():
os.environ[key] = value
yield env_vars
# Cleanup
for key in env_vars:
_ = os.environ.pop(key, default=None)
@pytest.fixture(scope="session")
def job_id_list() -> Generator[list[UUID]]:
"""Provide a list of job IDs for testing."""
yield [
UUID("d358cbba-75be-4c38-becd-7abf338ca941"),
UUID("56d20d25-b733-4ed3-880c-732d7257dcaa"),
UUID("8019292f-8353-43d7-a189-076c6630de7b"),
]
@pytest.fixture(scope="session")
def csv_data_file(
job_id_list: list[UUID],
) -> Generator[Path]:
"""Create a CSV data file for testing."""
# Create csv content
csv_data = pd.DataFrame(
{
"job_id": job_id_list,
"distance": [
"close",
"long",
"wrong",
],
"angle": ["high", "low", "wrong"],
"point_of_view": ["frontal", "oblique", "wrong"],
"contact": ["offer", "demand", "wrong"],
}
)
# Create data directory and file path
data_dir = Path(tempfile.mkdtemp())
data_path = data_dir / "annotations.csv"
# Write CSV content to file
csv_data.to_csv(data_path, header=True, index=False)
yield data_path
# Cleanup
shutil.rmtree(data_dir)
@pytest.fixture(scope="session")
def raw_minio_client(
set_environment_variables: dict[str, str],
) -> Generator[Minio]:
"""Provide a MinIO client for testing."""
# Create MinIO client
client = Minio(
endpoint=set_environment_variables["MINIO_ENDPOINT"],
access_key=set_environment_variables["MINIO_ACCESS_KEY"],
secret_key=set_environment_variables["MINIO_SECRET_KEY"],
secure=False,
)
# Create bucket if it doesn't exist
bucket_name = set_environment_variables["MINIO_BUCKET"]
try:
if not client.bucket_exists(bucket_name):
client.make_bucket(bucket_name)
except S3Error as e:
raise RuntimeError(f"Failed to create or access bucket: {e}")
yield client
# No explicit cleanup needed for MinIO client
@pytest.fixture(scope="session")
def image_repository() -> Generator[ImageRepository]:
"""Provide an ImageRepository instance for testing."""
# Instantiate and connect repository
repo = ImageRepository()
repo.connect()
yield repo
# Cleanup
repo.disconnect()
@pytest.fixture(scope="session")
def image() -> Generator[Image.Image]:
"""Provide a sample image for testing."""
img = Image.new("RGB", (10, 10))
yield img
@pytest.fixture(scope="function")
def image_in_minio(
set_environment_variables: dict[str, str],
raw_minio_client: Minio,
image: Image.Image,
job_id_list: list[UUID],
) -> Generator[tuple[UUID, Image.Image]]:
"""Upload a sample image to MinIO and provide its ID for testing."""
# Prepare object name and buffer
bucket_name = set_environment_variables["MINIO_BUCKET"]
image_id = job_id_list[0]
object_name = ImageRepository._object_name(image_id)
buffer = BytesIO()
image.save(buffer, ImageRepository.image_format)
num_bytes = buffer.getbuffer().nbytes
buffer.seek(0)
# Upload image to MinIO
raw_minio_client.put_object(
bucket_name=bucket_name,
object_name=object_name,
data=buffer,
length=num_bytes,
part_size=MinioAdapter.chunk_size,
)
buffer.close()
yield image_id, image
# Cleanup: remove image from MinIO
try:
raw_minio_client.remove_object(
bucket_name=bucket_name,
object_name=object_name,
)
except S3Error as e:
raise RuntimeError(f"Failed to remove object from bucket: {e}")
@pytest.fixture(scope="session")
def annotation_repository() -> Generator[AnnotationRepository]:
"""Provide an AnnotationRepository instance for testing."""
# Instantiate and connect repository
repo = AnnotationRepository()
repo.connect()
yield repo
# Cleanup
repo.disconnect()
@pytest.fixture(scope="session")
def annotation() -> Generator[Annotation]:
"""Provide a sample annotation for testing."""
annotation = Annotation(
distance=DistanceEnum.CLOSE,
angle=AngleEnum.HIGH,
point_of_view=PointOfViewEnum.FRONTAL,
contact=ContactEnum.OFFER,
)
yield annotation
@pytest.fixture(scope="function")
def annotation_in_minio(
set_environment_variables: dict[str, str],
raw_minio_client: Minio,
annotation: Annotation,
job_id_list: list[UUID],
) -> Generator[tuple[UUID, Annotation]]:
"""Upload a sample annotation to MinIO and provide its ID for testing."""
# Prepare object name and buffer
bucket_name = set_environment_variables["MINIO_BUCKET"]
job_id = job_id_list[0]
object_name = AnnotationRepository._object_name(job_id)
data_json = annotation.model_dump(mode="json")
data_bytes = json.dumps(data_json).encode(AnnotationRepository.encoding)
buffer = BytesIO(data_bytes)
num_bytes = buffer.getbuffer().nbytes
buffer.seek(0)
# Upload annotation to MinIO
raw_minio_client.put_object(
bucket_name=bucket_name,
object_name=object_name,
data=buffer,
length=num_bytes,
part_size=MinioAdapter.chunk_size,
)
buffer.close()
yield job_id, annotation
# Cleanup: remove annotation from MinIO
try:
raw_minio_client.remove_object(
bucket_name=bucket_name,
object_name=object_name,
)
except S3Error as e:
raise RuntimeError(f"Failed to remove object from bucket: {e}")
@pytest.fixture(scope="session")
def job_repository() -> Generator[JobRepository]:
"""Provide a JobRepository instance for testing."""
# Instantiate and connect repository
repo = JobRepository()
repo.connect()
yield repo
# Cleanup
repo.disconnect()
@pytest.fixture(scope="session")
def job(
job_id_list: list[UUID],
image: Image.Image,
annotation: Annotation,
) -> Generator[Job]:
"""Provide a sample job for testing."""
job = Job(
job_id=job_id_list[0],
image=image,
annotation=annotation,
)
yield job
@pytest.fixture(scope="function")
def job_in_minio(
job: Job,
image_in_minio: tuple[UUID, Image.Image],
annotation_in_minio: tuple[UUID, Annotation],
) -> Generator[Job]:
"""Provide a sample job with its components already in MinIO for testing."""
# A job consists of an image and an annotation and both must be in MinIO
# before the job can be considered to be in MinIO.
# Ensure the job_id matches those of the image and annotation
image_id, _ = image_in_minio
annotation_id, _ = annotation_in_minio
assert job.job_id == annotation_id
assert job.job_id == image_id
yield job
tests/integration/image_repository_interface_test.py
+110
View File
@@ -0,0 +1,110 @@
"""Integration tests for ImageRepositoryInterface."""
from uuid import UUID
import pytest
from PIL import Image
from data_store.interfaces import ImageRepositoryInterface
def test_get_raises_not_implemented_if_not_overwritten(
job_id_list: list[UUID],
) -> None:
"""Test that calling get on a subclass without get implemented raises NotImplementedError."""
class Incomplete(ImageRepositoryInterface):
"""A class missing the get method."""
def get(self, image_id: UUID) -> Image.Image | None:
return super().get(image_id) # type: ignore
def put(self, image: Image.Image, image_id: UUID) -> None:
return None
def delete(self, image_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.get(job_id_list[0])
def test_put_raises_not_implemented_if_not_overwritten(
image: Image.Image,
job_id_list: list[UUID],
) -> None:
"""Test that calling put on a subclass without put implemented raises NotImplementedError."""
class Incomplete(ImageRepositoryInterface):
"""A class missing the put method."""
def get(self, image_id: UUID) -> Image.Image | None:
return None
def put(self, image: Image.Image, image_id: UUID) -> None:
return super().put(image, image_id) # type: ignore
def delete(self, image_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.put(image, job_id_list[0])
def test_delete_raises_not_implemented_if_not_overwritten(
job_id_list: list[UUID],
) -> None:
"""Test that calling delete on a subclass without delete implemented raises NotImplementedError."""
class Incomplete(ImageRepositoryInterface):
"""A class missing the delete method."""
def get(self, image_id: UUID) -> Image.Image | None:
return None
def put(self, image: Image.Image, image_id: UUID) -> None:
return None
def delete(self, image_id: UUID) -> None:
return super().delete(image_id) # type: ignore
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.delete(job_id_list[0])
def test_list_all_raises_not_implemented_if_not_overwritten() -> None:
"""Test that calling list_all on a subclass without list_all implemented raises NotImplementedError."""
class Incomplete(ImageRepositoryInterface):
"""A class missing the list_all method."""
def get(self, image_id: UUID) -> Image.Image | None:
return None
def put(self, image: Image.Image, image_id: UUID) -> None:
return None
def delete(self, image_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return super().list_all() # type: ignore
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.list_all()
# Allows local debugging by running file as script
if __name__ == "__main__":
pytest.main(["-s", "-v", __file__])
tests/integration/image_repository_test.py
+95
View File
@@ -0,0 +1,95 @@
"""Integration tests for ImageRepository."""
from uuid import UUID
import pytest
from PIL import Image
from data_store.repositories.image_repository import ImageRepository
def same_image(img1: Image.Image, img2: Image.Image) -> bool:
"""Check if two images are the same based on their attributes."""
return (
img1.size == img2.size
and img1.mode == img2.mode
and list(img1.getdata()) == list(img2.getdata())
)
def test_get_image(
image_in_minio: tuple[UUID, Image.Image],
image_repository: ImageRepository,
) -> None:
"""Test retrieving an image from ImageRepository."""
# Arrange
image_id, expected_image = image_in_minio
# Act
retrieved_image = image_repository.get(image_id)
# Assert
assert retrieved_image is not None
assert isinstance(retrieved_image, Image.Image)
assert same_image(retrieved_image, expected_image)
def test_put_image(
image_repository: ImageRepository,
image: Image.Image,
job_id_list: list[UUID],
) -> None:
"""Test storing an image in ImageRepository."""
# Arrange
image_id = job_id_list[0]
assert image_repository.get(image_id) is None
# Act
image_repository.put(image, image_id)
# Assert
received_image = image_repository.get(image_id)
assert received_image is not None
assert isinstance(received_image, Image.Image)
assert received_image.size == image.size
assert received_image.mode == image.mode
assert list(received_image.getdata()) == list(image.getdata())
def test_delete_image(
image_in_minio: tuple[UUID, Image.Image],
image_repository: ImageRepository,
) -> None:
"""Test deleting an image from ImageRepository."""
# Arrange
image_id, _ = image_in_minio
assert image_repository.get(image_id) is not None
# Act
image_repository.delete(image_id)
# Assert
assert image_repository.get(image_id) is None
def test_list_all_images(
image_in_minio: tuple[UUID, Image.Image],
image_repository: ImageRepository,
) -> None:
"""Test listing all images in ImageRepository."""
# Arrange
image_id, _ = image_in_minio
# Act
image_ids = image_repository.list_all()
# Assert
assert isinstance(image_ids, list)
assert all(isinstance(i, UUID) for i in image_ids)
assert image_id in image_ids
def test_list_all_images_empty(
image_repository: ImageRepository,
) -> None:
"""Test listing all images in an empty ImageRepository."""
# Act
image_ids = image_repository.list_all()
# Assert
assert isinstance(image_ids, list)
assert len(image_ids) == 0
# Allows local debugging by running file as script
if __name__ == "__main__":
pytest.main(["-s", "-v", __file__])
tests/integration/job_repository_interface_test.py
+109
View File
@@ -0,0 +1,109 @@
"""Integration tests for the JobRepositoryInterface."""
from uuid import UUID
import pytest
from data_store.interfaces import JobRepositoryInterface
from data_store.dto import Job
def test_get_raises_not_implemented_if_not_overwritten(
job_id_list: list[UUID],
) -> None:
"""Test that calling get on a subclass without get implemented raises NotImplementedError."""
class Incomplete(JobRepositoryInterface):
"""A class missing the get method."""
def get(self, job_id: UUID) -> Job | None:
return super().get(job_id) # type: ignore
def put(self, job: Job) -> None:
return None
def delete(self, job_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.get(job_id_list[0])
def test_put_raises_not_implemented_if_not_overwritten(
job: Job,
) -> None:
"""Test that calling put on a subclass without put implemented raises NotImplementedError."""
class Incomplete(JobRepositoryInterface):
"""A class missing the put method."""
def get(self, job_id: UUID) -> Job | None:
return None
def put(self, job: Job) -> None:
return super().put(job) # type: ignore
def delete(self, job_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.put(job)
def test_delete_raises_not_implemented_if_not_overwritten(
job_id_list: list[UUID],
) -> None:
"""Test that calling delete on a subclass without delete implemented raises NotImplementedError."""
class Incomplete(JobRepositoryInterface):
"""A class missing the delete method."""
def get(self, job_id: UUID) -> Job | None:
return None
def put(self, job: Job) -> None:
return None
def delete(self, job_id: UUID) -> None:
return super().delete(job_id) # type: ignore
def list_all(self) -> list[UUID]:
return []
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.delete(job_id_list[0])
def test_list_all_raises_not_implemented_if_not_overwritten() -> None:
"""Test that calling list_all on a subclass without list_all implemented raises NotImplementedError."""
class Incomplete(JobRepositoryInterface):
"""A class missing the list_all method."""
def get(self, job_id: UUID) -> Job | None:
return None
def put(self, job: Job) -> None:
return None
def delete(self, job_id: UUID) -> None:
return None
def list_all(self) -> list[UUID]:
return super().list_all() # type: ignore
instance = Incomplete()
with pytest.raises(NotImplementedError):
instance.list_all()
# Allows local debugging by running file as script
if __name__ == "__main__":
pytest.main(["-s", "-v", __file__])
tests/integration/job_repository_test.py
+116
View File
@@ -0,0 +1,116 @@
"""Integration tests for JobRepository."""
from uuid import UUID
import pytest
from PIL import Image
from data_store.repositories.job_repository import JobRepository
from data_store.dto import Job
def same_image(img1: Image.Image, img2: Image.Image) -> bool:
"""Check if two images are the same based on their attributes."""
assert img1.size == img2.size
assert img1.mode == img2.mode
assert list(img1.getdata()) == list(img2.getdata())
return True
def same_job(job1: Job, job2: Job) -> bool:
"""Check if two Job instances are the same based on their attributes."""
assert job1.job_id == job2.job_id
assert same_image(job1.image, job2.image)
assert job1.annotation == job2.annotation
return True
def test_get_job(
job_in_minio: Job,
job_repository: JobRepository,
) -> None:
"""Test retrieving a job from JobRepository."""
# Arrange
job_id = job_in_minio.job_id
# Act
retrieved_job = job_repository.get(job_id)
# Assert
assert retrieved_job is not None
assert isinstance(retrieved_job, Job)
assert same_job(retrieved_job, job_in_minio)
def test_get_job_without_image_returns_none(
annotation_in_minio: tuple[UUID, object],
job_repository: JobRepository,
) -> None:
"""Test retrieving a job without an image returns None."""
# Arrange
job_id, _ = annotation_in_minio
# Act
retrieved_job = job_repository.get(job_id)
# Assert
assert retrieved_job is None
def test_get_job_without_annotation_returns_none(
image_in_minio: tuple[UUID, object],
job_repository: JobRepository,
) -> None:
"""Test retrieving a job without an annotation returns None."""
# Arrange
job_id, _ = image_in_minio
# Act
retrieved_job = job_repository.get(job_id)
# Assert
assert retrieved_job is None
def test_put_job(
job_repository: JobRepository,
job: Job,
) -> None:
"""Test storing a job in JobRepository."""
# Arrange
job_id = job.job_id
assert job_repository.get(job_id) is None
# Act
job_repository.put(job)
# Assert
received_job = job_repository.get(job_id)
assert received_job is not None
assert isinstance(received_job, Job)
assert same_job(received_job, job)
def test_delete_job(
job_in_minio: Job,
job_repository: JobRepository,
) -> None:
"""Test deleting a job from JobRepository."""
# Arrange
job_id = job_in_minio.job_id
assert job_repository.get(job_id) is not None
# Act
job_repository.delete(job_id)
# Assert
assert job_repository.get(job_id) is None
def test_list_all_jobs(
job_in_minio: Job,
job_repository: JobRepository,
) -> None:
"""Test listing all job IDs from JobRepository."""
# Arrange
job_id = job_in_minio.job_id
# Act
job_ids = job_repository.list_all()
# Assert
assert isinstance(job_ids, list)
assert all(isinstance(jid, UUID) for jid in job_ids)
assert job_id in job_ids
# Allows local debugging by running file as script
if __name__ == "__main__":
pytest.main(["-s", "-v", __file__])
tests/unit/test_annotation_repository.py
+128
View File
@@ -0,0 +1,128 @@
"""Unit tests for AnnotationRepository class."""
import unittest
from unittest.mock import patch, MagicMock, _patch_dict
from uuid import UUID
from data_store.repositories.annotation_repository import AnnotationRepository
from data_store.dto import (
Annotation,
DistanceEnum,
AngleEnum,
PointOfViewEnum,
ContactEnum,
)
class TestAnnotationRepository(unittest.TestCase):
"""Unit tests for AnnotationRepository class."""
patcher: _patch_dict
repo: AnnotationRepository
job_id: UUID
bad_job_id: str
annotation: Annotation
bad_annotation: dict
@classmethod
def setUpClass(cls) -> None:
"""Set up test case with a mocked MinioAdapter."""
# Patch environment variables for MinioAdapter
cls.patcher = patch.dict(
"os.environ",
{
"MINIO_ENDPOINT": "localhost:9000",
"MINIO_ACCESS_KEY": "minioadmin",
"MINIO_SECRET_KEY": "minioadmin",
"MINIO_BUCKET": "test-bucket",
},
)
cls.patcher.start()
# Prepare AnnotationRepository
cls.repo = AnnotationRepository()
cls.repo._get = MagicMock() # type: ignore[method-assign]
cls.repo._put = MagicMock() # type: ignore[method-assign]
cls.repo._delete = MagicMock() # type: ignore[method-assign]
# Prepare other test variables
cls.job_id = UUID("d87022f8-4169-4d69-8512-bbd65cc1cb23")
cls.bad_job_id = "not-a-uuid"
cls.annotation = Annotation(
distance=DistanceEnum.CLOSE,
angle=AngleEnum.HIGH,
point_of_view=PointOfViewEnum.FRONTAL,
contact=ContactEnum.OFFER,
)
cls.bad_annotation = {"not_an_annotation": "irrelevant value"}
@classmethod
def tearDownClass(cls) -> None:
# Stop patching
cls.patcher.stop()
def setUp(self) -> None:
"""Set up each test with a fresh AnnotationRepository instance."""
self.repo._get.reset_mock() # type: ignore[attr-defined]
self.repo._put.reset_mock() # type: ignore[attr-defined]
self.repo._delete.reset_mock() # type: ignore[attr-defined]
def test_should_adhere_to_interface(self) -> None:
"""Test that AnnotationRepository adheres to AnnotationRepositoryInterface."""
# Instantiation fails if interface not adhered to
_ = AnnotationRepository()
def test_should_have_encoding_set(self) -> None:
"""Test that AnnotationRepository has encoding when instantiated."""
self.assertTrue(hasattr(self.repo, "encoding"))
def test_should_have_data_format_set(self) -> None:
"""Test that AnnotationRepository has data_format when instantiated."""
self.assertTrue(hasattr(self.repo, "data_format"))
def test_should_have_folder_name_set(self) -> None:
"""Test that AnnotationRepository has folder_name when instantiated."""
self.assertTrue(hasattr(self.repo, "folder_name"))
def test_object_name_invalid_job_id_raises_value_error(self) -> None:
"""Test that _object_name() raises ValueError for invalid job_id."""
with self.assertRaises(ValueError):
_ = self.repo._object_name(self.bad_job_id) # type: ignore
def test_get_invalid_job_id_raises_value_error(self) -> None:
"""Test that get() raises ValueError for invalid job_id."""
with self.assertRaises(ValueError):
self.repo.get(self.bad_job_id) # type: ignore
def test_get_data_that_cannot_be_converted_returns_none(self) -> None:
"""Test that get() returns None if retrieved data cannot be converted to Annotation."""
# Arrange
bad_data = b'{"not":"valid annotation data"}'
mock_buffer = MagicMock()
mock_buffer.read.return_value = bad_data
self.repo._get.return_value = mock_buffer # type: ignore[attr-defined]
# Act
result = self.repo.get(self.job_id)
# Assert
self.assertIsNone(result)
self.repo._get.assert_called_once_with( # type: ignore[attr-defined]
f"{self.repo.folder_name}/{self.job_id}.{self.repo.data_format}"
)
def test_put_invalid_job_id_raises_value_error(self) -> None:
"""Test that put() raises ValueError for invalid job_id."""
with self.assertRaises(ValueError):
self.repo.put(self.annotation, self.bad_job_id) # type: ignore
def test_put_invalid_annotation_raises_value_error(self) -> None:
"""Test that put() raises ValueError for invalid annotation."""
with self.assertRaises(ValueError):
self.repo.put(self.bad_annotation, self.job_id) # type: ignore
def test_delete_invalid_job_id_raises_value_error(self) -> None:
"""Test that delete() raises ValueError for invalid job_id."""
with self.assertRaises(ValueError):
self.repo.delete(self.bad_job_id) # type: ignore
# Allows local debugging by running file as script
if __name__ == "__main__":
unittest.main()
tests/unit/test_image_repository.py
+98
View File
@@ -0,0 +1,98 @@
"""Unit tests for ImageRepository class."""
import unittest
from unittest.mock import patch, MagicMock, _patch_dict
from uuid import UUID
from PIL import Image
from data_store.repositories.image_repository import ImageRepository
class TestImageRepository(unittest.TestCase):
"""Unit tests for ImageRepository class."""
patcher: _patch_dict
repo: ImageRepository
image: Image.Image
bad_image: str
image_id: UUID
bad_image_id: str
@classmethod
def setUpClass(cls) -> None:
"""Set up test case with a mocked MinioAdapter."""
# Patch environment variables for MinioAdapter
cls.patcher = patch.dict(
"os.environ",
{
"MINIO_ENDPOINT": "localhost:9000",
"MINIO_ACCESS_KEY": "minioadmin",
"MINIO_SECRET_KEY": "minioadmin",
"MINIO_BUCKET": "test-bucket",
},
)
cls.patcher.start()
# Prepare ImageRepository with mocked MinioAdapter
cls.repo = ImageRepository()
cls.repo._get = MagicMock(return_value=None) # type: ignore[method-assign]
cls.repo._put = MagicMock(return_value=None) # type: ignore[method-assign]
cls.repo._delete = MagicMock(return_value=None) # type: ignore[method-assign]
# Prepare other test variables
cls.image = Image.new("RGB", (10, 10))
cls.image_id = UUID("d87022f8-4169-4d69-8512-bbd65cc1cb23")
cls.bad_image = "not_an_image"
cls.bad_image_id = "not-a-uuid"
@classmethod
def tearDownClass(cls) -> None:
# Stop patching
cls.patcher.stop()
def setUp(self) -> None:
"""Set up each test with a fresh ImageRepository instance."""
self.repo._get.reset_mock() # type: ignore[attr-defined]
self.repo._put.reset_mock() # type: ignore[attr-defined]
self.repo._delete.reset_mock() # type: ignore[attr-defined]
def test_should_adhere_to_interface(self) -> None:
"""Test that ImageRepository adheres to ImageRepositoryInterface."""
# Instantiation fails if interface not adhered to
_ = ImageRepository()
def test_should_have_image_format_set(self) -> None:
"""Test that ImageRepository has image_format set."""
self.assertTrue(hasattr(self.repo, "image_format"))
def test_should_have_folder_name_set(self) -> None:
"""Test that ImageRepository has folder_name set."""
self.assertTrue(hasattr(self.repo, "folder_name"))
def test_object_name_invalid_image_id_raises_value_error(self) -> None:
"""Test that _object_name() raises ValueError for invalid image_id."""
with self.assertRaises(ValueError):
_ = self.repo._object_name(self.bad_image_id) # type: ignore
def test_get_invalid_image_id_raises_value_error(self) -> None:
"""Test getting an image with invalid image_id."""
with self.assertRaises(ValueError):
self.repo.get(self.bad_image_id) # type: ignore
def test_put_invalid_image_id_raises_value_error(self) -> None:
"""Test putting an image with invalid image_id."""
with self.assertRaises(ValueError):
self.repo.put(self.image, self.bad_image_id) # type: ignore
def test_put_invalid_image_type_raises_value_error(self) -> None:
"""Test putting an invalid image type."""
with self.assertRaises(ValueError):
self.repo.put(self.bad_image, self.bad_image_id) # type: ignore
def test_delete_invalid_image_id_raises_value_error(self) -> None:
"""Test deleting an image with invalid image_id."""
with self.assertRaises(ValueError):
self.repo.delete(self.bad_image_id) # type: ignore
# Allows local debugging by running file as script
if __name__ == "__main__":
unittest.main()
tests/unit/test_job_repository.py
+85
View File
@@ -0,0 +1,85 @@
"""Unit tests for JobRepository class."""
import unittest
from unittest.mock import patch, MagicMock, _patch_dict
from uuid import UUID
from data_store.repositories.job_repository import JobRepository
class TestJobRepository(unittest.TestCase):
"""Unit tests for JobRepository class."""
patcher: _patch_dict
repo: JobRepository
job_id: UUID
bad_job_id: str
@classmethod
def setUpClass(cls) -> None:
"""Set up test case with a mocked MinioAdapter."""
# Patch environment variables for MinioAdapter
cls.patcher = patch.dict(
"os.environ",
{
"MINIO_ENDPOINT": "localhost:9000",
"MINIO_ACCESS_KEY": "minioadmin",
"MINIO_SECRET_KEY": "minioadmin",
"MINIO_BUCKET": "test-bucket",
},
)
cls.patcher.start()
# Prepare JobRepository
cls.repo = JobRepository()
cls.repo._get = MagicMock(return_value=None) # type: ignore[method-assign]
cls.repo._put = MagicMock(return_value=None) # type: ignore[method-assign]
cls.repo._delete = MagicMock(return_value=None) # type: ignore[method-assign]
# Prepare other test variables
cls.job_id = UUID("d87022f8-4169-4d69-8512-bbd65cc1cb23")
cls.bad_job_id = "not-a-uuid"
@classmethod
def tearDownClass(cls) -> None:
# Stop patching
cls.patcher.stop()
def setUp(self) -> None:
"""Set up each test with a fresh JobRepository instance."""
self.repo._get.reset_mock() # type: ignore[attr-defined]
self.repo._put.reset_mock() # type: ignore[attr-defined]
self.repo._delete.reset_mock() # type: ignore[attr-defined]
def test_should_adhere_to_interface(self) -> None:
"""Test that JobRepository adheres to JobRepositoryInterface."""
# Instantiation fails if interface not adhered to
_ = JobRepository()
def test_enter_calls_super_and_returns_self(self) -> None:
"""Test that __enter__ calls super().__enter__() and returns self."""
repo = JobRepository()
with patch(
"python_repositories.adapters.MinioAdapter.__enter__", return_value=None
) as mock_super_enter:
result = repo.__enter__()
mock_super_enter.assert_called_once()
self.assertIs(result, repo)
def test_get_invalid_job_id_raises_value_error(self) -> None:
"""Test that get with invalid job_id raises ValueError."""
with self.assertRaises(ValueError):
self.repo.get(self.bad_job_id) # type: ignore[arg-type]
def test_put_invalid_job_raises_value_error(self) -> None:
"""Test that put with invalid job raises ValueError."""
with self.assertRaises(ValueError):
self.repo.put("not_a_job") # type: ignore[arg-type]
def test_delete_invalid_job_id_raises_value_error(self) -> None:
"""Test that delete with invalid job_id raises ValueError."""
with self.assertRaises(ValueError):
self.repo.delete(self.bad_job_id) # type: ignore[arg-type]
# Allows local debugging by running file as script
if __name__ == "__main__":
unittest.main()
uv.lock
Generated
+1405
View File
File diff suppressed because it is too large Load Diff