"""Tests for orders model.""" from datetime import datetime import pytest from src.constants import fields from src.constants import services from src.models import orders from src.models import sql_queries mock_module = orders def request_result( created_at='2018-01-01T10:11:12', user_id=5, priority=2, meta_update='Y'): """Generate test data.""" result = {} result[fields.CREATED_AT] = created_at result[fields.USER_ID] = user_id if priority: result[fields.PRIORITY] = priority if meta_update: result[fields.META_UPDATE] = meta_update return result def get_queue_ids(count=2): """Generate fake encoding_queue ids.""" return [10000 + i for i in range(0, count)] def get_db_queue(count=2): """Generate fake encoding_queue db data.""" result = [] ids = get_queue_ids(count) for i in range(0, count): result.append({ fields.ENCODING_QUEUE_ID: ids[i], fields.ENCODING_ORDER_ID: 1000 + i, fields.ENCODING_ORDER_TYPE: 'release', fields.PRIORITY: 1, fields.META_UPDATE: 'Y', fields.ENCODER_ID: 10 }) return result def get_processed_queue(count=2, change_fields=True): """Generate encoding_queue fake processed data.""" if change_fields: fn_order_id = fields.ORDER_ID fn_order_type = fields.ORDER_TYPE else: fn_order_id = fields.ENCODING_ORDER_ID fn_order_type = fields.ENCODING_ORDER_TYPE result = {} ids = get_queue_ids(count) for i in range(0, count): queue_id = ids[i] result[queue_id] = { fn_order_id: 1000 + i, fn_order_type: 'release', fields.PRIORITY: 1, fields.META_UPDATE: True if change_fields else 'Y', fields.ENCODER_ID: 10 } if not change_fields: result[queue_id][fields.ENCODING_QUEUE_ID] = queue_id return result @pytest.mark.parametrize( 'argument, func_result, error', ( (get_db_queue(1)[0], list(get_processed_queue(1).values())[0], False), ({}, None, True) ) ) def test_extract_from_encoding_queue(argument, func_result, error): """Test _extract_from_encoding_queue function.""" if error: with pytest.raises(KeyError): orders._extract_from_encoding_queue(argument) else: assert orders._extract_from_encoding_queue( argument) == func_result def test_get_encoding_queue_detail_bulk_empty_input(): """Test bulk queries for dms_delivery_spec records for empty input.""" assert orders.get_encoding_queue_bulk(()) == {} @pytest.mark.parametrize( 'input_ids, db_query_result, change_fields, expected_result', ( # The case with all the records found. (get_queue_ids(), get_db_queue(), True, get_processed_queue()), (get_queue_ids(), get_db_queue(), False, get_processed_queue(change_fields=False)), # The case with some of the records found. (get_queue_ids(), get_db_queue(1), True, get_processed_queue(1)), # The case with no records found. (get_queue_ids(), [], True, {}), )) def test_get_encoding_queue_bulk( input_ids, db_query_result, change_fields, expected_result, patch_session): """Test get_encoding_queue_bulk function.""" mocked_session = patch_session( 'orders', 'direct_delivery', db_query_result) result = orders.get_encoding_queue_bulk(input_ids, change_fields) assert result == expected_result assert mocked_session.execute.call_count == 1 assert mocked_session.execute.call_args[0] == ( sql_queries.DD_SELECT_ENCODING_QUEUE_BULK, {fields.SQL_ID_LIST: input_ids} ) def get_db_order(str_date=False): """Generate fake order list.""" return { fields.ORDER_ID: 1001, fields.CREATED_AT: ( '2018-01-01T10:11:12' if str_date else datetime(2018, 1, 1, 10, 11, 12)), fields.USER_ID: 5 } def get_processed_order(str_date=False): """Generate fake processed order list.""" order_dict = get_db_order(str_date=str_date) pk = order_dict.pop(fields.ORDER_ID) return {pk: order_dict} @pytest.mark.parametrize( 'sql_result, func_result', ( ([get_db_order()], get_processed_order()), ([], {}) ) ) def test_get_encoding_order_db_bulk( sql_result, func_result, patch_session): """Test get_encoding_order_db function.""" test_ids = [1, 2] mocked_session = patch_session('orders', 'art_relations', sql_result) result = orders.get_encoding_order_db_bulk(test_ids) assert mocked_session.execute.call_count == 1 assert mocked_session.execute.call_args[0] == ( sql_queries.AR_SELECT_ENCODING_ORDER_BULK, {fields.SQL_ID_LIST: test_ids} ) assert result == func_result @pytest.mark.parametrize( 'service_response, expected_result', ( ( { 'items': [ { 'created_at': '2017-12-19T08:03:28', 'encoder_id': 18, 'meta_update': False, 'order_id': '1562059', 'priority': 2, 'user_id': 179 }, { 'created_at': '2018-01-24T14:34:25', 'encoder_id': 18, 'meta_update': True, 'order_id': '1605992', 'priority': 2, 'user_id': 179 } ] }, { 1562059: { 'created_at': '2017-12-19T08:03:28', 'user_id': 179 }, 1605992: { 'created_at': '2018-01-24T14:34:25', 'user_id': 179 } } ), ( {'items': []}, {}, ), ( None, {}, ), ) ) def test_get_encoding_order_via_api_bulk( service_response, expected_result, mocker): """Test retrieving multiple Vector orders from micro-service.""" order_ids = (1562059, 1605992) mocked_get = mocker.patch('src.models.orders.request.get') mocked_get.return_value = service_response result = orders.get_encoding_order_via_api_bulk(order_ids) assert result == expected_result mocked_get.assert_called_once_with( services.OWS_VECTORORDER, '/orders?order_id=1562059,1605992') @pytest.mark.parametrize( 'order_ids, api_result, expected_result', ( ([1, 2], {1: 'foo', 2: 'bar'}, {1: 'foo', 2: 'bar'}), ) ) def test_get_encoding_order_bulk( order_ids, api_result, expected_result, mocker): """Test get_encoding_order_bulk calls microservice.""" mocked_get_api = mocker.patch( 'src.models.orders.get_encoding_order_via_api_bulk') mocked_get_api.return_value = api_result mocked_get_db = mocker.patch('src.models.orders.get_encoding_order_db_bulk') result = orders.get_encoding_order_bulk(order_ids) assert result == expected_result assert mocked_get_api.call_count == 1 assert mocked_get_api.call_args[0] == (order_ids,) assert mocked_get_db.call_count == 0 @pytest.mark.parametrize( 'order_ids, api_result, missing_ids, db_result, expected_result', ( ([1, 2, 3], {1: 'a', 2: 'b'}, [3], {}, {1: 'a', 2: 'b'}), ([1, 2, 3], {1: 'a', 2: 'b'}, [3], {3: 'c'}, {1: 'a', 2: 'b', 3: 'c'}), ([1, 2, 3], {}, [1, 2, 3], {3: 'c'}, {3: 'c'}), ) ) def test_get_encoding_order_bulk_with_db_call( order_ids, api_result, missing_ids, db_result, expected_result, mocker): """Test get_encoding_order_bulk calls microservice and db.""" mocked_get_api = mocker.patch( 'src.models.orders.get_encoding_order_via_api_bulk') mocked_get_api.return_value = api_result mocked_get_db = mocker.patch('src.models.orders.get_encoding_order_db_bulk') mocked_get_db.return_value = db_result result = orders.get_encoding_order_bulk(order_ids) assert result == expected_result assert mocked_get_api.call_count == 1 assert mocked_get_api.call_args == [(order_ids,), {}] assert mocked_get_db.call_count == 1 assert mocked_get_db.call_args == [(missing_ids,), {}] @pytest.mark.parametrize( 'order_ids, api_result, missing_ids, db_result, use_ar_only, ' 'api_call_count, db_call_count, expected_result', ( ( [1, 2, 3], {1: 'a', 2: 'b'}, [3], {}, False, 1, 1, {1: 'a', 2: 'b'} ), ( [1, 2, 3], {1: 'a', 2: 'b'}, [3], {3: 'c'}, False, 1, 1, {1: 'a', 2: 'b', 3: 'c'}), ( [1, 2, 3], {}, [1, 2, 3], {3: 'c'}, False, 1, 1, {3: 'c'} ), ( [1, 2, 3], {}, [], {1: 'a'}, True, 0, 1, {1: 'a'} ), ) ) def test_get_encoding_order_bulk_ar_only( order_ids, api_result, missing_ids, db_result, use_ar_only, api_call_count, db_call_count, expected_result, mocker): """Test get_encoding_order_bulk calls AR only if configured so.""" config_mock = mocker.patch('src.models.orders.config') config_mock.FETCH_VECTOR_ORDERS_FROM_AR_ONLY = use_ar_only mocked_get_api = mocker.patch( 'src.models.orders.get_encoding_order_via_api_bulk') mocked_get_api.return_value = api_result mocked_get_db = mocker.patch('src.models.orders.get_encoding_order_db_bulk') mocked_get_db.return_value = db_result db_call_ids = order_ids if use_ar_only else missing_ids result = orders.get_encoding_order_bulk(order_ids) assert result == expected_result assert mocked_get_db.call_count == db_call_count assert mocked_get_db.call_args == [(db_call_ids,), {}] assert mocked_get_api.call_count == api_call_count if api_call_count: assert mocked_get_api.call_args == [(order_ids,), {}]