"""Tests for review_queue.""" import datetime from contextlib import nullcontext as does_not_raise import pytest from freezegun import freeze_time from freezegun.api import FakeDatetime from product_review.api import db from product_review.models import review_queue as review_queue_model from tests.utils.db_utils import ( create_test_approval, create_test_rejection, create_test_review_queue_item, create_test_score_history_item ) def test_review_queue_created(): """Test that object has inherited fields.""" test_object = create_test_review_queue_item(product_id=123456, status="new") assert test_object.product_id == 123456 assert test_object.status == "new" @pytest.mark.parametrize( ( "test_description", "inputs", "filter_status", "filter_review_queue_id", "expected_result", ), [ ( "get items by new status success", [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "new"}, {"product_id": 99999, "status": "complete"}, ], "new", None, {"expected_status": "new", "expected_number": 2}, ), ( "get items by complete status success", [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "new"}, {"product_id": 99999, "status": "complete"}, ], "complete", None, {"expected_status": "complete", "expected_number": 1}, ), ( "get items by invalid status", [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "new"}, {"product_id": 99999, "status": "complete"}, ], "garbage", None, {"expected_status": "new", "expected_number": 0}, ), ( "get items has no results", [], "new", None, {"expected_status": "new", "expected_number": 0}, ), ( "get single item", [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "new"}, {"product_id": 99999, "status": "complete"}, ], "new", 2, {"expected_status": "new", "expected_number": 1}, ), ( "get single item with non-matching status", [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "new"}, {"product_id": 99999, "status": "complete"}, ], "complete", 2, {"expected_number": 0}, ), ], ) def test_get_items( test_description, inputs, filter_status, filter_review_queue_id, expected_result, ): """Test that get_items filters items by status.""" for item in inputs: create_test_review_queue_item( product_id=item["product_id"], status=item["status"] ) results, results_total = review_queue_model.get_items( filter_status, review_queue_id=filter_review_queue_id ) for item in results: assert item.status == expected_result["expected_status"] single_result, single_result_count = review_queue_model.get_items( product_id=item.product_id ) assert single_result[0].product_id == item.product_id assert single_result_count == 1 assert results_total == expected_result["expected_number"] def test_get_review_queue_items_by_id(): """Test getting review queue items by id.""" create_test_review_queue_item( review_queue_id=123, product_id=123456, status="new", queue_name="initial" ) create_test_score_history_item( review_queue_id=123, validation_result="PASS", confidence_score=1, score_reasoning=[], score_version=1, created_datetime=datetime.datetime(2025, 11, 17, 15, 6, 6, 869093) ) results = review_queue_model.get_review_queue_items_by_id(["123"]) expected_result = { "review_queue_id": 123, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "locked_by_user_id": None, "locked_at_datetime": None, "locked_until_datetime": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "move_note": None, "moved_at_datetime": None, "approval": None, "rejection": None, "submitted_by_user_id": None, "submission_datetime": None, "score_history": { 'auto_approved': False, 'confidence_score': 1, 'created_datetime': datetime.datetime(2025, 11, 17, 15, 6, 6, 869093), 'failed_validations': None, 'review_queue_id': 123, 'score_history_id': 1, 'score_reasoning': [], 'score_version': 1.0, 'validation_result': 'PASS', } } assert isinstance(results[0].pop("created_datetime"), datetime.datetime) assert results[0] == expected_result def test_get_target_group_by_item_id(): """Test getting review queue items by id.""" create_test_review_queue_item( review_queue_id=123, product_id=123456, status="new", queue_name="initial" ) results = review_queue_model.get_target_group_by_item_id(["123"]) expected_result = { "review_queue_id": 123, "moved_to_target_id": None, } assert results[0] == expected_result def test_move_review_queue_item_to_queue(app_context): """Test move_review_queue_item_to_queue moves review queue item to queue.""" app_context.g.request_context.identity_id = "test_user_id" create_test_review_queue_item( review_queue_id=123, product_id=123456, status="new", queue_name="initial" ) expected_result = { "error_correction_id": None, "locked_at_datetime": None, "locked_by_user_id": None, "locked_until_datetime": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "move_note": None, "product_id": 123456, "queue_name": "under_investigation", "moved_to_queue_by_user_id": "test_user_id", "review_queue_id": 123, "status": "new", "submission_type": None, "submission_count": 1, "submitted_by_user_id": None, "submission_datetime": None } # test return value result = review_queue_model.move_review_queue_item_to_queue( review_queue_id=123, queue_name="under_investigation", move_note=None, moved_to_target_id=None, moved_to_target_name=None, moved_to_target_email=None, moved_to_orchadmin_user_id=None, ) assert isinstance(result.pop("created_datetime"), datetime.datetime) assert isinstance(result.pop("moved_at_datetime"), datetime.datetime) assert result == expected_result # test db value review_queue_item = review_queue_model.get_items(review_queue_id=123)[0][ 0 ].to_dict() assert isinstance(review_queue_item.pop("created_datetime"), datetime.datetime) assert isinstance(review_queue_item.pop("moved_at_datetime"), datetime.datetime) assert review_queue_item == expected_result @freeze_time("1234-02-03 00:30:00") def test_move_queue_item_unlocks_item(app_context): """Test move_review_queue_item_to_queue unlocks the item.""" app_context.g.request_context.identity_id = "test_user_id" create_test_review_queue_item( review_queue_id=124, product_id=123457, status="new", queue_name="initial", locked_by_user_id="test_user_id", locked_at_datetime=datetime.datetime.now(), locked_until_datetime=datetime.datetime.now(), ) expected_result = { "error_correction_id": None, "locked_at_datetime": None, "locked_by_user_id": None, "locked_until_datetime": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_at_datetime": FakeDatetime( 1234, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "moved_to_orchadmin_user_id": None, "move_note": None, "product_id": 123457, "queue_name": "under_investigation", "moved_to_queue_by_user_id": "test_user_id", "review_queue_id": 124, "status": "new", "submission_type": None, "submission_count": 1, "submitted_by_user_id": None, "submission_datetime": None, } result = review_queue_model.move_review_queue_item_to_queue( review_queue_id=124, queue_name="under_investigation", move_note=None, moved_to_target_id=None, moved_to_target_name=None, moved_to_target_email=None, moved_to_orchadmin_user_id=None, ) assert isinstance(result.pop("created_datetime"), datetime.datetime) assert result == expected_result def test_approve(mocker): """Test approve.""" mocker.patch.object(review_queue_model, "change_status", autospec=True) review_queue_model.approve(234) review_queue_model.change_status.assert_called_once_with( 234, review_queue_model.STATUS.COMPLETE ) @pytest.mark.parametrize( ( "test_description", "inputs", "is_integration_test_preparation", "is_db_resync", "status", "expected_new", "expected_dirty", "expected_deleted", "expected_raise", "expected_raise_message", ), [ ( "success", [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "new"}, {"product_id": 99999, "status": "complete"}, ], False, False, review_queue_model.STATUS.COMPLETE, [], [ { "review_queue_id": 2, "product_id": 45678, "error_correction_id": None, "status": "complete", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "created_datetime": datetime.datetime(1, 2, 3, 0, 0), "locked_by_user_id": None, "locked_at_datetime": None, "locked_until_datetime": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "submitted_by_user_id": None, "submission_datetime": None, } ], [], does_not_raise(), "None", ), ( "Review queue record found but invalid is_integration_test_preparation:false status change (complete => complete).", # noqa: E501 [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "complete"}, {"product_id": 99999, "status": "complete"}, ], False, False, review_queue_model.STATUS.COMPLETE, [], [], [], pytest.raises(Exception), 'Status change "complete => complete" is not valid', ), ( "Review queue record found but invalid is_integration_test_preparation:false status change (complete => new).", # noqa: E501 [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "complete"}, {"product_id": 99999, "status": "complete"}, ], False, False, review_queue_model.STATUS.NEW, [], [], [], pytest.raises(Exception), 'Status change "complete => new" is not valid', ), ( "Successful is_integration_test_preparation:true status change (complete => new).", # noqa: E501 [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "complete"}, {"product_id": 99999, "status": "complete"}, ], True, False, review_queue_model.STATUS.NEW, [], [ { "review_queue_id": 2, "product_id": 45678, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "created_datetime": datetime.datetime(1, 2, 3, 0, 0), "locked_by_user_id": None, "locked_at_datetime": None, "locked_until_datetime": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "submitted_by_user_id": None, "submission_datetime": None, } ], [], does_not_raise(), "None", ), ( "Successful is_db_resync:true status change (complete => new).", [ {"product_id": 123456, "status": "new"}, {"product_id": 45678, "status": "complete"}, {"product_id": 99999, "status": "complete"}, ], False, True, review_queue_model.STATUS.NEW, [], [ { "review_queue_id": 2, "product_id": 45678, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "created_datetime": datetime.datetime(1, 2, 3, 0, 0), "locked_by_user_id": None, "locked_at_datetime": None, "locked_until_datetime": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "submitted_by_user_id": None, "submission_datetime": None, } ], [], does_not_raise(), "None", ), ], ) @freeze_time("0001-02-03") def test_change_status( app_context, test_description, inputs, is_integration_test_preparation, is_db_resync, status, expected_new, expected_dirty, expected_deleted, expected_raise, expected_raise_message, ): """Test change_status.""" items = {} for item in inputs: tmp = create_test_review_queue_item( product_id=item["product_id"], status=item["status"] ) items[tmp.review_queue_id] = tmp app_context.g.request_context.is_integration_test_preparation = ( is_integration_test_preparation ) app_context.g.request_context.is_db_resync = is_db_resync with expected_raise as er: review_queue_model.change_status(items[2], status) assert str(getattr(er, "value", None)) == expected_raise_message assert [item.to_dict() for item in db.session.new] == expected_new assert [item.to_dict() for item in db.session.dirty] == expected_dirty assert [item.to_dict() for item in db.session.deleted] == expected_deleted @pytest.mark.parametrize( ( "test_description", "review_queue_items", "expected_new", "expected_dirty", "expected_deleted", "expected_raise", "expected_raise_message", "expected_result", ), [ ( "success, NEW, null locked_by_user_id", [{"product_id": 123456, "status": "new"}], [], [ { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1, 2, 3, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": FakeDatetime( 1, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, } ], [], does_not_raise(), "None", { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1, 2, 3, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": FakeDatetime( 1, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, }, ), ( "success, NEW, matching locked_by_user_id", [ { "product_id": 123456, "status": "new", "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", } ], [], [ { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1, 2, 3, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": FakeDatetime( 1, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, } ], [], does_not_raise(), "None", { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1, 2, 3, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": FakeDatetime( 1, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, }, ), ( "success, NEW, locked_until_datetime == current datetime", [ { "product_id": 123456, "status": "new", "created_datetime": datetime.datetime(1, 2, 2, 0, 0), "locked_until_datetime": datetime.datetime(1, 2, 3, 0, 0), } ], [], [ { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1, 2, 2, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": FakeDatetime( 1, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, } ], [], does_not_raise(), "None", { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1, 2, 2, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": FakeDatetime( 1, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, }, ), ( "success, NEW, locked_until_datetime < current datetime", [ { "product_id": 123456, "status": "new", "created_datetime": datetime.datetime(1, 2, 2, 0, 0), "locked_until_datetime": ( datetime.datetime(1, 2, 3, 0, 0) - datetime.timedelta(microseconds=1) ), } ], [], [ { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1, 2, 2, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": FakeDatetime( 1, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, } ], [], does_not_raise(), "None", { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1, 2, 2, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": FakeDatetime( 1, 2, 3, 0, 30, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, }, ), ( "error, NEW, non expired lock held by someone else", [ { "product_id": 123456, "status": "new", "locked_by_user_id": "7df21529-2026-4ee9-b823-99b69ade48fc", "locked_at_datetime": FakeDatetime( 1, 2, 3, 0, 0, tzinfo=datetime.timezone.utc ), "locked_until_datetime": ( datetime.datetime(1, 2, 3, 0, 0) + datetime.timedelta(microseconds=1) ), } ], [], [], [], pytest.raises(Exception), '{"status": 404, "code": "unable_to_lock", "message": "Lockable review queue item not found."}', # noqa None, ), ( "error, COMPLETE", [{"product_id": 123456, "status": "complete"}], [], [], [], pytest.raises(Exception), '{"status": 404, "code": "unable_to_lock", "message": "Lockable review queue item not found."}', # noqa None, ), ], ) @freeze_time("0001-02-03") def test_lock( app_context, test_description, review_queue_items, expected_new, expected_dirty, expected_deleted, expected_raise, expected_raise_message, expected_result, ): """Test locking a review queue item.""" app_context.g.request_context.identity_id = "5df21529-2026-4ee9-b823-99b69ade48fd" for review_queue_item in review_queue_items: create_test_review_queue_item(**review_queue_item) result = None with expected_raise as er: result = review_queue_model.lock(1) assert str(getattr(er, "value", None)) == expected_raise_message assert result == expected_result assert [item.to_dict() for item in db.session.new] == expected_new assert [item.to_dict() for item in db.session.dirty] == expected_dirty assert [item.to_dict() for item in db.session.deleted] == expected_deleted @pytest.mark.parametrize( ( "test_description", "review_queue_items", "expected_new", "expected_dirty", "expected_deleted", "expected_raise", "expected_raise_message", "expected_result", ), [ ( "success", [ { "product_id": 123456, "status": "new", "created_datetime": datetime.datetime(1234, 2, 3, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": datetime.datetime(1234, 2, 3, 0, 10), "locked_until_datetime": FakeDatetime( 1234, 2, 3, 0, 40, tzinfo=datetime.timezone.utc ), } ], [], [ { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1234, 2, 3, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": datetime.datetime(1234, 2, 3, 0, 10), "locked_until_datetime": FakeDatetime( 1234, 2, 3, 0, 20, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, } ], [], does_not_raise(), "None", { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": 1, "queue_name": "initial", "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "created_datetime": datetime.datetime(1234, 2, 3, 0, 0), "locked_by_user_id": "5df21529-2026-4ee9-b823-99b69ade48fd", "locked_at_datetime": datetime.datetime(1234, 2, 3, 0, 10), "locked_until_datetime": FakeDatetime( 1234, 2, 3, 0, 20, tzinfo=datetime.timezone.utc ), "submitted_by_user_id": None, "submission_datetime": None, }, ), ( "error, locked by someone else", [ { "product_id": 123456, "status": "new", "locked_by_user_id": "7df21529-2026-4ee9-b823-99b69ade48fc", } ], [], [], [], pytest.raises(Exception), '{"status": 404, "code": "unable_to_unlock", "message": "Unlockable review queue item not found."}', # noqa None, ), ], ) @freeze_time("1234-02-03 00:20:00") def test_unlock( app_context, test_description, review_queue_items, expected_new, expected_dirty, expected_deleted, expected_raise, expected_raise_message, expected_result, ): """Test unlocking a review queue item.""" app_context.g.request_context.identity_id = "5df21529-2026-4ee9-b823-99b69ade48fd" for review_queue_item in review_queue_items: create_test_review_queue_item(**review_queue_item) result = None with expected_raise as er: result = review_queue_model.unlock(1) assert str(getattr(er, "value", None)) == expected_raise_message assert result == expected_result assert [item.to_dict() for item in db.session.new] == expected_new assert [item.to_dict() for item in db.session.dirty] == expected_dirty assert [item.to_dict() for item in db.session.deleted] == expected_deleted @pytest.mark.parametrize( ( "test_description", "current_item", "expected_truncated_result" ), [ ("include current item", True, [ { "review_queue_id": 3, "product_id": 123456, "status": "new", "created_datetime": datetime.datetime(2021, 11, 1, 20, 37, 47) }, { "review_queue_id": 2, "product_id": 123456, "status": "complete", "created_datetime": datetime.datetime(2021, 10, 21, 20, 37, 47) }, { "review_queue_id": 1, "product_id": 123456, "status": "complete", "created_datetime": datetime.datetime(2019, 9, 21, 20, 37, 47) } ]), ("exclude current item", False, [ { "review_queue_id": 2, "product_id": 123456, "status": "complete", "created_datetime": datetime.datetime(2021, 10, 21, 20, 37, 47), }, { "review_queue_id": 1, "product_id": 123456, "status": "complete", "created_datetime": datetime.datetime(2019, 9, 21, 20, 37, 47), } ]) ] ) def test_review_history(test_description, current_item, expected_truncated_result, test_review_history_models): # noqa: E501 """Test review_history.""" for review_queue_item in test_review_history_models: db.session.add(review_queue_item) db.session.commit() results, _ = review_queue_model.review_history( 123456, 20, 0, current_item=current_item) for idx, entry in enumerate(results): item = entry.to_dict() assert all(item.get(key) == val for (key, val) in expected_truncated_result[idx].items()) @freeze_time("2023-05-17 21:30:17.300290") def test_reset_review_queue_item(): """Test reset_review_queue_item.""" db.session.add( review_queue_model.ReviewQueue( review_queue_id=1, product_id=123456, status="complete", queue_name="under_investigation", moved_to_queue_by_user_id="86f68cf2-5ee5-40f4-902b-ae1ef0398d72", created_datetime=datetime.datetime( 2023, 5, 16, 20, 29, 16, 200290, tzinfo=datetime.timezone.utc ), locked_by_user_id="86f68cf2-5ee5-40f4-902b-ae1ef0398d72", locked_at_datetime=datetime.datetime( 2023, 5, 16, 20, 29, 16, 200290, tzinfo=datetime.timezone.utc ), locked_until_datetime=datetime.datetime( 2023, 5, 16, 20, 29, 16, 200290, tzinfo=datetime.timezone.utc ), ) ) db.session.commit() ((pre_reset_review_queue_item,), _) = review_queue_model.get_items( review_queue_id=1 ) assert pre_reset_review_queue_item.to_dict() == { "created_datetime": datetime.datetime(2023, 5, 16, 20, 29, 16, 200290), "error_correction_id": None, "locked_at_datetime": datetime.datetime(2023, 5, 16, 20, 29, 16, 200290), "locked_by_user_id": "86f68cf2-5ee5-40f4-902b-ae1ef0398d72", "locked_until_datetime": datetime.datetime(2023, 5, 16, 20, 29, 16, 200290), "moved_to_queue_by_user_id": "86f68cf2-5ee5-40f4-902b-ae1ef0398d72", "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "product_id": 123456, "queue_name": "under_investigation", "review_queue_id": 1, "status": "complete", "submission_type": None, "submission_count": 1, "submitted_by_user_id": None, "submission_datetime": None, } review_queue_model.reset_review_queue_item(1) db.session.commit() ((post_reset_review_queue_item,), _) = review_queue_model.get_items( review_queue_id=1 ) assert post_reset_review_queue_item.to_dict() == { "created_datetime": datetime.datetime(2023, 5, 17, 21, 30, 17, 300290), "error_correction_id": None, "locked_at_datetime": None, "locked_by_user_id": None, "locked_until_datetime": None, "moved_to_queue_by_user_id": None, "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "moved_at_datetime": None, "move_note": None, "product_id": 123456, "queue_name": "initial", "review_queue_id": 1, "status": "new", "submission_type": None, "submission_count": 1, "submitted_by_user_id": None, "submission_datetime": None, } def test_set_to_new(mocker): """Test set_to_new.""" mocker.patch.object(review_queue_model, "change_status", autospec=True) mock_item = review_queue_model.ReviewQueue( review_queue_id=234, product_id=123456, status="complete", queue_name="under_investigation", moved_to_queue_by_user_id="86f68cf2-5ee5-40f4-902b-ae1ef0398d72", created_datetime=datetime.datetime( 2023, 5, 16, 20, 29, 16, 200290, tzinfo=datetime.timezone.utc ), locked_by_user_id="86f68cf2-5ee5-40f4-902b-ae1ef0398d72", locked_at_datetime=datetime.datetime( 2023, 5, 16, 20, 29, 16, 200290, tzinfo=datetime.timezone.utc ), locked_until_datetime=datetime.datetime( 2023, 5, 16, 20, 29, 16, 200290, tzinfo=datetime.timezone.utc ), ) db.session.add(mock_item) db.session.commit() review_queue_model.set_to_new(234) review_queue_model.change_status.assert_called_once_with( mock_item, review_queue_model.STATUS.NEW ) @pytest.mark.parametrize( ( "test_description", "item", "rejection", "approval", "expected_not_found", ), [ ( "get new item success", {"review_queue_id": 1, "product_id": 1, "status": "new"}, None, None, False, ), ( "get rejected item success", {"review_queue_id": 1, "product_id": 1, "status": "complete"}, {"review_queue_id": 1, "note": "i don't like it"}, None, False, ), ( "get approved item success", {"review_queue_id": 1, "product_id": 1, "status": "complete"}, None, {"review_queue_id": 1}, False, ), ( "get unsubmitted item", {"review_queue_id": 1, "product_id": 1, "status": "complete"}, None, None, True, ), ("get nonexistent item", None, None, None, True), ], ) def test_latest_review_by_product( test_description, item, rejection, approval, expected_not_found, ): """Test that get_items filters items by status.""" review_item = None approval_item = None rejection_item = None if item: review_item = create_test_review_queue_item(**item) if approval: approval_item = create_test_approval(**approval, user_id="def-456") if rejection: rejection_item = create_test_rejection(**rejection, user_id="def-456") result = review_queue_model.ReviewQueue.latest_review_by_product(1) if expected_not_found: assert result is None else: assert result == review_item assert result.approval == approval_item assert result.rejection == rejection_item