"""Logic test module.""" import json from unittest.mock import call, patch import pytest from src import logic class TestLogic: @pytest.fixture def sample_message(self): yield { 'timestamp': '2026-04-03T14:07:38Z', 'table': 'fraud_report', 'partition': { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-02', 'entity_name': 'ZQOR', 'service': 'UNLIMITED', 'music_territory': 'BR', }, 'status': 'new', } def sample_body(self, sample_message): return { 'Type': 'Notification', 'MessageId': 'a80dafe5-dc3b-54ac-907d-cdac17d18ef5', 'Message': json.dumps(sample_message), } @pytest.fixture(autouse=True) def mock_service(self): with ( patch.object(logic, 'lambda_metric') as lambda_metric_mock, patch.object(logic, 'jenkins_utils') as jenkins_utils_mock, patch.object(logic, 'logger') as logger_mock, ): self.lambda_metric_mock = lambda_metric_mock self.jenkins_utils_mock = jenkins_utils_mock self.logger_mock = logger_mock yield @pytest.mark.parametrize( 'environments, message, expected_jenkins_calls, expected_metrics_calls', [ pytest.param( ['env1', 'env2'], { 'timestamp': '2026-04-03T14:07:38Z', 'table': 'daily_play_events', 'partition': { 'Service': 'Service', 'Region': 'EU', 'Marketplace': '4', 'Territory': 'AT', 'Year': '2026', 'Month': '04', 'Day': '04', }, 'status': 'new', }, [ call( 'swf-amazon-datapulse-exec', { 'ENVIRONMENT': 'env1', 'REPORT': 'daily_play_events', 'CONTEXT_DATE': '2026-04-04', 'PARTITION': '{"Service": "Service", "Region": "EU", "Marketplace": ' '"4", "Territory": "AT", "Year": "2026", "Month": "04"' ', "Day": "04"}', }, ), call( 'swf-amazon-datapulse-exec', { 'ENVIRONMENT': 'env2', 'REPORT': 'daily_play_events', 'CONTEXT_DATE': '2026-04-04', 'PARTITION': '{"Service": "Service", "Region": "EU", "Marketplace": ' '"4", "Territory": "AT", "Year": "2026", "Month": "04"' ', "Day": "04"}', }, ), ], [ call( 'amazon_datapulse_trigger.routed', 1, tags=['report_type:daily_play_events', 'environment:env1'], ), call( 'amazon_datapulse_trigger.routed', 1, tags=['report_type:daily_play_events', 'environment:env2'], ), ], id='daily_play_events - trigger for 2 environments', ), pytest.param( ['env1'], { 'timestamp': '2026-04-03T14:07:38Z', 'table': 'fraud_report', 'partition': { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-02', 'entity_name': 'ZQOR', 'service': 'UNLIMITED', 'music_territory': 'BR', }, 'status': 'new', }, [ call( 'swf-amazon-datapulse-exec', { 'ENVIRONMENT': 'env1', 'REPORT': 'fraud_report', 'CONTEXT_DATE': '2026-04-02', 'PARTITION': json.dumps( { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-02', 'entity_name': 'ZQOR', 'service': 'UNLIMITED', 'music_territory': 'BR', } ), }, ), ], [ call( 'amazon_datapulse_trigger.routed', 1, tags=[ 'report_type:fraud_report', 'environment:env1', ], ), ], id='fraud_report ZQOR - 1 environment', ), pytest.param( ['env1', 'env2'], { 'timestamp': '2026-04-03T14:07:38Z', 'table': 'fraud_report', 'partition': { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-02', 'entity_name': 'ZQOR', 'service': 'UNLIMITED', 'music_territory': 'BR', }, 'status': 'new', }, [ call( 'swf-amazon-datapulse-exec', { 'ENVIRONMENT': 'env1', 'REPORT': 'fraud_report', 'CONTEXT_DATE': '2026-04-02', 'PARTITION': json.dumps( { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-02', 'entity_name': 'ZQOR', 'service': 'UNLIMITED', 'music_territory': 'BR', } ), }, ), call( 'swf-amazon-datapulse-exec', { 'ENVIRONMENT': 'env2', 'REPORT': 'fraud_report', 'CONTEXT_DATE': '2026-04-02', 'PARTITION': json.dumps( { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-02', 'entity_name': 'ZQOR', 'service': 'UNLIMITED', 'music_territory': 'BR', } ), }, ), ], [ call( 'amazon_datapulse_trigger.routed', 1, tags=[ 'report_type:fraud_report', 'environment:env1', ], ), call( 'amazon_datapulse_trigger.routed', 1, tags=[ 'report_type:fraud_report', 'environment:env2', ], ), ], id='fraud_report ZQOR - 2 environments', ), pytest.param( ['env1', 'env2'], { 'timestamp': '2026-04-03T14:07:38Z', 'table': 'fraud_report', 'partition': { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-29', 'entity_name': 'Sony', 'service': 'UNLIMITED', 'music_territory': 'US', }, 'status': 'new', }, [ call( 'swf-amazon-datapulse-exec', { 'ENVIRONMENT': 'env1', 'REPORT': 'fraud_report', 'CONTEXT_DATE': '2026-04-29', 'PARTITION': json.dumps( { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-29', 'entity_name': 'Sony', 'service': 'UNLIMITED', 'music_territory': 'US', } ), }, ), call( 'swf-amazon-datapulse-exec', { 'ENVIRONMENT': 'env2', 'REPORT': 'fraud_report', 'CONTEXT_DATE': '2026-04-29', 'PARTITION': json.dumps( { 'cadence': 'daily', 'region': 'na', 'report_date': '2026-04-29', 'entity_name': 'Sony', 'service': 'UNLIMITED', 'music_territory': 'US', } ), }, ), ], [ call( 'amazon_datapulse_trigger.routed', 1, tags=[ 'report_type:fraud_report', 'environment:env1', ], ), call( 'amazon_datapulse_trigger.routed', 1, tags=[ 'report_type:fraud_report', 'environment:env2', ], ), ], id='fraud_report Sony - 2 environments', ), ], ) def test_process_message_success( self, environments, message, expected_jenkins_calls, expected_metrics_calls, monkeypatch, ): monkeypatch.setattr(logic.config, 'JENKINS_ENVIRONMENTS', environments) logic.process_message(body=self.sample_body(message), message_id='msg-1') assert ( self.jenkins_utils_mock.trigger_job.call_args_list == expected_jenkins_calls ) assert self.lambda_metric_mock.call_args_list == expected_metrics_calls @pytest.mark.parametrize( 'environment, report_type, partition, expected', [ pytest.param( 'prod', 'fraud_report', { 'cadence': 'daily', 'region': 'fe', 'report_date': '2026-04-05', 'entity_name': 'Sony', 'service': 'UNLIMITED', 'music_territory': 'JP', }, { 'ENVIRONMENT': 'prod', 'REPORT': 'fraud_report', 'CONTEXT_DATE': '2026-04-05', 'PARTITION': json.dumps( { 'cadence': 'daily', 'region': 'fe', 'report_date': '2026-04-05', 'entity_name': 'Sony', 'service': 'UNLIMITED', 'music_territory': 'JP', } ), }, id='fraud_report Sony JP - prod', ), pytest.param( 'dev', 'daily_play_events', { 'Service': 'FreeTier', 'Region': 'EU', 'Marketplace': '4', 'Territory': 'AT', 'Year': '2026', 'Month': '04', 'Day': '04', }, { 'ENVIRONMENT': 'dev', 'REPORT': 'daily_play_events', 'CONTEXT_DATE': '2026-04-04', 'PARTITION': json.dumps( { 'Service': 'FreeTier', 'Region': 'EU', 'Marketplace': '4', 'Territory': 'AT', 'Year': '2026', 'Month': '04', 'Day': '04', } ), }, id='daily_play_events AT - dev', ), pytest.param( 'dev', 'daily_customer_metadata', { 'Region': 'ROW_FE', 'Marketplace': '111172', 'Territory': '00', 'Year': '2026', 'Month': '04', 'Day': '04', }, { 'ENVIRONMENT': 'dev', 'REPORT': 'daily_customer_metadata', 'CONTEXT_DATE': '2026-04-04', 'PARTITION': json.dumps( { 'Region': 'ROW_FE', 'Marketplace': '111172', 'Territory': '00', 'Year': '2026', 'Month': '04', 'Day': '04', } ), }, id='daily_customer_metadata 00 - dev', ), pytest.param( 'prod', 'daily_customer_metadata', { 'Region': 'ROW_FE', 'Marketplace': '111172', 'Territory': '00', 'Year': '2026', 'Month': '04', 'Day': '05', }, { 'ENVIRONMENT': 'prod', 'REPORT': 'daily_customer_metadata', 'CONTEXT_DATE': '2026-04-05', 'PARTITION': json.dumps( { 'Region': 'ROW_FE', 'Marketplace': '111172', 'Territory': '00', 'Year': '2026', 'Month': '04', 'Day': '05', } ), }, id='daily_customer_metadata 00 - prod', ), pytest.param( 'dev', 'daily_playlist_metadata', { 'Region': 'EU', 'Marketplace': '44551', 'Territory': 'ES', 'Year': '2026', 'Month': '04', 'Day': '04', }, { 'ENVIRONMENT': 'dev', 'REPORT': 'daily_playlist_metadata', 'CONTEXT_DATE': '2026-04-04', 'PARTITION': json.dumps( { 'Region': 'EU', 'Marketplace': '44551', 'Territory': 'ES', 'Year': '2026', 'Month': '04', 'Day': '04', } ), }, id='daily_playlist_metadata ES - dev', ), pytest.param( 'prod', 'daily_playlist_metadata', { 'Region': 'ROW_FE', 'Marketplace': '111172', 'Territory': '00', 'Year': '2026', 'Month': '04', 'Day': '05', }, { 'ENVIRONMENT': 'prod', 'REPORT': 'daily_playlist_metadata', 'CONTEXT_DATE': '2026-04-05', 'PARTITION': json.dumps( { 'Region': 'ROW_FE', 'Marketplace': '111172', 'Territory': '00', 'Year': '2026', 'Month': '04', 'Day': '05', } ), }, id='daily_playlist_metadata 00 - prod', ), ], ) def test_prepare_jenkins_args( self, environment, report_type, partition, expected ): assert ( logic._prepare_jenkins_args(environment, partition, report_type) == expected ) @pytest.mark.parametrize( 'report_type, partition, missing_key', [ pytest.param( 'daily_play_events', { 'Service': 'FreeTier', 'Region': 'EU', 'Marketplace': '4', 'Territory': 'AT', 'Month': '04', 'Day': '04', }, 'Year', id='daily_play_events - missing Year', ), pytest.param( 'daily_play_events', { 'Service': 'FreeTier', 'Region': 'EU', 'Marketplace': '4', 'Territory': 'AT', 'Year': '2026', 'month': '04', 'Day': '04', }, 'Month', id='daily_play_events - month instead of Month', ), pytest.param( 'fraud_report', { 'cadence': 'daily', 'region': 'na', 'entity_name': 'Sony', 'service': 'UNLIMITED', 'music_territory': 'BR', }, 'report_date', id='fraud_report - missing report_date', ), ], ) def test_prepare_jenkins_args_invalid_partition( self, report_type, partition, missing_key ): with pytest.raises(KeyError, match=missing_key): logic._prepare_jenkins_args('dev', partition, report_type) def test_process_message_unknown_report_type(self, sample_message): sample_message['table'] = 'unknown_report' body = self.sample_body(sample_message) with pytest.raises( ValueError, match='Unknown report type: unknown_report' ): logic.process_message(body=body, message_id='msg-1') assert self.lambda_metric_mock.call_args_list == [] assert self.jenkins_utils_mock.call_args_list == [] def test_process_message_non_notification(self): body = { 'Type': 'SubscriptionConfirmation', 'MessageId': '1234abcd-12ab-12ab-12ab-123456abcdef', 'Token': '...', 'TopicArn': 'arn:aws:sns:us-east-1:767398096717:TableUpdateTopic', 'Message': ( 'You have chosen to subscribe to the topic ' 'arn:aws:sns:us-east-1:767398096717:TableUpdateTopic.' ), 'SubscribeURL': 'https://sns.amazonaws.com/?Action=ConfirmSubscription', } logic.process_message(body=body, message_id='msg-1') self.logger_mock.warning.assert_called_once() assert self.jenkins_utils_mock.trigger_job.call_args_list == [] assert self.lambda_metric_mock.call_args_list == []