"""Shared test fixtures.""" from unittest.mock import MagicMock import config import pytest import requests TEST_PROJECT_CODE = "BEES555555" TEST_PROJECT_NAME = "Mambos And Cha Cha Chas" TEST_PROJECT_ID = 3878967 TEST_ARTIST_ID = 81151686 TEST_PRODUCT_CODE = "CD111222" TEST_PRODUCT_NAME = "The Whitefriar Street Christmas Service" TEST_UPC = 191773660765 TEST_VENDOR_ID = 23787 TEST_SUBACCOUNT_ID = 0 TEST_RELEASE_ID = 2104178 TEST_RELEASE_STATUS = "label_processing" TEST_ARTIST = "Paul Harrington" TEST_TRACK_ID = 26265357 TEST_ISRC = "QM6P41733255" @pytest.fixture def ows_not_found_response(): """Mock not found response from a microservice.""" return MagicMock(spec=requests.models.Response, status_code=404) @pytest.fixture def ows_error_response(): """Mock error response from a microservice.""" return MagicMock(spec=requests.models.Response, status_code=500) @pytest.fixture def test_release_id(): """Test input event.""" return TEST_RELEASE_ID @pytest.fixture def test_track_id(): """Test input event.""" return TEST_TRACK_ID @pytest.fixture def product_delete_input(): """Test input event.""" return { "database": "art_relations", "table": "releases", "type": "delete", "ts": 1762166240, "xid": 9801659793, "commit": True, "data": { "release_id": TEST_RELEASE_ID, "last_updated": "2025-11-03 10:36:21", "upc": TEST_UPC, "display_upc": f"{TEST_UPC}", "artist_id": TEST_ARTIST_ID, "subaccount_id": TEST_SUBACCOUNT_ID, }, } @pytest.fixture def product_db_row(): """Row from art_relations.releases.""" return { "release_id": TEST_RELEASE_ID, "last_updated": "2018-03-11 16:32:44", "upc": TEST_UPC, "display_upc": f"{TEST_UPC}", "artist_id": TEST_ARTIST_ID, "subaccount_id": None, "vendor_release_identifier": None, "release_name": TEST_PRODUCT_NAME, "label": "Bless Music Group", "description": None, "release_date": "2018-03-12", "listprice": None, "sellprice": None, "display": "Y", "deletions": "N", "distribution": "digital", "applfees": None, "referral_id": None, "date_added_REMOVE": None, "genre_id": 19, "format": "Full Length", "distribution_format_id": 1, "orig_release_year": None, "promocode": None, "c_line": "2018 Bless", "manufacturer_upc": None, "album_download": "Y", "download_price": None, "priority": 3, "territory": None, "sale_start_date": "2018-03-12", "vod_start_date": None, "original_release_date": None, "vendor_catalog_number": 15, "language_id": None, "promobox_no": None, "additional_id": None, "dracatalog": None, "channel_id": None, "product_type_id": 1, "trackdown_black_list": "N", "not_for_distribution": "N", "europe_public_domain": None, "karaoke": None, "unknown_artist": None, "new_release": None, "product_format": None, "product_code": None, "special_instructions": None, "promos_number": None, "promo_upc": None, "sync_only": "N", "country_of_origin": None, "short_synopsis": None, "keywords": None, "production_co": None, "preorder_date": None, "amd_id": None, "wholesale_price_tier": None, "season": None, "episode": None, "production_number": None, "network": None, "genre_note": None, "physical_release_date": None, "theatrical_release_date": None, "original_digital_release_date": None, "est": None, "vod": None, "rsd_comment": None, "int_mkt_manager": None, "tms_id": None, "brand": None, "digital_only": "N", "product_subtype_id": None, "version": None, "delivered_version": None, "ingestion_completed": None, "release_status": "transfer_to_content", "meta_language": "SPA", "itunes_previewable": None, "subtitle_language_id": None, "compilation": "N", "project_id": 3882049, } @pytest.fixture def product_update_input(product_db_row): """Test input event.""" return { "database": "art_relations", "table": "releases", "type": "update", "ts": 1762166181, "xid": 9801497622, "commit": True, "data": product_db_row, "old": { "release_id": TEST_RELEASE_ID, "release_status": "label_processing", "last_updated": "2025-10-30 15:30:07", "description": None, "format": "Single", "language_id": None, "europe_public_domain": None, "karaoke": None, "special_instructions": None, "promo_upc": None, "genre_note": None, "version": None, "itunes_previewable": None, "p_line": None, "delivered_version": None, }, } @pytest.fixture def product_insert_input(product_db_row): """Test input event.""" product_db_row["release_status"] = "in_content" return { "database": "art_relations", "table": "releases", "type": "insert", "ts": 1762166181, "xid": 9801497622, "commit": True, "data": product_db_row, } @pytest.fixture def track_db_row(): """Row from art_relations.track.""" return { "track_name": "Joy to the World EQ MSTR\u0014\u0019", "upc": 191773660765, "cd": 1, "track_id": 6, "orchdmclip": None, "sample": "N", "dmclips": None, "isrc": TEST_ISRC, "id": TEST_TRACK_ID, "length_minute": 1, "length_seconds": 56, "writer": None, "credit": "orig_comp", "mechanical": "no", "publisher": None, "publish_company": None, "dpd_license": "no", "explicit_lyrics": "N", "version": None, "purchasable": "Y", "download_price": None, "p_line": "2017 Beaumex", "hidden": None, "offer_type": "all", "dig_distribution_type": "digital_mobile", "third_party_publisher": "N", "royalty_collection": "Y", "publishing_admin": "N", "sync_admin": "Y", "additional_id": None, "track_type": "music", "mp3_url": None, "stereo_or_mono": None, "bonus": "N", "recommended": "N", "youtube_matches": "Monetize", "youtube_uploads": None, "preorder_only": "N", "original_file_name": None, "release_id": TEST_RELEASE_ID, "last_updated": "2017-08-28T11:00:22.000000", "meta_language": "ENG", "preview_start_time": None, "closed_caption_exists": None, "closed_caption_reason_id": None, "vendor_track_identifier": None, "recording_country": 90, "recording_year": None, "original_track_id": None, "us_publishing_obligation": "PublicDomain", } @pytest.fixture def track_insert_input(track_db_row): """Test input event.""" return { "database": "art_relations", "table": "track", "type": "insert", "ts": 1762166181, "xid": 9801497622, "commit": True, "data": track_db_row, } @pytest.fixture def product_document_fixture(): """Test data received from ows-product.""" return { "display": "Y", "artist_url": "", "vendor_status": "signed", "c_line": "2012 One Media Publishing", "project_id": TEST_PROJECT_ID, "release_status": "in_content", "vendor_owner": "SC", "product_type": "Music", "artist_id": TEST_ARTIST_ID, "artist_myspace_url": "", "track_type": [ "music", "music", "music", "music", ], "subgenre": "Classical", "product_type_id": 1, "vendor_catalog_number": "OMPCB64D0013", "configuration": "Digital", "label": "One Media IP", "product_code": TEST_PRODUCT_CODE, "release_date": "2012-08-12", "genre": "Classical", "release_name": TEST_PRODUCT_NAME, "featured": "N", "episode": None, "format": "Full Length", "release_year": 2012, "subaccount_id": None, "track_list": [ "La Jolie Fille De Perth (The Fair Maid of Perth) Suite I. Prelude", # noqa "La Jolie Fille De Perth (The Fair Maid of Perth) Suite II. Serenade No.1", # noqa "Carmen Suite No.1 Prelude. Allegro Giocoso", "Carmen Suite No.2:", ], "upc": TEST_UPC, "artist_name": "Various Artists", "display_upc": f"{TEST_UPC}", "sale_start_date": "2012-08-12", "vendor_id": 10693, "imprint": "OMP Classics", "release_id": TEST_RELEASE_ID, "product_configuration": "Digital Audio", "release_month": 8, "artist_description": "", "delivered_version": "Live at Wembley", "isrc_list": [ "GBQRF1107530", "GBQRF1107531", "GBQRF1107532", "GBQRF1107533", ], "deletions": "N", "project_code": TEST_PROJECT_CODE, "project_name": TEST_PROJECT_NAME, } @pytest.fixture def get_release_info_fixture(release_info_fixture): """Mock result of get_release_info() function.""" return { "release_name": release_info_fixture["product_name"], "release_status": release_info_fixture["status"], "upc": release_info_fixture["upc"], "vendor_id": release_info_fixture["vendor_id"], "subaccount_id": release_info_fixture["subaccount_id"], "deletions": release_info_fixture["deletions"], "vendorUUID": release_info_fixture.get("vendorUUID"), "subaccountUUID": release_info_fixture.get("subaccountUUID"), "companyBrandUUID": release_info_fixture.get("companyBrandUUID"), "parentCompanyUUID": release_info_fixture.get("parentCompanyUUID"), } @pytest.fixture def get_release_info_not_found_fixture(): """Mock not found result of get_release_info function.""" return { "release_name": "", "release_status": "", "upc": 0, "vendor_id": 0, "subaccount_id": 0, "deletions": "N", "vendorUUID": "", "subaccountUUID": "", "companyBrandUUID": "", "parentCompanyUUID": "", } @pytest.fixture def release_info_fixture(): """Test data received from ows-product.""" return { "context_type": "digital", "product_name": TEST_PRODUCT_NAME, "upc": TEST_UPC, "distribution_format_id": 1, "subaccount_id": TEST_SUBACCOUNT_ID, "status": TEST_RELEASE_STATUS, "product_id": TEST_RELEASE_ID, "release_date": "2017-09-11", "project_id": 3772799, "product_type_id": 1, "vendor_id": TEST_VENDOR_ID, "deletions": "N", "not_for_distribution": "N", "vendorUUID": "c995dafd-80b1-4183-930b-c18ee95ee802", "subaccountUUID": None, "companyBrandUUID": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", "parentCompanyUUID": "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", } @pytest.fixture def release_info_success_response(release_info_fixture): """Mock successful response from ows-product.""" return MagicMock( spec=requests.models.Response, status_code=200, json=lambda: release_info_fixture ) @pytest.fixture def release_info_with_spatial_fixture(release_info_fixture): """Test data received from ows-product.""" return { **release_info_fixture, "spatial_upc": 154558, "spatial_isrc_list": ["one", "two"], } @pytest.fixture def release_info_with_spatial_success_response(release_info_with_spatial_fixture): """Mock successful response from ows-product.""" return MagicMock( spec=requests.models.Response, status_code=200, json=lambda: release_info_with_spatial_fixture, ) @pytest.fixture def track_info_fixture(): """Mock track info received from ows-track.""" return { "tuid": TEST_TRACK_ID, "track_name": "Joy to the World", "volume_number": 1, "meta_language_code": "ENG", "version": "", "localizations": [], "writers": [{"type": "writer", "track_writer_id": 52703447, "name": "Public Domain"}], "product_id": 2104178, "explicit": "N", "track_number": 8, "artists": [ {"type": "performer", "track_artist_id": TEST_ARTIST_ID, "name": TEST_ARTIST}, { "type": "featuring", "track_artist_id": 81151700, "name": "The Whitefriar Street Church Choir", }, ], "isrc": TEST_ISRC, "recording_country_id": 90, "upc": 191773660765, "p_info": "2014 Beaumex", "original_rights_holder_country_id": 90, "ownership_rights": "owner", "preview_start_time": None, } @pytest.fixture def second_track_info_fixture(): """Mock second track info received from ows-track.""" return { "tuid": 26265358, "track_name": "Second Joy to the World", "volume_number": 1, "meta_language_code": "ENG", "version": "", "localizations": [], "writers": [{"type": "writer", "track_writer_id": 52703447, "name": "Public Domain"}], "product_id": 2104178, "explicit": "N", "track_number": 8, "artists": [ {"type": "performer", "track_artist_id": TEST_ARTIST_ID, "name": TEST_ARTIST}, {"type": "featuring", "track_artist_id": 81151701, "name": "Street Choir"}, ], "isrc": "QM6P41735155", "recording_country_id": 90, "upc": 191773660765, "p_info": "2014 Beaumex", "original_rights_holder_country_id": 90, "ownership_rights": "owner", "preview_start_time": None, } @pytest.fixture def track_info_success_response(track_info_fixture): """Mock success response from ows-track microservice.""" return MagicMock( spec=requests.models.Response, status_code=200, json=lambda: track_info_fixture ) @pytest.fixture def get_track_info_fixture(): """Mock result of get_track_info function.""" return {"artist_id": TEST_ARTIST_ID, "artist_name": TEST_ARTIST} @pytest.fixture def get_track_info_not_found_fixture(): """Mock not found result of get_track_info function.""" return {"artist_id": 0, "artist_name": ""} @pytest.fixture def tracks_by_product_id_success_response(track_info_fixture, second_track_info_fixture): """Mock success response from ows-track microservice.""" return MagicMock( spec=requests.models.Response, status_code=200, json=lambda: {"items": [track_info_fixture, second_track_info_fixture]}, ) @pytest.fixture def get_tracks_by_product_id_fixture(track_info_fixture, second_track_info_fixture): """Tracks data for product id.""" track_info_fixture["artist_id"] = TEST_ARTIST_ID track_info_fixture["artist_name"] = TEST_ARTIST second_track_info_fixture["artist_id"] = TEST_ARTIST_ID second_track_info_fixture["artist_name"] = TEST_ARTIST return [track_info_fixture, second_track_info_fixture] @pytest.fixture def track1_parsed_doc(track_info_fixture, product_update_input, release_info_fixture): """Tracks data for product id.""" return { "artist_id": 81151686, "artist_name": "Paul Harrington", "deletions": "N", "isrc": TEST_ISRC, "release_id": product_update_input["data"]["release_id"], "release_name": "The Whitefriar Street Christmas Service", "release_status": "in_content", "subaccount_id": 0, "track_name": "Joy to the World", "track_unique_id": track_info_fixture["tuid"], "upc": 191773660765, "vendor_id": 23787, "version": "", "vendorUUID": release_info_fixture.get("vendorUUID"), "companyBrandUUID": release_info_fixture.get("companyBrandUUID"), "parentCompanyUUID": release_info_fixture.get("parentCompanyUUID"), } @pytest.fixture def track2_parsed_doc(second_track_info_fixture, product_update_input, release_info_fixture): """Tracks data for product id.""" return { "artist_id": 81151686, "artist_name": "Paul Harrington", "deletions": "N", "isrc": "QM6P41735155", "release_id": product_update_input["data"]["release_id"], "release_name": "The Whitefriar Street Christmas Service", "release_status": "in_content", "subaccount_id": 0, "track_name": "Second Joy to the World", "track_unique_id": second_track_info_fixture["tuid"], "upc": 191773660765, "vendor_id": 23787, "version": "", "vendorUUID": release_info_fixture.get("vendorUUID"), "companyBrandUUID": release_info_fixture.get("companyBrandUUID"), "parentCompanyUUID": release_info_fixture.get("parentCompanyUUID"), } @pytest.fixture def track_document_fixture(track1_parsed_doc, release_info_fixture): """Test data received track.prepare_doc.""" return { **track1_parsed_doc, "cd": 1, "original_file_name": "", "track_id": 6, "track_name": "Joy to the World EQ MSTR", "track_type": "music", "release_status": "label_processing", "vendorUUID": release_info_fixture.get("vendorUUID"), "companyBrandUUID": release_info_fixture.get("companyBrandUUID"), "parentCompanyUUID": release_info_fixture.get("parentCompanyUUID"), } @pytest.fixture def prepare_release_tracks_docs_result( get_tracks_by_product_id_fixture, track1_parsed_doc, track2_parsed_doc ): """Test data received from prepare_release_tracks_docs.""" return [ { "_index": config.TRACK_INDEX_NAME, "_op_type": "update", "_id": get_tracks_by_product_id_fixture[0]["tuid"], "doc": track1_parsed_doc, "doc_as_upsert": True, }, { "_index": config.TRACK_INDEX_NAME, "_op_type": "update", "_id": get_tracks_by_product_id_fixture[1]["tuid"], "doc": track2_parsed_doc, "doc_as_upsert": True, }, ] @pytest.fixture def project_document_fixture(): """Test data received from ows-project-manager.""" return { "project_id": TEST_PROJECT_ID, "project_code": TEST_PROJECT_CODE, "vendor_id": TEST_VENDOR_ID, "subaccount_id": TEST_SUBACCOUNT_ID, "project_name": TEST_PROJECT_NAME, "created_date_utc": "2018-03-07 23:53:14.893249", "updated_date_utc": "2018-03-07 23:53:14.893249", "correlation_id": "123", "description": "Project description ...", "artist_id": TEST_ARTIST_ID, "name": TEST_ARTIST, "artist_description": "The best honky tonky", "artist_url": "http://www.billy.com/", } @pytest.fixture def project_db_row(): """Row from art_relations.project.""" return { "project_id": TEST_PROJECT_ID, "project_code": TEST_PROJECT_CODE, "vendor_id": TEST_VENDOR_ID, "subaccount_id": TEST_SUBACCOUNT_ID, "project_name": TEST_PROJECT_NAME, "created_date_utc": "2018-03-07 23:53:14.893249", "updated_date_utc": "2018-03-07 23:53:14.893249", "correlation_id": "beef4980-2262-11e8-9b46-0242ac110002.1", "artist_id": TEST_ARTIST_ID, "description": "Project description ...", } @pytest.fixture def project_insert_input(project_db_row): """Maxwell's Daemon project event converted to internal format.""" return {"data": project_db_row, "table": "project", "type": "insert"} @pytest.fixture def track_additional_isrc_db_row(): """Row from art_relations.track_additional_isrc.""" return { "track_id": TEST_TRACK_ID, "isrc": "QM6P42424506", "created_at": "2026-05-10 01:19:48", "deleted_at": None, "last_updated": "2025-11-03 10:36:21", } @pytest.fixture def track_additional_isrc_insert_input(track_additional_isrc_db_row): """Test input event for track_additional_isrc table.""" return { "database": "art_relations", "table": "track_additional_isrc", "type": "insert", "ts": 1762166181, "xid": 9801497622, "commit": True, "data": track_additional_isrc_db_row, } @pytest.fixture def get_track_by_tuid_fixture(track_info_fixture): """Mock result of get_track_by_tuid function.""" return {**track_info_fixture, "artist_id": TEST_ARTIST_ID, "artist_name": TEST_ARTIST} @pytest.fixture def product_additional_upc_db_row(): """Row from art_relations.product_additional_upc.""" return { "id": 1, "product_id": TEST_RELEASE_ID, "type": "atmos", "upc": 196872451234, "created_at": "2026-05-10 01:19:48", "updated_at": "2026-05-10 01:19:48", "deleted_at": None, } @pytest.fixture def product_additional_upc_insert_input(product_additional_upc_db_row): """Test input event for product_additional_upc table.""" return { "database": "art_relations", "table": "product_additional_upc", "type": "insert", "ts": 1762166181, "xid": 9801497622, "commit": True, "data": product_additional_upc_db_row, }