# pylint: disable=unused-argument from datetime import date, datetime, timedelta, timezone import pytest from db_schema.common import ActivityStatusEnum, CompletenessStatusEnum from db_schema.schemas.admin import AuditLog, AuditLogTypes from db_schema.schemas.slz import ContentStatus, ContentStatusEnum, UnitOfWork from sqlalchemy import null from delphi_slz_admin.modelviews.content_status import ( ContentStatusModelView, FilterContentStatus, FilterContentStatusId, FilterContentStatusIdReversed, FilterReportDateEarlierThan, FilterReportDateLaterThan, FilterSameDayOfWeek, FilterSameReportLicensorContext, FilterUnitOfWorkId, FilterUnitOfWorkIdReversed, FilterUoWCodeLike, ) from delphi_slz_admin.services.config import ReprocessingConfig FREEZED_DT = datetime(2015, 10, 15, 12, 23, 45, tzinfo=timezone.utc) @pytest.mark.freeze_time(FREEZED_DT) def test_action_complete_cs__ok(mocker, db, clean_db, app, content_status, unit_of_work): mocker.patch('boto3.client') db.begin() modelview = ContentStatusModelView(db) modelview.action_complete_cs([content_status.content_status_id]) db.rollback() db.refresh(content_status) assert content_status.content_status == ContentStatusEnum.COMPLETE assert content_status.completed_at == FREEZED_DT assert content_status.meta_data is None @pytest.mark.freeze_time(FREEZED_DT) def test_action_complete_cs_no_metadata__ok( mocker, db, clean_db, app, content_status, unit_of_work ): mocker.patch('boto3.client') content_status.meta_data = null() db.begin() modelview = ContentStatusModelView(db) assert content_status.meta_data is None modelview.action_complete_cs([content_status.content_status_id]) db.rollback() db.refresh(content_status) assert content_status.content_status == ContentStatusEnum.COMPLETE assert content_status.completed_at == FREEZED_DT assert content_status.meta_data is None @pytest.mark.freeze_time(FREEZED_DT) @pytest.mark.parametrize('status', [x for x in ContentStatusEnum if x != ContentStatusEnum.ON_HOLD]) def test_action_complete_cs__invalid_status( db, clean_db, app, content_status, unit_of_work, mocker, status ): mocker.patch('boto3.client') patched_flash = mocker.patch( 'delphi_slz_admin.modelviews.content_status.flash', spec=lambda *args, **kwargs: None ) content_status.content_status = status db.begin() modelview = ContentStatusModelView(db) modelview.action_complete_cs([content_status.content_status_id]) db.rollback() db.refresh(content_status) assert content_status.content_status == status assert content_status.completed_at != FREEZED_DT patched_flash.assert_called_once_with('Some items are not in ON_HOLD status', category='error') @pytest.mark.freeze_time(FREEZED_DT) def test_action_complete_cs__audit_log_created( mocker, db, clean_db, app, content_status, unit_of_work ): mocker.patch('boto3.client') db.begin() modelview = ContentStatusModelView(db) modelview.action_complete_cs([content_status.content_status_id]) db.rollback() audit_log: AuditLog = db.query(AuditLog).first() assert audit_log.event_type == AuditLogTypes.CONTENT_STATUS_SET_COMPLETED assert audit_log.created_at == FREEZED_DT assert audit_log.extra_data == {'content_status_ids': [content_status.content_status_id]} @pytest.mark.freeze_time(FREEZED_DT) def test_action_complete_uow__ok(db, clean_db, app, content_status, unit_of_work): db.add(unit_of_work) db.begin() modelview = ContentStatusModelView(db) modelview.action_complete_uow([content_status.content_status_id]) db.rollback() db.refresh(unit_of_work) assert unit_of_work.completeness_status == CompletenessStatusEnum.COMPLETE assert unit_of_work.last_updated_at == FREEZED_DT @pytest.mark.freeze_time(FREEZED_DT) def test_action_complete_uow__audit_log_created(db, clean_db, app, content_status, unit_of_work): db.begin() modelview = ContentStatusModelView(db) modelview.action_complete_uow([content_status.content_status_id]) db.rollback() audit_log: AuditLog = db.query(AuditLog).first() assert audit_log.event_type == AuditLogTypes.UOW_SET_COMPLETED assert audit_log.created_at == FREEZED_DT assert audit_log.extra_data == {'unit_of_work_ids': [unit_of_work.unit_of_work_id]} @pytest.mark.freeze_time(FREEZED_DT) def test_action_min_complete_uow__ok(db, clean_db, app, content_status, unit_of_work): db.add(unit_of_work) db.begin() modelview = ContentStatusModelView(db) modelview.action_min_complete_uow([content_status.content_status_id]) db.rollback() db.refresh(unit_of_work) assert unit_of_work.completeness_status == CompletenessStatusEnum.MIN_COMPLETE assert unit_of_work.last_updated_at == FREEZED_DT @pytest.mark.freeze_time(FREEZED_DT) def test_action_min_complete_uow__audit_log_created( db, clean_db, app, content_status, unit_of_work ): db.begin() modelview = ContentStatusModelView(db) modelview.action_min_complete_uow([content_status.content_status_id]) db.rollback() audit_log: AuditLog = db.query(AuditLog).first() assert audit_log.event_type == AuditLogTypes.UOW_SET_MIN_COMPLETED assert audit_log.created_at == FREEZED_DT assert audit_log.extra_data == {'unit_of_work_ids': [unit_of_work.unit_of_work_id]} @pytest.mark.freeze_time(FREEZED_DT) def test_action_schedule_from_idle_uow__ok(db, clean_db, app, content_status, unit_of_work): unit_of_work.activity_status = ActivityStatusEnum.IDLE db.add(unit_of_work) db.begin() modelview = ContentStatusModelView(db) modelview.action_schedule_from_idle([content_status.content_status_id]) db.rollback() db.refresh(unit_of_work) assert unit_of_work.activity_status == ActivityStatusEnum.NOT_IN_PROGRESS assert unit_of_work.completeness_status == CompletenessStatusEnum.ACTIVE assert unit_of_work.created_at == FREEZED_DT assert unit_of_work.last_updated_at == FREEZED_DT @pytest.mark.freeze_time(FREEZED_DT) def test_action_schedule_from_idle_uow__error(db, clean_db, app, content_status, unit_of_work): unit_of_work.activity_status = ActivityStatusEnum.IN_PROGRESS db.add(unit_of_work) db.begin() modelview = ContentStatusModelView(db) modelview.action_schedule_from_idle([content_status.content_status_id]) db.rollback() db.refresh(unit_of_work) assert unit_of_work.activity_status == ActivityStatusEnum.IN_PROGRESS assert unit_of_work.completeness_status == CompletenessStatusEnum.ACTIVE assert unit_of_work.created_at != FREEZED_DT assert unit_of_work.last_updated_at != FREEZED_DT def test_filter_uow_id__apply_included(db, clean_db, content_status): filter_ = FilterUnitOfWorkId() initial_query = db.query(ContentStatus) cleaned_value = filter_.clean( f'{content_status.unit_of_work_id}, {content_status.unit_of_work_id + 1}' ) filtered_query = filter_.apply(initial_query, value=cleaned_value) assert initial_query.count() == 1 assert filtered_query.count() == 1 def test_filter_uow_id__apply_excluded(db, clean_db, content_status): filter_ = FilterUnitOfWorkId() initial_query = db.query(ContentStatus) cleaned_value = filter_.clean( f'{content_status.unit_of_work_id + 1}, {content_status.unit_of_work_id + 2}' ) filtered_query = filter_.apply(initial_query, value=cleaned_value) assert initial_query.count() == 1 assert filtered_query.count() == 0 def test_filter_uow_id_reversed__apply_included(db, clean_db, content_status): filter_ = FilterUnitOfWorkIdReversed() initial_query = db.query(ContentStatus) cleaned_value = filter_.clean( f'{content_status.unit_of_work_id}, {content_status.unit_of_work_id + 1}' ) filtered_query = filter_.apply(initial_query, value=cleaned_value) assert initial_query.count() == 1 assert filtered_query.count() == 0 def test_filter_uow_id_reversed__apply_excluded(db, clean_db, content_status): filter_ = FilterUnitOfWorkIdReversed() initial_query = db.query(ContentStatus) cleaned_value = filter_.clean( f'{content_status.unit_of_work_id + 1}, {content_status.unit_of_work_id + 2}' ) filtered_query = filter_.apply(initial_query, value=cleaned_value) assert initial_query.count() == 1 assert filtered_query.count() == 1 def test_filter_cs_id__apply_included(db, clean_db, content_status): filter_ = FilterContentStatusId() initial_query = db.query(ContentStatus) cleaned_value = filter_.clean( f'{content_status.content_status_id}, {content_status.content_status_id + 1}' ) filtered_query = filter_.apply(initial_query, value=cleaned_value) assert initial_query.count() == 1 assert filtered_query.count() == 1 def test_filter_cs_id__apply_excluded(db, clean_db, content_status): filter_ = FilterContentStatusId() initial_query = db.query(ContentStatus) cleaned_value = filter_.clean( f'{content_status.content_status_id + 1}, {content_status.content_status_id + 2}' ) filtered_query = filter_.apply(initial_query, value=cleaned_value) assert initial_query.count() == 1 assert filtered_query.count() == 0 def test_filter_cs_id_reversed__apply_included(db, clean_db, content_status): filter_ = FilterContentStatusIdReversed() initial_query = db.query(ContentStatus) cleaned_value = filter_.clean( f'{content_status.content_status_id}, {content_status.content_status_id + 1}' ) filtered_query = filter_.apply(initial_query, value=cleaned_value) assert initial_query.count() == 1 assert filtered_query.count() == 0 def test_filter_cs_id_reversed__apply_excluded(db, clean_db, content_status): filter_ = FilterContentStatusIdReversed() initial_query = db.query(ContentStatus) cleaned_value = filter_.clean( f'{content_status.content_status_id + 1}, {content_status.content_status_id + 2}' ) filtered_query = filter_.apply(initial_query, value=cleaned_value) assert initial_query.count() == 1 assert filtered_query.count() == 1 def test_filter_content_status_on_hold__apply_included(db, clean_db, content_status): content_status.content_status = ContentStatusEnum.ON_HOLD filter_ = FilterContentStatus() initial_query = db.query(ContentStatus) filtered_query = filter_.apply(initial_query, value=ContentStatusEnum.ON_HOLD.value) assert initial_query.count() == 1 assert filtered_query.count() == 1 def test_filter_content_status_on_hold__apply_excluded(db, clean_db, content_status): content_status.content_status = ContentStatusEnum.COMPLETE filter_ = FilterContentStatus() initial_query = db.query(ContentStatus) filtered_query = filter_.apply(initial_query, value=ContentStatusEnum.ON_HOLD.value) assert initial_query.count() == 1 assert filtered_query.count() == 0 def test_filter_uow_code_like__apply_included(db, clean_db, content_status): filter_ = FilterUoWCodeLike() initial_query = db.query(ContentStatus).join(UnitOfWork) filtered_query = filter_.apply(initial_query, value='dim_track') assert initial_query.count() == 1 assert filtered_query.count() == 1 def test_filter_uow_code_like__apply_excluded(db, clean_db, content_status): filter_ = FilterUoWCodeLike() initial_query = db.query(ContentStatus).join(UnitOfWork) filtered_query = filter_.apply(initial_query, value='foo bar') assert initial_query.count() == 1 assert filtered_query.count() == 0 def test_filter_report_date_later_than__apply_included(db, clean_db, content_status): report_date = date(2015, 10, 15) filter_value = report_date - timedelta(days=1) content_status.unit_of_work.report_date = report_date filter_ = FilterReportDateLaterThan() initial_query = db.query(ContentStatus).join(UnitOfWork) filtered_query = filter_.apply(initial_query, value=filter_value) assert initial_query.count() == 1 assert filtered_query.count() == 1 def test_filter_report_date_later_than__apply_excluded(db, clean_db, content_status): report_date = date(2015, 10, 15) filter_value = report_date + timedelta(days=1) content_status.unit_of_work.report_date = report_date filter_ = FilterReportDateLaterThan() initial_query = db.query(ContentStatus).join(UnitOfWork) filtered_query = filter_.apply(initial_query, value=filter_value) assert initial_query.count() == 1 assert filtered_query.count() == 0 def test_filter_report_date_earlier_than__apply_included(db, clean_db, content_status): report_date = date(2015, 10, 15) filter_value = report_date + timedelta(days=1) content_status.unit_of_work.report_date = report_date filter_ = FilterReportDateEarlierThan() initial_query = db.query(ContentStatus).join(UnitOfWork) filtered_query = filter_.apply(initial_query, value=filter_value) assert initial_query.count() == 1 assert filtered_query.count() == 1 def test_filter_report_date_earlier_than__apply_excluded(db, clean_db, content_status): report_date = date(2015, 10, 15) filter_value = report_date - timedelta(days=1) content_status.unit_of_work.report_date = report_date filter_ = FilterReportDateEarlierThan() initial_query = db.query(ContentStatus).join(UnitOfWork) filtered_query = filter_.apply(initial_query, value=filter_value) assert initial_query.count() == 1 assert filtered_query.count() == 0 def test_filter_same_report_licensor_context__apply_included( db, clean_db, content_status, unit_of_work ): similar_unit_of_work = UnitOfWork( **{ 'unit_of_work_code': 'foo_code', 'reprocess_id': '', 'report_date': '2020-04-27', 'report_id': unit_of_work.report_id, 'licensor_id': unit_of_work.licensor_id, 'version': 'v1', 'activity_status': 'NOT_IN_PROGRESS', 'completeness_status': 'ACTIVE', 'next_run_at': '2020-04-28 01:06:09.715337', 'created_at': '2020-04-28 01:00:58.806272', 'last_updated_at': '2020-05-27 08:20:58.481813', 'is_force_complete': True, 'priority': 5 } ) db.add(similar_unit_of_work) db.flush() similar_content_status = ContentStatus( **{ 'unit_of_work_id': similar_unit_of_work.unit_of_work_id, 'context': content_status.context, 'content_status': 'ON_HOLD', 'content_name': 'gras_20200427_dim_track_participant.csv', 'content_size': 565118579, 'failure_count': 0, 'latest_job_id': '14260_2020-04-28T02.00.58', 'created_at': '2020-04-27 14:31:08.518350', 'completed_at': None, 'last_checked_at': '2020-04-27 15:39:33.622070', 'sub_content': 'null', 'record_count': 9188557, 'hash': '0265bed78e2fdccce1db1d4a28d926cf', 'metadata_process_status': 'COMPLETED', 'metadata_process_started_at': '2020-04-27 15:42:01.304492', 'metadata_process_completed_at': '2020-04-27 15:45:51.944453' } ) db.add(similar_content_status) db.flush() filter_ = FilterSameReportLicensorContext() initial_query = db.query(UnitOfWork).join(ContentStatus) filtered_query = filter_.apply(initial_query, value=content_status.content_status_id) assert initial_query.count() == 2 assert filtered_query.count() == 2 @pytest.mark.parametrize( 'report_id_shift, licensor_id_shift, context', [ (1, 0, 'dim_track_participant'), (0, 1, 'dim_track_participant'), (0, 0, 'foo_bar_baz'), ] ) def test_filter_same_report_licensor_context__apply_excluded( db, clean_db, content_status, unit_of_work, report_id_shift, licensor_id_shift, context ): similar_unit_of_work = UnitOfWork( **{ 'unit_of_work_code': 'foo_code', 'reprocess_id': '', 'report_date': '2020-04-27', 'report_id': unit_of_work.report_id + report_id_shift, 'licensor_id': unit_of_work.licensor_id + licensor_id_shift, 'version': 'v1', 'activity_status': 'NOT_IN_PROGRESS', 'completeness_status': 'ACTIVE', 'next_run_at': '2020-04-28 01:06:09.715337', 'created_at': '2020-04-28 01:00:58.806272', 'last_updated_at': '2020-05-27 08:20:58.481813', 'is_force_complete': True, 'priority': 5 } ) db.add(similar_unit_of_work) db.flush() similar_content_status = ContentStatus( **{ 'unit_of_work_id': similar_unit_of_work.unit_of_work_id, 'context': context, 'content_status': 'ON_HOLD', 'content_name': 'gras_20200427_dim_track_participant.csv', 'content_size': 565118579, 'failure_count': 0, 'latest_job_id': '14260_2020-04-28T02.00.58', 'created_at': '2020-04-27 14:31:08.518350', 'completed_at': None, 'last_checked_at': '2020-04-27 15:39:33.622070', 'sub_content': 'null', 'record_count': 9188557, 'hash': '0265bed78e2fdccce1db1d4a28d926cf', 'metadata_process_status': 'COMPLETED', 'metadata_process_started_at': '2020-04-27 15:42:01.304492', 'metadata_process_completed_at': '2020-04-27 15:45:51.944453' } ) db.add(similar_content_status) db.flush() filter_ = FilterSameReportLicensorContext() initial_query = db.query(UnitOfWork).join(ContentStatus) filtered_query = filter_.apply(initial_query, value=content_status.content_status_id) assert initial_query.count() == 2 assert filtered_query.count() == 1 def test_filter_same_day_of_week__apply_included(db, clean_db, content_status, unit_of_work): similar_unit_of_work = UnitOfWork( **{ 'unit_of_work_code': 'foo_code', 'reprocess_id': '', 'report_date': '2020-04-13', # shifted for two weeks back 'report_id': unit_of_work.report_id, 'licensor_id': unit_of_work.licensor_id, 'version': 'v1', 'activity_status': 'NOT_IN_PROGRESS', 'completeness_status': 'ACTIVE', 'next_run_at': '2020-04-28 01:06:09.715337', 'created_at': '2020-04-28 01:00:58.806272', 'last_updated_at': '2020-05-27 08:20:58.481813', 'is_force_complete': True, 'priority': 5 } ) db.add(similar_unit_of_work) db.flush() similar_content_status = ContentStatus( **{ 'unit_of_work_id': similar_unit_of_work.unit_of_work_id, 'context': content_status.context, 'content_status': 'ON_HOLD', 'content_name': 'gras_20200427_dim_track_participant.csv', 'content_size': 565118579, 'failure_count': 0, 'latest_job_id': '14260_2020-04-28T02.00.58', 'created_at': '2020-04-27 14:31:08.518350', 'completed_at': None, 'last_checked_at': '2020-04-27 15:39:33.622070', 'sub_content': 'null', 'record_count': 9188557, 'hash': '0265bed78e2fdccce1db1d4a28d926cf', 'metadata_process_status': 'COMPLETED', 'metadata_process_started_at': '2020-04-27 15:42:01.304492', 'metadata_process_completed_at': '2020-04-27 15:45:51.944453' } ) db.add(similar_content_status) db.flush() filter_ = FilterSameDayOfWeek() initial_query = db.query(UnitOfWork).join(ContentStatus) filtered_query = filter_.apply(initial_query, value=content_status.content_status_id) assert initial_query.count() == 2 assert filtered_query.count() == 2 @pytest.mark.parametrize( 'report_id_shift, licensor_id_shift, context, report_date', [ (1, 0, 'dim_track_participant', '2020-04-27'), (0, 1, 'dim_track_participant', '2020-04-27'), (0, 0, 'foo_bar_baz', '2020-04-27'), (0, 0, 'dim_track_participant', '2020-04-26'), ] ) def test_filter_same_day_of_week__apply_excluded( # pylint: disable=too-many-arguments db, clean_db, content_status, unit_of_work, report_id_shift, licensor_id_shift, context, report_date ): similar_unit_of_work = UnitOfWork( **{ 'unit_of_work_code': 'foo_code', 'reprocess_id': '', 'report_date': report_date, 'report_id': unit_of_work.report_id + report_id_shift, 'licensor_id': unit_of_work.licensor_id + licensor_id_shift, 'version': 'v1', 'activity_status': 'NOT_IN_PROGRESS', 'completeness_status': 'ACTIVE', 'next_run_at': '2020-04-28 01:06:09.715337', 'created_at': '2020-04-28 01:00:58.806272', 'last_updated_at': '2020-05-27 08:20:58.481813', 'is_force_complete': True, 'priority': 5 } ) db.add(similar_unit_of_work) db.flush() similar_content_status = ContentStatus( **{ 'unit_of_work_id': similar_unit_of_work.unit_of_work_id, 'context': context, 'content_status': 'ON_HOLD', 'content_name': 'gras_20200427_dim_track_participant.csv', 'content_size': 565118579, 'failure_count': 0, 'latest_job_id': '14260_2020-04-28T02.00.58', 'created_at': '2020-04-27 14:31:08.518350', 'completed_at': None, 'last_checked_at': '2020-04-27 15:39:33.622070', 'sub_content': 'null', 'record_count': 9188557, 'hash': '0265bed78e2fdccce1db1d4a28d926cf', 'metadata_process_status': 'COMPLETED', 'metadata_process_started_at': '2020-04-27 15:42:01.304492', 'metadata_process_completed_at': '2020-04-27 15:45:51.944453' } ) db.add(similar_content_status) db.flush() filter_ = FilterSameDayOfWeek() initial_query = db.query(UnitOfWork).join(ContentStatus) filtered_query = filter_.apply(initial_query, value=content_status.content_status_id) assert initial_query.count() == 2 assert filtered_query.count() == 1 def test_action_reprocess_cs__no_cs_in_complete_status(mocker, db, clean_db, app, content_status): patched_flash = mocker.patch( 'delphi_slz_admin.modelviews.content_status.flash', spec=lambda *args, **kwargs: None ) modelview = ContentStatusModelView(db) modelview.action_reprocess_cs_exp([content_status.content_status_id]) patched_flash.assert_called_once_with('Some items are not in COMPLETE status', category='error') def test_action_reprocess_cs__ok(mocker, db, clean_db, app, content_status): content_status.content_status = ContentStatusEnum.COMPLETE db.flush() config = ReprocessingConfig( slz_bucket_archive='bucket_archive', slz_bucket_decompressed='bucket_decompressed', sqs_exp_split_notification_queue='arn-1' ) patched_flash = mocker.patch( 'delphi_slz_admin.modelviews.content_status.flash', spec=lambda *args, **kwargs: None ) reprocessing_service_mock = mocker.patch( 'delphi_slz_admin.services.reprocessing.ExplorationReprocessingService.reprocess_cs', return_value={}, ) db.begin() modelview = ContentStatusModelView(db) modelview.action_reprocess_cs_exp([content_status.content_status_id]) db.rollback() reprocessing_service_mock.assert_called_once_with( content_statuses=[content_status], config=config ) patched_flash.assert_called_once_with('Content statuses were successfully reprocessed') audit_log: AuditLog = db.query(AuditLog).first() assert audit_log.event_type == AuditLogTypes.EXP_CONTENT_STATUS_REPROCESSED assert audit_log.extra_data == { 'slz_content_status_ids': [content_status.content_status_id], } def test_action_reprocess_cs__errors_handling(mocker, db, clean_db, app, content_status): content_status.content_status = ContentStatusEnum.COMPLETE db.flush() patched_flash = mocker.patch( 'delphi_slz_admin.modelviews.content_status.flash', spec=lambda *args, **kwargs: None ) mocker.patch( 'delphi_slz_admin.services.reprocessing.ExplorationReprocessingService.reprocess_cs', return_value={'111': 'some test error'}, ) db.begin() modelview = ContentStatusModelView(db) modelview.action_reprocess_cs_exp([content_status.content_status_id]) db.rollback() patched_flash.assert_called_once_with( 'Some content_statuses weren\'t reprocessed. Details: {\'111\': \'some test error\'}' ) def test_action_reprocess_uow__no_cs_in_complete_status(mocker, db, clean_db, app, content_status): patched_flash = mocker.patch( 'delphi_slz_admin.modelviews.content_status.flash', spec=lambda *args, **kwargs: None ) modelview = ContentStatusModelView(db) modelview.action_reprocess_uow_exp([content_status.content_status_id]) patched_flash.assert_called_once_with( 'No content_statuses in COMPLETE status for selected UOW(s)', category='error' ) def test_action_reprocess_uow__ok(mocker, db, clean_db, app, content_status): content_status.content_status = ContentStatusEnum.COMPLETE db.flush() config = ReprocessingConfig( slz_bucket_archive='bucket_archive', slz_bucket_decompressed='bucket_decompressed', sqs_exp_split_notification_queue='arn-1' ) patched_flash = mocker.patch( 'delphi_slz_admin.modelviews.content_status.flash', spec=lambda *args, **kwargs: None ) reprocessing_service_mock = mocker.patch( 'delphi_slz_admin.services.reprocessing.ExplorationReprocessingService.reprocess_cs', return_value={}, ) db.begin() modelview = ContentStatusModelView(db) modelview.action_reprocess_uow_exp([content_status.content_status_id]) db.rollback() reprocessing_service_mock.assert_called_once_with( content_statuses=[content_status], config=config ) patched_flash.assert_called_once_with( f'UOWs: {content_status.unit_of_work_id} were successfully reprocessed' ) audit_log: AuditLog = db.query(AuditLog).first() assert audit_log.event_type == AuditLogTypes.EXP_UOW_REPROCESSED assert audit_log.extra_data == { 'slz_unit_of_work_ids': [content_status.unit_of_work_id], 'slz_content_status_ids': [content_status.content_status_id], } def test_action_reprocess_uow__ok_several_cs( mocker, db, clean_db, app, unit_of_work, content_status ): similar_content_status = ContentStatus( **{ 'unit_of_work_id': unit_of_work.unit_of_work_id, 'context': 'another', 'content_status': 'COMPLETE', 'content_name': 'gras_20200427_another.csv', 'content_size': 565118579, 'failure_count': 0, 'latest_job_id': '14260_2020-04-28T02.00.58', 'created_at': '2020-04-27 14:31:08.518350', 'completed_at': None, 'last_checked_at': '2020-04-27 15:39:33.622070', 'sub_content': 'null', 'record_count': 9188557, 'hash': '0265bed78e2fdccce1db1d4a28d926cf', 'metadata_process_status': 'COMPLETED', 'metadata_process_started_at': '2020-04-27 15:42:01.304492', 'metadata_process_completed_at': '2020-04-27 15:45:51.944453' } ) db.add(similar_content_status) db.flush() config = ReprocessingConfig( slz_bucket_archive='bucket_archive', slz_bucket_decompressed='bucket_decompressed', sqs_exp_split_notification_queue='arn-1' ) patched_flash = mocker.patch( 'delphi_slz_admin.modelviews.content_status.flash', spec=lambda *args, **kwargs: None ) reprocessing_service_mock = mocker.patch( 'delphi_slz_admin.services.reprocessing.ExplorationReprocessingService.reprocess_cs', return_value={}, ) db.begin() modelview = ContentStatusModelView(db) modelview.action_reprocess_uow_exp([similar_content_status.content_status_id]) db.rollback() reprocessing_service_mock.assert_called_once_with( content_statuses=[similar_content_status], config=config ) patched_flash.assert_called_once_with( f'UOWs: {similar_content_status.unit_of_work_id} were successfully reprocessed' ) db.delete(similar_content_status) def test_action_reprocess_uow__errors_handling(mocker, db, clean_db, app, content_status): content_status.content_status = ContentStatusEnum.COMPLETE db.flush() patched_flash = mocker.patch( 'delphi_slz_admin.modelviews.content_status.flash', spec=lambda *args, **kwargs: None ) mocker.patch( 'delphi_slz_admin.services.reprocessing.ExplorationReprocessingService.reprocess_cs', return_value={'111': 'some test error'}, ) db.begin() modelview = ContentStatusModelView(db) modelview.action_reprocess_uow_exp([content_status.content_status_id]) db.rollback() patched_flash.assert_called_once_with( 'Some content_statuses weren\'t reprocessed. Details: {\'111\': \'some test error\'}' )