from datetime import date, datetime, timedelta, timezone from typing import Dict from unittest.mock import Mock import pytest from db_schema.common import ActivityStatusEnum, CompletenessStatusEnum from db_schema.schemas import slz from slz_house_keeper.services.db import DBService @pytest.mark.integration def test_get_subtract_contexts(db, db_licensors, db_reports): uow_1 = slz.UnitOfWork( unit_of_work_code='apple-20191123-theorchard-amEvent-v1_2', licensor=db_licensors['theorchard'], report=db_reports['amEvent'], report_date=date(2019, 11, 23), 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=datetime(2021, 11, 24, tzinfo=timezone.utc), last_updated_at=datetime(2021, 11, 24, tzinfo=timezone.utc), created_at=datetime(2021, 11, 24, tzinfo=timezone.utc), ) db.session.add(uow_1) db.session.commit() expected = { slz.ContentStatusEnum.COMPLETE: ['FR', 'IT', 'RU', 'US'], slz.ContentStatusEnum.CANCELLED: ['BR', 'GB'], slz.ContentStatusEnum.ON_HOLD: ['JP'], } for status, contexts in expected.items(): for context in contexts: cs = slz.ContentStatus( unit_of_work_id=uow_1.unit_of_work_id, context=context, content_name=f'{context}.txt', content_status=status, content_size=400, failure_count=0, created_at=datetime(2021, 11, 24, tzinfo=timezone.utc), sub_content='{}', metadata_process_status=slz.ContentMetadataStatusEnum.NOT_QUEUED.value, ) db.session.add(cs) db.session.commit() db_service = DBService( logger=Mock(), db_conn=db, now=Mock(), ) result = db_service.get_subtract_contexts(unit_of_work_id=uow_1.unit_of_work_id) assert result == expected @pytest.mark.integration @pytest.mark.parametrize( 'last_updated_at, threshold_min, field_updated', [ (datetime(2021, 11, 24, 15, tzinfo=timezone.utc), 45, False), (datetime(2021, 11, 24, 14, tzinfo=timezone.utc), 45, True), ] ) def test_fail_stuck( last_updated_at, threshold_min, field_updated, db, db_licensors, db_reports, logger_test ): now = datetime(2021, 11, 24, 15, 30, tzinfo=timezone.utc) uow_1 = slz.UnitOfWork( unit_of_work_code='apple-20191123-theorchard-amEvent-v1_2', licensor=db_licensors['theorchard'], report=db_reports['amEvent'], report_date=date(2019, 11, 23), version='v1_2', activity_status=ActivityStatusEnum.IN_PROGRESS.value, completeness_status=CompletenessStatusEnum.ACTIVE.value, is_force_complete=False, priority=slz.UnitOfWorkPriorityEnum.DEFAULT.value, next_run_at=datetime(2021, 11, 24, tzinfo=timezone.utc), last_updated_at=last_updated_at, created_at=datetime(2021, 11, 24, tzinfo=timezone.utc), ) db.session.add(uow_1) db.session.commit() db_service = DBService( logger=logger_test, db_conn=db, now=now, ) result = db_service.fail_stuck(threshold_min) assert result is True assert (uow_1.activity_status == ActivityStatusEnum.NOT_IN_PROGRESS) == field_updated assert (uow_1.last_updated_at == now) == field_updated @pytest.mark.integration def test_get_active_uow(db, db_licensors, db_reports, logger_test): uow_1 = slz.UnitOfWork( unit_of_work_code='apple-20191123-theorchard-amEvent-v1_2', licensor=db_licensors['theorchard'], report=db_reports['amEvent'], report_date=date(2019, 11, 23), version='v1_2', activity_status=ActivityStatusEnum.IN_PROGRESS.value, completeness_status=CompletenessStatusEnum.ACTIVE.value, is_force_complete=False, priority=slz.UnitOfWorkPriorityEnum.DEFAULT.value, next_run_at=datetime(2021, 11, 24, tzinfo=timezone.utc), last_updated_at=datetime(2021, 11, 24, tzinfo=timezone.utc), created_at=datetime(2021, 11, 24, tzinfo=timezone.utc), ) uow_2 = slz.UnitOfWork( unit_of_work_code='apple-20191124-theorchard-amEvent-v1_2', licensor=db_licensors['theorchard'], report=db_reports['amEvent'], report_date=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=datetime(2021, 11, 25, tzinfo=timezone.utc), last_updated_at=datetime(2021, 11, 25, tzinfo=timezone.utc), created_at=datetime(2021, 11, 25, tzinfo=timezone.utc), ) db.session.add(uow_1) db.session.add(uow_1) db.session.commit() db_service = DBService( logger=logger_test, db_conn=db, now=Mock(), ) result = db_service.get_active_uows() assert result == [uow_2] @pytest.mark.integration @pytest.mark.parametrize( 'force_complete, content_statuses_updated', [ (False, {'US', 'MX'}), (True, set()), ] ) def test_update_uow_completeness_status( force_complete, content_statuses_updated, db, logger_test, now, unit_of_work_stubs, content_status_stubs: Dict[int, slz.ContentStatus], ): now += timedelta(hours=2) uow_1 = slz.UnitOfWork(**unit_of_work_stubs[1]) db.session.add(uow_1) cs_1 = slz.ContentStatus(**content_status_stubs[1]) cs_1.content_status = slz.ContentStatusEnum.COMPLETE_VOLATILE cs_2 = slz.ContentStatus(**content_status_stubs[2]) cs_2.content_status = slz.ContentStatusEnum.COMPLETE_VOLATILE cs_3 = slz.ContentStatus(**content_status_stubs[3]) cs_3.content_status = slz.ContentStatusEnum.FAILED uow_1.content_statuses.append(cs_1) uow_1.content_statuses.append(cs_2) uow_1.content_statuses.append(cs_3) db.session.commit() db_service = DBService( logger=logger_test, db_conn=db, now=Mock(), ) db_service.complete_uow(uow_1, now, force=force_complete) found: slz.UnitOfWork = db.session.query( slz.UnitOfWork ).filter(slz.UnitOfWork.unit_of_work_id == uow_1.unit_of_work_id).one() assert found.completeness_status == CompletenessStatusEnum.COMPLETE assert found.is_force_complete is force_complete assert found.last_updated_at == now found = db.session.query(slz.ContentStatus).filter( slz.UnitOfWork.unit_of_work_id == uow_1.unit_of_work_id, slz.ContentStatus.content_status == slz.ContentStatusEnum.COMPLETE, ).all() assert content_statuses_updated == {cs.context for cs in found} def test_mock(): # mock test for docker/test. It requires at least 1 test otherwise fails. # All tests in this pkg are run through docker/integration cmd assert 1 == 1