"""Fixtures to use in tests.""" import datetime import json from unittest.mock import MagicMock import pytest from confluent_kafka import KafkaError from product_review.api import app, db from product_review.connectors import kafka as kafka_connector from product_review.models import approval as approval_model from product_review.models import canned_response as canned_response_model from product_review.models import canned_response_category from product_review.models import canned_response_note as canned_response_note_model from product_review.models import move_target as move_target_model from product_review.models import queue_move_history as queue_move_history_model from product_review.models import rejection as rejection_model from product_review.models import review_queue as review_queue_model from product_review.testing_accommodations import test_handlers # noqa: F401 class Release: """Release.""" def __init__( self, release_id, release_status, distribution_format_id, release_format="EP", product_type_id=1, ): """__init__.""" self.release_id = release_id self.release_status = release_status self.distribution_format_id = distribution_format_id self.format = release_format self.product_type_id = product_type_id def to_dict(self): """Produce dict version of Release.""" return { "release_id": self.release_id, "release_status": self.release_status, "distribution_format_id": self.distribution_format_id, "format": self.format, "product_type_id": self.product_type_id, } @pytest.fixture(autouse=True) def test_app(): """Application with sqlite database fixture.""" app.app_context().push() db.create_all() yield app db.drop_all() @pytest.fixture def app_context(test_app): """Return flask app context, including a mock logger.""" with test_app.app_context() as app_context: app_context.g.log = MagicMock() app_context.g.ows = MagicMock() app_context.g.request_context = MagicMock() yield app_context @pytest.fixture def fixture_client(test_app): """Create an api test client fixture.""" return test_app.test_client() @pytest.fixture def test_review_queue_models(): """Return new review queue models for testing.""" return [ review_queue_model.ReviewQueue( review_queue_id=1, created_datetime=datetime.date(2019, 4, 13), product_id=123456, status="new", ), review_queue_model.ReviewQueue( review_queue_id=2, created_datetime=datetime.date(2019, 4, 13), product_id=456789, status="new", ), ] @pytest.fixture def test_product_review_history_dict(): """Return a dict of review queue items as a product review history.""" return [ { "review_queue_id": 1, "product_id": 123456, "error_correction_id": None, "status": "complete", "submission_type": None, "submission_count": None, "queue_name": "initial", "moved_to_queue_by_user_id": None, "submitted_by_user_id": None, "created_datetime": datetime.datetime(2019, 9, 21, 20, 37, 47), "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, "submission_datetime": None, "approval": { "approval_id": 1, "review_queue_id": 1, "user_id": "123", "created_datetime": datetime.datetime(2019, 9, 23, 20, 37, 47), "note": None }, "rejection": None, "queue_move_history": None }, { "review_queue_id": 2, "product_id": 123456, "error_correction_id": None, "status": "complete", "submission_type": None, "submission_count": None, "queue_name": "initial", "moved_to_queue_by_user_id": None, "submitted_by_user_id": None, "created_datetime": datetime.datetime(2021, 10, 21, 20, 37, 47), "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": "test move back", "moved_at_datetime": datetime.datetime(2021, 10, 2, 9, 0), "submission_datetime": None, "approval": None, "rejection": { "rejection_id": 1, "review_queue_id": 2, "user_id": "123", "created_datetime": datetime.datetime(2021, 10, 26, 20, 37, 47), "note": "some rejection", "will_not_deliver": None }, "queue_move_history": [ { "queue_move_history_id": 2, "review_queue_id": 2, "queue_name": "initial", "previous_queue_name": "escalation", "moved_to_queue_by_user_id": "654", "moved_at_datetime": datetime.datetime(2021, 10, 2, 9, 0), "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "move_note": "test move back", "escalation_type": None }, { "queue_move_history_id": 1, "review_queue_id": 2, "queue_name": "escalation", "previous_queue_name": "initial", "moved_to_queue_by_user_id": "123", "moved_at_datetime": datetime.datetime(2021, 10, 1, 20, 0), "moved_to_target_id": 11, "moved_to_target_name": "person 2", "moved_to_target_email": "somebody@email", "moved_to_orchadmin_user_id": 222, "move_note": "test escalate", "escalation_type": "Escalation Type 1" } ] }, { "review_queue_id": 3, "product_id": 123456, "error_correction_id": None, "status": "new", "submission_type": None, "submission_count": None, "queue_name": "initial", "moved_to_queue_by_user_id": None, "submitted_by_user_id": None, "created_datetime": datetime.datetime(2021, 11, 1, 20, 37, 47), "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": "test move", "moved_at_datetime": datetime.datetime(2021, 11, 2, 9, 0), "submission_datetime": None, "approval": None, "rejection": None, "queue_move_history": [ { "queue_move_history_id": 3, "review_queue_id": 3, "queue_name": "under_investigation", "previous_queue_name": "initial", "moved_to_queue_by_user_id": "654", "moved_at_datetime": datetime.datetime(2021, 11, 2, 9, 0), "moved_to_target_id": None, "moved_to_target_name": None, "moved_to_target_email": None, "moved_to_orchadmin_user_id": None, "move_note": "test move", "escalation_type": None } ] } ] @pytest.fixture def test_review_history_models(): """Return review queue models for testing review history.""" fmt = "%Y-%m-%d %H:%M:%S" return [ review_queue_model.ReviewQueue( review_queue_id=1, created_datetime=datetime.datetime.strptime("2019-09-21 20:37:47", fmt), product_id=123456, status="complete", approval=approval_model.Approval( approval_id=1, review_queue_id=1, user_id="123", created_datetime=datetime.datetime.strptime("2019-09-23 20:37:47", fmt), ), queue_name="initial", ), review_queue_model.ReviewQueue( review_queue_id=2, created_datetime=datetime.datetime.strptime("2021-10-21 20:37:47", fmt), product_id=123456, status="complete", rejection=rejection_model.Rejection( rejection_id=1, review_queue_id=2, user_id="123", created_datetime=datetime.datetime.strptime("2021-10-26 20:37:47", fmt), note="some rejection", ), queue_name="initial", move_note="test move back", moved_at_datetime=datetime.datetime.strptime("2021-10-2 09:00:00", fmt), queue_move_history=[queue_move_history_model.QueueMoveHistory( queue_move_history_id=2, review_queue_id=2, queue_name="initial", previous_queue_name="escalation", moved_to_queue_by_user_id="654", moved_at_datetime=datetime.datetime.strptime("2021-10-2 09:00:00", fmt), moved_to_target_id=None, moved_to_target_name=None, moved_to_target_email=None, moved_to_orchadmin_user_id=None, move_note="test move back" ), queue_move_history_model.QueueMoveHistory( queue_move_history_id=1, review_queue_id=2, queue_name="escalation", previous_queue_name="initial", moved_to_queue_by_user_id="123", moved_at_datetime=datetime.datetime.strptime("2021-10-1 20:00:00", fmt), moved_to_target_id=11, moved_to_target_name="person 2", moved_to_target_email="somebody@email", moved_to_orchadmin_user_id=222, move_note="test escalate", move_target=move_target_model.MoveTarget( moved_to_target_id=11, display_name="Escalation Type 1", target_group_id=1, escalation_type="Escalation Type 1", ) )] ), review_queue_model.ReviewQueue( review_queue_id=3, created_datetime=datetime.datetime.strptime("2021-11-01 20:37:47", fmt), product_id=123456, status="new", queue_name="initial", move_note="test move", moved_at_datetime=datetime.datetime.strptime("2021-11-02 09:00:00", fmt), queue_move_history=[queue_move_history_model.QueueMoveHistory( queue_move_history_id=3, review_queue_id=3, queue_name="under_investigation", previous_queue_name="initial", moved_to_queue_by_user_id="654", moved_at_datetime=datetime.datetime.strptime( "2021-11-02 09:00:00", fmt), moved_to_target_id=None, moved_to_target_name=None, moved_to_target_email=None, moved_to_orchadmin_user_id=None, move_note="test move" )] ), ] @pytest.fixture def test_canned_response_note_models(): """Return new canned response note models for testing.""" return [ canned_response_note_model.CannedResponseNote( note_id=1, canned_response_id=1, note_keyword="abc", note_text="foo bar", language_code="en", modified_by_user_id="user-1", ), canned_response_note_model.CannedResponseNote( note_id=2, canned_response_id=2, note_keyword="def", note_text="foo bar", language_code="en", modified_by_user_id="user-2", ), canned_response_note_model.CannedResponseNote( note_id=3, canned_response_id=2, note_keyword="def", note_text="foo bar", language_code="fr", modified_by_user_id="user-2", ), ] @pytest.fixture def test_canned_response_models(): """Return new canned response models for testing.""" return [ canned_response_model.CannedResponse( canned_response_id=1, category_id=1, keyword="abc", modified_by_user_id="user-1", review_context="rejection", canned_response_category=canned_response_category.CannedResponseCategory( category_id=1, name="cat 1" ), canned_response_notes=[ canned_response_note_model.CannedResponseNote( note_id=1, canned_response_id=1, note_keyword="abc", note_text="foo bar", language_code="en", modified_by_user_id="user-1", ) ], ), canned_response_model.CannedResponse( canned_response_id=2, category_id=1, keyword="def", modified_by_user_id="user-2", review_context="rejection", canned_response_category=canned_response_category.CannedResponseCategory( category_id=1, name="cat 1" ), canned_response_notes=[ canned_response_note_model.CannedResponseNote( note_id=2, canned_response_id=2, note_keyword="def", note_text="foo bar", language_code="en", modified_by_user_id="user-2", ) ], ), ] def format_kafka_message(*, topic, key, value, offset=None): """Format kafka message.""" return { "topic": topic, "key": key.encode(), "value": json.dumps(value).encode(), "partition": -1, "offset": offset, "timestamp": 0, "headers": None, "error": None, } @pytest.fixture def get_event_ows_content_review_product(): """Get mock event.owsContentReview.product message.""" def _get_event_ows_content_review_product(*, has_error=False): error = KafkaError(error=KafkaError.TOPIC_EXCEPTION) if has_error else None return error, format_kafka_message( topic="event.owsContentReview.product", key="product_456", value={ "operation": { "type": "update", "timestamp": 1684359017300.29, "context": "queue_move", }, "payload": { "product_id": 456, "review_queue_id": 123, "user_id": "some-identity-id", "queue_name": "under_investigation", }, }, offset=error, ) return _get_event_ows_content_review_product @pytest.fixture def get_event_ows_content_review_review(): """Get mock event.owsContentReview.review message.""" def _get_event_ows_content_review_review(has_error=False): error = KafkaError(error=KafkaError.TOPIC_EXCEPTION) if has_error else None return error, format_kafka_message( topic="event.owsContentReview.review", key="product_456", value={ "operation": { "type": "create", "timestamp": 1684359017300.29, "context": "approve", }, "payload": { "product_id": 456, "review_queue_id": 123, "user_id": "some-identity-id", "review_note": "note", }, }, offset=error, ) return _get_event_ows_content_review_review @pytest.fixture(autouse=True) def clear_cached_kafka_producer(): """Clear the cached kafka producer before each test.""" kafka_connector.get_producer.cache_clear()