"""Functional tests for thumbnail S3 logic using the moto server.""" import boto3 from video.constants.thumbnails import ( DEFAULT_IMAGE_SIZE, THUMBNAILS_BUCKET, THUMBNAILS_CDN_URL, ) from video.logic import thumbnails as thumbnails_logic def test_get_available_thumbnails_filters_by_job_id() -> None: """Only thumbnails for the requested job_id are returned.""" job_id = 123 other_job_id = 999 s3 = boto3.client("s3") s3.create_bucket(Bucket=THUMBNAILS_BUCKET) for key in [ f"thumbnails/{DEFAULT_IMAGE_SIZE}/{job_id}/0000.jpg", f"thumbnails/{DEFAULT_IMAGE_SIZE}/{job_id}/0001.jpg", f"thumbnails/{DEFAULT_IMAGE_SIZE}/{other_job_id}/0000.jpg", ]: s3.put_object(Bucket=THUMBNAILS_BUCKET, Key=key, Body=b"") result = thumbnails_logic.get_available_thumbnails(job_id, None) assert set(result) == { f"{THUMBNAILS_CDN_URL}/{DEFAULT_IMAGE_SIZE}/{job_id}/0000.jpg", f"{THUMBNAILS_CDN_URL}/{DEFAULT_IMAGE_SIZE}/{job_id}/0001.jpg", } def test_get_available_thumbnails_respects_image_size() -> None: """image_size param controls which prefix is queried.""" job_id = 42 s3 = boto3.client("s3") s3.create_bucket(Bucket=THUMBNAILS_BUCKET) for key in [ f"thumbnails/small/{job_id}/0000.jpg", f"thumbnails/{DEFAULT_IMAGE_SIZE}/{job_id}/0000.jpg", ]: s3.put_object(Bucket=THUMBNAILS_BUCKET, Key=key, Body=b"") result = thumbnails_logic.get_available_thumbnails(job_id, "small") assert result == [f"{THUMBNAILS_CDN_URL}/small/{job_id}/0000.jpg"] def test_get_available_thumbnails_empty_bucket() -> None: """Empty bucket returns an empty list without error.""" s3 = boto3.client("s3") s3.create_bucket(Bucket=THUMBNAILS_BUCKET) result = thumbnails_logic.get_available_thumbnails(1, None) assert result == []