"""Auto approval logic tests.""" import json from http import HTTPStatus from typing import Any from unittest.mock import MagicMock, patch import jsonschema import pytest from pytest_mock import MockerFixture from video.config import AUTO_APPROVE_USER from video.constants import ( approval as approval_constants, notifications as notifications_constants, ) from video.logic import ( approval as approval_logic, auto_approval as auto_approval_logic, job as job_logic, product as product_logic, ) from video.models.sql.classes import ( auto_carveout_dms as auto_carveout_dms_model, product_video, video_type as video, ) @pytest.mark.parametrize( ("final_release_status", "duped_isrc_bool"), [ # Case where both release and content approval are auto-applied # so final approval is also auto-applied ("in_content", False), # Case where the isrc is already used # This case exists because the logic for isrc check lives in # the submit function ("transfer_to_content", True), ], ) def test_submit_auto_approval( mocker: MockerFixture, final_release_status: Any, duped_isrc_bool: Any, mock_submit_pre_auto_approval_funcs: Any, ) -> None: """Test auto approval on product submission.""" # setup test data upc = 12 isrc = 15 product_id = 1 account_type = "vendor" account_id = 1 product_video_object = { "is_valid": True, "upc": upc, "isrc": isrc, "submitted_at": None, "special_instructions": None, "channel_selection": "test", "latest_pipeline_run_id": 1234, "type_of_video": "Official Music Video", } release_response = {"release_status": "transfer_to_content"} release_response_with_vendor = { "release_status": "transfer_to_content", "vendor_id": 12345, } # setup mocks method in confest.py mock_submit_pre_auto_approval_funcs( product_video_object, release_response, release_response_with_vendor ) # Mocking of auto approval logic starts here mocked_auto_apply_label_mgr_check = mocker.patch.object( auto_approval_logic, "should_auto_apply_label_mgr_approval", return_value=True ) mocked_auto_apply_label_mgr_approval = mocker.patch.object( auto_approval_logic, "auto_apply_label_mgr_approval" ) mocked_auto_apply_carveouts = mocker.patch.object( auto_approval_logic, "auto_apply_video_type_carveouts" ) mocked_auto_apply_vidops_check = mocker.patch.object( auto_approval_logic, "should_auto_apply_vidops_approval", return_value=True ) mocked_auto_apply_vidops_approval = mocker.patch.object( auto_approval_logic, "auto_apply_vidops_approval" ) mocked_should_final_approval_apply = mocker.patch.object( auto_approval_logic, "should_final_approval_be_applied", return_value=True ) mocked_auto_apply_final_approval = mocker.patch.object( auto_approval_logic, "apply_final_approval", return_value={"release_status": final_release_status}, ) mocker.patch.object( getattr(product_logic, "ows_track"), "is_isrc_used", return_value={"used": duped_isrc_bool}, ) # Mocked for the case when the isrc is already used mocked_revert_label_mgr_approval = mocker.patch.object( approval_logic, "revert_label_mgr_approval" ) submit_response = product_logic.submit(product_id, account_type, account_id) # Assertions mocked_auto_apply_label_mgr_check.assert_called_with(product_video_object) assert mocked_auto_apply_label_mgr_check.call_count == 2 mocked_auto_apply_label_mgr_approval.assert_called_once_with(product_id) mocked_auto_apply_carveouts.assert_called_once_with(product_id) mocked_auto_apply_vidops_check.assert_called_once_with(product_id) mocked_auto_apply_vidops_approval.assert_called_once_with(product_id) mocked_should_final_approval_apply.assert_called_once_with(product_id) mocked_auto_apply_final_approval.assert_called_once_with( product_id, product_video_object, release_response_with_vendor ) if duped_isrc_bool: mocked_revert_label_mgr_approval.assert_called_once_with(product_id) assert submit_response.get("release_status") == final_release_status @pytest.mark.parametrize( ("special_instructions", "channel_selection", "expected_return"), [ # Happy path (None, "https://www.youtube.com/somechannel", True), # Sad Path ("some instructions here", "some channel", False), # Sad Path (None, "I don't see the channel I want", False), # Sad Path ("some instructions", "I don't see the channel I want", False), ], ) def test_should_auto_apply_label_mgr_approval( mocker: MockerFixture, special_instructions: Any, channel_selection: Any, expected_return: Any, ) -> None: """Test should_auto_apply_label_mgr_approval.""" upc = 12 isrc = 15 product_video_object = { "is_valid": True, "upc": upc, "isrc": isrc, "submitted_at": None, "special_instructions": special_instructions, "channel_selection": channel_selection, } actual_return = auto_approval_logic.should_auto_apply_label_mgr_approval( product_video_object ) assert expected_return == actual_return @patch("video.logic.auto_approval.g", spec=["log"]) def test_auto_apply_label_mgr_approval( mocked_logger: Any, mocker: MockerFixture ) -> None: """Test auto_apply_label_mgr_approval.""" product_id = 12 mocked_approval_change = mocker.patch.object( getattr(auto_approval_logic, "approval_logic"), "change", ) auto_approval_logic.auto_apply_label_mgr_approval(product_id) mocked_approval_change.assert_called_once_with( product_id, AUTO_APPROVE_USER, data={"approval_type": approval_constants.RELEASE_TYPE, "value": True}, ) mocked_logger.log.info.assert_called_once_with( "Auto-applied release approval (label manager) for product id: {}".format( product_id ) ) def test_should_auto_apply_vidops_approval(mocker: MockerFixture) -> None: """Test should_auto_apply_vidops_approval.""" product_id = 12 mocked_itunes_carveout_check = mocker.patch.object( getattr(auto_approval_logic, "carveouts_python_logic"), "is_itunes_carved_out" ) auto_approval_logic.should_auto_apply_vidops_approval(product_id) mocked_itunes_carveout_check.assert_called_once_with(product_id) @patch("video.logic.auto_approval.g", spec=["log"]) def test_auto_apply_vidops_approval(mocked_logger: Any, mocker: MockerFixture) -> None: """Test auto_apply_vidops_approval.""" # Setup data and mocks product_id = 12 mocked_approval_change = mocker.patch.object( getattr(auto_approval_logic, "approval_logic"), "change", ) # Execute function auto_approval_logic.auto_apply_vidops_approval(product_id) # Assertions mocked_approval_change.assert_called_once_with( product_id, AUTO_APPROVE_USER, data={"approval_type": approval_constants.CONTENT_TYPE, "value": True}, ) mocked_logger.log.info.assert_called_once_with( "Auto-applied content approval (vid ops) for product id: {}".format(product_id) ) @pytest.mark.parametrize( ("release_approver", "content_approver", "expected_return"), [ # Happy path (AUTO_APPROVE_USER, AUTO_APPROVE_USER, True), # Sad path (123, AUTO_APPROVE_USER, False), # Sad path (AUTO_APPROVE_USER, 123, False), # Sad path (123, 321, False), ], ) def test_should_final_approval_be_applied( mocker: MockerFixture, release_approver: Any, content_approver: Any, expected_return: Any, ) -> None: """Test should_final_approval_be_applied.""" product_id = 12 approval_get = { "release_approved_by": release_approver, "content_approved_by": content_approver, } mock_approval_get: MagicMock = mocker.patch.object( approval_logic, "get", return_value=approval_get ) actual_return = auto_approval_logic.should_final_approval_be_applied(product_id) mock_approval_get.assert_called_once_with(product_id) assert expected_return == actual_return @patch("video.logic.auto_approval.g", spec=["log"]) def test_apply_final_approval(mocked_logger: Any, mocker: MockerFixture) -> None: """Test apply_final_approval.""" # Setup data and mocks product_id = 12 upc = 321 isrc = 15 release_message: dict[str, Any] = {"message": "blah"} latest_pipeline_run_id = 543 video_product_message = { "upc": upc, "isrc": isrc, "latest_pipeline_run_id": latest_pipeline_run_id, } approval_change_data = { "approval_type": approval_constants.FINAL_APPROVAL_TYPE, "value": True, "upc": upc, "isrc": isrc, "approval_in_progress": True, } mocked_approval_change = mocker.patch.object( getattr(auto_approval_logic, "approval_logic"), "change", return_value=True ) mocked_create_workflow = mocker.patch.object( auto_approval_logic, "_create_approval_workflow_job" ) mocked_send_notification = mocker.patch.object( auto_approval_logic, "send_final_approval_notification" ) mocked_g = mocker.patch.object(auto_approval_logic, "g") mocked_get_release = mocker.patch.object( getattr(auto_approval_logic, "release"), "get" ) # Execute function auto_approval_logic.apply_final_approval( product_id, video_product_message, release_message ) # Assertions mocked_approval_change.assert_called_once_with( product_id, AUTO_APPROVE_USER, data=approval_change_data ) mocked_create_workflow.assert_called_once_with( product_id, upc, latest_pipeline_run_id ) mocked_send_notification.assert_called_once_with( video_product_message, release_message ) mocked_g.log.info.assert_called_once_with( "Auto-applied final approval for product id: {}".format(product_id) ) mocked_get_release.assert_called_once_with(product_id) def test_apply_final_approval_new_flow(mocker: MockerFixture) -> None: """Test apply_final_approval.""" # Setup data and mocks product_id = 12 upc = 321 isrc = 15 release_message: dict[str, Any] = {"message": "blah"} latest_pipeline_run_id = 543 video_product_message = { "upc": upc, "isrc": isrc, "latest_pipeline_run_id": latest_pipeline_run_id, } approval_change_data = { "approval_type": approval_constants.FINAL_APPROVAL_TYPE, "value": True, "upc": upc, "isrc": isrc, "approval_in_progress": True, } mocked_approval_change = mocker.patch.object( getattr(auto_approval_logic, "approval_logic"), "change", return_value=True ) mocked_create_workflow = mocker.patch.object( auto_approval_logic, "_create_approval_workflow_job" ) mocked_send_notification = mocker.patch.object( auto_approval_logic, "send_final_approval_notification" ) mocked_g = mocker.patch.object(auto_approval_logic, "g") mocked_get_release = mocker.patch.object( getattr(auto_approval_logic, "release"), "get" ) # Execute function auto_approval_logic.apply_final_approval( product_id, video_product_message, release_message ) # Assertions mocked_approval_change.assert_called_once_with( product_id, AUTO_APPROVE_USER, data=approval_change_data ) mocked_create_workflow.assert_called_once_with( product_id, upc, latest_pipeline_run_id ) mocked_send_notification.assert_called_once_with( video_product_message, release_message ) mocked_g.log.info.assert_called_once_with( "Auto-applied final approval for product id: {}".format(product_id) ) mocked_get_release.assert_called_once_with(product_id) def test_create_approval_workflow_job(mocker: MockerFixture) -> None: """Test creation of approval workflow job.""" # Setup data and mocks product_id = 1 upc = "12" latest_pipeline_run_id = 1234 expected_workflow_response = [ { "context": { "correlation_id": "3163f53e-e2c9-11e8-865e-0242ac110002.1", "datetime": "2018-11-07T20:10:38Z", "id": 3750, "subaccount_id": None, "user_id": "oa:123", "vendor_id": None, }, "id": 5992, "inputs": None, "outputs": None, "parent_id": 5986, "status": "SUBMITTED", "type": "workflow_approval", } ] data = { "type": "workflow_approval", "context": {"product_id": product_id, "upc": upc}, "workflow_ingest_job_id": latest_pipeline_run_id, } mock_setup_workflow: MagicMock = mocker.patch.object( job_logic, "setup_approval_workflow", return_value=expected_workflow_response ) # Execute function and Assert assert auto_approval_logic._create_approval_workflow_job( product_id, upc, latest_pipeline_run_id ) mock_setup_workflow.assert_called_with(data) # test that invalid json response raises exception mock_setup_workflow.return_value = [ { "context": { "correlation_id": "3163f53e-e2c9-11e8-865e-0242ac110002.1", "datetime": "2018-11-07T20:10:38Z", "id": 3750, "subaccount_id": None, "user_id": "oa:123", "vendor_id": None, }, "id": 5992, "inputs": None, "outputs": None, "parent_id": 5986, "status": "INVALID STATUS", "type": "workflow_approval", } ] with pytest.raises(jsonschema.exceptions.ValidationError): auto_approval_logic._create_approval_workflow_job( product_id, upc, latest_pipeline_run_id ) @pytest.mark.parametrize( ("subaccount_id", "feed_id"), [(None, "vendor_1337"), ("123456", "subaccount_123456")], ) def test_notification_feed_id( mocker: MockerFixture, subaccount_id: Any, feed_id: Any ) -> None: """Test creation of notification feed id.""" release = { "subaccount_id": subaccount_id, "vendor_id": 1337, "project_id": 133737, } returned_feed_id = auto_approval_logic.notification_feed_id(release) assert returned_feed_id == feed_id def test_send_final_approval_notification(mocker: MockerFixture) -> None: """Test create notification on successful final approval.""" # Setup data and mocks product = { "product_id": 13333337, "video_title": "1337 title", "upc": 523489723468, "isrc": "FAC44359872134", "primary_artist_id": 9999, } release = { "subaccount_id": 12345, "vendor_id": 1337, "project_id": 133737, } mocker.patch.object( getattr(auto_approval_logic, "ows_artist"), "get_artist", return_value={"name": "billy bob"}, ) mocker.patch.object( auto_approval_logic, "notification_feed_id", return_value={"subaccount_12345"} ) expected_arg = { "feed_name": notifications_constants.APPROVAL_FEED_NAME, "feed_id": auto_approval_logic.notification_feed_id({}), "payload": { "actor": notifications_constants.APPROVAL_ACTOR, "verb": notifications_constants.APPROVAL_VERB, "object": notifications_constants.APPROVAL_OBJECT, "project_id": 133737, "product_id": 13333337, "video_title": "1337 title", "upc": 523489723468, "isrc": "FAC44359872134", "artist_name": "billy bob", }, } create_notification_mock = mocker.patch.object( getattr(auto_approval_logic, "notifications_model"), "create_notification", return_value=MagicMock(content="hurray", status_code=200), ) mocked_sentry_send = mocker.patch.object( getattr(auto_approval_logic, "sentry_sdk"), "capture_message", ) # Execute function auto_approval_logic.send_final_approval_notification(product, release) # Assertions create_notification_mock.assert_called_once() create_notification_mock.assert_called_with(expected_arg) assert mocked_sentry_send.called def test_send_final_approval_notification_handles_exceptions( mocker: MockerFixture, ) -> None: """Test exception handling of send_final_approval_notification.""" # Setup data and mocks subaccount_id = 1337 mocked_notification_response = mocker.Mock() mocked_notification_response.status_code = HTTPStatus.INTERNAL_SERVER_ERROR mocked_notification_response.content = "womp womp" mocker.patch.object( getattr(auto_approval_logic, "notifications_model"), "create_notification", return_value=mocked_notification_response, ) product = { "product_id": 13333337, "video_title": "1337 title", "upc": 523489723468, "isrc": "FAC44359872134", "primary_artist_id": 9999, } release = { "subaccount_id": subaccount_id, "vendor_id": 1337, "project_id": 133737, } mocker.patch.object( getattr(auto_approval_logic, "ows_artist"), "get_artist", return_value={"name": "billy bob"}, ) mocked_sentry_send = mocker.patch.object( getattr(auto_approval_logic, "sentry_sdk"), "capture_message", ) # Execute function auto_approval_logic.send_final_approval_notification(product, release) # Assertions mocked_sentry_send.assert_called_with( json.dumps( { "message": "Call to ows-notifications has failed - womp womp", "status": 500, } ) ) def test_auto_apply_video_type_carveouts(mocker: MockerFixture) -> None: """Test auto_apply_video_type_carveouts.""" # Setup data and mocks product_id = 12 distribution_type_id = 3 upc = 134567 isrc = 123 type_of_video = "Behind the Scenes" video_types_result = { "video_type_id": 11, "video_type": type_of_video, } dms_ids = [ {"auto_carveout_dms_id": 64, "video_type_id": 11, "dms_id": 1320}, {"auto_carveout_dms_id": 71, "video_type_id": 11, "dms_id": 1450}, {"auto_carveout_dms_id": 78, "video_type_id": 11, "dms_id": 1451}, {"auto_carveout_dms_id": 87, "video_type_id": 11, "dms_id": 1}, {"auto_carveout_dms_id": 94, "video_type_id": 11, "dms_id": 1208}, {"auto_carveout_dms_id": 101, "video_type_id": 11, "dms_id": 187}, {"auto_carveout_dms_id": 108, "video_type_id": 11, "dms_id": 1532}, {"auto_carveout_dms_id": 115, "video_type_id": 11, "dms_id": 1533}, {"auto_carveout_dms_id": 122, "video_type_id": 11, "dms_id": 1551}, ] carveout_post_data = { "service": [ {"service_id": 1320, "distribution_types": [distribution_type_id]}, {"service_id": 1450, "distribution_types": [distribution_type_id]}, {"service_id": 1451, "distribution_types": [distribution_type_id]}, {"service_id": 1, "distribution_types": [distribution_type_id]}, {"service_id": 1208, "distribution_types": [distribution_type_id]}, {"service_id": 187, "distribution_types": [distribution_type_id]}, {"service_id": 1532, "distribution_types": [distribution_type_id]}, {"service_id": 1533, "distribution_types": [distribution_type_id]}, {"service_id": 1551, "distribution_types": [distribution_type_id]}, ], "updated_by": "1911", } product_video_object = { "is_valid": True, "upc": upc, "isrc": isrc, "submitted_at": None, "special_instructions": None, "channel_selection": "test", "latest_pipeline_run_id": 1234, "type_of_video": type_of_video, } mocked_get_product_video = mocker.patch.object( product_video, "get", return_value=product_video_object ) mocked_get_video_types = mocker.patch.object( video, "get_video_type", return_value=video_types_result ) mocked_get_auto_carveout_dms = mocker.patch.object( auto_carveout_dms_model, "get_auto_carveouts_by_video_type_id", return_value=dms_ids, ) mocked_add_carveout_call = mocker.patch.object( getattr(auto_approval_logic, "ows_carveouts_python"), "save_carveouts" ) mocked_logger = mocker.patch.object(auto_approval_logic, "g") # Execute method auto_approval_logic.auto_apply_video_type_carveouts(product_id) # Assertions mocked_get_product_video.assert_called_once_with(product_id) mocked_get_video_types.assert_called_once_with(type_of_video) mocked_get_auto_carveout_dms.assert_called_once_with( video_types_result["video_type_id"] ) mocked_add_carveout_call.assert_called_once_with(product_id, carveout_post_data) mocked_logger.log.info.assert_called_once_with( f"Auto-applied video carveouts for product id: {product_id}" )