Brian Bjarke JensenandCursor
dc6f8a3e89
PR Title Check / check-title (pull_request) Successful in 7s
Code Quality Pipeline / code-quality (pull_request) Failing after 19s
Test Python Package / unit-tests (pull_request) Successful in 47s
Test Python Package / integration-tests (pull_request) Successful in 44s
Test Python Package / coverage-report (pull_request) Successful in 11s
Introduce PostgresAdapter for dict-based row CRUD via psycopg3, including config, lazy exports, unit/integration tests with testcontainers, and an example UserTableRepository. Co-authored-by: Cursor <cursoragent@cursor.com>
125 lines
3.8 KiB
Python
125 lines
3.8 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 psycopg
|
|
import pytest
|
|
|
|
from python_repositories.examples.artifact_object_repository import (
|
|
ArtifactObjectRepository,
|
|
)
|
|
from python_repositories.examples.user_json_repository import UserJsonRepository
|
|
from python_repositories.examples.user_table_repository import UserTableRepository
|
|
from tests.integration.postgres._containers import TEST_TABLE
|
|
|
|
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", autouse=True)
|
|
def set_postgres_example_env(
|
|
postgres_container: str,
|
|
raw_postgres_client: psycopg.Connection,
|
|
) -> Generator[None, None, None]:
|
|
"""Set Postgres env vars for example repositories using from_env() defaults."""
|
|
env_vars = {
|
|
"POSTGRES_URI": postgres_container,
|
|
"POSTGRES_TABLE": TEST_TABLE,
|
|
"POSTGRES_PRIMARY_KEY": "id",
|
|
}
|
|
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": "alice@example.com"}
|
|
|
|
|
|
@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()
|
|
|
|
|
|
@pytest.mark.needs_postgres
|
|
def test_user_table_repository_save_and_get(
|
|
user_data: dict[str, str],
|
|
) -> None:
|
|
"""Test that UserTableRepository can save and retrieve a user."""
|
|
with UserTableRepository() as repo:
|
|
repo.save_user("alice", user_data)
|
|
user = repo.get_user("alice")
|
|
assert user is not None
|
|
assert user["name"] == user_data["name"]
|
|
assert user["email"] == user_data["email"]
|
|
|
|
|
|
@pytest.mark.needs_postgres
|
|
def test_user_table_repository_delete(
|
|
user_data: dict[str, str],
|
|
) -> None:
|
|
"""Test that UserTableRepository can delete a user."""
|
|
with UserTableRepository() as repo:
|
|
repo.save_user("alice", user_data)
|
|
repo.delete_user("alice")
|
|
assert repo.get_user("alice") is None
|