"""Tests for main handlers /v2/(asset|image)/product base.""" import copy import json from typing import Any from unittest.mock import MagicMock, _Call, call import pytest from flask import Response from flask.testing import FlaskClient from flexmock import flexmock from owsresponse import response from owsresponse.adaptors.flask import flaskify from pytest_mock import MockerFixture from requests.exceptions import HTTPError from sqlalchemy.exc import SQLAlchemyError from assets.exceptions import AssetDeleteFailure from assets.logic import ( ai_detection as ai_audio_logic, delete as delete_logic_v2, match_audio as match_audio_logic, ownership, product as product_v2, ) OA_HEADERS = {"Orchard-User-Id": "oa:12345"} ALW_HEADERS = { "Orchard-User-Id": "alw:12345", "Grass-Account-Id": "234234", "Grass-Account-Type": "vendor", } @pytest.mark.parametrize("headers", [OA_HEADERS, ALW_HEADERS, {}]) def test_get_product_by_id_v2_success( fixture_client: FlaskClient, headers: dict[str, str] ) -> None: """Test route for getting product assets.""" product_id = "1" request_url = "v2/asset/product/{}".format(product_id) ownership_mock_calls = 1 if headers == ALW_HEADERS else 0 flexmock(ownership).should_receive("check_ownership").times( ownership_mock_calls ).and_return(response.Response()) asset_with_uploading_status = { "asset_type": "WAV", "filename": "4cd3d30a_11d6_4ee9_9f39_08f982b39023", "original_filename": "45s_24_bit_noise_and_test_tones.wav", "status": "uploading", "stream": {}, "track_unique_id": 41604110, } assets_without_uploading_status = { "assets": [ { "asset_type": "WAV", "filename": "eb6a500c_9f7e_45f6_9b73_c2c120c157dd", "original_filename": "45s_24_bit_noise_and_test_tones.wav", "status": "upload_complete", "stream": {}, "track_unique_id": 41604097, }, { "asset_type": "WAV", "filename": "ee84d149_eb01_48c9_b2f1_8ec763049108", "original_filename": "45s_24_bit_noise_and_test_tones.wav", "status": "encoding_completed", "stream": {"duration": 45.035, "track_unique_id": 41604104}, "track_unique_id": 41604104, }, ], "product_id": 4921330, "upc": 195081444086, } product_data: dict[str, Any] = copy.deepcopy(assets_without_uploading_status) product_data["assets"].append(asset_with_uploading_status) flexmock(product_v2).should_receive("get_assets_by_product_id").and_return( product_data ) data = {} headers = headers.copy() if not headers: data = {"Orchard-User-Id": "alw:321"} headers["Content-Type"] = "application/json" result = fixture_client.get(request_url, data=json.dumps(data), headers=headers) assert result.status_code == 200 assert result.headers.get("Correlation-Id") assert assets_without_uploading_status == json.loads(result.data.decode("utf-8")) def test_get_product_by_id_v2_success_no_user_id(fixture_client: FlaskClient) -> None: """Test route for getting product assets with no user_id.""" product_id = "1" request_url = "v2/asset/product/{}".format(product_id) flexmock(ownership).should_receive("check_ownership").times(0).and_return( response.Response() ) product_data = {"product_id": 1, "assets": []} flexmock(product_v2).should_receive("get_assets_by_product_id").and_return( product_data ) result = fixture_client.get(request_url) assert result.status_code == 200 assert result.headers.get("Correlation-Id") assert product_data == json.loads(result.data.decode("utf-8")) def test_get_product_v2_permission_failure(fixture_client: FlaskClient) -> None: """Test permission failure of route for getting product assets.""" product_id = "1" request_url = f"v2/asset/product/{product_id}" flexmock(ownership).should_receive("check_ownership").and_return( response.create_error_response( code="not_owned", message="not_owned", status=403 ) ) result = fixture_client.get(request_url, headers=ALW_HEADERS) assert result.status_code == 403 assert result.headers.get("Correlation-Id") def test_get_product_v2_logic_failure(fixture_client: FlaskClient) -> None: """Test permission failure of route for getting product assets.""" product_id = "1" request_url = "v2/asset/product/{}".format(product_id) flexmock(ownership).should_receive("check_ownership").and_return( response.Response() ) error_message = "query error" flexmock(product_v2).should_receive("get_assets_by_product_id").and_raise( SQLAlchemyError(error_message) ) result = fixture_client.get(request_url, headers=ALW_HEADERS) assert result.status_code == 500 assert result.headers.get("Correlation-Id") @pytest.mark.parametrize( ( "test_description", "request_headers", "request_data", "expected_check_ownership_calls", "check_ownership_response", "expected_get_match_audio_results_calls", "get_match_audio_results_response", "expected_result", ), [ ( "success with OA user (headers_auth)", OA_HEADERS, None, [], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "success with ALW user (headers_auth)", ALW_HEADERS, None, [call(12321, account_type="vendor", account_id="234234")], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "success with OA user (body_auth)", None, OA_HEADERS, [], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "success with ALW user (body_auth)", None, ALW_HEADERS, [], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "success with no user", {}, None, [], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "error ownership check with ALW user (headers_auth)", ALW_HEADERS, None, [call(12321, account_type="vendor", account_id="234234")], response.create_error_response( code="not_owned", message="not_owned", status=403 ), [], None, flaskify( response.create_error_response( code="not_owned", message="not_owned", status=403 ) ), ), ( "error get_match_audio_results", {}, None, [], None, [call(12321)], HTTPError(response=MagicMock(status_code=404)), flaskify( response.create_error_response( code="bad_gateway", status=502, message="" ) ), ), ], ) def test_get_content_review_match_audio( mocker: MockerFixture, fixture_client: FlaskClient, test_description: str, request_headers: dict[str, str], request_data: dict[str, Any], expected_check_ownership_calls: list[_Call], check_ownership_response: Any, expected_get_match_audio_results_calls: list[_Call], get_match_audio_results_response: Any, expected_result: Any, ) -> None: """Test get_content_review_match_audio.""" product_id = 12321 request_url = f"v2/asset/product/{product_id}/match-audio" check_ownership_mock = mocker.patch.object( ownership, "check_ownership", return_value=check_ownership_response, ) get_match_audio_results_mock = mocker.patch.object( match_audio_logic, "get_match_audio_results", side_effect=[get_match_audio_results_response], ) result = fixture_client.get( request_url, data=json.dumps(request_data), headers=request_headers ) assert check_ownership_mock.mock_calls == expected_check_ownership_calls assert get_match_audio_results_mock.mock_calls == ( expected_get_match_audio_results_calls ) assert result.status_code == expected_result.status_code assert result.json == expected_result.json @pytest.mark.parametrize( ( "test_description", "request_headers", "request_data", "expected_check_ownership_calls", "check_ownership_response", "expected_get_ai_generated_audio_results_by_product_calls", "get_ai_generated_audio_results_by_product_response", "expected_result", ), [ ( "success with OA user (headers_auth)", OA_HEADERS, None, [], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "success with ALW user (headers_auth)", ALW_HEADERS, None, [call(12321, account_type="vendor", account_id="234234")], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "success with OA user (body_auth)", None, OA_HEADERS, [], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "success with ALW user (body_auth)", None, ALW_HEADERS, [], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "success with no user", {}, None, [], response.Response(), [call(12321)], [{"some": "data"}], flaskify(response.Response({"items": [{"some": "data"}]})), ), ( "error ownership check with ALW user (headers_auth)", ALW_HEADERS, None, [call(12321, account_type="vendor", account_id="234234")], response.create_error_response( code="not_owned", message="not_owned", status=403 ), [], None, flaskify( response.create_error_response( code="not_owned", message="not_owned", status=403 ) ), ), ( "error get_ai_generated_audio_results", {}, None, [], None, [call(12321)], HTTPError(response=MagicMock(status_code=404)), flaskify( response.create_error_response( code="bad_gateway", status=502, message="" ) ), ), ], ) def test_get_content_review_ai_generated_audio( mocker: MockerFixture, fixture_client: FlaskClient, test_description: str, request_headers: dict[str, str], request_data: dict[str, Any], expected_check_ownership_calls: list[_Call], check_ownership_response: Any, expected_get_ai_generated_audio_results_by_product_calls: list[_Call], get_ai_generated_audio_results_by_product_response: Response, expected_result: Any, ) -> None: """Test get_content_review_ai_generated_audio.""" product_id = 12321 request_url = f"v2/asset/product/{product_id}/ai-generated-audio" check_ownership_mock = mocker.patch.object( ownership, "check_ownership", return_value=check_ownership_response, ) get_ai_generated_audio_results_by_product_mock = mocker.patch.object( ai_audio_logic, "get_ai_generated_audio_results_by_product", side_effect=[get_ai_generated_audio_results_by_product_response], ) result = fixture_client.get( request_url, data=json.dumps(request_data), headers=request_headers ) assert check_ownership_mock.mock_calls == expected_check_ownership_calls assert ( get_ai_generated_audio_results_by_product_mock.mock_calls == expected_get_ai_generated_audio_results_by_product_calls ) assert result.status_code == expected_result.status_code assert result.json == expected_result.json @pytest.mark.parametrize("headers", [OA_HEADERS, ALW_HEADERS, {}]) def test_remove_product_image_v2_success( fixture_client: FlaskClient, headers: dict[str, str] ) -> None: """Test route for deleting product image assets.""" # mocking product_id = "1" request_url = "/v2/image/{product_id}".format(product_id=product_id) ownership_mock_calls = 1 if headers == ALW_HEADERS else 0 flexmock(ownership).should_receive("check_ownership").times( ownership_mock_calls ).and_return(response.Response()) successful_response = {"status": "ok"} flexmock(product_v2).should_receive("remove_image_assets_by_product_id").and_return( successful_response ) data = {} headers = headers.copy() if not headers: data = {"Orchard-User-Id": "alw:321"} headers["Content-Type"] = "application/json" # test function call result = fixture_client.delete(request_url, data=json.dumps(data), headers=headers) # checking assert result.status_code == 200 assert result.headers.get("Correlation-Id") assert successful_response == json.loads(result.data.decode("utf-8")) def test_remove_product_image_v2_success_no_user_id( fixture_client: FlaskClient, ) -> None: """Test route for deleting image assets with no user_id.""" # mocking product_id = "1" request_url = "/v2/image/{product_id}".format(product_id=product_id) flexmock(ownership).should_receive("check_ownership").times(0).and_return( response.Response() ) successful_response = {"status": "ok"} flexmock(product_v2).should_receive("remove_image_assets_by_product_id").and_return( successful_response ) # test function call result = fixture_client.delete(request_url) # checking assert result.status_code == 200 assert result.headers.get("Correlation-Id") assert successful_response == json.loads(result.data.decode("utf-8")) def test_remove_product_image_v2_permission_failure( fixture_client: FlaskClient, ) -> None: """Test permission failure of route for deleting image assets.""" # mocking product_id = "1" request_url = "/v2/image/{product_id}".format(product_id=product_id) flexmock(ownership).should_receive("check_ownership").and_return( response.create_error_response( code="not_owned", message="not_owned", status=403 ) ) flexmock(product_v2).should_receive("remove_image_assets_by_product_id").and_return( response.Response() ) # test function call result = fixture_client.delete(request_url, headers=ALW_HEADERS) # checking assert result.status_code == 403 assert result.headers.get("Correlation-Id") def test_remove_product_image_v2_logic_failure(fixture_client: FlaskClient) -> None: """Test permission failure of route for deleting image assets.""" # mocking product_id = "1" request_url = "/v2/image/{product_id}".format(product_id=product_id) flexmock(ownership).should_receive("check_ownership").and_return( response.Response() ) flexmock(product_v2).should_receive("remove_image_assets_by_product_id").and_raise( AssetDeleteFailure ) # test function call result = fixture_client.delete(request_url, headers=ALW_HEADERS) # checking assert result.status_code == 500 assert result.headers.get("Correlation-Id") @pytest.mark.parametrize("headers", [OA_HEADERS, ALW_HEADERS, {}]) def test_remove_product_v2_success( fixture_client: FlaskClient, headers: dict[str, str] ) -> None: """Test route for deleting product assets.""" # mocking product_id = "1" request_url = f"/v2/product/{product_id}" ownership_mock_calls = 1 if headers == ALW_HEADERS else 0 flexmock(ownership).should_receive("check_ownership").times( ownership_mock_calls ).and_return(response.Response()) successful_response = {"status": "ok"} flexmock(delete_logic_v2).should_receive("delete_product").and_return( successful_response ) data = {} headers = headers.copy() if not headers: data = {"Orchard-User-Id": "alw:321"} headers["Content-Type"] = "application/json" # test function call result = fixture_client.delete(request_url, data=json.dumps(data), headers=headers) # checking assert result.status_code == 200 assert result.headers.get("Correlation-Id") assert successful_response == json.loads(result.data.decode("utf-8")) def test_remove_product_v2_success_no_user_id(fixture_client: FlaskClient) -> None: """Test route for deleting product assets with no user_id.""" # mocking product_id = "1" request_url = f"/v2/product/{product_id}" flexmock(ownership).should_receive("check_ownership").times(0).and_return( response.Response() ) successful_response = {"status": "ok"} flexmock(delete_logic_v2).should_receive("delete_product").and_return( successful_response ) # test function call result = fixture_client.delete(request_url) # checking assert result.status_code == 200 assert result.headers.get("Correlation-Id") assert successful_response == json.loads(result.data.decode("utf-8")) def test_remove_product_v2_permission_failure(fixture_client: FlaskClient) -> None: """Test permission failure of route for deleting product assets.""" # mocking product_id = "1" request_url = f"/v2/product/{product_id}" flexmock(ownership).should_receive("check_ownership").and_return( response.create_error_response( code="not_owned", message="not_owned", status=403 ) ) flexmock(delete_logic_v2).should_receive("delete_product").and_return( response.Response() ) # test function call result = fixture_client.delete(request_url, headers=ALW_HEADERS) # checking assert result.status_code == 403 assert result.headers.get("Correlation-Id") def test_remove_product_v2_logic_failure(fixture_client: FlaskClient) -> None: """Test permission failure of route for deleting product assets.""" # mocking product_id = "1" request_url = f"/v2/product/{product_id}" flexmock(ownership).should_receive("check_ownership").and_return( response.Response() ) flexmock(delete_logic_v2).should_receive("delete_product").and_raise( AssetDeleteFailure ) # test function call result = fixture_client.delete(request_url, headers=ALW_HEADERS) # checking assert result.status_code == 500 assert result.headers.get("Correlation-Id") @pytest.mark.parametrize("headers", [OA_HEADERS, ALW_HEADERS, {}]) def test_discard_product_corrections_v2_success( fixture_client: FlaskClient, headers: dict[str, str] ) -> None: """Test route for deleting product assets.""" # mocking product_id = "1" request_url = "/v2/product/{}/corrections".format(product_id) successful_response = {"status": "ok"} flexmock(delete_logic_v2).should_receive("delete_product_corrections").and_return( successful_response ) data = {} headers = headers.copy() if not headers: data = {"Orchard-User-Id": "alw:321"} headers["Content-Type"] = "application/json" # test function call result = fixture_client.delete(request_url, data=json.dumps(data), headers=headers) # checking assert result.status_code == 200 assert result.headers.get("Correlation-Id") assert successful_response == json.loads(result.data.decode("utf-8")) def test_discard_product_corrections_v2_success_ownership_check( fixture_client: FlaskClient, ) -> None: """Test route for deleting product assets.""" # mocking product_id = "1" request_url = "/v2/product/{}/corrections".format(product_id) successful_response = {"status": "ok"} flexmock(delete_logic_v2).should_receive("delete_product_corrections").and_return( successful_response ) data = {} headers = ALW_HEADERS.copy() if not headers: data = {"Orchard-User-Id": "alw:321"} headers["Content-Type"] = "application/json" # test function call result = fixture_client.delete(request_url, data=json.dumps(data), headers=headers) # checking assert result.status_code == 200 assert result.headers.get("Correlation-Id") assert successful_response == json.loads(result.data.decode("utf-8"))