"""Integration API tests for ows-product-digital.""" import json from os import getenv from time import sleep from auth.auth import GrassAuth from auth.auth import UserType from tests.integration.conftest import unsubmit_product_id from tests.utils.api_client.api_client import APIClient def workstation_session_token(user_id): """Return workstation grass auth token for a particular user.""" return GrassAuth(UserType.WORKSTATION, user_id).grass_token def oa_session_token(): """Return OA grass auth token for automation QA user.""" return GrassAuth(UserType.OA).grass_token def ws_api_client(workstation_user): """Initialize product digital APIClient with Workstation session token of provided user.""" return APIClient( getenv('BASE_QA_WORKSTATION_URL'), workstation_session_token( workstation_user['user_id'])) def oa_api_client(): """Initialize product digital APIClient with OA session token of provided user.""" return APIClient( getenv('BASE_QA_WORKSTATION_URL'), oa_session_token()) def product_data(project_id, subaccount_id): """Return a dictionary with product data.""" return {'format': 'Full Length', 'product_name': 'My Product', 'distribution_format_type': 'digital', 'upc': '', 'project_id': project_id, 'subaccount_id': subaccount_id} def create_product(user, project_id): """Create a new product and check response.""" product_response = ws_api_client(user).create(product_data(project_id, user['subaccount_id'])) assert product_response.status_code == 201, \ 'Result of POST was {}, expected 201.'.format(product_response.status_code) return json.loads(product_response.content) def get_product(user, product_id): """Run a GET for a product and check response.""" product_response = ws_api_client(user).get(product_id) assert product_response.status_code == 200, \ 'Result of GET was {}, expected 200.'.format(product_response.status_code) return json.loads(product_response.content) # These tests use the WS API client # ------------------------------------------------ def test_move_product_as_d3( d3_subaccount_1_user, project_id_sub_1_proj1, project_id_sub_2, d3_user): """Move product as d3 from one subaccount to another.""" product_id = create_product(d3_subaccount_1_user, project_id_sub_1_proj1)['product_id'] move_response = ws_api_client(d3_user).move(product_id, project_id_sub_2) assert move_response.status_code == 200, 'Result of PUT was {}, expected 200.'.format( move_response.status_code) assert json.loads(move_response.content)['project_id'] == project_id_sub_2, \ 'Expected move product project id to be {}, was {}'.format( project_id_sub_2, json.loads(move_response.content)['project_id']) get_result_project_id = get_product(d3_user, product_id)['project_id'] assert get_result_project_id == project_id_sub_2, \ 'Expected get product project id to be {}, was {}'.format( project_id_sub_2, get_result_project_id) def test_move_product_as_d3_same_sub( d3_subaccount_1_user, project_id_sub_1_proj1, project_id_sub_1_proj2, d3_user): """Move product as d3 from one subaccount to the same subaccount.""" product_id = create_product(d3_subaccount_1_user, project_id_sub_1_proj1)['product_id'] move_response = ws_api_client(d3_user).move(product_id, project_id_sub_1_proj2) assert move_response.status_code == 200, 'Result of PUT was {}, expected 200.'.format( move_response.status_code) assert json.loads(move_response.content)['project_id'] == project_id_sub_1_proj2, \ 'Expected move product project id to be {}, was {}'.format( project_id_sub_1_proj2, json.loads(move_response.content)['project_id']) get_result_project_id = get_product(d3_user, product_id)['project_id'] assert get_result_project_id == project_id_sub_1_proj2, \ 'Expected get product project id to be {}, was {}'.format( project_id_sub_1_proj2, get_result_project_id) def test_move_product_as_subaccount( d3_subaccount_1_user, project_id_sub_1_proj1, project_id_sub_1_proj2): """Move product as subaccount between projects.""" product_id = create_product(d3_subaccount_1_user, project_id_sub_1_proj1)['product_id'] move_response = ws_api_client(d3_subaccount_1_user).move(product_id, project_id_sub_1_proj2) assert move_response.status_code == 200, 'Result of PUT was {}, expected 200.'.format( move_response.status_code) assert json.loads(move_response.content)['project_id'] == project_id_sub_1_proj2, \ 'Expected move product project id to be {}, was {}'.format( project_id_sub_1_proj2, json.loads(move_response.content)['project_id']) get_result_project_id = get_product(d3_subaccount_1_user, product_id)['project_id'] assert get_result_project_id == project_id_sub_1_proj2, \ 'Expected get product project id to be {}, was {}'.format( project_id_sub_1_proj2, get_result_project_id) def test_move_product_as_subaccount_to_sub_2( d3_subaccount_1_user, project_id_sub_1_proj1, project_id_sub_2): """Ensure one subaccount can not move product to another subaccount.""" product_id = create_product(d3_subaccount_1_user, project_id_sub_1_proj1)['product_id'] move_response = ws_api_client(d3_subaccount_1_user).move(product_id, project_id_sub_2) assert move_response.status_code == 403, 'Result of PUT was {}, expected 403.'.format( move_response.status_code) def test_unsubmit_ec_product(test_user, product_in_error_correction): """Test that an error_correction product can be unsubmitted.""" client = ws_api_client(test_user) product_id = product_in_error_correction submit_response = client.submit(product_id) assert submit_response.status_code == 200, \ 'Submit PUT was {}, expected 200. Body: {}'.format( submit_response.status_code, submit_response.text) # The workflow service checks in the product asynchronously after submission. # Wait for that to complete before calling unsubmit. sleep(10) get_response = client.get(product_id) assert get_response.status_code == 200, \ 'GET after submit was {}, expected 200.'.format(get_response.status_code) status_after_submit = get_response.json()['release_status'] assert status_after_submit == 'transfer_to_content', \ 'Expected transfer_to_content after submit, was {}'.format(status_after_submit) unsubmit_response = client.unsubmit(product_id) assert unsubmit_response.status_code == 200, \ 'Unsubmit PUT was {}, expected 200. Body: {}'.format( unsubmit_response.status_code, unsubmit_response.text) get_response = client.get(product_id) assert get_response.status_code == 200, 'Result of GET was {}, expected 200.'.format( get_response.status_code) status = get_response.json()['release_status'] assert status == 'error_correction', \ 'Expected product to be in error_correction status, was {}'.format(status) def test_unsubmit_product(test_user): """Test that a label_processing product can be unsubmitted.""" product_id = unsubmit_product_id() client = ws_api_client(test_user) get_response = client.get(product_id) assert get_response.status_code == 200, 'Result of GET was {}, expected 200.'.format( get_response.status_code) if get_response.json()['release_status'] == 'label_processing': submit_response = client.submit(product_id) assert submit_response.status_code == 200, 'Result of PUT was {}, expected 200.'.format( submit_response.status_code) unsubmit_response = client.unsubmit(product_id) assert unsubmit_response.status_code == 200, 'Result of PUT was {}, expected 200.'.format( unsubmit_response.status_code) get_response = client.get(product_id) assert get_response.status_code == 200, 'Result of GET was {}, expected 200.'.format( get_response.status_code) status = get_response.json()['release_status'] assert status == 'label_processing', \ 'Expected product to be in label_processing status, was {}'.format(status) # These tests use the OA API client # ------------------------------------------------ def create_product_as_oa(user, project_id): """Create a new product and check response as an OA user.""" print('user', user) product_response = oa_api_client().create(product_data(project_id, user['subaccount_id'])) assert product_response.status_code == 201, \ 'Result of POST was {}, expected 201.'.format(product_response.status_code) return json.loads(product_response.content) def get_product_as_oa(product_id): """Run a GET for a product and check response as an OA user.""" product_response = oa_api_client().get(product_id) assert product_response.status_code == 200, \ 'Result of GET was {}, expected 200.'.format(product_response.status_code) return json.loads(product_response.content) def test_get_product_as_oa(d3_subaccount_1_user, project_id_sub_1_proj1): """Get product as an OA user.""" product_id = create_product_as_oa(d3_subaccount_1_user, project_id_sub_1_proj1)['product_id'] get_result = get_product_as_oa(product_id) assert get_result['product_id'] == product_id, \ 'Expected get product id to be {}, was {}'.format( product_id, get_result['product_id']) # These tests use a JWT-based client # ------------------------------------------------ def test_dataload_product_audio_validation(client, checked_out_product_id, vendor_id): """Test dataloader validation can be fetched for a product.""" response = client.dataload_product_audio_validation( [checked_out_product_id], account_type="vendor", account_id=vendor_id, orchard_user_id="" ) assert response.status_code == 200, 'Result of POST was {}, expected 200. Body: {}'.format( response.status_code, response.text) body = response.json() assert isinstance(body, list) assert len(body) == 1 assert 'validations' in body[0]