"""Unit tests for the Snowflake ACK processor connector.""" import math from unittest.mock import MagicMock from unittest.mock import patch from src.common.ack_processor.connectors import snowflake as snowflake_ack # Rows returned by Snowflake cursor as tuples: (sound_recording_id, version_id, sfn_execution_id, timestamp) SAMPLE_TUPLE_ROWS = [ ('SR001', 'USESK1315718', 'exec-001', '2025-07-08T05:54:07.750132Z'), ('SR002', 'USESK1315717', 'exec-002', '2025-07-08T05:58:07.750132Z'), ] SAMPLE_SUCCESS_ACKS = [ {'version_id': 'USESK1315718', 'datetime': '2025-07-08T05:54:07.750132Z'}, ] SAMPLE_ERROR_ACKS = [ { 'version_id': 'USESK1315717', 'datetime': '2025-07-08T05:58:07.750132Z', 'message': 'MetadataMissing', }, ] def _make_cursor(fetchall_return=None): """Return a mock cursor usable as a context manager.""" cursor = MagicMock() cursor.__enter__ = lambda s: s cursor.__exit__ = MagicMock(return_value=False) if fetchall_return is not None: cursor.fetchall.return_value = fetchall_return return cursor def _make_conn(cursor=None): """Return a mock Snowflake connection.""" conn = MagicMock() if cursor is not None: conn.cursor.return_value = cursor return conn # --------------------------------------------------------------------------- # fetch_unacked_delivery_history # --------------------------------------------------------------------------- @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_fetch_unacked_delivery_history_returns_mapped_rows(mock_get_conn): """fetch_unacked_delivery_history maps tuple rows to dicts correctly.""" cursor = _make_cursor(fetchall_return=SAMPLE_TUPLE_ROWS) mock_get_conn.return_value = _make_conn(cursor) result = snowflake_ack.fetch_unacked_delivery_history(service=['TikTok']) assert len(result) == 2 assert result[0] == { 'sound_recording_id': 'SR001', 'version_id': 'USESK1315718', 'sfn_execution_id': 'exec-001', 'datetime': '2025-07-08T05:54:07.750132Z', 'ack': None, } mock_get_conn.return_value.close.assert_called_once() @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_fetch_unacked_delivery_history_returns_empty_on_no_rows(mock_get_conn): """fetch_unacked_delivery_history returns [] when Snowflake has no rows.""" cursor = _make_cursor(fetchall_return=[]) mock_get_conn.return_value = _make_conn(cursor) result = snowflake_ack.fetch_unacked_delivery_history() assert result == [] mock_get_conn.return_value.close.assert_called_once() @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_fetch_unacked_delivery_history_uses_three_month_filter(mock_get_conn): """fetch_unacked_delivery_history applies a 3-month lookback in the SQL.""" cursor = _make_cursor(fetchall_return=[]) mock_get_conn.return_value = _make_conn(cursor) snowflake_ack.fetch_unacked_delivery_history() executed_sql = cursor.execute.call_args[0][0] assert 'dateadd(month, -3, getdate())' in executed_sql @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_fetch_unacked_delivery_history_applies_limit_when_given(mock_get_conn): """fetch_unacked_delivery_history adds a LIMIT clause when limit is provided.""" cursor = _make_cursor(fetchall_return=[]) mock_get_conn.return_value = _make_conn(cursor) snowflake_ack.fetch_unacked_delivery_history(limit=1000) executed_sql = cursor.execute.call_args[0][0] assert 'LIMIT 1000' in executed_sql @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_fetch_unacked_delivery_history_omits_limit_by_default(mock_get_conn): """fetch_unacked_delivery_history adds no LIMIT clause when limit is None.""" cursor = _make_cursor(fetchall_return=[]) mock_get_conn.return_value = _make_conn(cursor) snowflake_ack.fetch_unacked_delivery_history() executed_sql = cursor.execute.call_args[0][0] assert 'LIMIT' not in executed_sql # --------------------------------------------------------------------------- # fetch_awaited_delivery_history # --------------------------------------------------------------------------- @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_fetch_awaited_delivery_history_returns_mapped_rows(mock_get_conn): """fetch_awaited_delivery_history maps tuple rows to dicts with ack='awaited'.""" cursor = _make_cursor(fetchall_return=SAMPLE_TUPLE_ROWS) mock_get_conn.return_value = _make_conn(cursor) result = snowflake_ack.fetch_awaited_delivery_history( service=['TikTok'], filter_awaited_hours=72, ) assert len(result) == 2 assert result[0]['ack'] == 'awaited' assert result[0]['version_id'] == 'USESK1315718' mock_get_conn.return_value.close.assert_called_once() @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_fetch_awaited_delivery_history_returns_empty_on_no_rows(mock_get_conn): """fetch_awaited_delivery_history returns [] when Snowflake has no rows.""" cursor = _make_cursor(fetchall_return=[]) mock_get_conn.return_value = _make_conn(cursor) result = snowflake_ack.fetch_awaited_delivery_history( service=['TikTok'], filter_awaited_hours=72, ) assert result == [] # --------------------------------------------------------------------------- # mark_unacked_as_awaited # --------------------------------------------------------------------------- @patch('src.common.ack_processor.connectors.snowflake.fetch_unacked_delivery_history') @patch('src.common.ack_processor.connectors.snowflake.fetch_awaited_delivery_history') @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_mark_unacked_as_awaited_returns_combined_list(mock_get_conn, mock_fetch_awaited, mock_fetch_unacked): """mark_unacked_as_awaited inserts unacked rows and returns awaited + unacked combined.""" mock_fetch_awaited.return_value = [ { 'sound_recording_id': 'SR003', 'version_id': 'USESK1315720', 'sfn_execution_id': 'exec-003', 'datetime': '2025-07-08T06:00:00.000000Z', 'ack': 'awaited', }, ] mock_fetch_unacked.return_value = [ { 'sound_recording_id': 'SR001', 'version_id': 'USESK1315718', 'sfn_execution_id': 'exec-001', 'datetime': '2025-07-08T05:54:07.750132Z', 'ack': None, }, { 'sound_recording_id': 'SR002', 'version_id': 'USESK1315717', 'sfn_execution_id': 'exec-002', 'datetime': '2025-07-08T05:58:07.750132Z', 'ack': None, }, ] cursor = _make_cursor() conn = _make_conn(cursor) mock_get_conn.return_value = conn result = snowflake_ack.mark_unacked_as_awaited() # 1 awaited + 2 unacked = 3 total assert len(result) == 3 version_ids = [r['version_id'] for r in result] assert 'USESK1315720' in version_ids assert 'USESK1315718' in version_ids assert 'USESK1315717' in version_ids # INSERT should be called once for the 2 unacked rows cursor.execute.assert_called_once() conn.commit.assert_called_once() conn.close.assert_called_once() @patch('src.common.ack_processor.connectors.snowflake.fetch_unacked_delivery_history') @patch('src.common.ack_processor.connectors.snowflake.fetch_awaited_delivery_history') @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_mark_unacked_as_awaited_returns_empty_when_no_rows(mock_get_conn, mock_fetch_awaited, mock_fetch_unacked): """mark_unacked_as_awaited returns [] and skips INSERT when nothing to process.""" mock_fetch_awaited.return_value = [] mock_fetch_unacked.return_value = [] result = snowflake_ack.mark_unacked_as_awaited() assert result == [] mock_get_conn.assert_not_called() @patch('src.common.ack_processor.connectors.snowflake.fetch_unacked_delivery_history') @patch('src.common.ack_processor.connectors.snowflake.fetch_awaited_delivery_history') @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_mark_unacked_as_awaited_batches_inserts_for_large_unacked_list( mock_get_conn, mock_fetch_awaited, mock_fetch_unacked, ): """mark_unacked_as_awaited issues one INSERT and one commit per BATCH_INSERT_SIZE chunk.""" mock_fetch_awaited.return_value = [] mock_fetch_unacked.return_value = [ { 'sound_recording_id': f'SR{i}', 'version_id': f'ID{i}', 'sfn_execution_id': f'exec-{i}', 'datetime': '2025-07-08T05:54:07.750132Z', 'ack': None, } for i in range(snowflake_ack.BATCH_INSERT_SIZE + 1) ] cursor = _make_cursor() conn = _make_conn(cursor) mock_get_conn.return_value = conn snowflake_ack.mark_unacked_as_awaited() expected_batches = math.ceil((snowflake_ack.BATCH_INSERT_SIZE + 1) / snowflake_ack.BATCH_INSERT_SIZE) # Each batch commits independently so a partial run still makes durable progress. assert cursor.execute.call_count == expected_batches assert conn.commit.call_count == expected_batches conn.close.assert_called_once() @patch('src.common.ack_processor.connectors.snowflake.fetch_unacked_delivery_history') @patch('src.common.ack_processor.connectors.snowflake.fetch_awaited_delivery_history') @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_mark_unacked_as_awaited_skips_insert_when_all_awaited(mock_get_conn, mock_fetch_awaited, mock_fetch_unacked): """mark_unacked_as_awaited skips INSERT when unacked list is empty.""" mock_fetch_awaited.return_value = [ { 'sound_recording_id': 'SR003', 'version_id': 'USESK1315720', 'sfn_execution_id': 'exec-003', 'datetime': '2025-07-08T06:00:00.000000Z', 'ack': 'awaited', }, ] mock_fetch_unacked.return_value = [] result = snowflake_ack.mark_unacked_as_awaited() assert len(result) == 1 mock_get_conn.assert_not_called() # --------------------------------------------------------------------------- # resolve_awaited_as_success / resolve_awaited_as_error # --------------------------------------------------------------------------- @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_resolve_awaited_as_success_commits(mock_get_conn): """resolve_awaited_as_success should execute UPDATE and commit.""" cursor = _make_cursor() conn = _make_conn(cursor) mock_get_conn.return_value = conn snowflake_ack.resolve_awaited_as_success(SAMPLE_SUCCESS_ACKS) cursor.execute.assert_called_once() conn.commit.assert_called_once() conn.close.assert_called_once() @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_resolve_awaited_as_error_includes_message(mock_get_conn): """resolve_awaited_as_error should interpolate the error message into the SQL.""" cursor = _make_cursor() conn = _make_conn(cursor) mock_get_conn.return_value = conn snowflake_ack.resolve_awaited_as_error(SAMPLE_ERROR_ACKS) cursor.execute.assert_called_once() executed_sql = cursor.execute.call_args[0][0] assert 'MetadataMissing' in executed_sql conn.commit.assert_called_once() conn.close.assert_called_once() @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_resolve_awaited_as_success_empty_list_skips(mock_get_conn): """resolve_awaited_as_success with empty list should not open a connection.""" snowflake_ack.resolve_awaited_as_success([]) mock_get_conn.assert_not_called() @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_resolve_awaited_batches_large_list(mock_get_conn): """Large ACK lists should produce one execute per batch for success updates.""" cursor = _make_cursor() conn = _make_conn(cursor) mock_get_conn.return_value = conn large_acks = [ {'version_id': f'ID{i}', 'datetime': '2025-07-08T05:54:07.750132Z'} for i in range(snowflake_ack.BATCH_SIZE + 1) ] snowflake_ack.resolve_awaited_as_success(large_acks) expected_batches = math.ceil(len(large_acks) / snowflake_ack.BATCH_SIZE) assert cursor.execute.call_count == expected_batches conn.commit.assert_called_once() conn.close.assert_called_once() # --------------------------------------------------------------------------- # resolve_awaited_as_missing # --------------------------------------------------------------------------- @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_resolve_awaited_as_missing_commits(mock_get_conn): """resolve_awaited_as_missing should execute UPDATE and commit.""" cursor = _make_cursor() conn = _make_conn(cursor) mock_get_conn.return_value = conn snowflake_ack.resolve_awaited_as_missing([{'version_id': 'v1', 'datetime': '2024-01-01 00:00:00'}]) cursor.execute.assert_called_once() conn.commit.assert_called_once() conn.close.assert_called_once() @patch('src.common.ack_processor.connectors.snowflake.snowflake_connector.get_connection') def test_resolve_awaited_as_missing_batches_large_list(mock_get_conn): """Large missing ACK lists should produce one execute per batch.""" cursor = _make_cursor() conn = _make_conn(cursor) mock_get_conn.return_value = conn large_acks = [ {'version_id': f'ID{i}', 'datetime': '2024-01-01 00:00:00'} for i in range(snowflake_ack.BATCH_SIZE + 1) ] snowflake_ack.resolve_awaited_as_missing(large_acks) expected_batches = math.ceil(len(large_acks) / snowflake_ack.BATCH_SIZE) assert cursor.execute.call_count == expected_batches conn.commit.assert_called_once() conn.close.assert_called_once()