# pylint: disable=no-member import os import unittest from datetime import datetime, timezone import pytest from db_schema.postgres import connection from db_schema.schemas import slz from parameterized import parameterized from slz_storage.conv import UoWCreateHelper from slz_storage.exceptions import ReportNotFoundError @pytest.mark.integration class UoWConvToPGTestCase(unittest.TestCase): @classmethod def setUpClass(cls): cls._pg = connection.get_session( host=os.environ.get('PG_HOST', '0.0.0.0'), port=os.environ.get('PG_PORT', 5432), db=os.environ.get('PG_DB', 'slz'), user=os.environ.get('PG_USER', 'admin'), password=os.environ.get('PG_PASSWORD', 'admin'), ) cls._uow_conv = UoWCreateHelper(cls._pg) @classmethod def tearDownClass(cls): for model in [ slz.UnitOfWorkMigrationLog, slz.ContentStatusMigrationLog, slz.ContentFailureLog, slz.ContentStatus, slz.UnitOfWork, ]: cls._pg.query(model).delete() cls._pg.commit() cls._pg.flush() cls._pg.close() @parameterized.expand( [ ( 'apple_theorchard_amEvent_v1_2_reprocessing', 'apple-20191124-theorchard-amEvent-v1_2-rp20191202T111111', '20191202T111111' ), ( 'apple_theorchard-amArtistDemographics-v1_0_reprocessing', 'apple-20191123-theorchard-amArtistDemographics-v1_0-rp20191125T121213', '20191125T121213' ), ( 'apple_theorchard-amContent-v1_2_reprocessing', 'apple-20191122-theorchard-amContent-v1_2-rp20191202T111111', '20191202T111111' ), ( 'apple_theorchard-amContent-v1_2_reprocessing', 'apple-20191122-theorchard-amContent-v1_2-rp20191202T111111', '20191202T111111' ), ('apple_theorchard-amContent-v1_2', 'apple-20191122-theorchard-amContent-v1_2', ''), ('spotify-sme-tracks-v1', 'spotify-20191114-sme-tracks-v1', ''), ('amazonadsupported-sme-activity-v1', 'amazonadsupported-20191101-sme-activity-v1', ''), ('amazonprime-sme-user-v1', 'amazonprime-20191122-sme-user-v1', ''), ('amazonadsupported-sme-user-v1', 'amazonadsupported-20191205-sme-user-v1', ''), ( 'youtubemusic-sme-daily_activity_summary-v1', 'youtubemusic-20191101-sme-daily_activity_summary-v1', '' ), ] ) def test_get_uow(self, _, input_, expected): self.assertEqual(self._uow_conv.get_uow_reprocess_id(input_), expected) @parameterized.expand( [ ( 'apple_theorchard_amEvent_v1_2_reprocessing', 'apple-20191124-theorchard-amEvent-v1_2-rp20191202T111111', 'apple-20191124-theorchard-amEvent-v1_2' ), ( 'apple_theorchard-amArtistDemographics-v1_0_reprocessing', 'apple-20191123-theorchard-amArtistDemographics-v1_0-rp20191125T121213', 'apple-20191123-theorchard-amArtistDemographics-v1_0' ), ( 'apple_theorchard-amContent-v1_2_reprocessing', 'apple-20191122-theorchard-amContent-v1_2-rp20191202T111111', 'apple-20191122-theorchard-amContent-v1_2' ), ( 'apple_theorchard-amContent-v1_2', 'apple-20191122-theorchard-amContent-v1_2', 'apple-20191122-theorchard-amContent-v1_2' ), ( 'spotify-sme-tracks-v1', 'spotify-20191114-sme-tracks-v1', 'spotify-20191114-sme-tracks-v1' ), ( 'amazonadsupported-sme-activity-v1', 'amazonadsupported-20191101-sme-activity-v1', 'amazonadsupported-20191101-sme-activity-v1' ), ( 'amazonprime-sme-user-v1', 'amazonprime-20191122-sme-user-v1', 'amazonprime-20191122-sme-user-v1' ), ( 'amazonadsupported-sme-user-v1', 'amazonadsupported-20191205-sme-user-v1', 'amazonadsupported-20191205-sme-user-v1' ), ( 'youtubemusic-sme-daily_activity_summary-v1', 'youtubemusic-20191101-sme-daily_activity_summary-v1', 'youtubemusic-20191101-sme-daily_activity_summary-v1' ), ] ) def test_get_original_uow_id(self, _, input_, expected): self.assertEqual(self._uow_conv.get_original_uow_id(input_), expected) @parameterized.expand( [ ( 'date', 'apple-20191124-theorchard-amEvent-v1_2-rp20191202T111111', datetime(year=2019, month=11, day=24) ), ( 'date', 'apple-20191124-theorchard-amEvent-v1_2', datetime(year=2019, month=11, day=24) ), ] ) def test_get_original_uow_id_2(self, _, input_, expected): self.assertEqual(self._uow_conv.get_uow_report_date(input_), expected) @parameterized.expand( [ ( 'apple_theorchard_amEvent_v1_2_reprocessing', 'apple-20191124-theorchard-amEvent-v1_2-rp20191202T111111', ( 'apple-20191124-theorchard-amEvent-v1_2', datetime(year=2019, month=11, day=24), '20191202T111111' ) ), ( 'apple_theorchard_amEvent_v1_2', 'apple-20191124-theorchard-amEvent-v1_2', ( 'apple-20191124-theorchard-amEvent-v1_2', datetime(year=2019, month=11, day=24), '' ) ), ] ) def test_get_compound_id(self, _, input_, expected): self.assertEqual(self._uow_conv.get_compound_id(input_), expected) @parameterized.expand( [ ( 'apple_theorchard_amEvent_v1_2_reprocessing', 'apple-20191124-theorchard-amEvent-v1_2-rp20191202T111111', 'v1_2' ), ( 'apple_theorchard-amArtistDemographics-v1_0_reprocessing', 'apple-20191123-theorchard-amArtistDemographics-v1_0-rp20191125T121213', 'v1_0' ), ( 'apple_theorchard-amContent-v1_2_reprocessing', 'apple-20191122-theorchard-amContent-v1_2-rp20191202T111111', 'v1_2' ), ('apple_theorchard-amContent-v1_2', 'apple-20191122-theorchard-amContent-v1_2', 'v1_2'), ('spotify-sme-tracks-v1', 'spotify-20191114-sme-tracks-v1', 'v1'), ( 'amazonadsupported-sme-activity-v1', 'amazonadsupported-20191101-sme-activity-v1', 'v1' ), ('amazonprime-sme-user-v1', 'amazonprime-20191122-sme-user-v1', 'v1'), ('amazonadsupported-sme-user-v1', 'amazonadsupported-20191205-sme-user-v1', 'v1'), ( 'youtubemusic-sme-daily_activity_summary-v1', 'youtubemusic-20191101-sme-daily_activity_summary-v1', 'v1' ), ] ) def test_get_original_uow_id_3(self, _, input_, expected): self.assertEqual(self._uow_conv.get_version(input_), expected) def test_get_report_ok(self): report_id = self._uow_conv.get_report_id( 'apple-20191124-theorchard-amEvent-v1_2-rp20191202T111111' ) expected = self._pg.query(slz.Report).filter(slz.Report.report_name == 'amEvent').one() self.assertEqual(report_id, expected.report_id) def test_get_report_raise_exc(self): uow_id = 'apple-20191124-theorchard-amUnknownReport-v1_2' self.assertRaises(ReportNotFoundError, self._uow_conv.get_report_id, uow_id) def test_get_licensor_id(self): licensor_id = self._uow_conv.get_licensor_id( 'apple-20191124-theorchard-amEvent-v1_2-rp20191202T111111' ) expected = self._pg.query(slz.Licensor).filter(slz.Licensor.licensor_name == 'theorchard' ).one() self.assertEqual(licensor_id, expected.licensor_id) def test_uow(self): now = datetime.now() expected_uow = slz.UnitOfWork( unit_of_work_code='apple-20191124-theorchard-amEvent-v1_2', reprocess_id='1234', report_date=datetime(year=2019, month=11, day=24), report_id=3, licensor_id=3, version='v1_2', activity_status=slz.ActivityStatusEnum.NOT_IN_PROGRESS.value, completeness_status=slz.CompletenessStatusEnum.ACTIVE.value, next_run_at=now.astimezone(timezone.utc), created_at=now.astimezone(timezone.utc), last_updated_at=now.astimezone(timezone.utc), is_force_complete=False, priority=slz.UnitOfWorkPriorityEnum.DEFAULT.value ) uow = self._uow_conv.generate_uow( 'apple-20191124-theorchard-amEvent-v1_2-rp1234', now, now, now ) self.assertEqual(uow.unit_of_work_code, expected_uow.unit_of_work_code) self.assertEqual(uow.reprocess_id, expected_uow.reprocess_id) self.assertEqual(uow.report_date, expected_uow.report_date) self.assertEqual(uow.report_id, expected_uow.report_id) self.assertEqual(uow.licensor_id, expected_uow.licensor_id) self.assertEqual(uow.version, expected_uow.version) self.assertEqual(uow.activity_status, expected_uow.activity_status) self.assertEqual(uow.completeness_status, expected_uow.completeness_status) self.assertEqual(uow.next_run_at, expected_uow.next_run_at) self.assertEqual(uow.created_at, expected_uow.created_at) self.assertEqual(uow.last_updated_at, expected_uow.last_updated_at) self.assertEqual(uow.is_force_complete, expected_uow.is_force_complete) self.assertEqual(uow.priority, expected_uow.priority)