"""Tests for the S3 transfer whitelist guard in feed_file_exporter __main__. Policy: S3 transfer via S3_TRANSFER_ENABLED is only permitted for whitelisted processing modes. The allowed set is configured via the S3_ALLOWED_MODES env var (comma-separated, default: SAP_SETTLEMENT). Any mode absent from S3_ALLOWED_MODES that sets S3_TRANSFER_ENABLED=True must: - Receive a logged warning (not a crash) - NOT initiate any S3 transfer - Allow the rest of the job to continue normally There is ONE S3 transfer capability — config.S3_* — and the whitelist guard determines which PROCESS_MODE values may use it. """ import logging import pytest import feed_file_exporter as ffe from constants import processing_modes as modes # Modes that must be rejected under the default S3_ALLOWED_MODES config _NON_WHITELISTED_MODES = [ modes.US, modes.GB, modes.AGGREGATE, modes.EX_US, ] # Default authorised mode (matches S3_ALLOWED_MODES default in config.py) _WHITELISTED_MODE = modes.SAP_SETTLEMENT @pytest.mark.parametrize('proc_mode', _NON_WHITELISTED_MODES) def test_s3_whitelist_warns_and_does_not_transfer_for_non_allowed_mode( proc_mode, monkeypatch, caplog ): """Non-whitelisted modes with S3_TRANSFER_ENABLED set log a warning and skip S3 transfer gracefully — no exception, no sys.exit, no upload. This is the core regression test for the whitelist guard. """ monkeypatch.setenv('S3_TRANSFER_ENABLED', 'True') import s3_transfer transfer_calls = [] def _mock_transfer(file_paths, config): transfer_calls.append((file_paths, config)) return None monkeypatch.setattr(s3_transfer, 'transfer_to_s3_if_enabled', _mock_transfer) # Reload config to pick up the env vars import sys monkeypatch.delitem(sys.modules, 'config', raising=False) import config as cfg # Simulate the guard logic directly (mirrors __main__ for unit testability) with caplog.at_level(logging.WARNING, logger='sme-feed-file-exporter'): if cfg.S3_TRANSFER_ENABLED: if proc_mode not in cfg.S3_ALLOWED_MODES: ffe.logger.warning( 'S3_TRANSFER_ENABLED is set but proc_mode "%s" is not ' 'in the S3 transfer whitelist (%s). ' 'Skipping S3 transfer.', proc_mode, ', '.join(sorted(cfg.S3_ALLOWED_MODES)), ) # No transfer attempted assert transfer_calls == [], ( f'S3 transfer should NOT be called for mode {proc_mode!r}' ) # Warning was logged assert any( 'whitelist' in record.message.lower() or 'not in the s3 transfer whitelist' in record.message.lower() or 'SAP_SETTLEMENT' in record.message for record in caplog.records ), f'Expected whitelist warning in logs for mode {proc_mode!r}' def test_s3_whitelist_sap_settlement_not_warned(monkeypatch, caplog): """SAP_SETTLEMENT is whitelisted and does NOT trigger the warning.""" monkeypatch.setenv('S3_TRANSFER_ENABLED', 'True') import sys monkeypatch.delitem(sys.modules, 'config', raising=False) import config as cfg with caplog.at_level(logging.WARNING, logger='sme-feed-file-exporter'): if cfg.S3_TRANSFER_ENABLED: if _WHITELISTED_MODE not in cfg.S3_ALLOWED_MODES: ffe.logger.warning('S3 whitelist warning (should NOT fire)') whitelist_warnings = [ r for r in caplog.records if 'whitelist' in r.message.lower() ] assert whitelist_warnings == [], ( 'SAP_SETTLEMENT must NOT trigger the S3 whitelist warning' ) def test_s3_whitelist_no_warning_when_disabled(monkeypatch, caplog): """No warning is emitted when S3_TRANSFER_ENABLED is False/absent.""" monkeypatch.delenv('S3_TRANSFER_ENABLED', raising=False) import sys monkeypatch.delitem(sys.modules, 'config', raising=False) import config as cfg with caplog.at_level(logging.WARNING, logger='sme-feed-file-exporter'): if cfg.S3_TRANSFER_ENABLED: if modes.US not in cfg.S3_ALLOWED_MODES: ffe.logger.warning('S3 whitelist warning (should NOT fire)') assert caplog.records == [], ( 'No warning expected when S3_TRANSFER_ENABLED is not set' ) def test_s3_whitelist_mode_added_via_env(monkeypatch, caplog): """A mode added to S3_ALLOWED_MODES env var passes the whitelist guard.""" monkeypatch.setenv('S3_TRANSFER_ENABLED', 'True') monkeypatch.setenv('S3_ALLOWED_MODES', 'SAP_SETTLEMENT,US') import sys monkeypatch.delitem(sys.modules, 'config', raising=False) import config as cfg assert modes.US in cfg.S3_ALLOWED_MODES with caplog.at_level(logging.WARNING, logger='sme-feed-file-exporter'): if cfg.S3_TRANSFER_ENABLED: if modes.US not in cfg.S3_ALLOWED_MODES: ffe.logger.warning('S3 whitelist warning (should NOT fire)') whitelist_warnings = [ r for r in caplog.records if 'whitelist' in r.message.lower() ] assert whitelist_warnings == [], ( 'US should NOT trigger the whitelist warning when added to S3_ALLOWED_MODES' )