117 lines
3.0 KiB
Python
117 lines
3.0 KiB
Python
"""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__])
|