from datetime import datetime, timedelta, timezone from unittest import TestCase, mock import pytest from db_schema.common import ActivityStatusEnum, CompletenessStatusEnum from db_schema.schemas import slz from slz_downloader.content_status_service import ContentStatusService from slz_downloader.entities import Job @pytest.fixture def job_mock(uow_mock): return Job( uow_id='appreciationengine-20210825-sme-membersvisittotals-v1', unit_of_work_id=uow_mock.unit_of_work_id, dsp='appreciationengine', report_type='membersvisittotals', subtype=None, version='v1', report_date='2021-08-25', licensor='sme', extension='csv', config_bucket='delphi-configs', context='UK', context_params=None, job_id="test_job_id", ) @pytest.fixture def uow_mock(db, db_licensors, db_reports): now = datetime.now(timezone.utc) - timedelta(days=1) uow = slz.UnitOfWork( unit_of_work_code='appreciationengine-20210825-sme-membersvisittotals-v1', licensor=db_licensors['sme'], report=db_reports['membersvisittotals'], report_date='2019-11-24', version='v1', activity_status=ActivityStatusEnum.NOT_IN_PROGRESS.value, completeness_status=CompletenessStatusEnum.ACTIVE.value, is_force_complete=False, priority=slz.UnitOfWorkPriorityEnum.DEFAULT.value, next_run_at=now, last_updated_at=now, created_at=now, ) db.session.add(uow) db.session.commit() return uow @pytest.fixture def cs_mock(db, uow_mock, job_mock): now = datetime.now(timezone.utc) - timedelta(days=1) cs = slz.ContentStatus( unit_of_work_id=uow_mock.unit_of_work_id, content_name='report.txt.gz', context=job_mock.context, content_status=slz.ContentStatusEnum.ACTIVE.value, latest_job_id=job_mock.job_id, content_size=44, created_at=now, last_checked_at=now, failure_count=0, metadata_process_status=slz.ContentMetadataStatusEnum.NOT_QUEUED.value, ) db.session.add(cs) db.session.commit() return cs @pytest.mark.integration @mock.patch('slz_downloader.content_status_service.datetime') def test_save_failed(datetime_mock, logger_test, db, job_mock, cs_mock): datetime_mock.now.return_value = datetime(2019, 4, 4, 1) service = ContentStatusService(logger=logger_test, db_conn=db) assert db.session.query(slz.ContentFailureLog).count() == 0 result = service.save_failed( job_mock, 'content_name', 'ValidationError', 'some fields are invalid', ) assert result is True assert cs_mock.content_status == slz.ContentStatusEnum.FAILED assert db.session.query(slz.ContentFailureLog).count() == 1 @pytest.mark.integration @mock.patch('slz_downloader.content_status_service.datetime') def test_save_success(datetime_mock, logger_test, db, job_mock, cs_mock): datetime_mock.now.return_value = datetime(2019, 4, 4, 1) service = ContentStatusService(logger=logger_test, db_conn=db) result = service.save_success( job_mock, 'content_name', 'content_name', 444, [], ) assert result is True assert cs_mock.content_status == slz.ContentStatusEnum.COMPLETE assert cs_mock.content_size == 444 @pytest.mark.integration @mock.patch('slz_downloader.content_status_service.datetime') def test_save_active(datetime_mock, logger_test, db, job_mock, cs_mock): datetime_mock.now.return_value = datetime(2019, 4, 4, 1) service = ContentStatusService(logger=logger_test, db_conn=db) result = service.save_active( job_mock, 'content_name', ) assert result is True assert cs_mock.content_status == slz.ContentStatusEnum.ACTIVE @pytest.mark.integration @mock.patch('slz_downloader.content_status_service.datetime') def test_save_missing(datetime_mock, logger_test, db, job_mock, cs_mock): cs_mock.content_status = slz.ContentStatusEnum.FAILED db.session.commit() assert cs_mock.content_status == slz.ContentStatusEnum.FAILED datetime_mock.now.return_value = datetime(2019, 4, 4, 1) service = ContentStatusService(logger=logger_test, db_conn=db) result = service.save_missing( job_mock, 'content_name', ) assert result is True assert cs_mock.content_status == slz.ContentStatusEnum.MISSING @pytest.mark.integration @mock.patch('slz_downloader.content_status_service.datetime') def test_start_processing(datetime_mock, logger_test, db, job_mock, cs_mock): datetime_mock.now.return_value = now = datetime(2019, 4, 4, 1, tzinfo=timezone.utc) service = ContentStatusService(logger=logger_test, db_conn=db) # already completed cs_mock.content_status = slz.ContentStatusEnum.COMPLETE db.session.commit() result = service.start_processing( job_mock, 'content_name', ) assert result is False assert db.session.query(slz.ContentStatus).count() == 1 assert cs_mock.last_checked_at != now # Updated cs_mock.content_status = slz.ContentStatusEnum.FAILED db.session.commit() result = service.start_processing( job_mock, 'content_name', ) assert result is True assert db.session.query(slz.ContentStatus).count() == 1 assert cs_mock.last_checked_at == now # new job_mock.context = 'Alamo' result = service.start_processing( job_mock, 'content_name', ) assert result is True assert db.session.query(slz.ContentStatus).count() == 2 cs = db.session.query(slz.ContentStatus).filter(slz.ContentStatus.context == job_mock.context ).one() assert cs.created_at == now @pytest.mark.integration def test_find_content_status_same_day_of_week(logger_test, db, db_licensors, db_reports): service = ContentStatusService(logger=logger_test, db_conn=db) now = datetime(2019, 4, 4, 1, tzinfo=timezone.utc) uow_1 = slz.UnitOfWork( unit_of_work_code='apple-20191124-theorchard-amEvent-v1_2', reprocess_id='', licensor=db_licensors['theorchard'], report=db_reports['amEvent'], report_date='2019-11-24', version='v1_2', activity_status=ActivityStatusEnum.NOT_IN_PROGRESS.value, completeness_status=CompletenessStatusEnum.ACTIVE.value, is_force_complete=False, priority=slz.UnitOfWorkPriorityEnum.DEFAULT.value, next_run_at='2019-11-07T01:10:38.377644+00:00', created_at='2019-11-07T01:01:00.599872+00:00', last_updated_at='2019-11-21T18:41:39.985786+00:00', ) uow_2 = slz.UnitOfWork( unit_of_work_code='apple-20191110-theorchard-amEvent-v1_2', reprocess_id='', licensor=db_licensors['theorchard'], report=db_reports['amEvent'], report_date='2019-11-10', version='v1_2', activity_status=ActivityStatusEnum.NOT_IN_PROGRESS.value, completeness_status=CompletenessStatusEnum.ACTIVE.value, is_force_complete=False, priority=slz.UnitOfWorkPriorityEnum.DEFAULT.value, next_run_at='2019-11-07T01:10:38.377644+00:00', created_at='2019-11-07T01:01:00.599872+00:00', last_updated_at='2019-11-21T18:41:39.985786+00:00', ) uow_3 = slz.UnitOfWork( unit_of_work_code='apple-20191116-theorchard-amEvent-v1_2', reprocess_id='', licensor=db_licensors['theorchard'], report=db_reports['amEvent'], report_date='2019-11-16', version='v1_2', activity_status=ActivityStatusEnum.NOT_IN_PROGRESS.value, completeness_status=CompletenessStatusEnum.ACTIVE.value, is_force_complete=False, priority=slz.UnitOfWorkPriorityEnum.DEFAULT.value, next_run_at='2019-11-07T01:10:38.377644+00:00', created_at='2019-11-07T01:01:00.599872+00:00', last_updated_at='2019-11-21T18:41:39.985786+00:00', ) uow_4 = slz.UnitOfWork( unit_of_work_code='apple-20191201-theorchard-amEvent-v1_2', reprocess_id='', licensor=db_licensors['theorchard'], report=db_reports['amEvent'], report_date='2019-12-01', version='v1_2', activity_status=ActivityStatusEnum.NOT_IN_PROGRESS.value, completeness_status=CompletenessStatusEnum.ACTIVE.value, is_force_complete=False, priority=slz.UnitOfWorkPriorityEnum.DEFAULT.value, next_run_at='2019-11-07T01:10:38.377644+00:00', created_at='2019-11-07T01:01:00.599872+00:00', last_updated_at='2019-11-21T18:41:39.985786+00:00', ) cs_1 = slz.ContentStatus( context='US', content_name='US.txt', content_status=slz.ContentStatusEnum.COMPLETE, content_size=100, created_at=now, failure_count=0, sub_content='{}', metadata_process_status=slz.ContentMetadataStatusEnum.NOT_QUEUED.value, ) cs_2 = slz.ContentStatus( context='MX', content_name='MX.txt', content_status=slz.ContentStatusEnum.COMPLETE, created_at=now, content_size=200, failure_count=0, sub_content='{}', metadata_process_status=slz.ContentMetadataStatusEnum.NOT_QUEUED.value, ) cs_3 = slz.ContentStatus( context='US', content_name='US.txt', content_status=slz.ContentStatusEnum.COMPLETE, content_size=300, failure_count=0, created_at=now, sub_content='{}', metadata_process_status=slz.ContentMetadataStatusEnum.NOT_QUEUED.value, ) cs_4 = slz.ContentStatus( context='US', content_name='US.txt', content_status=slz.ContentStatusEnum.FAILED, content_size=400, failure_count=0, created_at=now, sub_content='{}', metadata_process_status=slz.ContentMetadataStatusEnum.NOT_QUEUED.value, ) cs_5 = slz.ContentStatus( context='US', content_name='US.txt', content_status=slz.ContentStatusEnum.COMPLETE, content_size=500, failure_count=0, created_at=now, sub_content='{}', metadata_process_status=slz.ContentMetadataStatusEnum.NOT_QUEUED.value, ) db.session.add(uow_1) db.session.add(uow_2) db.session.add(uow_3) db.session.add(uow_4) uow_1.content_statuses.append(cs_1) uow_1.content_statuses.append(cs_2) uow_2.content_statuses.append(cs_3) uow_3.content_statuses.append(cs_4) uow_4.content_statuses.append(cs_5) db.session.add(cs_1) db.session.add(cs_2) db.session.add(cs_3) db.session.add(cs_4) db.session.add(cs_5) db.session.commit() job = Job.empty() job.unit_of_work_id = uow_4.unit_of_work_id job.context = cs_5.context css = service.get_file_size_history(job, limit=3) assert [cs_1.content_size, cs_3.content_size] == css