Brian Bjarke JensenandCursor
b28ac6e803
PR Title Check / check-title (pull_request) Successful in 7s
Test Python Package / unit-tests (pull_request) Successful in 13s
Test Python Package / integration-tests (pull_request) Successful in 23s
Test Python Package / coverage-report (pull_request) Successful in 10s
Code Quality Pipeline / code-quality (pull_request) Successful in 49s
Split Redis and MinIO container setup into backend-specific conftest modules so only the containers needed for collected tests are imported and started. Co-authored-by: Cursor <[email protected]>
80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
"""Integration tests for example domain repositories."""
|
|
|
|
from collections.abc import Generator
|
|
from io import BytesIO
|
|
import os
|
|
import random
|
|
|
|
from minio import Minio
|
|
import pytest
|
|
|
|
from python_repositories.examples.artifact_object_repository import (
|
|
ArtifactObjectRepository,
|
|
)
|
|
from python_repositories.examples.user_json_repository import UserJsonRepository
|
|
|
|
pytestmark = [
|
|
pytest.mark.integration,
|
|
pytest.mark.needs_redis,
|
|
pytest.mark.needs_minio,
|
|
]
|
|
|
|
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def set_example_env(
|
|
redis_container: str,
|
|
minio_container: dict[str, str],
|
|
raw_minio_client: Minio,
|
|
) -> Generator[None, None, None]:
|
|
"""Set env vars so example repositories can use from_env() defaults."""
|
|
_ = raw_minio_client
|
|
env_vars = {"REDIS_URI": redis_container, **minio_container}
|
|
for key, value in env_vars.items():
|
|
os.environ[key] = value
|
|
yield
|
|
for key in env_vars:
|
|
_ = os.environ.pop(key, default=None)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def user_data() -> Generator[dict[str, str], None, None]:
|
|
"""Provide sample user data for tests."""
|
|
yield {"name": "Alice", "email": "[email protected]"}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def artifact_data() -> Generator[BytesIO, None, None]:
|
|
"""Provide sample artifact data for tests."""
|
|
yield BytesIO(random.randbytes(2**20))
|
|
|
|
|
|
def test_user_json_repository_save_and_get(
|
|
user_data: dict[str, str],
|
|
) -> None:
|
|
"""Test that UserJsonRepository can save and retrieve a user."""
|
|
with UserJsonRepository() as repo:
|
|
repo.save_user("alice", user_data)
|
|
assert repo.get_user("alice") == user_data
|
|
|
|
|
|
def test_user_json_repository_delete(
|
|
user_data: dict[str, str],
|
|
) -> None:
|
|
"""Test that UserJsonRepository can delete a user."""
|
|
with UserJsonRepository() as repo:
|
|
repo.save_user("alice", user_data)
|
|
repo.delete_user("alice")
|
|
assert repo.get_user("alice") is None
|
|
|
|
|
|
def test_artifact_object_repository_store_and_get(
|
|
artifact_data: BytesIO,
|
|
) -> None:
|
|
"""Test that ArtifactObjectRepository can store and retrieve an artifact."""
|
|
with ArtifactObjectRepository() as repo:
|
|
repo.store_artifact("report-1", artifact_data)
|
|
received = repo.get_artifact("report-1")
|
|
assert received is not None
|
|
artifact_data.seek(0)
|
|
assert received.read() == artifact_data.read()
|