"""Test for Image Location Model.""" from typing import Any import pytest from assets import config from assets.models import image_location from tests.unit.models import au_operations ENTITY_IDS_TO_IMAGE_FILENAMES = { 123456: "product/cover/e10adc3949ba59abbe56e057f20f883e.jpg", 112233: "product/cover/d0970714757783e6cf17b26fb8e2298f.jpg", } @pytest.fixture def fixture_entity_ids() -> list[int]: """Fixture of entity IDs.""" return list(ENTITY_IDS_TO_IMAGE_FILENAMES.keys()) @pytest.fixture def fixture_image_filenames() -> list[str]: """Fixture of image filenames.""" return list(ENTITY_IDS_TO_IMAGE_FILENAMES.values()) @pytest.fixture def fixture_entity_id(fixture_entity_ids: list[int]) -> int: """Fixture of single entity ID.""" return fixture_entity_ids[0] @pytest.fixture def fixture_image_filename(fixture_image_filenames: list[str]) -> str: """Fixture of single image filename.""" return fixture_image_filenames[0] @pytest.fixture def fixture_cdn_url() -> str: """Fixture to set CDN url.""" config.CDN_URL = "https://fakesite.cloudfront.net" return config.CDN_URL def test_get_image_filename( fixture_entity_id: int, fixture_image_filename: str ) -> None: """Test getting image filename for product.""" response = image_location.get_image_filename( entity_id=fixture_entity_id, image_type="product", image_format="cover" ) assert response == fixture_image_filename def test_get_image_filenames( fixture_entity_ids: list[int], fixture_image_filenames: list[str] ) -> None: """Test getting image filenames for list of products.""" response = image_location.get_image_filenames( entity_ids=fixture_entity_ids, image_type="product", image_format="cover" ) assert response == fixture_image_filenames def test_get_image_location_by_filename( fixture_image_filename: str, fixture_cdn_url: str ) -> None: """Test getting image location on CDN for s3 file.""" response = image_location.get_image_location_by_filename(fixture_image_filename) assert response == "{cdn}/{filename}".format( cdn=fixture_cdn_url, filename=fixture_image_filename ) def test_get_image_location_by_filenames( fixture_image_filenames: list[str], fixture_cdn_url: str ) -> None: """Test getting image locations on CDN for s3 files.""" results = image_location.get_image_location_by_filenames(fixture_image_filenames) assert results == [ "{cdn}/{filename}".format(cdn=fixture_cdn_url, filename=filename) for filename in fixture_image_filenames ] def test_get_image_location_by_filename_and_cdn(fixture_image_filename: str) -> None: """Test getting image location on given CDN for s3 file.""" xlarge_cdn = "https://xlargecovers.cloudfront.net" response = image_location.get_image_location_by_filename( fixture_image_filename, xlarge_cdn ) assert response == "{cdn}/{filename}".format( cdn=xlarge_cdn, filename=fixture_image_filename ) def test_get_image_location_by_filenames_and_cdn( fixture_image_filenames: list[str], ) -> None: """Test getting image locations on given CDN for s3 files.""" xlarge_cdn = "https://xlargecovers.cloudfront.net" results = image_location.get_image_location_by_filenames( fixture_image_filenames, xlarge_cdn ) assert results == [ "{cdn}/{filename}".format(cdn=xlarge_cdn, filename=filename) for filename in fixture_image_filenames ] ASSET_UPLOAD_PK_0 = 1000 ASSET_UPLOAD_PK_1 = 1001 ASSET_UPLOAD_PK_2 = 1002 ASSET_UPLOAD_PK_3 = 1003 ASSET_UPLOAD_PK_4 = 1004 ASSET_UPLOAD_PK_5 = 1005 ASSET_UPLOAD_PK_6 = 1006 ASSET_UPLOAD_PK_7 = 1007 ASSET_UPLOAD_PK_8 = 1008 PRODUCT_ID_1 = 1001 PRODUCT_ID_2 = 1002 PRODUCT_ID_3 = 1003 PRODUCT_ID_4 = 1004 PRODUCT_ID_5 = 1005 PRODUCT_UPC_1 = "123" PRODUCT_UPC_2 = "456" PRODUCT_UPC_3 = "789" PRODUCT_UPC_4 = "101" @pytest.fixture def bulk_data() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Fixture for bulk image location.""" asset_uploads = [ { "id": ASSET_UPLOAD_PK_0, "user_id": "oa:123456", "token": "session_token_0".encode(), "filename": "unique_filename_0", "api_version": 2, "track_unique_id": 0, "product_id": PRODUCT_ID_1, "deleted": False, }, { "id": ASSET_UPLOAD_PK_8, "user_id": "oa:123456", "token": "session_token_0".encode(), "filename": "unique_filename_0", "api_version": 2, "track_unique_id": 123, "product_id": PRODUCT_ID_1, "deleted": False, }, { "id": ASSET_UPLOAD_PK_1, "user_id": "oa:123456", "token": "session_token_1".encode(), "filename": "unique_filename_1", "api_version": 2, "track_unique_id": 0, "product_id": PRODUCT_ID_1, "deleted": True, }, { "id": ASSET_UPLOAD_PK_2, "user_id": "oa:123456", "token": "session_token_2".encode(), "filename": "unique_filename_2", "api_version": 2, "track_unique_id": 0, "product_id": PRODUCT_ID_1, "deleted": False, }, { "id": ASSET_UPLOAD_PK_3, "user_id": "oa:123456", "token": "session_token_3".encode(), "filename": "unique_filename_3", "api_version": 2, "track_unique_id": 0, "product_id": PRODUCT_ID_2, "deleted": False, }, { "id": ASSET_UPLOAD_PK_4, "user_id": "oa:123456", "token": "session_token_4".encode(), "filename": "unique_filename_4", "api_version": 2, "track_unique_id": 0, "product_id": PRODUCT_ID_2, "deleted": False, }, { "id": ASSET_UPLOAD_PK_5, "user_id": "oa:123456", "token": "session_token_5".encode(), "filename": "unique_filename_5", "api_version": 2, "track_unique_id": 0, "product_id": PRODUCT_ID_3, "deleted": False, }, { "id": ASSET_UPLOAD_PK_6, "filename": "unique_filename_6", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_6", "track_unique_id": 0, "product_id": PRODUCT_ID_5, "deleted": False, "is_correction": False, }, { "id": ASSET_UPLOAD_PK_7, "filename": "unique_filename_19", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_7", "track_unique_id": 0, "product_id": PRODUCT_ID_5, "deleted": False, "is_correction": True, }, ] asset_finals = [ { "asset_upload_id": ASSET_UPLOAD_PK_0, "asset_type": "JPG", "asset_subtype": "large_cover", "filename": "images/v2/product/large_cover/unique_asset_final_0.jpg", "bucket": "bucket", }, { "asset_upload_id": ASSET_UPLOAD_PK_1, "asset_type": "JPG", "asset_subtype": "large_cover", "filename": "images/v2/product/large_cover/unique_asset_final_1.jpg", "bucket": "bucket", }, { "asset_upload_id": ASSET_UPLOAD_PK_2, "asset_type": "JPG", "asset_subtype": "large_cover", "filename": "images/v2/product/large_cover/unique_asset_final_2.jpg", "bucket": "bucket", }, { "asset_upload_id": ASSET_UPLOAD_PK_2, "asset_type": "JPG", "asset_subtype": "xlarge_cover", "filename": "images/v2/product/xlarge_cover/unique_asset_final_2.jpg", "bucket": "bucket", }, { "asset_upload_id": ASSET_UPLOAD_PK_3, "asset_type": "JPG", "asset_subtype": "large_cover", "filename": "images/v2/product/large_cover/unique_asset_final_3.jpg", "bucket": "bucket", }, { "asset_upload_id": ASSET_UPLOAD_PK_4, "asset_type": "JPG", "asset_subtype": "large_cover", "filename": "images/v2/product/large_cover/unique_asset_final_4.jpg", "bucket": "bucket", }, { "asset_upload_id": ASSET_UPLOAD_PK_6, "asset_type": "JPG", "asset_subtype": "large_cover", "filename": "images/v2/product/large_cover/unique_asset_final_6.jpg", "bucket": "bucket", }, { "asset_upload_id": ASSET_UPLOAD_PK_7, "asset_type": "JPG", "asset_subtype": "large_cover", "filename": "images/v2/product/large_cover/unique_asset_final_7.jpg", "bucket": "bucket", }, ] return asset_uploads, asset_finals @pytest.fixture def bulk_db_fixture( bulk_data: tuple[list[dict[str, Any]], list[dict[str, Any]]], ) -> None: """Seed bulk data.""" asset_uploads, asset_finals = bulk_data au_operations.truncate_tables() au_operations.seed_asset_upload_table(asset_uploads) au_operations.seed_asset_final_table(asset_finals) def test_get_image_locations_large_cover(bulk_db_fixture: None) -> None: """Test bulk location endpoint for large_cover.""" products = [PRODUCT_ID_1, PRODUCT_ID_2, PRODUCT_ID_3, PRODUCT_ID_4] results = image_location.get_image_locations(products, "large_cover") assert results assert set(results.keys()) == {1001, 1002, 1003, 1004} assert "v2/product/large_cover/unique_asset_final_2.jpg" in results[1001] assert "v2" not in results[1003] def test_get_image_locations_large_cover_no_fallback(bulk_db_fixture: None) -> None: """Test bulk location endpoint for large_cover with no fallback.""" products = [PRODUCT_ID_1, PRODUCT_ID_2, PRODUCT_ID_3] results = image_location.get_image_locations( products, "large_cover", fallback=False ) assert results assert set(results.keys()) == {1001, 1002} def test_get_image_locations_large_cover_omit_corrections( bulk_db_fixture: None, ) -> None: """Test bulk location endpoint for large_cover with/without corrections.""" products = [PRODUCT_ID_5] results = image_location.get_image_locations( products, "large_cover", omit_corrections=False ) assert "v2/product/large_cover/unique_asset_final_7.jpg" in results[PRODUCT_ID_5] results = image_location.get_image_locations( products, "large_cover", omit_corrections=True ) assert "v2/product/large_cover/unique_asset_final_6.jpg" in results[PRODUCT_ID_5] def test_get_image_locations_xlarge_cover(bulk_db_fixture: None) -> None: """Test bulk location endpoint for xlarge_cover.""" product_ids = [PRODUCT_ID_1, PRODUCT_ID_2, PRODUCT_ID_3, PRODUCT_ID_4] results = image_location.get_image_locations(product_ids, "xlarge_cover") assert results assert set(results.keys()) == {1001, 1002, 1003, 1004} assert "v2/product/xlarge_cover/unique_asset_final_2.jpg" in results[1001] assert "placeholders/xlarge_cover.png" in results[1002] assert "placeholders/xlarge_cover.png" in results[1003] assert "placeholders/xlarge_cover.png" in results[1004] def test_get_image_locations_single_product(bulk_db_fixture: None) -> None: """Test bulk location endpoint for a single product which oddly called sometimes.""" results = image_location.get_image_locations([PRODUCT_ID_1], "large_cover") assert results def test_get_last_uploaded_image_locations_large_cover(bulk_db_fixture: None) -> None: """Test bulk location endpoint for large_cover which returns the latest uploaded image Locations.""" products = [PRODUCT_ID_1, PRODUCT_ID_2, PRODUCT_ID_3, PRODUCT_ID_4] results = image_location.get_image_locations(products, "large_cover") assert results assert set(results.keys()) == set(products) assert "v2/product/large_cover/unique_asset_final_2.jpg" in results[PRODUCT_ID_1] assert "v2/product/large_cover/unique_asset_final_4.jpg" in results[PRODUCT_ID_2] assert "v2" not in results[PRODUCT_ID_3] assert "v2" not in results[PRODUCT_ID_4]