"""Job logic tests.""" from datetime import datetime from typing import Any from unittest.mock import MagicMock, call from zoneinfo import ZoneInfo import pytest from flask import g from freezegun import freeze_time from pytest_mock import MockerFixture from requests.exceptions import HTTPError from sqlalchemy.exc import SQLAlchemyError from werkzeug.exceptions import BadRequest from video.constants import release as release_constants from video.exceptions import ProductOwnershipError, ProjectOwnershipError from video.logic import approval as approval_logic, metadata as metadata_logic from video.logic.metadata import USER_SPECIFIABLE_FIELDS from video.models.ows import ( product as ows_product, project as ows_project, track as ows_track, ) from video.models.sql.classes import ( artist_delivery_settings, carveouts, product_manager_mapping_product, product_video, release, youtube_channel, ) from video.utils import status_checker from tests.unit import ar_db_utils from tests.unit.ar_db_utils import mock_db_session @ar_db_utils.test_schema def test_load_metadata_success(mocker: MockerFixture) -> None: """Test load metadata.""" mocker.patch.object( product_video, "get", return_value={"data": "hirthy"}, ) load_metadata_response = metadata_logic.load_metadata(1, None, None) assert load_metadata_response == {"data": "hirthy"} @ar_db_utils.test_schema def test_load_metadata_success_ff_enable(mocker: MockerFixture) -> None: """Test load metadata.""" product_video_response = {"product_video_id": 1} mocker.patch.object( product_video, "get", return_value=product_video_response, ) mocker.patch.object( product_manager_mapping_product, "get_product_manager", return_value={"product_manager_id": 69}, ) load_metadata_response = metadata_logic.load_metadata(1, None, None) assert load_metadata_response == { "product_manager_id": 69, "product_video_id": 1, } def test_load_metadata_no_ownership(mocker: MockerFixture) -> None: """Test load metadata.""" mocker.patch.object( ows_product, "check_product_ownership", side_effect=ProductOwnershipError(), ) with pytest.raises(ProductOwnershipError): metadata_logic.load_metadata(1, None, None) @ar_db_utils.test_schema def test_load_metadata_oa_user(mocker: MockerFixture) -> None: """Test load metadata with oa account type.""" success = {"release_id": 1} mock_check_ownership: MagicMock = mocker.patch.object( ows_product, "check_product_ownership", ) mock_pv_get: MagicMock = mocker.patch.object( product_video, "get", return_value=success ) load_metadata_response = metadata_logic.load_metadata(1, "oa", 12345) assert mock_check_ownership.call_count == 0 assert mock_pv_get.call_count == 1 assert load_metadata_response == success def test_get_available_channels_no_ownership(mocker: MockerFixture) -> None: """Test available channels no ownership.""" mocker.patch.object( ows_product, "check_product_ownership", side_effect=ProductOwnershipError(), ) with pytest.raises(ProductOwnershipError): metadata_logic.get_available_channels(1, None, None, None) @pytest.mark.parametrize("user_id", [None, "oa:12345"]) def test_get_available_channels_by_artist_and_vendor( mocker: MockerFixture, user_id: Any ) -> None: """Test available channels.""" mocker.patch.object( release, "get_by_id_with_vendor", return_value={"vendor_id": 12, "project_id": 13}, ) mocker.patch.object( ows_project, "get_project_by_id", return_value={"artist_id": 12}, ) mocker.patch.object( artist_delivery_settings, "get", return_value=[ { "channel_name": "name", "channel_id": "1", "vevo_controlled": "Yes", }, { "channel_name": None, "channel_id": "2", "vevo_controlled": "Yes", }, { "channel_name": "2", "channel_id": "4", "vevo_controlled": "No", }, { "channel_name": None, "channel_id": "5", "vevo_controlled": "No", }, ], ) mocker.patch.object( youtube_channel, "get", return_value=[ { "channel_name": "name", "channel_id": "1", }, { "channel_name": "vendor", "channel_id": "3", }, ], ) channels_response = metadata_logic.get_available_channels(1, None, None, user_id) assert channels_response == [ {"channel_name": "name", "channel_id": "1", "vevo_controlled": "Yes"}, {"channel_name": None, "channel_id": "2", "vevo_controlled": "Yes"}, {"channel_name": "2", "channel_id": "4", "vevo_controlled": "No"}, {"channel_name": None, "channel_id": "5", "vevo_controlled": "No"}, {"channel_name": "vendor", "channel_id": "3", "vevo_controlled": "No"}, ] def test_get_available_channels_by_artist_only(mocker: MockerFixture) -> None: """Test available channels.""" mocker.patch.object( ows_product, "check_product_ownership", ) mocker.patch.object( release, "get_by_id_with_vendor", return_value={"vendor_id": 12, "project_id": 13}, ) mocker.patch.object( ows_project, "get_project_by_id", return_value={"artist_id": 12}, ) mocker.patch.object( artist_delivery_settings, "get", return_value=[ { "channel_name": "name", "channel_id": "1", "vevo_controlled": "No", }, ], ) channels_response = metadata_logic.get_available_channels( 1, "subaccount", "12", None ) assert channels_response == [ {"channel_name": "name", "channel_id": "1", "vevo_controlled": "No"} ] def test_get_available_channels_by_project_id_no_ownership( mocker: MockerFixture, ) -> None: """Test available project channels no ownership.""" mocker.patch.object( ows_project, "check_project_ownership", side_effect=ProjectOwnershipError(), ) with pytest.raises(ProjectOwnershipError): metadata_logic.get_available_channels_by_project_id(1, None, None) def test_get_available_channels_by_project_id_project_error( mocker: MockerFixture, ) -> None: """Test available project channels with get project error.""" mocker.patch.object( ows_project, "check_project_ownership", ) mocker.patch.object( ows_project, "get_project_by_id", side_effect=HTTPError("fatal_project_error"), ) with pytest.raises(HTTPError): metadata_logic.get_available_channels_by_project_id(1, None, None) def test_get_available_channels_by_project_id_success(mocker: MockerFixture) -> None: """Test available project channels success.""" mocker.patch.object( ows_project, "check_project_ownership", ) mocker.patch.object( ows_project, "get_project_by_id", return_value={"artist_id": 12}, ) _get_channels_response = [ {"channel_name": "name", "channel_id": "1", "vevo_controlled": "No"} ] mocker.patch.object( metadata_logic, "_get_channels_for_artist", return_value=_get_channels_response ) channels_response = metadata_logic.get_available_channels_by_project_id( 1, None, None ) assert channels_response == _get_channels_response def test_validate_bad_isrc() -> None: """Test bad isrc format.""" data, errors = metadata_logic._validate( 1, {"isrc": "dango"}, None, None, ) assert data == {} assert errors == ["badIsrcFormat"] def test_validate_already_used_isrc(mocker: MockerFixture) -> None: """Test isrc already used.""" mocker.patch.object( ows_track, "is_isrc_used", return_value={"used": True, "used_by_product_id": 1000}, ) data, errors = metadata_logic._validate( 1, {"isrc": "AB1231212345"}, None, None, ) assert data == {} assert errors == ["isrcUsedByMusic"] def test_validate_already_used_isrc_by_same_product(mocker: MockerFixture) -> None: """Test isrc used by the same product.""" mocker.patch.object( ows_track, "is_isrc_used", return_value={"used": True, "used_by_product_id": 1}, ) data, errors = metadata_logic._validate( 1, {"isrc": "AB1231212345"}, None, None, ) assert data == {"isrc": "AB1231212345"} assert errors == [] def test_product_code_in_use(mocker: MockerFixture) -> None: """Test product code in use.""" mocker.patch.object( ows_product, "check_product_code_exists", return_value=True, ) data, errors = metadata_logic._validate( 1, {"product_code": "porridge"}, 1, 2, ) assert data == {} assert errors == ["productCodeInUse"] """Test product code not in use.""" mocker.patch.object( ows_product, "check_product_code_exists", return_value=False, ) data, errors = metadata_logic._validate( 1, {"product_code": "porridge"}, 1, 2, ) assert data == {"product_code": "porridge"} assert errors == [] def test_upc_format() -> None: """Test upc bad format.""" data, errors = metadata_logic._validate( 1, {"upc": "format"}, None, None, ) assert data == {} assert errors == ["badUpcFormat"] def test_upc_duplicate(mocker: MockerFixture) -> None: """Test duplicate upc.""" mocker.patch.object( release, "get_by_upc", return_value={"id": 12}, ) data, errors = metadata_logic._validate( 1, {"upc": "123456789012"}, None, None, ) assert data == {} assert errors == ["duplicateUpc"] def test_upc_duplicate_used_by_self(mocker: MockerFixture) -> None: """Test duplicate upc when it's the same product using it.""" release_id = 12 mocker.patch.object( release, "get_by_upc", return_value={"id": release_id}, ) data, errors = metadata_logic._validate( release_id, {"upc": "123456789012"}, None, None, ) assert data == {"upc": "123456789012"} assert errors == [] def test_invalid_upc() -> None: """Test invalid upc.""" data, errors = metadata_logic._validate( 1, {"upc": "123456789016"}, None, None, ) assert data == {} assert errors == ["invalidUpc"] def test_invalid_line_years() -> None: """Test invalid pline and cline.""" data, errors = metadata_logic._validate( 1, {"p_line_year": "201a", "c_line_year": "201"}, None, None, ) assert data == {} assert errors == ["pLineError", "cLineError"] def test_invalid_line_copyrights() -> None: """Test invalid pline and cline.""" data, errors = metadata_logic._validate( 1, {"p_line_copyright_holder": " yum", "c_line_copyright_holder": "201"}, None, None, ) assert data == {"c_line_copyright_holder": "201"} assert errors == ["pLineError"] def test_validate_release_date() -> None: """Test invalid release_date.""" data, errors = metadata_logic._validate( 1, {"release_date": "2017-01-01"}, None, None, ) assert data == {} assert errors == ["releaseDateInPast"] def test_validate_release_date_bulk() -> None: """Test invalid release_date when validating a bulk item.""" data, errors = metadata_logic._validate( 1, {"release_date": "2017-01-01"}, None, None, is_bulk=True, ) assert data == {"release_date": "2017-01-01"} assert errors == [] @freeze_time("2021-11-23 02:08:00") def test_validate_release_date_with_est() -> None: """Test invalid release_date.""" utc = ZoneInfo("UTC") data, errors = metadata_logic._validate( 1, {"release_date": datetime.now(utc).date().strftime("%Y-%m-%d")}, None, None, ) assert data == { "release_date": datetime.now(utc).date().strftime("%Y-%m-%d"), } assert errors == [] @pytest.mark.parametrize( "form_data", [ ([{"name": "Peter", "role": "Director"}, {"name": "Clint", "role": None}]), ([{"name": "Peter", "role": "Director"}, {"name": None, "role": "Producer"}]), ([{"name": "Conrad", "role": "Director"}, {"name": None, "role": None}]), ], ) def test_validate_contributor(form_data: Any) -> None: """Test validates contributors.""" data, errors = metadata_logic._validate( 1, {"contributors": form_data}, None, None, ) assert data == {} assert errors == ["contributorNameAndRoleMustBePresent"] def test_validate_contributors_performer() -> None: """Test validate contributors with a performer.""" data = { "contributors": [ {"role": "performer", "name": None}, {"role": "Composer", "name": "Jane"}, {"role": "Producer", "name": "Jack"}, ] } errors: list[str] = [] metadata_logic._validate_contributors(data, errors) assert data["contributors"] == [ {"role": "Composer", "name": "Jane"}, {"role": "Producer", "name": "Jack"}, ] assert errors == [] def test_validate_contributors_missing_composer(feature_flags: Any) -> None: """Test validate contributors without a composer returns error.""" feature_flags["ccm_vpb_composer_required"] = True data = { "contributors": [ {"role": "Producer", "name": "Jack"}, {"role": "Director", "name": "Jane"}, ] } errors: list[str] = [] metadata_logic._validate_contributors(data, errors) assert errors == ["composerRequired"] def test_validate_contributors_composer_case_insensitive(feature_flags: Any) -> None: """Test validates contributors accept both Composer and composer.""" feature_flags["ccm_vpb_composer_required"] = True # Test with lowercase data = { "contributors": [ {"role": "composer", "name": "Jane Smith"}, {"role": "Producer", "name": "Jack"}, ] } errors: list[str] = [] metadata_logic._validate_contributors(data, errors) assert errors == [] # Test with uppercase data = { "contributors": [ {"role": "COMPOSER", "name": "Jane Smith"}, {"role": "Producer", "name": "Jack"}, ] } errors = [] metadata_logic._validate_contributors(data, errors) assert errors == [] def test_bulk_ingest_metadata_invalid_item(mocker: MockerFixture) -> None: """Test bulk ingest metadata with an invalid item.""" mock_db_session(mocker) data = [{"vendor_id": 1, "video_title": "E1", "p_line_year": "THIS IS INVALID"}] result = metadata_logic.bulk_ingest_metadata(data) errors = result["errors"] assert errors is not None assert errors[0]["type"] == "ValidationError" assert errors[0]["errors"] == "pLineError" assert errors[0]["item"] == data[0] @freeze_time(datetime(2019, 1, 1, 0, 0, 0)) def test_bulk_ingest_metadata_with_release_id(mocker: MockerFixture) -> None: """Test bulk ingest metadata for items with a release_id.""" mock_db_session(mocker) data = [ # This item will fail because it's missing a UPC { "vendor_id": 1, "video_title": "E1", "release_id": 1, "channel_name": "Channel 1", }, # This item will fail because a product_video already exists { "vendor_id": 1, "video_title": "E2", "release_id": 1, "upc": 111111111111, "channel_name": "Channel 1", }, # This item will fail because the release doesn't exist { "vendor_id": 1, "video_title": "E3", "release_id": 2, "upc": 222222222222, "channel_name": "Channel 1", }, # This item will fail because the product_video fails to be created { "vendor_id": 1, "video_title": "E4", "release_id": 3, "upc": 333333333333, "channel_name": "Channel 1", }, # This item will succeed { "vendor_id": 1, "video_title": "S1", "release_id": 4, "upc": 444444444444, "channel_name": "Channel 1", }, ] get_product_video_side_effect = [ {"id": 1}, None, None, None, ] get_release_side_effect = [ None, {"id": 1}, {"id": 1}, ] get_channels_side_effect: list[Any] = [ [{"channel_name": "Channel 1", "vevo_controlled": "No"}], [{"channel_name": "Channel 1", "vevo_controlled": "No"}], ] insert_product_video_side_effect = [ None, {"id": 4}, ] mocker.patch.object(metadata_logic, "_validate_upc") mocker.patch.object(product_video, "get", side_effect=get_product_video_side_effect) mocker.patch.object( release, "get_by_id_with_vendor", side_effect=get_release_side_effect ) mocker.patch.object( metadata_logic, "_get_channels_for_artist", side_effect=get_channels_side_effect ) mocker.patch.object( product_video, "create", side_effect=insert_product_video_side_effect ) result = metadata_logic.bulk_ingest_metadata(data) errors = result["errors"] successes = result["successes"] assert errors is not None assert successes is not None assert errors[0]["type"] == "ReleaseError" assert errors[0]["errors"] == "If a release_id is provided, a UPC is required." assert errors[0]["item"] == data[0] assert errors[1]["type"] == "ReleaseError" assert errors[1]["errors"] == ( "The provided release_id already exists in product_video." ) assert errors[1]["item"] == data[1] assert errors[2]["type"] == "ReleaseError" assert errors[2]["errors"] == "The provided release_id does not exist." assert errors[2]["item"] == data[2] assert errors[3]["type"] == "ProductVideoError" assert errors[3]["errors"] == "Failed to insert product video" assert errors[3]["item"] == data[3] assert successes[0]["result"] == {"id": 4} # product_video.create is patched via mock_db_session; access via module attr call_args = dict(product_video.create.call_args_list[1][0][0]) # type: ignore[attr-defined] call_args.pop("migrated_metadata_at") assert call_args == { "vendor_id": 1, "video_title": "S1", "release_id": 4, "upc": 444444444444, "channel_name": "Channel 1", "channel_selection": "Channel 1", "vevo_controlled": "No", } def test_bulk_ingest_metadata_with_project_id(mocker: MockerFixture) -> None: """Test bulk ingest metadata for items with a project_id.""" mock_db_session(mocker) data = [ # This item will fail because the project doesn't exist { "vendor_id": 1, "video_title": "E1", "project_id": 1, "channel_name": "Channel 1", }, # This item will fail because the project does not match the vendor { "vendor_id": 1, "video_title": "S1", "project_id": 2, "channel_name": "Channel 1", }, # This item will fail because the UPC will fail to be generated. { "vendor_id": 1, "video_title": "E2", "project_id": 2, "channel_name": "Channel 1", }, # This item will fail because the release will fail to be created { "vendor_id": 1, "video_title": "E3", "project_id": 3, "channel_name": "Channel 1", }, # This item will succeed { "vendor_id": 1, "video_title": "S1", "project_id": 2, "channel_name": "Channel 1", }, ] get_project_side_effect = [ HTTPError("Project error code"), {"project_id": 2, "project_code": "2", "vendor_id": 2}, {"project_id": 2, "project_code": "2", "vendor_id": 1}, {"project_id": 2, "project_code": "2", "vendor_id": 1}, {"project_id": 2, "project_code": "2", "vendor_id": 1}, ] generate_placeholder_upc_side_effect = [ HTTPError("Product error code"), 111111111111, 222222222222, ] create_release_side_effect = [ SQLAlchemyError("Failed to create release"), {"release_id": 1}, ] get_channels_side_effect: list[Any] = [ [{"channel_name": "Channel 1", "vevo_controlled": "No"}] ] insert_product_video_side_effect = [{"id": 1}] mocker.patch.object( ows_project, "get_project_by_id", side_effect=get_project_side_effect ) mocker.patch.object( ows_product, "generate_placeholder_upc", side_effect=generate_placeholder_upc_side_effect, ) mocker.patch.object(release, "create", side_effect=create_release_side_effect) mocker.patch.object( metadata_logic, "_get_channels_for_artist", side_effect=get_channels_side_effect ) mocker.patch.object( product_video, "create", side_effect=insert_product_video_side_effect ) result = metadata_logic.bulk_ingest_metadata(data) errors = result["errors"] successes = result["successes"] assert errors is not None assert successes is not None assert errors[0]["type"] == "ProjectError" assert errors[0]["errors"] == "Project error code" assert errors[0]["item"] == data[0] assert errors[1]["type"] == "ProjectError" assert errors[1]["errors"] == ( "The provided project_id does not belong to the requesting account." ) assert errors[1]["item"] == data[1] assert errors[2]["type"] == "ProductError" assert errors[2]["errors"] == "Product error code" assert errors[2]["item"] == data[2] assert errors[3]["type"] == "ReleaseError" assert errors[3]["errors"] == "Failed to create release" assert errors[3]["item"] == data[3] assert successes[0]["result"] == {"id": 1} def test_bulk_ingest_metadata_with_project_data(mocker: MockerFixture) -> None: """Test bulk ingest metadata for items with some project data.""" mock_db_session(mocker) data = [ # This item will fail because the project will fail to be created { "vendor_id": 1, "video_title": "E1", "project_name": "One", "project_code": "1", "artist_id": 1, "channel_name": "Channel 1", }, # This item will succeed. { "vendor_id": 1, "video_title": "S1", "project_name": "One", "project_code": "1", "artist_id": 1, "channel_name": "Channel 1", }, # This item will succeed. { "vendor_id": 1, "subaccount_id": 1, "video_title": "S2", "project_name": "Two", "project_code": "2", "artist_id": 2, "channel_name": "Channel 1", }, ] create_project_side_effect = [ HTTPError("Project error code"), {"project_id": 1, "project_code": "1"}, {"project_id": 2, "project_code": "2"}, ] generate_placeholder_upc_side_effect = [ 111111111111, 222222222222, ] create_release_side_effect = [ {"release_id": 1}, {"release_id": 2}, ] get_channels_side_effect: list[Any] = [ [{"channel_name": "Channel 1", "vevo_controlled": "No"}], [{"channel_name": "Channel 1", "vevo_controlled": "No"}], ] insert_product_video_side_effect = [ {"id": 1}, {"id": 2}, ] mocker.patch.object( ows_project, "create_project", side_effect=create_project_side_effect ) mocker.patch.object( ows_product, "generate_placeholder_upc", side_effect=generate_placeholder_upc_side_effect, ) mocker.patch.object(release, "create", side_effect=create_release_side_effect) mocker.patch.object( metadata_logic, "_get_channels_for_artist", side_effect=get_channels_side_effect ) mocker.patch.object( product_video, "create", side_effect=insert_product_video_side_effect ) result = metadata_logic.bulk_ingest_metadata(data) errors = result["errors"] successes = result["successes"] assert errors is not None assert successes is not None assert errors[0]["type"] == "ProjectError" assert errors[0]["errors"] == "Project error code" assert errors[0]["item"] == data[0] assert successes[0]["result"] == {"id": 1} assert successes[1]["result"] == {"id": 2} def test_bulk_ingest_metadata_channel_errors(mocker: MockerFixture) -> None: """Test bulk ingest metadata when there are some channel errors.""" mock_db_session(mocker) data = [ # This item will fail because the channels will fail to load { "vendor_id": 1, "video_title": "E1", "release_id": 1, "upc": 111111111111, "primary_artist_id": 1, "channel_name": "Channel 1", }, # This item will fail because the channel will not be found { "vendor_id": 1, "video_title": "E2", "release_id": 2, "upc": 222222222222, "primary_artist_id": 2, "channel_name": "Channel 1", }, ] get_product_video_side_effect = [ None, None, ] get_release_side_effect = [{"id": 1}, {"id": 1}] get_channels_side_effect: list[Any] = [ None, [], ] mocker.patch.object(metadata_logic, "_validate_upc") mocker.patch.object(product_video, "get", side_effect=get_product_video_side_effect) mocker.patch.object( release, "get_by_id_with_vendor", side_effect=get_release_side_effect ) mocker.patch.object( metadata_logic, "_get_channels_for_artist", side_effect=get_channels_side_effect ) result = metadata_logic.bulk_ingest_metadata(data) errors = result["errors"] assert errors is not None assert errors[0]["type"] == "ChannelError" assert errors[0]["errors"] == "No channels available" assert errors[0]["item"] == data[0] assert errors[1]["type"] == "ChannelError" assert errors[1]["errors"] == "No channels available" assert errors[1]["item"] == data[1] def test_prepare_associated_track_dict() -> None: """Test prepare_associated_track_dict.""" required_fields = ["tuid", "isrc", "track_name"] response_data = dict.fromkeys(required_fields) response_data.update({"excess_field": 1}) prepared_dict = metadata_logic.prepare_associated_track_dict(response_data) assert prepared_dict is not None assert list(prepared_dict.keys()) == required_fields assert "excess_field" not in prepared_dict def test_migrate_project(mocker: MockerFixture) -> None: """Test migrate the project of product.""" mocker.patch.object(ows_product, "check_product_ownership") mocker.patch.object(ows_project, "check_project_ownership") mocker.patch.object( ows_project, "get_project_by_id", return_value={ "release_id": 1234, "artist_id": 1234567, "project_id": 12, "subaccount_id": 4321, }, ) mocker.patch.object(product_video, "update_primary_artist", return_value=True) update_mock = mocker.patch.object( release, "update", return_value={"release_status": "label_processing"} ) migrate_project_response = metadata_logic.migrate_project(1234, 12, "vendor", 25824) assert migrate_project_response == {"release_status": "label_processing"} update_mock.assert_called_once_with( 1234, {"project_id": 12, "artist_id": 1234567, "subaccount_id": 4321} ) # Start _make_new_placeholder_upc tests def test_make_new_placeholder_upc_with_existing_placeholder_upc( mocker: MockerFixture, ) -> None: """Tests that existing placeholder UPCs are not overwritten.""" get_by_id_with_vendor_mock = mocker.patch.object( release, "get_by_id_with_vendor", autospec=True, return_value={ "upc": 99_999_999_999_999, }, ) generate_placeholder_upc_mock = mocker.patch.object( ows_product, "generate_placeholder_upc", autospec=True, ) product_id = 99 input_release_data: dict[str, Any] = {} metadata_logic._make_new_placeholder_upc(product_id, input_release_data) get_by_id_with_vendor_mock.assert_called_with(product_id) assert "upc" not in input_release_data assert "display_upc" not in input_release_data generate_placeholder_upc_mock.assert_not_called() def test_make_new_placeholder_upc_with_existing_real_upc(mocker: MockerFixture) -> None: """Tests that real UPC get replaced by newly generated placeholders.""" get_by_id_with_vendor_mock = mocker.patch.object( release, "get_by_id_with_vendor", autospec=True, return_value={ "upc": 999_999_999_999, }, ) generate_placeholder_upc_mock = mocker.patch.object( ows_product, "generate_placeholder_upc", autospec=True, return_value=10_000_000_000_000, ) product_id = 99 input_release_data: dict[str, Any] = {} metadata_logic._make_new_placeholder_upc(product_id, input_release_data) get_by_id_with_vendor_mock.assert_called_with(product_id) assert input_release_data["upc"] == 10_000_000_000_000 assert input_release_data["display_upc"] == 10_000_000_000_000 generate_placeholder_upc_mock.assert_called_with() # End _make_new_placeholder_upc tests # Start save_metadata tests def test_save_metadata_fails_if_not_owner(mocker: MockerFixture) -> None: """Tests that ownership errors are propagated.""" product_id = 100 account_type = "test_account_type" account_id = "test_account_id" user_id = None mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, side_effect=ProductOwnershipError(), ) with pytest.raises(ProductOwnershipError): metadata_logic.save_metadata( product_id, {}, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) def test_save_metadata_checks_release_status_for_no_user_id( mocker: MockerFixture, ) -> None: """Tests that release status mismatch is propagated with no user.""" product_id = 100 account_type = "test_account_type" account_id = "test_account_id" user_id = None session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, side_effect=BadRequest(), ) with pytest.raises(BadRequest): metadata_logic.save_metadata( product_id, {}, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_called_with( product_id, frozenset({release_constants.LABEL_PROCESSING}), session=session, ) def test_save_metadata_checks_release_status_for_non_oa_user_id( mocker: MockerFixture, ) -> None: """Tests that release status mismatch is propagated for non-oa user.""" product_id = 100 account_type = "test_account_type" account_id = "test_account_id" user_id = "workstation:123" session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, side_effect=BadRequest(), ) with pytest.raises(BadRequest): metadata_logic.save_metadata( product_id, {}, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_called_with( product_id, frozenset({release_constants.LABEL_PROCESSING}), session=session, ) def test_save_metadata_product_update_failure(mocker: MockerFixture) -> None: """Tests that product_video upsert is called even if it returns falsy.""" product_id = 100 account_type = "test_account_type" account_id = "test_account_id" user_id = "workstation:123" errors = ["mock_error"] data = { "video_title": "Test Video Title", } session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "Post Validation Test Video Title", }, errors, ), ) product_video_data = {"video_title": "Post Validation Test Video Title"} upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.LABEL_PROCESSING}, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_called_with( product_id, frozenset({release_constants.LABEL_PROCESSING}), session=session, ) _validate_mock.assert_called_with( product_id, { "video_title": "Test Video Title", }, account_type, account_id, ) upsert_mock.assert_called_with( { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, session=session, ) assert result == { "video_title": "Post Validation Test Video Title", "errors": errors, } def test_save_metadata_release_update_failure(mocker: MockerFixture) -> None: """Tests that update_release is called and result is returned from video_data.""" product_id = 100 account_type = "test_account_type" account_id = "test_account_id" user_id = "workstation:123" errors = ["mock_error"] data = { "video_title": "Test Video Title", } session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "Post Validation Test Video Title", }, errors, ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.LABEL_PROCESSING}, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_called_with( product_id, frozenset({release_constants.LABEL_PROCESSING}), session=session, ) _validate_mock.assert_called_with( product_id, { "video_title": "Test Video Title", }, account_type, account_id, ) upsert_mock.assert_called_with( { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, session=session, ) update_release_mock.assert_called_with( product_id, { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, product_video_data, ) assert result == { "product_video_return_key": "product_video_return_value", "errors": errors, } def test_save_metadata_not_writing_to_delivery_table(mocker: MockerFixture) -> None: """Tests that workstation user updates don't write to delivery tables.""" product_id = 100 account_type = "test_account_type" account_id = "test_account_id" user_id = "workstation:123" errors = ["mock_error"] data = { "video_title": "Test Video Title", } session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "Post Validation Test Video Title", }, errors, ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.LABEL_PROCESSING}, ) write_to_delivery_tables_mock = mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_called_with( product_id, frozenset({release_constants.LABEL_PROCESSING}), session=session, ) _validate_mock.assert_called_with( product_id, { "video_title": "Test Video Title", }, account_type, account_id, ) upsert_mock.assert_called_with( { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, session=session, ) update_release_mock.assert_called_with( product_id, { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, product_video_data, ) write_to_delivery_tables_mock.assert_not_called() assert result == { "errors": ["mock_error"], "product_video_return_key": "product_video_return_value", } def test_save_metadata_renames_release_name_field( mocker: MockerFixture, app_context: Any ) -> None: """Tests that release_name field is renamed to video_title.""" product_id = 100 account_type = "test_account_type" account_id = "test_account_id" user_id = "workstation:123" errors = ["mock_error"] data = { "release_name": "input_release_name", } session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "release_name": "post_validation_release_name", }, errors, ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.LABEL_PROCESSING}, ) write_to_delivery_tables_mock = mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_called_with( product_id, frozenset({release_constants.LABEL_PROCESSING}), session=session, ) _validate_mock.assert_called_with( product_id, { "release_name": "input_release_name", }, account_type, account_id, ) upsert_mock.assert_called_with( { "release_id": product_id, "video_title": "post_validation_release_name", }, session=session, ) update_release_mock.assert_called_with( product_id, { "release_id": product_id, "video_title": "post_validation_release_name", }, product_video_data, ) write_to_delivery_tables_mock.assert_not_called() assert g.log.warning.mock_calls == [ call( "Unexpected product metadata fields: {'release_name': 'input_release_name'}" ), ] assert result == { "errors": ["mock_error"], "product_video_return_key": "product_video_return_value", } @pytest.mark.parametrize( ("release_status", "expected_errors"), ( (release_constants.LABEL_PROCESSING, ["other", "releaseDateInPast"]), (release_constants.IN_CONTENT, ["other"]), ), ) def test_save_metadata_removes_release_date_in_past_error_for_in_content( mocker: MockerFixture, release_status: Any, expected_errors: Any, ) -> None: """Tests that an in_content release has no releaseDateInPast error.""" product_id = 100 account_type = "test_account_type" account_id = "test_account_id" user_id = "workstation:123" errors = ["other", "releaseDateInPast"] data = { "video_title": "Test Video Title", } session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "Post Validation Test Video Title", }, errors, ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_status}, ) write_to_delivery_tables_mock = mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_called_with( product_id, frozenset({release_constants.LABEL_PROCESSING}), session=session, ) _validate_mock.assert_called_with( product_id, { "video_title": "Test Video Title", }, account_type, account_id, ) upsert_mock.assert_called_with( { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, session=session, ) update_release_mock.assert_called_with( product_id, { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, product_video_data, ) write_to_delivery_tables_mock.assert_not_called() assert result == { "errors": expected_errors, "product_video_return_key": "product_video_return_value", } def test_save_metadata_write_to_delivery_tables_failure(mocker: MockerFixture) -> None: """Tests that write_to_delivery table failure is propagated.""" product_id = 100 account_type = None account_id = None user_id = "oa:123" errors = ["mock_error"] data = { "video_title": "Test Video Title", } session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "Post Validation Test Video Title", }, errors, ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.IN_CONTENT}, ) write_to_delivery_tables_mock = mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, return_value=None, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_not_called() _validate_mock.assert_called_with( product_id, { "video_title": "Test Video Title", }, account_type, account_id, ) upsert_mock.assert_called_with( { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, session=session, ) update_release_mock.assert_called_with( product_id, { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, product_video_data, ) write_to_delivery_tables_mock.assert_called_with( product_id, session=session, ) assert result == { "product_video_return_key": "product_video_return_value", "errors": errors, } def test_save_metadata_writes_to_delivery_tables_if_oa_user_and_in_content( mocker: MockerFixture, ) -> None: """Tests that an OA user updating in_content writes to delivery tables.""" product_id = 100 account_type = None account_id = None user_id = "oa:123" errors = ["mock_error"] data = { "video_title": "Test Video Title", } session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "Post Validation Test Video Title", }, errors, ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.IN_CONTENT}, ) write_to_delivery_tables_mock = mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, return_value=None, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_not_called() _validate_mock.assert_called_with( product_id, { "video_title": "Test Video Title", }, account_type, account_id, ) upsert_mock.assert_called_with( { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, session=session, ) update_release_mock.assert_called_with( product_id, { "video_title": "Post Validation Test Video Title", "release_id": product_id, }, product_video_data, ) write_to_delivery_tables_mock.assert_called_with( product_id, session=session, ) assert result == { "errors": ["mock_error"], "product_video_return_key": "product_video_return_value", } def test_save_metadata_logs_unexpected_fields( mocker: MockerFixture, app_context: Any ) -> None: """Tests save_metadata logs unexpected fields.""" product_id = 100 account_type = None account_id = None user_id = "oa:123" errors = ["mock_error"] expected_unexpected_fields = { "latest_approval_job_id": 123, "product_manager_id": 456, "release_name": "release name", "not_for_distribution": "SwitchboardDummy", "vendor_release_identifier": "vendor release identifier", } data = { **{key: key for key in USER_SPECIFIABLE_FIELDS}, **expected_unexpected_fields, } session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "Test Video Title", }, errors, ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } product_video_upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) product_manager_mapping_product_upsert_mock = mocker.patch.object( product_manager_mapping_product, "upsert", autospec=True, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.IN_CONTENT}, ) write_to_delivery_tables_mock = mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, return_value=None, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_not_called() _validate_mock.assert_called_with(product_id, data, account_type, account_id) product_video_upsert_mock.assert_called_with( { "video_title": "Test Video Title", "release_id": product_id, }, session=session, ) product_manager_mapping_product_upsert_mock.assert_not_called() update_release_mock.assert_called_with( product_id, { "video_title": "Test Video Title", "release_id": product_id, }, product_video_data, ) write_to_delivery_tables_mock.assert_called_with(product_id, session=session) assert g.log.warning.mock_calls == [ call(f"Unexpected product metadata fields: {expected_unexpected_fields}"), ] assert result == { "errors": errors, "product_video_return_key": "product_video_return_value", } def test_save_metadata_updates_nothing_when_no_data(mocker: MockerFixture) -> None: """Tests save_metadata updates nothing when no data.""" product_id = 100 account_type = None account_id = None user_id = "oa:123" errors = ["mock_error"] data: dict[str, Any] = {} mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( {}, errors, ), ) product_video_upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, ) product_manager_mapping_product_upsert_mock = mocker.patch.object( product_manager_mapping_product, "upsert", autospec=True, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, ) write_to_delivery_tables_mock = mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_not_called() _validate_mock.assert_called_with(product_id, {}, account_type, account_id) product_video_upsert_mock.assert_not_called() product_manager_mapping_product_upsert_mock.assert_not_called() update_release_mock.assert_not_called() write_to_delivery_tables_mock.assert_not_called() assert result assert result == { "errors": errors, } def test_save_metadata_success_with_product_manager_id( mocker: MockerFixture, app_context: Any ) -> None: """Test save metadata with product manager id.""" product_id = 100 account_type = None account_id = None user_id = "oa:123" data = {"video_title": "Test Video Title", "product_manager_id": 69} session = mock_db_session(mocker) check_product_ownership_mock = mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) check_release_status_before_action_mock = mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) _validate_mock = mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "Post Validation Test Video Title", "product_manager_id": 69, }, [], ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } product_video_upsert_mock = mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) product_manager_mapping_product_upsert_mock = mocker.patch.object( product_manager_mapping_product, "upsert", autospec=True, return_value=None, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.IN_CONTENT}, ) write_to_delivery_tables_mock = mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, return_value=None, ) result = metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) check_product_ownership_mock.assert_called_with( product_id, account_type, account_id, ) check_release_status_before_action_mock.assert_not_called() _validate_mock.assert_called_with( product_id, {"video_title": "Test Video Title", "product_manager_id": 69}, account_type, account_id, ) product_video_upsert_mock.assert_called_with( { "video_title": "Post Validation Test Video Title", "release_id": product_id, "product_manager_id": 69, }, session=session, ) product_manager_mapping_product_upsert_mock.assert_called_with( product_id, 69, session=session ) update_release_mock.assert_called_with( product_id, { "video_title": "Post Validation Test Video Title", "release_id": product_id, "product_manager_id": 69, }, product_video_data, ) write_to_delivery_tables_mock.assert_called_with(product_id, session=session) assert g.log.warning.mock_calls == [ call("Unexpected product metadata fields: {'product_manager_id': 69}"), ] assert result == { "errors": [], "product_video_return_key": "product_video_return_value", "product_manager_id": 69, } # End save_metadata tests # Start update_upc_for_carveouts tests def test_update_upc_for_carveouts_upc_fetch_failure(mocker: MockerFixture) -> None: """Tests that update_upc_for_carveouts continues even when UPCs differ.""" product_id = 100 saved_data = { "upc": "saved_upc", } session = mock_db_session(mocker) get_upc_by_product_id_mock = mocker.patch.object( release, "get_upc_by_product_id", autospec=True, return_value={"upc": "different_upc"}, ) update_upc_mock = mocker.patch.object( carveouts, "update_upc", autospec=True, ) metadata_logic.update_upc_for_carveouts(product_id, saved_data) get_upc_by_product_id_mock.assert_called_with(product_id) update_upc_mock.assert_called_with(product_id, "saved_upc", session) def test_update_upc_for_carveouts_upc_matches_noop(mocker: MockerFixture) -> None: """Tests that saving the already set UPC has no effect.""" product_id = 100 saved_data = { "upc": "release_upc", } get_upc_by_product_id_mock = mocker.patch.object( release, "get_upc_by_product_id", autospec=True, return_value={"upc": "release_upc"}, ) update_upc_mock = mocker.patch.object( carveouts, "update_upc", autospec=True, ) metadata_logic.update_upc_for_carveouts(product_id, saved_data) get_upc_by_product_id_mock.assert_called_with(product_id) update_upc_mock.assert_not_called() def test_update_upc_for_carveouts_upc_changed_updates_tables( mocker: MockerFixture, ) -> None: """Tests that updating the UPC updates the carveout tables.""" product_id = 100 saved_data = { "upc": "saved_upc", } session = mock_db_session(mocker) get_upc_by_product_id_mock = mocker.patch.object( release, "get_upc_by_product_id", autospec=True, return_value={"upc": "release_upc"}, ) update_upc_mock = mocker.patch.object( carveouts, "update_upc", autospec=True, return_value=None, ) metadata_logic.update_upc_for_carveouts(product_id, saved_data) get_upc_by_product_id_mock.assert_called_with(product_id) update_upc_mock.assert_called_with(product_id, "saved_upc", session) # End update_upc_for_carveouts tests # Start update_release tests def test_update_release_maps_data(mocker: MockerFixture) -> None: """Tests that saved_data is mapped to the releases table format.""" test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", } complete_video_data: dict[str, Any] = {} _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", } ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", }, ) assert result is test_response def test_update_release_sets_keywords_from_saved_data(mocker: MockerFixture) -> None: """Tests that keywords are converted to a comma-separated string.""" test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "keywords": ["keyword1", "keyword2"], } complete_video_data: dict[str, Any] = {} _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "keywords": ["keyword1", "keyword2"], } ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "keywords": "keyword1, keyword2", }, ) assert result is test_response def test_update_release_does_not_set_c_line_if_year_updated_with_no_holder( mocker: MockerFixture, ) -> None: """ Tests c_line field behavior. Tests that the c_line field of release is not set if the c_line_year is updated and c_line_copyright_holder isn't set. """ test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "c_line_year": "c_line_year_value", } complete_video_data = { "c_line_year": "9999", "c_line_copyright_holder": None, } _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "c_line_year": "c_line_year_value", } ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", }, ) assert result is test_response def test_update_release_sets_c_line_if_c_line_year_updated( mocker: MockerFixture, ) -> None: """ Tests that the c_line field of release is set if c_line_year is updated. The c_line field should be set based upon whatever is saved in product_video, not what was sent in as a value. """ test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "c_line_year": "c_line_year_value", } complete_video_data = { "c_line_year": "9999", "c_line_copyright_holder": "Example Copyright Holder", } _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "c_line_year": "c_line_year_value", } ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "c_line": "9999 Example Copyright Holder", }, ) assert result is test_response def test_update_release_does_not_set_c_line_if_holder_updated_with_no_year( mocker: MockerFixture, ) -> None: """ Tests c_line field behavior. Tests that the c_line field of release is not set if the c_line_copyright_holder is updated and c_line_copyright_year isn't set. """ test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "c_line_copyright_holder": "c_line_copyright_holder_value", } complete_video_data = { "c_line_year": None, "c_line_copyright_holder": "Example Copyright Holder", } _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "c_line_copyright_holder": "c_line_copyright_holder_value", } ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", }, ) assert result is test_response def test_update_release_sets_c_line_if_c_line_copyright_holder_updated( mocker: MockerFixture, ) -> None: """ Tests c_line field behavior. Tests that the c_line field of release is set if c_line_copyright_holder is updated. The c_line field should be set based upon whatever is saved in product_video, not what was sent in as a value. """ test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "c_line_copyright_holder": "c_lin_copyright_holder_value", } complete_video_data = { "c_line_year": "9999", "c_line_copyright_holder": "Example Copyright Holder", } _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "c_line_copyright_holder": "c_lin_copyright_holder_value", } ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "c_line": "9999 Example Copyright Holder", }, ) assert result is test_response @pytest.mark.parametrize( ("saved_new_release_value", "expected_new_release_value"), ( (False, release_constants.NEW_RELEASE_CATALOG), (True, release_constants.NEW_RELEASE_NEW), ), ) def test_update_release_sets_new_release( mocker: MockerFixture, saved_new_release_value: Any, expected_new_release_value: Any, ) -> None: """Tests the new_release field is set based on the input boolean.""" test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "new_release": saved_new_release_value, } complete_video_data: dict[str, Any] = {} _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "new_release": saved_new_release_value, } ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "new_release": expected_new_release_value, }, ) assert result is test_response def test_update_release_upc_empty_string_ignored(mocker: MockerFixture) -> None: """Tests that a blank UPC has no effect.""" test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "upc": "", } complete_video_data: dict[str, Any] = {} _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) _make_new_placeholder_upc_mock = mocker.patch.object( metadata_logic, "_make_new_placeholder_upc", autospec=True, ) update_upc_for_carveouts_mock = mocker.patch.object( metadata_logic, "update_upc_for_carveouts", autospec=True, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "upc": "", } ) _make_new_placeholder_upc_mock.assert_not_called() update_upc_for_carveouts_mock.assert_not_called() update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", }, ) assert result is test_response def test_update_release_upc_set_to_none_sets_placeholder_upc( mocker: MockerFixture, ) -> None: """Tests that setting UPC to None sets a placeholder UPC.""" test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "upc": None, } complete_video_data: dict[str, Any] = {} _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) def make_new_placeholder_upc_side_effect( _product_id: Any, release_data: Any ) -> None: assert release_data == { "mapped_release_data_key": "mapped_release_data_value", } release_data["upc"] = "placeholder_upc" release_data["display_upc"] = "placeholder_upc" _make_new_placeholder_upc_mock = mocker.patch.object( metadata_logic, "_make_new_placeholder_upc", autospec=True, side_effect=make_new_placeholder_upc_side_effect, ) update_upc_for_carveouts_mock = mocker.patch.object( metadata_logic, "update_upc_for_carveouts", autospec=True, return_value=None, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "upc": None, } ) # Call values asserted in make_new_placeholder_upc_side_effect due to # mutable arguments _make_new_placeholder_upc_mock.assert_called_once() update_upc_for_carveouts_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "upc": "placeholder_upc", "display_upc": "placeholder_upc", }, ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "upc": "placeholder_upc", "display_upc": "placeholder_upc", }, ) assert result is test_response def test_update_release_upc_set_to_value_propagates_carveouts_and_updates_release( mocker: MockerFixture, ) -> None: """Tests that setting a UPC calls update_upc_for_carveouts and updates release.""" test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "upc": "input_upc", } complete_video_data: dict[str, Any] = {} _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) _make_new_placeholder_upc_mock = mocker.patch.object( metadata_logic, "_make_new_placeholder_upc", autospec=True, ) update_upc_for_carveouts_mock = mocker.patch.object( metadata_logic, "update_upc_for_carveouts", autospec=True, return_value=None, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "upc": "input_upc", } ) _make_new_placeholder_upc_mock.assert_not_called() update_upc_for_carveouts_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "upc": "input_upc", "display_upc": "input_upc", }, ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "upc": "input_upc", "display_upc": "input_upc", }, ) assert result is test_response def test_update_release_upc_set_to_value_updates_carveouts_and_release( mocker: MockerFixture, ) -> None: """Tests that setting a UPC works and updated the carveouts tables.""" test_response = {"release_status": "label_processing"} product_id = 100 saved_data = { "saved_data_key": "saved_data_value", "upc": "input_upc", } complete_video_data: dict[str, Any] = {} _map_product_video_data_to_release_data_mock = mocker.patch.object( metadata_logic, "_map_product_video_data_to_release_data", autospec=True, return_value={ "mapped_release_data_key": "mapped_release_data_value", }, ) update_mock = mocker.patch.object( release, "update", autospec=True, return_value=test_response, ) _make_new_placeholder_upc_mock = mocker.patch.object( metadata_logic, "_make_new_placeholder_upc", autospec=True, ) update_upc_for_carveouts_mock = mocker.patch.object( metadata_logic, "update_upc_for_carveouts", autospec=True, return_value=None, ) result = metadata_logic.update_release( product_id, saved_data, complete_video_data, ) _map_product_video_data_to_release_data_mock.assert_called_with( { "saved_data_key": "saved_data_value", "upc": "input_upc", } ) _make_new_placeholder_upc_mock.assert_not_called() update_upc_for_carveouts_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "upc": "input_upc", "display_upc": "input_upc", }, ) update_mock.assert_called_with( product_id, { "mapped_release_data_key": "mapped_release_data_value", "upc": "input_upc", "display_upc": "input_upc", }, ) assert result is test_response # End update_release # Start _map_product_video_data_to_release_data tests @pytest.mark.parametrize( ("test_input", "test_output"), ( # All fields individually ( { "imprint": "test", }, { "label": "test", }, ), ( { "video_title": "test", }, { "release_name": "test", }, ), ( { "language_of_video_title": "test", }, { "meta_language": "test", }, ), ( { "description": "test", }, { "description": "test", }, ), ( { "product_code": "test", }, { "product_code": "test", }, ), ( { "language_of_video_content": "test", }, { "language_id": "test", }, ), ( { "original_release_date": "test", }, { "original_release_date": "test", }, ), ( { "release_date": "test", }, { "release_date": "test", "sale_start_date": "test", }, ), ( { "not_for_distribution": "test", }, { "not_for_distribution": "test", }, ), ( { "genre_id": "test", }, { "genre_id": "test", }, ), ( { "version": "test", }, { "version": "test", }, ), ( { "special_instructions": "test", }, { "special_instructions": "test", }, ), # All fields at once ( { "imprint": "imprint_input", "video_title": "video_title_input", "language_of_video_title": "language_of_video_title_input", "description": "description_input", "product_code": "product_code_input", "language_of_video_content": "language_of_video_content_input", "original_release_date": "original_release_date_input", "release_date": "release_date_input", "not_for_distribution": "not_for_distribution_input", "genre_id": "genre_id_input", "version": "version_input", "special_instructions": "special_instructions_input", "vendor_catalog_number": "vendor_catalog_number", }, { "label": "imprint_input", "release_name": "video_title_input", "meta_language": "language_of_video_title_input", "description": "description_input", "product_code": "product_code_input", "language_id": "language_of_video_content_input", "original_release_date": "original_release_date_input", "release_date": "release_date_input", "sale_start_date": "release_date_input", "not_for_distribution": "not_for_distribution_input", "genre_id": "genre_id_input", "version": "version_input", "special_instructions": "special_instructions_input", "vendor_catalog_number": "vendor_catalog_number", }, ), ), ) def test_map_product_video_data_to_release_data( test_input: Any, test_output: Any ) -> None: """Tests that the input value is mapped to the output value.""" result = metadata_logic._map_product_video_data_to_release_data(test_input) assert result == test_output @ar_db_utils.test_schema def test_save_metadata_success_with_vendor_release_identifier( mocker: MockerFixture, app_context: Any, ) -> None: """Test save metadata with vendor_release_identifier.""" product_id = 100 account_type = None account_id = None user_id = "oa:123" data = {"video_title": "video title", "vendor_release_identifier": "G014354523452"} mocker.patch.object( ows_product, "check_product_ownership", autospec=True, return_value=True, ) mocker.patch.object( status_checker, "check_release_status_before_action", autospec=True, return_value=None, ) mocker.patch.object( metadata_logic, "_validate", autospec=True, return_value=( { "video_title": "video title", "vendor_release_identifier": "G014354523452", }, [], ), ) product_video_data = { "product_video_return_key": "product_video_return_value", } mocker.patch.object( product_video, "upsert", autospec=True, return_value=product_video_data, ) mocker.patch.object( product_manager_mapping_product, "upsert", autospec=True, return_value=None, ) update_release_mock = mocker.patch.object( metadata_logic, "update_release", autospec=True, return_value={"release_status": release_constants.IN_CONTENT}, ) mocker.patch.object( approval_logic, "write_to_delivery_tables", autospec=True, return_value=None, ) metadata_logic.save_metadata( product_id, data, account_type, account_id, user_id, ) update_release_mock.assert_called_with( product_id, { "video_title": "video title", "release_id": product_id, "vendor_release_identifier": "G014354523452", }, product_video_data, ) assert g.log.warning.mock_calls == [ call( "Unexpected product metadata fields: {'vendor_release_identifier': 'G014354523452'}" ), ]