# pylint: disable=unused-argument, protected-access from datetime import datetime, timezone from unittest.mock import Mock import pytest from db_schema.schemas import slz from psycopg2.tz import FixedOffsetTimezone from slz_job_manager import ExecutionError from slz_job_manager.services.db import CommandDBService from slz_job_manager.services.uow_creator import UowCreatorService @pytest.mark.integration @pytest.mark.parametrize( 'now, expected_uows_created', [ ( datetime(2019, 4, 13, tzinfo=timezone.utc), [ 'amazonadsupported-20190412-sme-activity-v1', 'gras-20190412-sme-dim_country-v1', ] ), ( datetime(2021, 10, 26, tzinfo=timezone.utc), [ 'amazonadsupported-20211025-sme-activity-v1', 'youtubereporting-20211024-sme-active_claims-a1', 'gras-20211025-sme-dim_country-v1', ] ), ( datetime(2021, 10, 29, tzinfo=timezone.utc), [ 'amazonadsupported-20211028-sme-activity-v1', 'youtubereporting-20211027-sme-active_claims-a1', # meet frequency pattern 'appreciationengine-20211029-sme-request_to_forget-v1', 'gras-20211028-sme-dim_country-v1', ] ), ] ) def test_uow_creator_service( now, expected_uows_created, db, clean_db, logger_test, merged_config_service_test ): db_service = CommandDBService( logger=logger_test, db_conn=db, now=now, ) errors_collector = [] uow_creator = UowCreatorService( logger=logger_test, now=now, merged_config_service=merged_config_service_test, db_service=db_service, errors_collector=errors_collector, ) result = uow_creator.create_uows() assert sorted(expected_uows_created) == sorted(result) assert db.session.query(slz.UnitOfWork).count() == len(expected_uows_created) # call func next time, uow should not be created result = uow_creator.create_uows() assert len(result) == 0 # count of objects in db should remain the same assert db.session.query(slz.UnitOfWork).count() == len(expected_uows_created) def test_uow_create_correct_handle_exc(logger_test, merged_config_service_test): now = datetime(2021, 10, 29, tzinfo=timezone.utc) # choose 1 random config key = list(merged_config_service_test.merged_config.keys())[0] config = merged_config_service_test.merged_config[key] db_service = Mock() db_service.create_uow.side_effect = Exception('Test') errors_collector = [] uow_creator = UowCreatorService( logger=logger_test, now=now, merged_config_service=merged_config_service_test, db_service=db_service, errors_collector=errors_collector, ) result = uow_creator._create_uows([config]) assert result == [] assert len(errors_collector) == 1 expected_error = ExecutionError(error=Exception('Test'), context={'uow_config': config}) assert errors_collector[0] == expected_error @pytest.mark.integration @pytest.mark.parametrize( 'hours_offset, expected_start_dates', [ ( 0, [ datetime(2018, 4, 3, 6, tzinfo=FixedOffsetTimezone()), datetime(2018, 4, 3, 12, tzinfo=FixedOffsetTimezone()), datetime(2018, 4, 3, 18, tzinfo=FixedOffsetTimezone()), # because of 6-23 in schedule datetime(2018, 4, 4, 6, tzinfo=FixedOffsetTimezone()), ] ), ( 2, [ datetime(2018, 4, 3, 8, tzinfo=FixedOffsetTimezone()), datetime(2018, 4, 3, 14, tzinfo=FixedOffsetTimezone()), datetime(2018, 4, 3, 20, tzinfo=FixedOffsetTimezone()), datetime(2018, 4, 4, 6, tzinfo=FixedOffsetTimezone()), ] ) ] ) def test_uow_creator_service_timeframes( hours_offset, expected_start_dates, db, clean_db, logger_test, merged_config_service_test, ): merged_config_service_test.merged_config['appreciationengine-{yyyymmdd}-sme-memberslogin-v1'] \ .dsp_config.hours_offset = hours_offset now = datetime(2018, 4, 3, tzinfo=timezone.utc) expected_uows_created = [ 'appreciationengine-20180403-sme-memberslogin-v1', 'appreciationengine-20180403-sme-memberslogin-v1', 'appreciationengine-20180403-sme-memberslogin-v1', 'appreciationengine-20180403-sme-memberslogin-v1', ] db_service = CommandDBService( logger=logger_test, db_conn=db, now=now, ) errors_collector = [] uow_creator = UowCreatorService( logger=logger_test, now=now, merged_config_service=merged_config_service_test, db_service=db_service, errors_collector=errors_collector, ) assert db.session.query(slz.UnitOfWork).count() == 0 assert db.session.query(slz.UnitOfWorkGroup).count() == 0 assert db.session.query(slz.UnitOfWorkPipeline).count() == 0 result = uow_creator.create_uows() uows = db.session.query(slz.UnitOfWork).order_by(slz.UnitOfWork.unit_of_work_id).all() group_id = uows[0].unit_of_work_pipeline.unit_of_work_group_id assert sorted(expected_uows_created) == sorted(result) assert len(uows) == 4 assert db.session.query(slz.UnitOfWorkGroup).count() == 1 assert all(map(lambda x: x.unit_of_work_pipeline.unit_of_work_group_id == group_id, uows)) assert db.session.query(slz.UnitOfWorkPipeline).count() == 4 # check next_run_at in created UoWs assert uows[0].next_run_at == expected_start_dates[0] assert uows[1].next_run_at == expected_start_dates[1] assert uows[2].next_run_at == expected_start_dates[2] assert uows[3].next_run_at == expected_start_dates[3] # call func next time, uow should not be created result = uow_creator.create_uows() assert len(result) == 0 # count of objects in db should remain the same assert db.session.query(slz.UnitOfWork).count() == 4 assert db.session.query(slz.UnitOfWorkGroup).count() == 1 assert db.session.query(slz.UnitOfWorkPipeline).count() == 4 @pytest.mark.integration def test_uow_creator_timeframes_several_dates_correct_group( db, clean_db, logger_test, merged_config_service_test ): for now in [ datetime(2018, 4, 3, tzinfo=timezone.utc), datetime(2018, 4, 4, tzinfo=timezone.utc) ]: db_service = CommandDBService( logger=logger_test, db_conn=db, now=now, ) errors_collector = [] uow_creator = UowCreatorService( logger=logger_test, now=now, merged_config_service=merged_config_service_test, db_service=db_service, errors_collector=errors_collector, ) uow_creator.create_uows() assert db.session.query(slz.UnitOfWork).count() == 8 assert db.session.query(slz.UnitOfWorkGroup).count() == 2 assert db.session.query(slz.UnitOfWorkPipeline).count() == 8 # check that all UoWs grouped in correct group groups = db.session.query(slz.UnitOfWorkGroup).all() group_id_1 = groups[0].unit_of_work_group_id uows_1 = db.session.query(slz.UnitOfWork).join( slz.UnitOfWorkPipeline ).filter(slz.UnitOfWorkPipeline.unit_of_work_group_id == groups[0].unit_of_work_group_id).all() assert len(uows_1) == 4 assert all(map(lambda x: x.unit_of_work_pipeline.unit_of_work_group_id == group_id_1, uows_1)) assert all([ uow.unit_of_work_code == 'appreciationengine-20180403-sme-memberslogin-v1' for uow in uows_1 ]) group_id_2 = groups[1].unit_of_work_group_id uows_2 = db.session.query(slz.UnitOfWork).join( slz.UnitOfWorkPipeline ).filter(slz.UnitOfWorkPipeline.unit_of_work_group_id == groups[1].unit_of_work_group_id).all() assert len(uows_2) == 4 assert all(map(lambda x: x.unit_of_work_pipeline.unit_of_work_group_id == group_id_2, uows_2)) assert all([ uow.unit_of_work_code == 'appreciationengine-20180404-sme-memberslogin-v1' for uow in uows_2 ])