"""Tests for main handlers. /v2/asset base.""" import json from typing import Any from unittest.mock import MagicMock, patch import pytest from flask.ctx import AppContext from flask.testing import FlaskClient from flexmock import flexmock from owsrequest import flask_request from owsresponse import response from sqlalchemy.exc import SQLAlchemyError from assets.constants import authorization, error, field_const from assets.exceptions import AssetFinalNotFound, AssetUploadNotFound, ProductNotFound from assets.logic import ( asset_info, asset_status, asset_upload, ownership, stereo_reference, ) OA_HEADERS = {"Orchard-User-Id": "oa:12345"} ALW_HEADERS = { "Orchard-User-Id": "alw:12345", "Grass-Account-Id": "234234", "Grass-Account-Type": "vendor", } @pytest.fixture def fixture_final_assets() -> list[dict[str, Any]]: """Return final assets data.""" return [ { "asset_type": "JPEG", "key": "img/jpeg/test_image_2000_2000.jpeg", "bucket": "some_bucket", }, {"container": "FLAC", "key": "test_flac.flac", "bucket": "some_bucket"}, ] def test_post_asset_status_success( fixture_client: FlaskClient, correct_post_asset_status_v2_body: dict[str, Any] ) -> None: """Test route that updates asset general status.""" flexmock(asset_status).should_receive("create_status_by_filename").with_args( filename=correct_post_asset_status_v2_body["filename"], status=correct_post_asset_status_v2_body["status"], description=correct_post_asset_status_v2_body["description"], message=correct_post_asset_status_v2_body["message"], timestamp=correct_post_asset_status_v2_body["timestamp"], ).and_return("1") request_url = "/v2/asset/status" headers = {"Content-Type": "application/json"} data = json.dumps(correct_post_asset_status_v2_body) result = fixture_client.post(request_url, data=data, headers=headers) assert result @pytest.mark.parametrize( "asset_type, expected_response", [ ( "MP3_192", { "product_id": 12345, "assets": [ { "bucket": "some_bucket", "filename": "filename.mp3", "track_unique_id": 54321, } ], }, ), ( "WAV", { "product_id": 12345, "assets": [ { "bucket": "some_bucket", "filename": "filename.wav", "track_unique_id": 54321, } ], }, ), ( "FLAC", { "product_id": 12345, "assets": [ { "bucket": "some_bucket", "filename": "filename.flac", "track_unique_id": 54321, } ], }, ), ( "TIF", { "product_id": 12345, "assets": [{"bucket": "some_bucket", "filename": "filename.tif"}], }, ), ], ) def test_get_single_track_audio_asset_info_success( fixture_client: FlaskClient, asset_type: str, expected_response: dict[str, Any] ) -> None: """Test success of route that gets asset info by asset type.""" product_id = 12345 track_id = 54321 flexmock(asset_info).should_receive("get_v2_assets_info_by_asset_type").and_return( expected_response ) request_url = f"/v2/assets/{product_id}/asset_type/{asset_type}?track_id={track_id}" result = fixture_client.get(request_url) response_data = json.loads(result.data.decode()) assert result.status_code == 200 assert response_data == expected_response def test_invalid_asset_typ_failure(fixture_client: FlaskClient) -> None: """Test success of route that gets asset info by asset type.""" product_id = 12345 track_id = 54321 invalid_asset_type = "abc" request_url = ( f"/v2/assets/{product_id}/asset_type/{invalid_asset_type}?track_id={track_id}" ) result = fixture_client.get(request_url) response_data = json.loads(result.data.decode()) assert result.status_code == 400 assert response_data["message"] == "Unsupported Asset File Type" def test_post_asset_status_failure( fixture_client: FlaskClient, correct_post_asset_status_v2_body: dict[str, Any] ) -> None: """Test failure of route that updates asset general status.""" flexmock(asset_status).should_receive("create_status_by_filename").and_raise( SQLAlchemyError("query error") ) request_url = "/v2/asset/status" headers = {"Content-Type": "application/json"} data = json.dumps(correct_post_asset_status_v2_body) result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 500 @pytest.mark.parametrize("state", field_const.ASSET_STATES) def test_get_asset_info_success( fixture_client: FlaskClient, fixture_get_status_filename: str, state: str ) -> None: """Test failure of route that get asset info.""" flexmock(asset_info).should_receive("get_asset_info").and_return({}) request_url = "/v2/asset?filename={filename}&state={state}".format( filename=fixture_get_status_filename, state=state ) result = fixture_client.get(request_url) assert result.status_code == 200 def test_get_asset_info_logic_failure( fixture_client: FlaskClient, fixture_get_status_filename: str ) -> None: """Test failure of route that get asset info.""" flexmock(asset_info).should_receive("get_asset_info").and_raise( SQLAlchemyError("fatal_error") ) request_url = "/v2/asset?filename={filename}&state={state}".format( filename=fixture_get_status_filename, state="raw" ) result = fixture_client.get(request_url) assert result.status_code == 500 def test_post_asset_final_success( fixture_client: FlaskClient, correct_post_asset_final_v2_body: dict[str, Any], fixture_final_assets: list[dict[str, Any]], ) -> None: """Test route that saves info about final asset.""" expected_status = "status" correct_post_asset_final_v2_body["final_assets"] = fixture_final_assets flexmock(asset_status).should_receive("map_encoding_state_to_status").with_args( correct_post_asset_final_v2_body["status"] ).and_return(expected_status) flexmock(asset_status).should_receive("create_status_and_final_assets").with_args( filename=correct_post_asset_final_v2_body["filename"], status=expected_status, description="", message=correct_post_asset_final_v2_body["message"], timestamp=correct_post_asset_final_v2_body["timestamp"], final_assets=correct_post_asset_final_v2_body["final_assets"], ).and_return(None) request_url = "/v2/asset/final" headers = {"Content-Type": "application/json"} data = json.dumps(correct_post_asset_final_v2_body) result = fixture_client.post(request_url, data=data, headers=headers) response_message = json.loads(result.data.decode()) assert response assert response_message == {"status": error.SUCCESS_CODE} def test_post_asset_final_failure( fixture_client: FlaskClient, correct_post_asset_final_v2_body: dict[str, Any] ) -> None: """Test route that saves info about final asset with status error.""" flexmock(asset_status).should_receive("map_encoding_state_to_status").with_args( correct_post_asset_final_v2_body["status"] ).and_return("status") flexmock(asset_status).should_receive("create_status_and_final_assets").and_raise( SQLAlchemyError("query error") ) request_url = "/v2/asset/final" headers = {"Content-Type": "application/json"} data = json.dumps(correct_post_asset_final_v2_body) result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 500 def test_post_asset_v2_success_with_metadata( fixture_client: FlaskClient, correct_post_assets_v2_body: dict[str, Any] ) -> None: """Test sending a filename along with metadata to associate with that filename.""" ( flexmock(asset_upload) .should_receive("update_asset_upload") .with_args( product_id=12345, upc=123456, track_unique_id=54321, filename="unique_filename.wav", original_filename="original_filename.mp3", is_correction=True, ) .and_return({"status": error.SUCCESS_CODE}) .once() ) request_url = "/v2/asset" headers = {"Content-Type": "application/json"} data = json.dumps(correct_post_assets_v2_body) result = fixture_client.post(request_url, data=data, headers=headers) assert result def test_post_asset_v2_success_without_metadata(fixture_client: FlaskClient) -> None: """Test sending a filename without metadata to associate with that filename.""" ( flexmock(asset_upload) .should_receive("update_asset_upload") .with_args( filename="some_filename", upc=None, track_unique_id=None, product_id=None, original_filename=None, is_correction=False, ) .and_return({"status": error.SUCCESS_CODE}) .once() ) request_url = "/v2/asset" headers = {"Content-Type": "application/json"} data = json.dumps({"filename": "some_filename"}) result = fixture_client.post(request_url, data=data, headers=headers) assert result def test_post_asset_v2_alw_and_oa_users_blocked( fixture_client: FlaskClient, correct_post_assets_v2_body: dict[str, Any] ) -> None: """Test route that saves new asset info blocks alw and oa users.""" request_url = "/v2/asset" headers = { "Content-Type": "application/json", "Grass-Account-Id": "123", } data = json.dumps(correct_post_assets_v2_body) result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 403 assert ( result.data == b'{"code": "authorization_error", "message": "Request not allowed"}' ) def test_post_asset_v2_success_is_correction_missing( fixture_client: FlaskClient, correct_post_assets_v2_body: dict[str, Any] ) -> None: """Test v2 post asset call without is_missing body arg.""" update_asset_upload_args = correct_post_assets_v2_body.copy() del update_asset_upload_args[field_const.IS_CORRECTION] update_asset_upload_args[field_const.UPC] = 123 update_asset_upload_args[field_const.IS_CORRECTION] = False ( flexmock(asset_upload) .should_receive("update_asset_upload") .with_args(**update_asset_upload_args) .and_return(response.Response()) ) request_url = "/v2/asset" headers = {"Content-Type": "application/json"} data = json.dumps(correct_post_assets_v2_body) result = fixture_client.post(request_url, data=data, headers=headers) assert result def test_post_asset_v2_failure( fixture_client: FlaskClient, correct_post_assets_v2_body: dict[str, Any] ) -> None: """Test route that saves new asset info from lambda call failure.""" ( flexmock(asset_upload) .should_receive("update_asset_upload") .and_raise(SQLAlchemyError("Query error")) ) request_url = "/v2/asset" headers = {"Content-Type": "application/json"} data = json.dumps(correct_post_assets_v2_body) result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 500 @patch("assets.utils.handlers.g") def test_get_asset_download_url_v2_success( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = authorization.HIVE_AI_DETECTION_UUID asset_final_id = 123 request_url = f"/v2/asset/download_url?asset_final_id={asset_final_id}" flexmock(asset_info).should_receive("get_asset_download_url_by_id").with_args( asset_final_id, 900 ).and_return({}) result = fixture_client.get(request_url) assert result assert result.status_code == 200 @patch("assets.utils.handlers.g") def test_get_asset_download_url_v2_with_expires_success( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = authorization.HIVE_AI_DETECTION_UUID asset_final_id = 123 expires_in = 17 request_url = f"/v2/asset/download_url?asset_final_id={asset_final_id}&expires_in={expires_in}" flexmock(asset_info).should_receive("get_asset_download_url_by_id").with_args( asset_final_id, expires_in ).and_return({}) result = fixture_client.get(request_url) assert result assert result.status_code == 200 @patch("assets.utils.handlers.g") def test_get_asset_download_url_v2_hive_text_recognition_jwt_success( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = authorization.HIVE_TEXT_RECOGNITION_UUID asset_final_id = 123 request_url = f"/v2/asset/download_url?asset_final_id={asset_final_id}" flexmock(asset_info).should_receive("get_asset_download_url_by_id").with_args( asset_final_id, 900 ).and_return({}) result = fixture_client.get(request_url) assert result.status_code == 200 @patch("assets.utils.handlers.g") def test_get_asset_download_url_v2_missing_jwt_returns_401( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = None flexmock(asset_info).should_receive("get_asset_download_url_by_id").never() result = fixture_client.get("/v2/asset/download_url?asset_final_id=123") assert result.status_code == 401 @patch("assets.utils.handlers.g") def test_get_asset_download_url_v2_wrong_jwt_identity_returns_403( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = "not-an-allowed-identity-uuid" flexmock(asset_info).should_receive("get_asset_download_url_by_id").never() result = fixture_client.get("/v2/asset/download_url?asset_final_id=123") assert result.status_code == 403 @patch("assets.utils.handlers.g") def test_get_asset_download_url_v2_invalid_id( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = authorization.HIVE_AI_DETECTION_UUID request_url = "/v2/asset/download_url?asset_final_id=abc" flexmock(asset_info).should_receive("get_asset_download_url_by_id").never() result = fixture_client.get(request_url) result_json = json.loads(result.data.decode()) assert result assert result.status_code == 400 assert result_json["code"] == "error_query_validation" assert result_json["message"].startswith("'abc' does not match '^[0-9]+$'") @patch("assets.utils.handlers.g") def test_get_asset_download_url_v2_invalid_expires( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = authorization.HIVE_AI_DETECTION_UUID request_url = "/v2/asset/download_url?expires_in=abc&asset_final_id=123" flexmock(asset_info).should_receive("get_asset_download_url_by_id").never() result = fixture_client.get(request_url) result_json = json.loads(result.data.decode()) assert result assert result.status_code == 400 assert result_json["code"] == "error_query_validation" assert result_json["message"].startswith("'abc' does not match '^[0-9]+$'") @patch("assets.utils.handlers.g") def test_get_asset_download_url_v2_missing_param( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = authorization.HIVE_AI_DETECTION_UUID request_url = "/v2/asset/download_url" flexmock(asset_info).should_receive("get_asset_download_url_by_id").never() result = fixture_client.get(request_url) assert result assert result.status_code == 400 @pytest.mark.parametrize("headers", [OA_HEADERS, ALW_HEADERS]) def test_get_asset_status_v2_success( fixture_client: FlaskClient, fixture_get_status_filename: str, headers: dict[str, str], ) -> None: """Test route that get asset general status.""" expected_args = {"filename": fixture_get_status_filename} if headers == ALW_HEADERS: expected_args.update( { "account_id": ALW_HEADERS["Grass-Account-Id"], "account_type": ALW_HEADERS["Grass-Account-Type"], } ) flexmock(asset_upload).should_receive("get_asset_upload_by_filename").with_args( fixture_get_status_filename ).and_return({"id": 123}) flexmock(asset_status).should_receive("get_current_status").with_args( 123 ).and_return({}) request_url = "/v2/asset/status/{filename}".format( filename=fixture_get_status_filename ) result = fixture_client.get(request_url, headers=headers) assert result @pytest.mark.parametrize("headers", [OA_HEADERS, ALW_HEADERS]) def test_get_asset_status_v2_failure( fixture_client: FlaskClient, fixture_get_status_filename: str, headers: dict[str, str], ) -> None: """Test failure of route that get asset general status.""" flexmock(asset_upload).should_receive("get_asset_upload_by_filename").with_args( fixture_get_status_filename ).and_return({"id": 123}) flexmock(asset_status).should_receive("get_current_status").and_raise( SQLAlchemyError("query error") ) request_url = "/v2/asset/status/{filename}".format( filename=fixture_get_status_filename ) result = fixture_client.get(request_url, headers=headers) response_json = json.loads(result.data.decode()) assert result.status_code == 500 assert ( response_json["message"] == "The server encountered an internal error and was unable to complete your request." ) def test_get_current_status_with_nonexistent_asset( fixture_client: FlaskClient, status_data: dict[str, Any] ) -> None: """Test for getting asset status for nonexistent asset.""" ( flexmock(asset_upload) .should_receive("get_asset_upload_by_filename") .with_args(filename=status_data["filename"]) .and_raise(AssetUploadNotFound(error.ERROR_ASSET_UPLOAD_NOT_FOUND)) ) request_url = f"/v2/asset/status/{status_data['filename']}" result = fixture_client.get(request_url, headers=OA_HEADERS) response_json = json.loads(result.data.decode()) assert result.status_code == 404 assert response_json["message"] == error.ERROR_ASSET_UPLOAD_NOT_FOUND def test_get_current_status_with_ownership_check_failure( asset_upload_data: list[dict[str, Any]], status_data: dict[str, Any], fixture_client: FlaskClient, ) -> None: """Test for getting asset status with ownership check failure.""" ( flexmock(asset_upload) .should_receive("get_asset_upload_by_filename") .with_args(status_data["filename"]) .and_return(asset_upload_data[0]) ) ( flexmock(ownership) .should_receive("check_ownership") .with_args( asset_upload_data[0]["product_id"], ALW_HEADERS["Grass-Account-Type"], int(ALW_HEADERS["Grass-Account-Id"]), ) .and_return( response.create_error_response( error.ERROR_CODE_NOT_OWNED, error.ERROR_PRODUCT_NOT_OWNED, 403 ) ) ) request_url = f"/v2/asset/status/{status_data['filename']}" result = fixture_client.get(request_url, headers=ALW_HEADERS) response_json = json.loads(result.data.decode()) assert result.status_code == 403 assert response_json["message"] == error.ERROR_PRODUCT_NOT_OWNED @pytest.mark.parametrize("headers", [OA_HEADERS, ALW_HEADERS]) def test_apply_asset_corrections( fixture_client: FlaskClient, headers: dict[str, str] ) -> None: """Test product assets approve correction success.""" product_id = 1234 flexmock(asset_upload).should_receive("apply_asset_corrections").with_args( product_id=product_id ).and_return([{"asset_upload_id": 1}]) flexmock(flask_request).should_receive("verify_grass_ownership").and_return( response.Response() ) request_url = "/v2/asset/product/{product_id}/corrections/apply".format( product_id=product_id ) headers.update({"Content-Type": "application/json"}) result = fixture_client.put(request_url, headers=headers) assert result.status_code == 200 @pytest.mark.parametrize("headers", [OA_HEADERS, ALW_HEADERS]) def test_apply_asset_corrections_error( fixture_client: FlaskClient, headers: dict[str, str] ) -> None: """Test product assets approve correction error.""" product_id = "1234" flexmock(asset_upload).should_receive("apply_asset_corrections").and_raise( SQLAlchemyError("Query error") ) flexmock(flask_request).should_receive("verify_grass_ownership").and_return( response.Response() ) request_url = "/v2/asset/product/{product_id}/corrections/apply".format( product_id=product_id ) headers.update({"Content-Type": "application/json"}) result = fixture_client.put(request_url, headers=headers) assert result.status_code == 500 def test_get_asset_owner_success(fixture_client: FlaskClient) -> None: """Test get owner returns 200 with owner information.""" filename = "some_asset.wav" expected_owner = {"vendor_id": 42, "subaccount_id": 53} ( flexmock(ownership) .should_receive("get_asset_product_owner") .with_args(filename) .and_return(expected_owner) .once() ) result = fixture_client.get(f"/internal/v2/asset/{filename}/owner") result_json = json.loads(result.data.decode()) assert result.status_code == 200 assert result_json == expected_owner def test_get_asset_owner_asset_not_found(fixture_client: FlaskClient) -> None: """Test GET /v2/asset//owner returns 404 when asset does not exist.""" filename = "missing_asset.wav" ( flexmock(ownership) .should_receive("get_asset_product_owner") .with_args(filename) .and_raise(AssetUploadNotFound(error.ERROR_ASSET_UPLOAD_NOT_FOUND)) .once() ) result = fixture_client.get(f"/internal/v2/asset/{filename}/owner") result_json = json.loads(result.data.decode()) assert result.status_code == 404 assert result_json["message"] == error.ERROR_ASSET_UPLOAD_NOT_FOUND def test_get_asset_owner_no_product_id(fixture_client: FlaskClient) -> None: """Test GET /v2/asset//owner returns 404 when asset has no product id.""" filename = "asset_no_product.wav" ( flexmock(ownership) .should_receive("get_asset_product_owner") .with_args(filename) .and_raise(ProductNotFound(error.ERROR_RELEASE_NOT_FOUND)) .once() ) result = fixture_client.get(f"/internal/v2/asset/{filename}/owner") result_json = json.loads(result.data.decode()) assert result.status_code == 404 assert result_json["message"] == error.ERROR_RELEASE_NOT_FOUND @patch("assets.utils.handlers.g") def test_get_stereo_for_asset_success( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = ( authorization.LAMBDA_ASSETS_SPATIAL_AUDIO_VALIDATION_UUID ) asset_upload_filename = "some_input.wav" expected = {"bucket": "prod-encoded-assets", "key": "stereo.flac"} ( flexmock(stereo_reference) .should_receive("get_stereo_for_asset") .with_args(asset_upload_filename) .and_return(expected) .once() ) result = fixture_client.get(f"/internal/assets/{asset_upload_filename}/stereo") assert result.status_code == 200 assert json.loads(result.data.decode()) == expected @patch("assets.utils.handlers.g") def test_get_stereo_for_asset_missing_jwt_returns_401( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = None flexmock(stereo_reference).should_receive("get_stereo_for_asset").never() result = fixture_client.get("/internal/assets/some_input.wav/stereo") assert result.status_code == 401 @patch("assets.utils.handlers.g") def test_get_stereo_for_asset_wrong_jwt_identity_returns_403( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = authorization.HIVE_AI_DETECTION_UUID flexmock(stereo_reference).should_receive("get_stereo_for_asset").never() result = fixture_client.get("/internal/assets/some_input.wav/stereo") assert result.status_code == 403 @patch("assets.utils.handlers.g") def test_get_stereo_for_asset_input_not_found( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = ( authorization.LAMBDA_ASSETS_SPATIAL_AUDIO_VALIDATION_UUID ) asset_upload_filename = "missing_input.wav" ( flexmock(stereo_reference) .should_receive("get_stereo_for_asset") .with_args(asset_upload_filename) .and_raise(AssetUploadNotFound(error.ERROR_ASSET_UPLOAD_NOT_FOUND)) .once() ) result = fixture_client.get(f"/internal/assets/{asset_upload_filename}/stereo") assert result.status_code == 404 assert ( json.loads(result.data.decode())["message"] == error.ERROR_ASSET_UPLOAD_NOT_FOUND ) @patch("assets.utils.handlers.g") def test_get_stereo_for_asset_stereo_flac_not_found( mock_g: MagicMock, fixture_client: FlaskClient, fixture_app: AppContext ) -> None: mock_g.request_context.jwt_identity_id = ( authorization.LAMBDA_ASSETS_SPATIAL_AUDIO_VALIDATION_UUID ) asset_upload_filename = "input_without_flac.wav" ( flexmock(stereo_reference) .should_receive("get_stereo_for_asset") .with_args(asset_upload_filename) .and_raise(AssetFinalNotFound(error.ERROR_ASSET_FINAL_NOT_FOUND)) .once() ) result = fixture_client.get(f"/internal/assets/{asset_upload_filename}/stereo") assert result.status_code == 404 assert ( json.loads(result.data.decode())["message"] == error.ERROR_ASSET_FINAL_NOT_FOUND )