"""Tests for Handlers.""" import json from unittest.mock import MagicMock from unittest.mock import patch from flexmock import flexmock from oto import response import pytest from assets import handlers from assets import logic from assets.constants import error from assets.logic import asset_status from assets.logic import asset_upload from assets.logic import ownership from assets.logic import post_asset from assets.logic import validate_upc_for_product from assets.logic.legacy import copy_product_assets from assets.logic.legacy import import_asset from assets.logic.legacy import product from assets.logic.legacy import stream_info from assets.logic.legacy import remove_asset from assets.logic.legacy import track @patch('assets.handlers.g') def test_exception_handler(mock_g): """Verify exception_Handler returns 500 status code and json payload.""" message = ( 'The server encountered an internal error ' 'and was unable to complete your request.') mock_error = MagicMock() server_response = handlers.exception_handler(mock_error) mock_g.log.exception.assert_called_with(mock_error) # assert status code is 500 assert server_response.status_code == 500 # assert json payload response_message = json.loads(server_response.data.decode()) assert response_message['message'] == message assert response_message['code'] == response.error.ERROR_CODE_INTERNAL_ERROR @pytest.fixture def get_asset_status_successful_response(): """Create an api test client fixture.""" return response.Response({'status': 'finished'}) @pytest.fixture def get_streaming_link_successul_response(): """Return successful response for get_streaming_link endpoint.""" return response.Response({ 'stream': { 'track_unique_id': 1, 'url': 'url', 'duration': 10.10 }, 'access': { } }) @pytest.fixture def fixture_response_ok(): """Fixture for successful response.""" return response.Response({ 'status': error.SUCCESS_CODE }) @pytest.fixture def track_copying_input_data(): """Data for use in track copying call.""" return { 'from_tuid': '12345_67891', 'to_tuid': '21435_65879', 'full_user_id': 'alw:12345' } @pytest.fixture def track_copying_response_response(): """Return successful response for copy_track endpoint.""" return response.Response({ 'status': error.SUCCESS_CODE }) @pytest.fixture def get_product_by_id_successul_response(): """Fixture for getting product assets successful response.""" return response.Response({ 'product_id': 10, 'upc': 123456789, 'assets': [{ 'filename': 'filename1.jpg', 'asset_type': 'image', 'status': 'new' }, { 'filename': 'filename2.wav', 'asset_type': 'audio', 'status': 'new', 'track_unique_id': 99 }] }) MOCK_LOGIC_DATA = { 'credentials': { 'token': 'a real token', 'aws_secret_access_key': 'a real secret', 'aws_secret_key_id': 'a real key', 'expiration': '2017-02-28 21:14:43' }, 'bucket': 'test-orcd-raw-assets', 'filename': 'fred123' } 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_upload_token_success(fixture_client, headers): """Test route that gets the upload token.""" mock_logic_response = response.Response(MOCK_LOGIC_DATA) (flexmock(logic.generate_upload_data) .should_receive('get_upload_permission') .and_return(mock_logic_response)) result = fixture_client.get('/upload-token', headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') upload_token = json.loads(result.data.decode('utf-8')) assert upload_token.get('credentials') assert upload_token.get('bucket') assert upload_token.get('filename') @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_upload_token_logic_layer_failure(fixture_client, headers): """Test route fails when there is logic layer failure.""" mock_logic_response = response.create_fatal_response('fatal_error') (flexmock(logic.generate_upload_data) .should_receive('get_upload_permission') .and_return(mock_logic_response)) headers = headers.copy() headers['Correlation-id'] = 'correlation_id' result = fixture_client.get('/upload-token', headers=headers) assert result.status_code == 500 response_message = json.loads(result.data.decode()) assert response_message['code'] == 'internal_error' assert response_message['message'] == 'fatal_error' assert result.headers.get('Correlation-Id') == 'correlation_id' @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_asset_status_success( monkeypatch, fixture_client, headers, get_asset_status_successful_response): """Test route that gets file uploading status.""" monkeypatch.setattr( import_asset, 'get_status_by_filename', value=MagicMock( return_value=get_asset_status_successful_response)) valid_filename = 'test_file_name' request_url = '/asset/status?filename={}'.format(valid_filename) result = fixture_client.get(request_url, headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') response_body = json.loads(result.data.decode('utf-8')) assert 'status' in response_body assert 'code' not in response_body assert response_body['status'] == ( get_asset_status_successful_response.message['status']) @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_asset_status_failure(fixture_client, headers): """Test route fails when there is invalid Orchard-User-Id in header.""" valid_filename = 'test_file_name' request_url = '/asset/status?filename={}'.format(valid_filename) headers = headers.copy() headers['Orchard-User-Id'] = 'invalid_id' result = fixture_client.get(request_url, headers=headers) assert result.status_code == 400 response_message = json.loads(result.data.decode()) assert response_message['code'] == error.ERROR_CODE_HEADER_VALIDATION @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_product_link_failure(fixture_client, headers): """Test route fails when there is invalid Orchard-User-Id in header.""" product_id = '1' request_url = '/asset/product/'.format(product_id) headers = headers.copy() headers['Orchard-User-Id'] = 'invalid_id' result = fixture_client.get(request_url, headers=headers) assert result.status_code == 400 response_message = json.loads(result.data.decode()) assert response_message['code'] == error.ERROR_CODE_HEADER_VALIDATION @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_streaming_link_success( monkeypatch, fixture_client, headers, get_streaming_link_successul_response): """Test route that gets streaming link.""" ownership_mock = MagicMock( return_value=response.Response()) monkeypatch.setattr( ownership, 'check_track_ownership', value=ownership_mock) monkeypatch.setattr( stream_info, 'get_stream', value=MagicMock( return_value=get_streaming_link_successul_response)) track_id = '1' request_url = '/stream/track/'.format(track_id) result = fixture_client.get(request_url, headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') message = json.loads(result.data.decode('utf-8')) assert message == get_streaming_link_successul_response.message if headers == ALW_HEADERS: ownership_mock.assert_called() @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_streaming_link_failure(fixture_client, headers): """Test route fails when there is invalid Orchard-User-Id in header.""" track_id = '1' request_url = '/stream/track/'.format(track_id) headers = headers.copy() headers['Orchard-User-Id'] = 'invalid_id' result = fixture_client.get(request_url, headers=headers) assert result.status_code == 400 response_message = json.loads(result.data.decode()) assert response_message['code'] == error.ERROR_CODE_HEADER_VALIDATION def test_get_streaming_link_with_invalid_ownership( monkeypatch, fixture_client): """Test route that gets streaming link.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' monkeypatch.setattr( ownership, 'check_track_ownership', value=MagicMock( return_value=response.create_error_response( code='not_owned', message='not_owned', status=403))) track_id = '1' request_url = '/stream/track/'.format(track_id) result = fixture_client.get(request_url, headers=headers) assert result.status_code == 403 assert result.headers.get('Correlation-Id') def test_validate_bit_depth_by_upc(fixture_client): """Test route for validating bit-depth for provided UPC.""" product_id = 2011621 mock_logic_response = response.Response(message='ok') (flexmock(logic.validators) .should_receive('validate_bit_depth_by_upc') .and_return(mock_logic_response)) request_url = '/validators/product/{product_id}/bit-depth?'\ 'physical_location_id=1&asset_type_id=1'.\ format(product_id=product_id) result = fixture_client.get(request_url) result_message = result.data.decode('utf8') assert result.status_code == 200 assert result.headers.get('Correlation-Id') assert mock_logic_response.message == result_message def test_validate_bit_depth_with_invalid_upc(fixture_client): """Test route for validating bit-depth with invalid UPC.""" product_id = 000 mock_logic_response = response.Response(status=400, message='not ok') (flexmock(logic.validators) .should_receive('validate_bit_depth_by_upc') .and_return(mock_logic_response)) request_url = '/validators/product/{product_id}/bit-depth?'\ 'physical_location_id=1&asset_type_id=1'.\ format(product_id=product_id) result = fixture_client.get(request_url) result_message = result.data.decode('utf8') assert result.status_code == 400 assert result.headers.get('Correlation-Id') assert mock_logic_response.message == result_message @pytest.fixture def correct_post_assets_body(): """Correct post asset body fixture.""" return { 'asset_type': 'WAV', 'product_id': 12345, 'upc': '123', 'track_unique_id': 54321, 'filename': 'unique_filename.wav', 'token': 'unique_token' } @pytest.fixture def correct_post_assets_body_image(asset_type): """Correct post asset body fixture for image upload.""" return { 'asset_type': asset_type, 'product_id': 12345, 'upc': '123', 'track_unique_id': 0, 'filename': 'unique_filename.{}'.format(asset_type.lower()), 'token': 'unique_token' } @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_post_asset_success( fixture_client, headers, correct_post_assets_body): """"Test route that saves new asset info.""" headers['Content-Type'] = 'application/json' mock_logic_response = response.Response({'status': 'ok'}) (flexmock(post_asset) .should_receive('post_asset') .and_return(mock_logic_response)) if headers == OA_HEADERS: ownership_times = 0 else: ownership_times = 1 (flexmock(ownership) .should_receive('check_ownership') .times(ownership_times) .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_upc_for_product') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_product_track_ownership') .and_return(response.Response())) request_url = '/asset' data = json.dumps(correct_post_assets_body) result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') @pytest.mark.parametrize('payload', [ correct_post_assets_body_image('JPG'), correct_post_assets_body_image('JPEG'), correct_post_assets_body_image('jpg'), correct_post_assets_body_image('jpeg'), correct_post_assets_body_image('TIF'), correct_post_assets_body_image('TIFF'), correct_post_assets_body_image('tif'), correct_post_assets_body_image('tiff') ]) def test_post_image_asset_success(fixture_client, payload): """"Test route that saves new asset info.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' mock_logic_response = response.Response({'status': 'ok'}) (flexmock(post_asset) .should_receive('post_asset') .and_return(mock_logic_response)) (flexmock(ownership) .should_receive('check_ownership') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_upc_for_product') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_product_track_ownership').never()) request_url = '/asset' data = json.dumps(payload) result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 200 @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_post_asset_failure( fixture_client, headers, correct_post_assets_body): """"Test route that saves new asset info logic layer failure.""" headers['Content-Type'] = 'application/json' mock_logic_response = response.create_fatal_response('fatal_response') (flexmock(post_asset) .should_receive('post_asset') .and_return(mock_logic_response)) (flexmock(ownership) .should_receive('check_ownership') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_upc_for_product') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_product_track_ownership') .and_return(response.Response())) request_url = '/asset' data = json.dumps(correct_post_assets_body) assert headers result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 500 assert result.headers.get('Correlation-Id') def test_post_asset_ownership_failure( fixture_client, correct_post_assets_body): """"Test route that saves new asset info ownership check failure.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' mock_logic_response = response.Response({'status': 'ok'}) (flexmock(post_asset) .should_receive('post_asset') .and_return(mock_logic_response)) (flexmock(ownership) .should_receive('check_ownership') .and_return(response.create_error_response( code='not_owned', message='not_owned', status=403))) request_url = '/asset' data = json.dumps(correct_post_assets_body) assert headers result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 403 assert result.headers.get('Correlation-Id') def test_post_asset_product_upc_match_failure( fixture_client, correct_post_assets_body): """"Test route that saves new asset info product upc matching failure.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' mock_logic_response = response.Response({'status': 'ok'}) (flexmock(post_asset) .should_receive('post_asset') .and_return(mock_logic_response)) (flexmock(ownership) .should_receive('check_ownership') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_upc_for_product') .and_return(response.create_error_response( 'error_code', 'error_msg', 403))) request_url = '/asset' data = json.dumps(correct_post_assets_body) assert headers result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 403 assert result.headers.get('Correlation-Id') def test_post_asset_product_validate_product_track_ownership_failure( fixture_client, correct_post_assets_body): """"Test route that saves new asset track ownership validation failure.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' mock_logic_response = response.Response({'status': 'ok'}) (flexmock(post_asset) .should_receive('post_asset') .and_return(mock_logic_response)) (flexmock(ownership) .should_receive('check_ownership') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_upc_for_product') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_product_track_ownership') .and_return(response.create_error_response( 'error_code', 'error_msg', 403))) request_url = '/asset' data = json.dumps(correct_post_assets_body) assert headers result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == 403 assert result.headers.get('Correlation-Id') def test_post_asset_product_success_image( fixture_client, correct_post_assets_body): """"Test route that saves new asset image.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' correct_post_assets_body['asset_type'] = 'TIF' mock_logic_response = response.Response({'status': 'ok'}) (flexmock(post_asset) .should_receive('post_asset') .and_return(mock_logic_response)) (flexmock(ownership) .should_receive('check_ownership') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_upc_for_product') .and_return(response.Response())) (flexmock(validate_upc_for_product) .should_receive('validate_product_track_ownership') .and_return(response.create_error_response( 'error_code', 'error_msg', 403))) request_url = '/asset' data = json.dumps(correct_post_assets_body) assert headers result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code assert result.headers.get('Correlation-Id') def test_get_image_location(fixture_client): """Test route for getting the location of an image file.""" entity_id = 123456 mock_logic_response = response.Response(message='ok') (flexmock(logic.image_location) .should_receive('get_image_location') .and_return(mock_logic_response)) request_url = '/image/{image_type}/{image_format}/{entity_id}/location'.\ format( entity_id=entity_id, image_type='product', image_format='cover') result = fixture_client.get(request_url) result_message = result.data.decode('utf8') assert result.status_code == 200 assert result.headers.get('Correlation-Id') assert mock_logic_response.message == result_message def test_check_for_desired_bit_depth(fixture_client): """Test route to check desired bit-depth for provided UPC.""" product_id = 2011621 desired_bit_depth = 64 mock_logic_response = response.Response(message='ok') (flexmock(logic.validators) .should_receive('check_for_desired_bit_depth') .and_return(mock_logic_response)) request_url = '/validators/product/{product_id}'\ '/desired-bit-depth/{bit_depth}?'\ 'physical_location_id=1&asset_type_id=1'.\ format(product_id=product_id, bit_depth=desired_bit_depth) result = fixture_client.get(request_url) result_message = result.data.decode('utf8') assert result.status_code == 200 assert result.headers.get('Correlation-Id') assert mock_logic_response.message == result_message def test_check_for_desired_bit_depth_with_invalid_upc(fixture_client): """Test route to check for deisered bit-depth with invalid UPC.""" product_id = 000 desired_bit_depth = 64 mock_logic_response = response.Response(status=400, message='not ok') (flexmock(logic.validators) .should_receive('check_for_desired_bit_depth') .and_return(mock_logic_response)) request_url = '/validators/product/{product_id}'\ '/desired-bit-depth/{bit_depth}?'\ 'physical_location_id=1&asset_type_id=1'.\ format(product_id=product_id, bit_depth=desired_bit_depth) result = fixture_client.get(request_url) result_message = result.data.decode('utf8') assert result.status_code == 400 assert result.headers.get('Correlation-Id') assert mock_logic_response.message == result_message @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS, {} ]) def test_copy_track_success( monkeypatch, fixture_client, track_copying_input_data, track_copying_response_response, headers): """"Test route that copy track asset.""" request_url_pattern = '/track/{0}/copy/{1}' request_url = request_url_pattern.format( track_copying_input_data['from_tuid'], track_copying_input_data['to_tuid'] ) ownership_mock = MagicMock(return_value=response.Response()) monkeypatch.setattr( ownership, 'check_track_ownership', value=ownership_mock) data = {} if not headers: data = {'Orchard-User-Id': 'alw:321'} track_copying_input_data['full_user_id'] = 'alw:321' else: track_copying_input_data['full_user_id'] = headers['Orchard-User-Id'] (flexmock(track) .should_receive('copy_track_asset') .with_args(**track_copying_input_data) .and_return(track_copying_response_response)) headers['Content-Type'] = 'application/json' result = fixture_client.post( request_url, data=json.dumps(data), headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') if headers == ALW_HEADERS: ownership_mock.assert_called() @pytest.mark.parametrize('headers, status, error_code', [ (OA_HEADERS, 400, error.ERROR_CODE_HEADER_VALIDATION), (ALW_HEADERS, 400, error.ERROR_CODE_HEADER_VALIDATION), ({}, 401, error.ERROR_CODE_AUTHORIZATION) ]) def test_copy_track_failure( fixture_client, track_copying_input_data, headers, status, error_code): """"Test route that copy track asset failure.""" request_url_pattern = '/track/{0}/copy/{1}' request_url = request_url_pattern.format( track_copying_input_data['from_tuid'], track_copying_input_data['to_tuid'] ) headers = headers.copy() data = json.dumps({}) if headers: headers['Orchard-User-Id'] = 'invalid_id' else: headers = {'Content-Type': 'application/json'} result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == status response_message = json.loads(result.data.decode()) assert response_message['code'] == error_code @pytest.mark.parametrize('failure_mock_call_order', [0, 1]) def test_copy_track_with_invalid_ownership( failure_mock_call_order, monkeypatch, fixture_client, track_copying_input_data): """"Test route that copy track asset failure.""" request_url_pattern = '/track/{0}/copy/{1}' request_url = request_url_pattern.format( track_copying_input_data['from_tuid'], track_copying_input_data['to_tuid'] ) headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' ownership_responses = [response.Response(), response.Response()] ownership_responses[failure_mock_call_order] = ( response.create_error_response( code='not_owned', message='not_owned', status=403)) monkeypatch.setattr( ownership, 'check_track_ownership', value=MagicMock( side_effect=ownership_responses)) result = fixture_client.post(request_url, headers=ALW_HEADERS) assert result.status_code == 403 assert result.headers.get('Correlation-Id') @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS, {} ]) def test_copy_product_success(monkeypatch, fixture_client, headers): """"Test copy product assets.""" ownership_mock = MagicMock(return_value=response.Response()) monkeypatch.setattr( ownership, 'check_ownership', value=ownership_mock) monkeypatch.setattr( copy_product_assets, 'copy_product_assets', value=MagicMock( return_value=response.Response())) request_url_pattern = '/product/{0}/copy/{1}' request_url = request_url_pattern.format(10, 20) data = {} if not headers: data = {'Orchard-User-Id': 'alw:321'} headers['Content-Type'] = 'application/json' result = fixture_client.post( request_url, data=json.dumps(data), headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') if headers == ALW_HEADERS: ownership_mock.assert_called() @pytest.mark.parametrize('headers, status, error_code', [ (OA_HEADERS, 400, error.ERROR_CODE_HEADER_VALIDATION), (ALW_HEADERS, 400, error.ERROR_CODE_HEADER_VALIDATION), ({}, 401, error.ERROR_CODE_AUTHORIZATION) ]) def test_copy_product_failure(fixture_client, headers, status, error_code): """"Test copy product assets failure.""" request_url_pattern = '/track/{0}/copy/{1}' request_url = request_url_pattern.format(10, 20) headers = headers.copy() data = json.dumps({}) if headers: headers['Orchard-User-Id'] = 'invalid_id' else: headers = {'Content-Type': 'application/json'} result = fixture_client.post(request_url, data=data, headers=headers) assert result.status_code == status response_message = json.loads(result.data.decode()) assert response_message['code'] == error_code def test_copy_product_with_invalid_ownership(monkeypatch, fixture_client): """"Test copy product assets.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' monkeypatch.setattr( ownership, 'check_ownership', value=MagicMock( return_value=response.create_error_response( code='not_owned', message='not_owned', status=403))) request_url_pattern = '/product/{0}/copy/{1}' request_url = request_url_pattern.format(10, 20) result = fixture_client.post(request_url, headers=headers) assert result.status_code == 403 assert result.headers.get('Correlation-Id') @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_copy_product_failure_logic_layer( monkeypatch, fixture_client, headers): """"Test copy product assets logic layer failure.""" monkeypatch.setattr( ownership, 'check_ownership', value=MagicMock( return_value=response.Response())) monkeypatch.setattr( copy_product_assets, 'copy_product_assets', value=MagicMock( return_value=response.create_fatal_response('fatal_error'))) request_url_pattern = '/product/{0}/copy/{1}' request_url = request_url_pattern.format(10, 20) result = fixture_client.post(request_url, headers=headers) assert result.status_code == 500 assert result.headers.get('Correlation-Id') @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_remove_product_image_success( monkeypatch, fixture_client, headers, fixture_response_ok): """"Test route that remove product image.""" ownership_mock = MagicMock(return_value=response.Response()) monkeypatch.setattr(ownership, 'check_ownership', value=ownership_mock) monkeypatch.setattr( remove_asset, 'remove_image_asset_by_product_id', value=MagicMock( return_value=fixture_response_ok)) product_id = 1 request_url = '/image/'.format(product_id) result = fixture_client.delete(request_url, headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') message = json.loads(result.data.decode('utf-8')) assert message == fixture_response_ok.message if headers == ALW_HEADERS: ownership_mock.assert_called() def test_remove_product_image_with_invalid_ownership( monkeypatch, fixture_client, fixture_response_ok): """"Test route that remove product image.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' monkeypatch.setattr( ownership, 'check_ownership', value=MagicMock( return_value=response.create_error_response( code='not_owned', message='not_owned', status=403))) monkeypatch.setattr( remove_asset, 'remove_image_asset_by_product_id', value=MagicMock( return_value=fixture_response_ok)) product_id = 1 request_url = '/image/'.format(product_id) result = fixture_client.delete(request_url, headers=headers) assert result.status_code == 403 assert result.headers.get('Correlation-Id') @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_remove_track_asset_success( monkeypatch, fixture_client, headers, fixture_response_ok): """"Test route that remove track asset.""" track_id = 1 request_url = '/track/'.format(track_id) ownership_mock = MagicMock(return_value=response.Response()) monkeypatch.setattr(ownership, 'check_track_ownership', ownership_mock) monkeypatch.setattr( remove_asset, 'remove_track_asset', value=MagicMock( return_value=fixture_response_ok)) result = fixture_client.delete(request_url, headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') message = json.loads(result.data.decode('utf-8')) assert message == fixture_response_ok.message if headers == ALW_HEADERS: ownership_mock.assert_called() def test_remove_track_asset_with_invalid_ownership( monkeypatch, fixture_client): """"Test route that remove track asset.""" # test for stub endpoint headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' monkeypatch.setattr( ownership, 'check_track_ownership', value=MagicMock( return_value=response.create_error_response( code='not_owned', message='not_owned', status=403))) track_id = 1 request_url = '/track/'.format(track_id) result = fixture_client.delete(request_url, headers=headers) assert result.status_code == 403 assert result.headers.get('Correlation-Id') @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_product_by_id_success( monkeypatch, fixture_client, headers, get_product_by_id_successul_response): """Test route for getting product assets.""" ownership_mock = MagicMock( return_value=response.Response()) monkeypatch.setattr( ownership, 'check_ownership', value=ownership_mock) monkeypatch.setattr( product, 'get_product_assets', value=MagicMock( return_value=get_product_by_id_successul_response)) product_id = '1' request_url = '/asset/product/'.format(product_id) result = fixture_client.get(request_url, headers=headers) assert result.status_code == 200 assert result.headers.get('Correlation-Id') message = json.loads(result.data.decode('utf-8')) assert message == get_product_by_id_successul_response.message if headers == ALW_HEADERS: ownership_mock.assert_called() @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_product_by_id_failure(fixture_client, headers): """Test route fails when there is invalid Orchard-User-Id in header.""" product_id = '1' request_url = '/asset/product/'.format(product_id) headers = headers.copy() headers['Orchard-User-Id'] = 'invalid_id' result = fixture_client.get(request_url, headers=headers) assert result.status_code == 400 response_message = json.loads(result.data.decode()) assert response_message['code'] == error.ERROR_CODE_HEADER_VALIDATION def test_get_product_by_id_with_invalid_ownership( monkeypatch, fixture_client): """Test route for getting product assets with invalid product ownership.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' monkeypatch.setattr( ownership, 'check_ownership', value=MagicMock( return_value=response.create_error_response( code='not_owned', message='not_owned', status=403))) product_id = '1' request_url = '/asset/product/'.format(product_id) result = fixture_client.get(request_url, headers=headers) assert result.status_code == 403 assert result.headers.get('Correlation-Id') def test_rename_correction_images_alw_user(fixture_client): """Test that a request from an alw user is rejected.""" headers = ALW_HEADERS.copy() headers['Content-Type'] = 'application/json' product_id = 1 request_url = '/correction/images/approval/{}'.format(product_id) result = fixture_client.post(request_url, headers=headers) assert result.status_code == 403 def test_rename_correction_images_oa_user(fixture_client): """Test that a request from an oa user is accepted.""" headers = OA_HEADERS.copy() headers['Content-Type'] = 'application/json' product_id = 1 request_url = '/correction/images/approval/{}'.format(product_id) (flexmock(logic.correction_images) .should_receive('rename_image_assets') .with_args(product_id) .and_return(response.Response())) result = fixture_client.post(request_url, headers=headers) assert result.status_code == 200 def test_rename_correction_images_no_user_id(fixture_client): """Test that a request without an orchard user id is rejected.""" headers = {'Content-Type': 'application/json'} product_id = 1 request_url = '/correction/images/approval/{}'.format(product_id) result = fixture_client.post(request_url, headers=headers) assert result.status_code == 400 @pytest.fixture def correct_post_assets_v2_body(): """Correct post asset body fixture.""" return { 'asset_type': 'WAV', 'product_id': 12345, 'upc': '123', 'track_unique_id': 54321, 'filename': 'unique_filename.wav' } def test_post_asset_v2_success(fixture_client, correct_post_assets_v2_body): """"Test route that saves new asset info from lambda call success.""" (flexmock(asset_upload) .should_receive('update_asset_upload') .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, correct_post_assets_v2_body): """"Test route that saves new asset info from lambda call failure.""" (flexmock(asset_upload) .should_receive('update_asset_upload') .and_return(response.create_fatal_response('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 @pytest.fixture def fixture_get_status_filename(): """Fixture to return filename for get asset endpoint param.""" return 'unique_filename.ext' @pytest.mark.parametrize('headers', [ OA_HEADERS, ALW_HEADERS ]) def test_get_asset_status_v2_success( fixture_client, fixture_get_status_filename, headers): """"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_status) .should_receive('get_current_status') .with_args( **expected_args ) .and_return(response.Response())) 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, fixture_get_status_filename, headers): """"Test failure of route that get asset general status.""" (flexmock(asset_status) .should_receive('get_current_status') .and_return(response.create_fatal_response('fatal_error'))) request_url = '/v2/asset/status/{filename}'.format( filename=fixture_get_status_filename) result = fixture_client.get(request_url, headers=headers) assert result.status_code == 500