"""Tests for review_queue logic. Field display_name in move_target model would be deprecated soon. Field escalation_type has been added and is a replica of display_name. """ from contextlib import nullcontext as does_not_raise from datetime import datetime from unittest.mock import MagicMock, call import pytest from owsrequest import request from owsrequest.test_utils import MockOwsResponse from product_review.api import db from product_review.constants import services from product_review.logic import review_queue as review_queue_logic from product_review.models import move_target as move_target_model from product_review.models import ows_account as ows_account_model from product_review.models import ows_product as ows_product_model from product_review.models import ows_users as ows_users_model from product_review.models import queue_move_history as queue_move_history_model from product_review.models import review_queue as review_queue_model from product_review.models import splitio as splitio_model from product_review.models import graphql_router as graphql_router_model from tests.utils.db_utils import ( create_test_approval, create_test_move_target, create_test_rejection, create_test_review_queue_item, create_test_target_group, ) some_exception = Exception("some exception") def test_get_review_queue_item(mocker, test_review_queue_models): """Test getting a an review queue item.""" review_queue_dict = [item.to_dict() for item in test_review_queue_models] expected_return_value = test_review_queue_models, len( test_review_queue_models) expected_queue_response = { "items": review_queue_dict, "pagination": { "type": "standard", "offset": 0, "limit": 50, "total_records": len(test_review_queue_models), }, } mocker.patch.object( review_queue_model, "get_items", return_value=expected_return_value, autospec=True, ) results = review_queue_logic.get_review_queue_items(review_queue_id=2) review_queue_model.get_items.assert_called_with( "new", page_offset=None, page_limit=None, review_queue_id=2, product_id=None, ) assert results.status == 200 assert results.message == expected_queue_response for item in results.message["items"]: assert item["status"] == "new" def test_review_queue_items_by_id(mocker, test_review_queue_models): """Test getting review queue items by id.""" mocker.patch.object( review_queue_model, "get_review_queue_items_by_id", autospec=True, ) review_queue_logic.get_review_queue_items_by_id( review_queue_ids=["1", "2"]) review_queue_model.get_review_queue_items_by_id.assert_called_once() def test_get_target_group_by_item_id(mocker, test_review_queue_models): """Test getting review queue items by id.""" mocker.patch.object( review_queue_model, "get_target_group_by_item_id", autospec=True, ) review_queue_logic.get_target_group_by_item_id(review_queue_ids=["1", "2"]) review_queue_model.get_target_group_by_item_id.assert_called_once() def test_get_default_review_queue_items_success(mocker, test_review_queue_models): """Test get list of new review_queue items with model layer success.""" review_queue_dict = [item.to_dict() for item in test_review_queue_models] expected_return_value = test_review_queue_models, len( test_review_queue_models) expected_queue_response = { "items": review_queue_dict, "pagination": { "type": "standard", "offset": 0, "limit": 50, "total_records": len(test_review_queue_models), }, } mocker.patch.object( review_queue_model, "get_items", return_value=expected_return_value, autospec=True, ) results = review_queue_logic.get_review_queue_items() review_queue_model.get_items.assert_called_with( "new", page_offset=None, page_limit=None, review_queue_id=None, product_id=None, ) assert results.status == 200 assert results.message == expected_queue_response for item in results.message["items"]: assert item["status"] == "new" def test_get_default_review_queue_items_with_pagination( mocker, test_review_queue_models ): """Test get list of new review_queue items with pagination limits.""" review_queue_dict = [test_review_queue_models[0].to_dict()] expected_return_value = [test_review_queue_models[0]], 1 expected_queue_response = { "items": review_queue_dict, "pagination": { "type": "standard", "offset": 1, "limit": 1, "total_records": 1, }, } mocker.patch.object( review_queue_model, "get_items", return_value=expected_return_value, autospec=True, ) results = review_queue_logic.get_review_queue_items( page_offset=1, page_limit=1) review_queue_model.get_items.assert_called_with( "new", page_offset=1, page_limit=1, review_queue_id=None, product_id=None, ) assert results.message == expected_queue_response def test_get_review_queue_items_with_status(mocker): """Test get list of new review_queue items with model layer success.""" expected_return_value = [], 0 mocker.patch.object( review_queue_model, "get_items", return_value=expected_return_value, autospec=True, ) results = review_queue_logic.get_review_queue_items(status="complete") review_queue_model.get_items.assert_called_with( "complete", page_offset=None, page_limit=None, review_queue_id=None, product_id=None, ) assert results.status == 200 def test_returns_empty_list_success(mocker): """Test returns empty list when no new items are available in teh queue.""" expected_return_value = [], 0 expected_queue_response = { "items": [], "pagination": { "type": "standard", "offset": 0, "limit": 50, "total_records": 0, }, } mocker.patch.object( review_queue_model, "get_items", return_value=expected_return_value, autospec=True, ) results = review_queue_logic.get_review_queue_items() review_queue_model.get_items.assert_called_with( "new", page_offset=None, page_limit=None, review_queue_id=None, product_id=None, ) assert results.status == 200 assert results.message == expected_queue_response def test_move_review_queue_item_to_queue_success_already_in_queue(mocker): """Test move_review_queue_item_to_queue success already in queue.""" mocker.patch.object( review_queue_logic, "get_review_queue_item_for_update", return_value=review_queue_model.ReviewQueue( review_queue_id=123, product_id=456, status="new", queue_name="under_investigation", ), ) mocker.patch.object(review_queue_logic, "get_move_target") mocker.patch.object(review_queue_model, "move_review_queue_item_to_queue") mocker.patch.object(queue_move_history_model, "add_queue_move") result = review_queue_logic.move_review_queue_item_to_queue( review_queue_id=123, queue_name="under_investigation", move_note=None, moved_to_target_id=None, ) review_queue_logic.get_review_queue_item_for_update.assert_called_once_with( 123) review_queue_model.move_review_queue_item_to_queue.assert_not_called() queue_move_history_model.add_queue_move.assert_not_called() assert result == { "message": "Review queue item already in under_investigation queue", } def test_move_review_queue_item_to_queue_success_moved_to_queue(mocker, app_context): """Test move_review_queue_item_to_queue success moved to queue.""" app_context.g.request_context.identity_id = "some-user-id" mocker.patch.object( review_queue_logic, "get_review_queue_item_for_update", return_value=review_queue_model.ReviewQueue( review_queue_id=123, product_id=456, status="new", queue_name="initial", ), ) mocker.patch.object(review_queue_model, "move_review_queue_item_to_queue") mocker.patch.object(queue_move_history_model, "add_queue_move") mocker.patch.object(review_queue_logic, "publish_queue_move_event") result = review_queue_logic.move_review_queue_item_to_queue( review_queue_id=123, queue_name="under_investigation", move_note="some note", moved_to_target_id=None, ) review_queue_model.move_review_queue_item_to_queue.assert_called_once_with( review_queue_id=123, queue_name="under_investigation", move_note="some note", moved_to_target_id=None, moved_to_target_name=None, moved_to_target_email=None, moved_to_orchadmin_user_id=None, ) queue_move_history_model.add_queue_move.assert_called_once_with( queue_name="under_investigation", previous_queue_name="initial", review_queue_id=123, move_note="some note", moved_to_target_id=None, moved_to_target_name=None, moved_to_target_email=None, moved_to_orchadmin_user_id=None, ) review_queue_logic.get_review_queue_item_for_update.assert_called_once_with( 123) review_queue_logic.publish_queue_move_event.assert_called_once_with( product_id=456, review_queue_id=123, queue_name="under_investigation" ) assert result == review_queue_model.move_review_queue_item_to_queue.return_value def test_move_review_queue_item_to_queue_success_move_target(mocker, app_context): """Test move_review_queue_item_to_queue w/ move target fields.""" app_context.g.request_context.identity_id = "some-user-id" mocker.patch.object( review_queue_logic, "get_review_queue_item_for_update", return_value=review_queue_model.ReviewQueue( review_queue_id=123, product_id=456, status="new", queue_name="initial", ), ) mocker.patch.object( review_queue_logic, "get_move_target", return_value={ "orchadmin_user_id": 1001, "display_name": "test name", "email": "test@orchard.com", "moved_to_target_id": 1, }, ) mocker.patch.object(review_queue_model, "move_review_queue_item_to_queue") mocker.patch.object(queue_move_history_model, "add_queue_move") mocker.patch.object(review_queue_logic, "publish_queue_move_event") result = review_queue_logic.move_review_queue_item_to_queue( review_queue_id=123, queue_name="under_investigation", move_note="some note", moved_to_target_id=1, ) review_queue_logic.get_move_target.assert_called_once_with(1, 456) review_queue_model.move_review_queue_item_to_queue.assert_called_once_with( review_queue_id=123, queue_name="under_investigation", move_note="some note", moved_to_target_id=1, moved_to_target_name="test name", moved_to_target_email="test@orchard.com", moved_to_orchadmin_user_id=1001, ) queue_move_history_model.add_queue_move.assert_called_once_with( queue_name="under_investigation", previous_queue_name="initial", review_queue_id=123, move_note="some note", moved_to_target_id=1, moved_to_target_name="test name", moved_to_target_email="test@orchard.com", moved_to_orchadmin_user_id=1001, ) review_queue_logic.get_review_queue_item_for_update.assert_called_once_with( 123) review_queue_logic.publish_queue_move_event.assert_called_once_with( product_id=456, review_queue_id=123, queue_name="under_investigation" ) assert result == review_queue_model.move_review_queue_item_to_queue.return_value @pytest.mark.parametrize( ( "test_description", "mock_move_target_object", "expected_get_product_calls", "expected_get_vendor_assigned_to_calls", "expected_get_vendor_assigned_reviewer_calls", "expected_get_orchadmin_user_calls", "expected_result", ), [ ( "label manager", {"target_group_id": 1}, [call(456)], [call(111)], [], [call("oa:1001")], { "orchadmin_user_id": 1001, "display_name": "Some Test User", "email": "test@orchard.com", "target_group_id": 1 }, ), ( "assigned reviewer", {"target_group_id": 3}, [call(456)], [], [call(111)], [], { "orchadmin_user_id": 1002, "display_name": "First Last", "email": "test2@orchard.com", "target_group_id": 3 }, ), ( "ows-users orchadmin user", { "orchadmin_user_id": 1000, "target_group_id": 2, }, [], [], [], [call("oa:1000")], { "orchadmin_user_id": 1000, "display_name": "Some Test User", "email": "test@orchard.com", "target_group_id": 2 }, ), ], ) def test_get_move_target_success( mocker, test_description, mock_move_target_object, expected_get_product_calls, expected_get_vendor_assigned_to_calls, expected_get_vendor_assigned_reviewer_calls, expected_get_orchadmin_user_calls, expected_result, ): """Test get_move_target success.""" mocker.patch.object( splitio_model, "is_enabled_escalate_to_product_manager", return_value=False, ) mocker.patch.object( move_target_model, "get_move_target", return_value=mock_move_target_object, ) mocker.patch.object( ows_product_model, "get_product", return_value={"vendor_id": 111} ) mocker.patch.object( ows_account_model, "get_vendor_assigned_to", return_value={ "id": 1001, "f_name": "test", "l_name": "name", }, ) mocker.patch.object( ows_account_model, "get_vendor_assigned_reviewer", return_value={ "id": 1002, "f_name": "First", "l_name": "Last", "email": "test2@orchard.com", }, ) mocker.patch.object( ows_users_model, "get_orchadmin_user", return_value={ "user_id": "oa:1001", "first_name": "Some", "last_name": "Test User", "email": "test@orchard.com" }, ) result = review_queue_logic.get_move_target(1, 456) move_target_model.get_move_target.assert_called_once_with(1) assert ows_product_model.get_product.mock_calls == expected_get_product_calls assert ( ows_account_model.get_vendor_assigned_to.mock_calls == expected_get_vendor_assigned_to_calls ) assert ( ows_account_model.get_vendor_assigned_reviewer.mock_calls == expected_get_vendor_assigned_reviewer_calls ) assert ( ows_users_model.get_orchadmin_user.mock_calls == expected_get_orchadmin_user_calls ) assert result == expected_result @pytest.mark.parametrize( ( "test_description", "company_brand", "expected_get_product_calls", "expected_get_vendor_company_brand_calls", "expected_get_vendor_product_manager_calls", "expected_get_vendor_assigned_to_calls", "expected_get_orchadmin_user_calls", "expected_result", ), [ ( "product manager", {"name": "theorchard"}, [call(456)], [call(111)], [call(111)], [], [call("oa:1003")], { "orchadmin_user_id": 1003, "display_name": "Test User", "email": "test@orchard.com", "target_group_id": 1 }, ), ( "label manager", {"name": "awal"}, [call(456)], [call(111)], [], [call(111)], [call("oa:1003")], { "orchadmin_user_id": 1003, "display_name": "Test User", "email": "test@orchard.com", "target_group_id": 1 }, ), ] ) def test_get_move_target_escalate_to_pm( mocker, test_description, company_brand, expected_get_product_calls, expected_get_vendor_company_brand_calls, expected_get_vendor_product_manager_calls, expected_get_vendor_assigned_to_calls, expected_get_orchadmin_user_calls, expected_result ): """Test get_move_target with escalate to pm feature flag.""" mocker.patch.object( splitio_model, "is_enabled_escalate_to_product_manager", return_value=True, ) mocker.patch.object( move_target_model, "get_move_target", return_value={"target_group_id": 1}, ) mocker.patch.object( ows_product_model, "get_product", return_value={"vendor_id": 111} ) mocker.patch.object( graphql_router_model, "get_vendor_company_brand", return_value=company_brand, ) mocker.patch.object( graphql_router_model, "get_vendor_product_manager", return_value=1003, ) mocker.patch.object( ows_account_model, "get_vendor_assigned_to", return_value={ "id": 1003, "f_name": "test", "l_name": "name", }, ) mocker.patch.object( ows_users_model, "get_orchadmin_user", return_value={ "user_id": "oa:1003", "first_name": "Test", "last_name": "User", "email": "test@orchard.com" } ) result = review_queue_logic.get_move_target(1, 456) move_target_model.get_move_target.assert_called_once_with(1) assert ows_product_model.get_product.mock_calls == expected_get_product_calls assert ( graphql_router_model.get_vendor_company_brand.mock_calls == expected_get_vendor_company_brand_calls ) assert ( graphql_router_model.get_vendor_product_manager.mock_calls == expected_get_vendor_product_manager_calls ) assert ( ows_account_model.get_vendor_assigned_to.mock_calls == expected_get_vendor_assigned_to_calls ) assert ( ows_users_model.get_orchadmin_user.mock_calls == expected_get_orchadmin_user_calls ) assert result == expected_result @pytest.mark.parametrize( ("test_description", "mock_move_target_item"), [ ("move_target_item not found", None), ("target group not found", {"target_group_id": 4}) ] ) def test_get_move_target_error(mocker, test_description, mock_move_target_item): """Test get_move_target error.""" mocker.patch.object( move_target_model, "get_move_target", return_value=mock_move_target_item ) with pytest.raises(Exception): review_queue_logic.get_move_target(2, 456) move_target_model.get_move_target.assert_called_once_with(2) def test_get_queue_move_history(mocker): """Test get_queue_move_history.""" mocker.patch.object(queue_move_history_model, "get_queue_move_history") result = review_queue_logic.get_queue_move_history( review_queue_id=123, ) queue_move_history_model.get_queue_move_history.assert_called_once_with( review_queue_id=123, ) assert result == { "items": queue_move_history_model.get_queue_move_history.return_value, } def test_insert_review_queue_item(mocker, test_review_queue_models): """Test inserting an review queue item.""" mocker.patch.object( review_queue_logic, "get_error_correction_id", return_value=None ) result = review_queue_logic.insert_review_queue_item(8, "initial") assert result.status == 201 with pytest.raises(Exception): review_queue_logic.insert_review_queue_item(8, "initial") result = review_queue_logic.insert_review_queue_item(9, "initial") assert result.status == 201 def test_get_product_review_history( mocker, test_review_history_models, test_product_review_history_dict): """Test returns completed reviews for a product.""" expected_return_value = test_review_history_models, len( test_review_history_models ) expected_response = { "items": test_product_review_history_dict, "pagination": { "type": "standard", "offset": 0, "limit": 50, "total_records": len(test_product_review_history_dict), }, } mocker.patch.object( review_queue_model, "review_history", return_value=expected_return_value, autospec=True, ) results = review_queue_logic.get_product_review_history(123456) review_queue_model.review_history.assert_called_with( page_offset=None, page_limit=None, product_id=123456, current_item=True ) assert results.status == 200 assert results.message == expected_response @pytest.mark.parametrize( ( "test_description", "expected_review_queue_model_calls", "expected_review_queue_model_side_effect", "expected_db_session_commit_calls", "expected_db_session_rollback_calls", "expected_raise", "expected_raise_message", "expected_result", ), [ ( "success", [call(123)], None, [call()], [], does_not_raise(), "None", "some result", ), ( "review_queue model exception", [call(123)], some_exception, [], [call()], pytest.raises(Exception), str(some_exception), None, ), ], ) def test_lock( mocker, test_description, expected_review_queue_model_calls, expected_review_queue_model_side_effect, expected_db_session_commit_calls, expected_db_session_rollback_calls, expected_raise, expected_raise_message, expected_result, ): """Test locking a review queue item.""" mocker.patch.object( review_queue_model, "lock", return_value=expected_result, side_effect=expected_review_queue_model_side_effect, autospec=True, ) mocker.patch.object(db.session, "rollback", autospec=True) mocker.patch.object(db.session, "commit", autospec=True) mock_response = MockOwsResponse(200, '{"status":200}') mocker.patch.object(request, "put", return_value=mock_response) result = None with expected_raise as er: result = review_queue_logic.lock(123) assert result == expected_result assert str(getattr(er, "value", None)) == expected_raise_message assert review_queue_model.lock.mock_calls == expected_review_queue_model_calls assert db.session.rollback.mock_calls == expected_db_session_rollback_calls assert db.session.commit.mock_calls == expected_db_session_commit_calls @pytest.mark.parametrize( ( "test_description", "expected_review_queue_model_calls", "expected_review_queue_model_side_effect", "expected_db_session_commit_calls", "expected_db_session_rollback_calls", "expected_raise", "expected_raise_message", "expected_result", ), [ ( "success", [call(123)], None, [call()], [], does_not_raise(), "None", "some result", ), ( "review_queue model exception", [call(123)], some_exception, [], [call()], pytest.raises(Exception), str(some_exception), None, ), ], ) def test_unlock( mocker, test_description, expected_review_queue_model_calls, expected_review_queue_model_side_effect, expected_db_session_commit_calls, expected_db_session_rollback_calls, expected_raise, expected_raise_message, expected_result, ): """Test unlocking a review queue item.""" mocker.patch.object( review_queue_model, "unlock", return_value=expected_result, side_effect=expected_review_queue_model_side_effect, autospec=True, ) mocker.patch.object(db.session, "rollback", autospec=True) mocker.patch.object(db.session, "commit", autospec=True) mock_response = MockOwsResponse(200, '{"status":200}') mocker.patch.object(request, "put", return_value=mock_response) result = None with expected_raise as er: result = review_queue_logic.unlock(123) assert result == expected_result assert str(getattr(er, "value", None)) == expected_raise_message assert review_queue_model.unlock.mock_calls == expected_review_queue_model_calls assert db.session.rollback.mock_calls == expected_db_session_rollback_calls assert db.session.commit.mock_calls == expected_db_session_commit_calls def test_get_status_logic(): """Test getting a review queue item with status.""" mock_item = create_test_review_queue_item( **{"review_queue_id": 1, "product_id": 1, "status": "complete"} ) mock_rejection = create_test_rejection( **{ "review_queue_id": mock_item.review_queue_id, "user_id": "bar", "created_datetime": datetime(2023, 1, 21, 13, 34, 5, 80), "note": "foobar", } ) expected_return_value = { "id": mock_item.product_id, "ec_and_ar": False, "release_status": "action_required", "release_approval_status": "rejected", "last_updated": "2023-01-21 13:34:05.000080", "note": mock_rejection.note, } results = review_queue_logic.get_status(1) assert results.message == expected_return_value assert results.status == 200 def test_get_applying_status_logic(): """Test getting a review queue item with status.""" mock_item = create_test_review_queue_item( **{ "review_queue_id": 1, "product_id": 1, "status": "applying", "created_datetime": datetime(2023, 1, 21, 13, 34, 5, 80), } ) expected_return_value = { "id": mock_item.product_id, "ec_and_ar": False, "release_status": "transfer_to_content", "release_approval_status": "checked_out", "last_updated": "2023-01-21 13:34:05.000080", } results = review_queue_logic.get_status(1) assert results.message == expected_return_value assert results.status == 200 def test_get_will_not_deliver_status_logic(): """Test getting a review queue item rejected with will_not_deliver.""" mock_item = create_test_review_queue_item( **{ "review_queue_id": 1, "product_id": 1, "status": "complete", "created_datetime": datetime(2023, 1, 21, 13, 34, 5, 80), } ) create_test_rejection( **{ "review_queue_id": 1, "will_not_deliver": True, "note": "foo", "user_id": "abc", "created_datetime": datetime(2023, 1, 21, 13, 34, 5, 80), } ) expected_return_value = { "id": mock_item.product_id, "ec_and_ar": False, "release_status": "transfer_to_content", "release_approval_status": "checked_out", "last_updated": "2023-01-21 13:34:05.000080", } results = review_queue_logic.get_status(1) assert results.message == expected_return_value assert results.status == 200 def test_send_email_notification_rejection(mocker): """Test sending a product rejection email.""" create_test_review_queue_item( **{"review_queue_id": 1, "product_id": 1, "status": "complete"} ) create_test_rejection( **{ "review_queue_id": 1, "user_id": "bar", "note": "foobar", } ) mock_response = MockOwsResponse(200, ["LabelProfile_123"]) mocker.patch.object(request, "post", return_value=mock_response) expected_return_value = { "notification_type": "rejection", "profiles": ["LabelProfile_123"], } results = review_queue_logic.send_email_notification(1) assert results.message == expected_return_value assert results.status == 201 request.post.assert_called_once_with( services.OWS_NOTIFICATIONS, "/activity/product_rejection", json={ "product_id": 1, "rejection_reasons": [{"title": "General Comments", "comments": "foobar"}], }, ) def test_send_email_notification_will_not_deliver(mocker): """Test send_email_notification will not deliver.""" create_test_review_queue_item( **{"review_queue_id": 1, "product_id": 1, "status": "complete"} ) create_test_rejection( **{ "review_queue_id": 1, "user_id": "bar", "note": "foobar", "will_not_deliver": 1, } ) mocker.patch.object(request, "post") expected_return_value = {"will_not_deliver": True} results = review_queue_logic.send_email_notification(1) assert results.message == expected_return_value assert results.status == 200 request.post.assert_not_called() def test_send_email_notification_bad_review(mocker): """Test error raised if review has no approval or rejection.""" create_test_review_queue_item( **{"review_queue_id": 1, "product_id": 1, "status": "complete"} ) mocker.patch.object(request, "post", return_value=MagicMock()) results = review_queue_logic.send_email_notification(1) assert results.status == 400 assert results.errors == { "code": "bad_request", "message": "Review 1 is missing data.", } request.post.assert_not_called() @pytest.mark.parametrize( ("test_description", "mock_vendor_owner", "mock_company_brand"), [ ("test vendor owner", "awal", "random"), ("test company brand", "random", "awal") ] ) def test_send_email_notification_approval( test_description, mock_vendor_owner, mock_company_brand, mocker ): """Test approval notification.""" create_test_review_queue_item( **{"review_queue_id": 1, "product_id": 1, "status": "complete"} ) create_test_approval( **{ "review_queue_id": 1, "user_id": "bar", "note": "foobar", } ) mock_response = MockOwsResponse(200, ["LabelProfile_123"]) mocker.patch.object(ows_product_model, "get_product", return_value={"vendor_id": 1}) mocker.patch.object(ows_account_model, "get_vendor", return_value={ "owner": mock_vendor_owner, "company_brand": mock_company_brand }) mocker.patch.object(request, "post", return_value=mock_response) expected_return_value = { "notification_type": "approval", "profiles": ["LabelProfile_123"], } results = review_queue_logic.send_email_notification(1) assert results.message == expected_return_value assert results.status == 201 request.post.assert_called_once_with( services.OWS_NOTIFICATIONS, "/activity/product_approval", json={"product_id": 1} ) def test_send_email_notification_approval_not_sent(mocker): """Test approval notification not sent.""" create_test_review_queue_item( **{"review_queue_id": 1, "product_id": 1, "status": "complete"} ) create_test_approval( **{ "review_queue_id": 1, "user_id": "bar", "note": "foobar", } ) mocker.patch.object(ows_product_model, "get_product", return_value={"vendor_id": 1}) mocker.patch.object( ows_account_model, "get_vendor", return_value={ "owner": "random", "company_brand": "random" } ) mocker.patch.object(request, "post") results = review_queue_logic.send_email_notification(1) assert results.message == { "notification_type": "approval", "profiles": [], } assert results.status == 200 request.post.assert_not_called() def test_send_email_notification_not_found(mocker): """Test error raised if review is not found.""" mocker.patch.object(request, "post", return_value=MockOwsResponse(200, {})) results = review_queue_logic.send_email_notification(1) assert results.status == 404 assert results.errors == { "code": "not_found_error", "message": "Review 1 not found.", } request.post.assert_not_called() def test_get_all_target_groups(mocker): """Test get all target groups.""" mock_target_group_0 = create_test_target_group( target_group_id=1, display_name="some group" ) mock_target_group_1 = create_test_target_group( target_group_id=2, display_name="another group" ) mock_move_target_0 = create_test_move_target( moved_to_target_id=10, display_name="abc", target_group_id=1, escalation_type="abc" ) mock_move_target_1 = create_test_move_target( moved_to_target_id=11, display_name="somebody", target_group_id=1, show_in_ui=0, escalation_type="somebody" ) mock_move_target_2 = create_test_move_target( moved_to_target_id=12, display_name="label manager", target_group_id=2, show_in_ui=0, escalation_type="label manager" ) expected_result = [ { **mock_target_group_0.to_dict(), "targets": [mock_move_target_0.to_dict(), mock_move_target_1.to_dict()], }, { **mock_target_group_1.to_dict(), "targets": [mock_move_target_2.to_dict()], }, ] result = review_queue_logic.get_all_target_groups() assert result == expected_result def test_get_error_correction_id(app_context, request_engine): """Test getting an error correction id successfully.""" request_engine["ows-product"].add_spec( "GET", "/product/8", response={"status": "in_content"} ) request_engine["ows-product-workflow"].add_spec( "GET", "/product/8/correction", response={ "release_correction_id": 3, "release_id": 8, "status": "submitted", "items": [ { "release_correction_detail_id": 1, "table_name": "releases", "field_name": "release_name", "key_id": 8, "key_value": "New Name", } ], }, ) actual = review_queue_logic.get_error_correction_id(8) assert actual == 3 def test_get_error_correction_id_new_submission(app_context, request_engine): """Test trying to get the error correction id for a new submission.""" request_engine["ows-product"].add_spec( "GET", "/product/8", response={"status": "transfer_to_content"} ) actual = review_queue_logic.get_error_correction_id(8) assert actual is None def test_get_error_correction_ows_product_error(app_context, request_engine): """Test encountering an error when fetching the product status.""" request_engine["ows-product"].add_spec( "GET", "/product/8", status=500, response={"error_code": "foo"} ) with pytest.raises(Exception) as e: review_queue_logic.get_error_correction_id(8) assert e.value.args[0] == ( '{"status": 500, "code": "ows_product_error", ' '"message": "{\\"error_code\\": \\"foo\\"}"}' ) def test_get_error_correction_ows_product_workflow_error(app_context, request_engine): """Test encountering an error when fetching the product error correction.""" request_engine["ows-product"].add_spec( "GET", "/product/8", response={"status": "in_content"} ) request_engine["ows-product-workflow"].add_spec( "GET", "/product/8/correction", status=500, response={"error_code": "foo"} ) with pytest.raises(Exception) as e: review_queue_logic.get_error_correction_id(8) assert e.value.args[0] == ( '{"status": 500, "code": "ows_product_workflow_error", ' '"message": "{\\"error_code\\": \\"foo\\"}"}' )