"""Tests for Collector Worker Lambda handler.""" import gzip import json import os from io import BytesIO from unittest.mock import MagicMock, patch, call import pytest from handler import ( handler, _deserialize_ddb_item, _refresh_token, _call_spotify_api, _process_fan, _process_fan_batch, _process_file, _normalize_fan_record, _create_kafka_producer, _produce_to_kafka, _delivery_report, TokenRefreshError, SpotifyApiError, SpotifyRateLimitError, ) # --------------------------------------------------------------------------- # Sample data # --------------------------------------------------------------------------- BUCKET = "dev-mymac80" FILE_KEY = "resonance-engine/ddb-export/01771804222395-8dd0c6a7/data/aaa.json.gz" CLIENT_ID = "test-client-id" CLIENT_SECRET = "test-client-secret" TOPIC = "resonance-engine.spotify-data" PRESAVE_RECORD_DDB = { "Item": { "partitionKey": {"S": "group:album12345"}, "sortKey": {"S": "task:spotify-presave:user123"}, "refreshToken": {"S": "refresh-tok-abc"}, "spotifyUserId": {"S": "spotify-user-123"}, } } EMAIL_RECORD_DDB = { "Item": { "partitionKey": {"S": "group:album12345"}, "sortKey": {"S": "task:email-notification:abc"}, "email": {"S": "fan@example.com"}, } } GROUP_RECORD_DDB = { "Item": { "partitionKey": {"S": "group:album12345"}, "sortKey": {"S": "group:album12345"}, "count": {"N": "42"}, } } APPLE_RECORD_DDB = { "Item": { "partitionKey": {"S": "group:album12345"}, "sortKey": {"S": "task:apple-music-presave:xyz"}, "appleToken": {"S": "apple-tok"}, } } TOP_ARTISTS_RESPONSE = { "items": [{"name": "Artist A", "id": "art1"}], "total": 1, } RECENTLY_PLAYED_RESPONSE = { "items": [{"track": {"name": "Song X", "id": "trk1"}}], } DESERIALIZED_PRESAVE = { "partitionKey": "group:album12345", "sortKey": "task:spotify-presave:user123", "refreshToken": "refresh-tok-abc", "spotifyUserId": "spotify-user-123", } def _make_gzip_body(records): """Build a gzip-compressed body from a list of DDB JSON records.""" lines = [json.dumps(r) for r in records] raw = "\n".join(lines).encode("utf-8") buf = BytesIO() with gzip.GzipFile(fileobj=buf, mode="wb") as gz: gz.write(raw) return buf.getvalue() def _mock_s3_response(body_bytes): """Create a mock S3 get_object response. Returns a BytesIO as the Body so gzip.GzipFile can stream from it. """ return {"Body": BytesIO(body_bytes)} def _make_sqs_event(messages): """Build an SQS event with the given message bodies.""" records = [] for i, msg in enumerate(messages): records.append({ "messageId": f"msg-{i}", "body": json.dumps(msg), }) return {"Records": records} def _make_context(remaining_ms=300000): """Create a mock Lambda context.""" ctx = MagicMock() ctx.get_remaining_time_in_millis.return_value = remaining_ms return ctx # --------------------------------------------------------------------------- # TestDeserializeDdbItem # --------------------------------------------------------------------------- class TestDeserializeDdbItem: def test_string_type(self): assert _deserialize_ddb_item({"S": "hello"}) == "hello" def test_number_int(self): assert _deserialize_ddb_item({"N": "42"}) == 42 def test_number_float(self): assert _deserialize_ddb_item({"N": "3.14"}) == 3.14 def test_boolean_true(self): assert _deserialize_ddb_item({"BOOL": True}) is True def test_boolean_false(self): assert _deserialize_ddb_item({"BOOL": False}) is False def test_null(self): assert _deserialize_ddb_item({"NULL": True}) is None def test_list(self): raw = {"L": [{"S": "a"}, {"N": "1"}, {"BOOL": True}]} assert _deserialize_ddb_item(raw) == ["a", 1, True] def test_nested_map(self): raw = {"M": {"name": {"S": "Alice"}, "age": {"N": "30"}}} assert _deserialize_ddb_item(raw) == {"name": "Alice", "age": 30} def test_full_presave_record(self): result = _deserialize_ddb_item(PRESAVE_RECORD_DDB["Item"]) assert result == DESERIALIZED_PRESAVE def test_passthrough_non_dict(self): assert _deserialize_ddb_item("plain") == "plain" assert _deserialize_ddb_item(123) == 123 def test_multi_key_dict_is_not_type_descriptor(self): raw = { "partitionKey": {"S": "pk"}, "sortKey": {"S": "sk"}, } result = _deserialize_ddb_item(raw) assert result == {"partitionKey": "pk", "sortKey": "sk"} # --------------------------------------------------------------------------- # TestNormalizeFanRecord # --------------------------------------------------------------------------- class TestNormalizeFanRecord: def test_backfill_format_passthrough(self): """camelCase keys are returned as-is (with defaults for missing).""" raw = { "spotifyUserId": "user1", "refreshToken": "tok1", "partitionKey": "pk1", "sortKey": "sk1", } result = _normalize_fan_record(raw) assert result == raw def test_snowflake_format_mapped(self): """UPPERCASE keys are mapped to camelCase.""" raw = { "SPOTIFY_USER_ID": "user1", "REFRESH_TOKEN": "tok1", "PARTITION_KEY": "pk1", "SORT_KEY": "sk1", } result = _normalize_fan_record(raw) assert result == { "spotifyUserId": "user1", "refreshToken": "tok1", "partitionKey": "pk1", "sortKey": "sk1", } def test_missing_optional_fields(self): """partitionKey and sortKey default to empty string.""" raw = { "spotifyUserId": "user1", "refreshToken": "tok1", } result = _normalize_fan_record(raw) assert result["partitionKey"] == "" assert result["sortKey"] == "" assert result["spotifyUserId"] == "user1" assert result["refreshToken"] == "tok1" # --------------------------------------------------------------------------- # TestRefreshToken # --------------------------------------------------------------------------- class TestRefreshToken: @patch("handler.requests.post") def test_success(self, mock_post): mock_resp = MagicMock() mock_resp.status_code = 200 mock_resp.json.return_value = { "access_token": "new-access-tok", } mock_resp.raise_for_status.return_value = None mock_post.return_value = mock_resp access, new_refresh = _refresh_token( "old-refresh", CLIENT_ID, CLIENT_SECRET ) assert access == "new-access-tok" assert new_refresh is None mock_post.assert_called_once() call_kwargs = mock_post.call_args assert "accounts.spotify.com/api/token" in call_kwargs[0][0] assert call_kwargs[1]["data"]["grant_type"] == "refresh_token" @patch("handler.requests.post") def test_success_with_rotated_refresh_token(self, mock_post): mock_resp = MagicMock() mock_resp.json.return_value = { "access_token": "new-access", "refresh_token": "rotated-refresh", } mock_resp.raise_for_status.return_value = None mock_post.return_value = mock_resp access, new_refresh = _refresh_token( "old-refresh", CLIENT_ID, CLIENT_SECRET ) assert access == "new-access" assert new_refresh == "rotated-refresh" @patch("handler.requests.post") def test_http_error_with_known_error_type(self, mock_post): import requests mock_resp = MagicMock() mock_resp.status_code = 400 mock_resp.json.return_value = { "error": "invalid_grant", "error_description": "Refresh token revoked", } mock_resp.raise_for_status.side_effect = requests.HTTPError( response=mock_resp ) mock_post.return_value = mock_resp with pytest.raises(TokenRefreshError, match="Refresh token revoked"): _refresh_token("bad-token", CLIENT_ID, CLIENT_SECRET) @patch("handler.requests.post") def test_http_error_with_unknown_error_type_sanitized(self, mock_post): import requests mock_resp = MagicMock() mock_resp.status_code = 400 mock_resp.json.return_value = { "error": "unexpected_error_type", "error_description": "sensitive info here", } mock_resp.raise_for_status.side_effect = requests.HTTPError( response=mock_resp ) mock_post.return_value = mock_resp with pytest.raises(TokenRefreshError, match="HTTP 400"): _refresh_token("bad-token", CLIENT_ID, CLIENT_SECRET) @patch("handler.requests.post") def test_network_error_raises_token_refresh_error(self, mock_post): import requests mock_post.side_effect = requests.ConnectionError("DNS failure") with pytest.raises(TokenRefreshError, match="request failed"): _refresh_token("tok", CLIENT_ID, CLIENT_SECRET) @patch("handler.requests.post") def test_basic_auth_header_is_correct(self, mock_post): import base64 mock_resp = MagicMock() mock_resp.json.return_value = {"access_token": "tok"} mock_resp.raise_for_status.return_value = None mock_post.return_value = mock_resp _refresh_token("refresh", CLIENT_ID, CLIENT_SECRET) headers = mock_post.call_args[1]["headers"] expected_b64 = base64.b64encode( f"{CLIENT_ID}:{CLIENT_SECRET}".encode("ascii") ).decode("ascii") assert headers["Authorization"] == f"Basic {expected_b64}" # --------------------------------------------------------------------------- # TestCallSpotifyApi # --------------------------------------------------------------------------- class TestCallSpotifyApi: @patch("handler.requests.get") def test_success(self, mock_get): mock_resp = MagicMock() mock_resp.status_code = 200 mock_resp.json.return_value = TOP_ARTISTS_RESPONSE mock_resp.raise_for_status.return_value = None mock_get.return_value = mock_resp result = _call_spotify_api("access-tok", "/me/top/artists") assert result == TOP_ARTISTS_RESPONSE mock_get.assert_called_once() call_args = mock_get.call_args assert "api.spotify.com/v1/me/top/artists" in call_args[0][0] assert call_args[1]["headers"]["Authorization"] == "Bearer access-tok" @patch("handler.time.sleep") @patch("handler.requests.get") def test_rate_limit_then_success(self, mock_get, mock_sleep): rate_resp = MagicMock() rate_resp.status_code = 429 rate_resp.headers = {"Retry-After": "2"} ok_resp = MagicMock() ok_resp.status_code = 200 ok_resp.json.return_value = {"items": []} ok_resp.raise_for_status.return_value = None mock_get.side_effect = [rate_resp, ok_resp] result = _call_spotify_api("tok", "/me/top/artists") assert result == {"items": []} assert mock_get.call_count == 2 mock_sleep.assert_called_once() sleep_time = mock_sleep.call_args[0][0] # attempt 0: min(2 * 2^0, 30) + jitter = 2 + [0,1) assert 2.0 <= sleep_time <= 3.0 @patch("handler.time.sleep") @patch("handler.requests.get") def test_rate_limit_exhausted_after_retries(self, mock_get, mock_sleep): """All retry attempts return 429 — should exhaust after 4 calls.""" rate_resp = MagicMock() rate_resp.status_code = 429 rate_resp.headers = {"Retry-After": "1"} mock_get.return_value = rate_resp with pytest.raises(SpotifyApiError, match="Rate limit exhausted"): _call_spotify_api("tok", "/me/top/artists") # 1 initial + 3 retries = 4 calls total assert mock_get.call_count == 4 assert mock_sleep.call_count == 3 @patch("handler.time.sleep") @patch("handler.requests.get") def test_rate_limit_exponential_backoff(self, mock_get, mock_sleep): """Backoff should increase exponentially across retries.""" rate_resp = MagicMock() rate_resp.status_code = 429 rate_resp.headers = {"Retry-After": "2"} ok_resp = MagicMock() ok_resp.status_code = 200 ok_resp.json.return_value = {"items": []} ok_resp.raise_for_status.return_value = None # 429 twice, then success on third call mock_get.side_effect = [rate_resp, rate_resp, ok_resp] result = _call_spotify_api("tok", "/me/top/artists") assert result == {"items": []} assert mock_sleep.call_count == 2 # attempt 0: min(2 * 2^0, 30) + jitter = 2 + [0,1) first_sleep = mock_sleep.call_args_list[0][0][0] assert 2.0 <= first_sleep <= 3.0 # attempt 1: min(2 * 2^1, 30) + jitter = 4 + [0,1) second_sleep = mock_sleep.call_args_list[1][0][0] assert 4.0 <= second_sleep <= 5.0 @patch("handler.requests.get") def test_unauthorized_raises(self, mock_get): mock_resp = MagicMock() mock_resp.status_code = 401 mock_get.return_value = mock_resp with pytest.raises(SpotifyApiError, match="Unauthorized"): _call_spotify_api("bad-tok", "/me/top/artists") @patch("handler.requests.get") def test_server_error(self, mock_get): import requests mock_resp = MagicMock() mock_resp.status_code = 500 mock_resp.raise_for_status.side_effect = requests.HTTPError( "500 Server Error" ) mock_get.return_value = mock_resp with pytest.raises(SpotifyApiError, match="Spotify API error"): _call_spotify_api("tok", "/me/top/artists") @patch("handler.requests.get") def test_network_timeout(self, mock_get): import requests mock_get.side_effect = requests.Timeout("Connection timed out") with pytest.raises(SpotifyApiError, match="Connection timed out"): _call_spotify_api("tok", "/me/top/artists") @patch("handler.time.sleep") @patch("handler.requests.get") def test_rate_limit_non_numeric_retry_after(self, mock_get, mock_sleep): """Non-numeric Retry-After header should default to 5.""" rate_resp = MagicMock() rate_resp.status_code = 429 rate_resp.headers = {"Retry-After": "Wed, 21 Oct 2025 07:28:00 GMT"} ok_resp = MagicMock() ok_resp.status_code = 200 ok_resp.json.return_value = {"items": []} ok_resp.raise_for_status.return_value = None mock_get.side_effect = [rate_resp, ok_resp] result = _call_spotify_api("tok", "/me/top/artists") assert result == {"items": []} sleep_time = mock_sleep.call_args[0][0] assert 5.0 <= sleep_time <= 6.0 # default 5 + jitter(0-1) @patch("handler.requests.get") def test_passes_params(self, mock_get): mock_resp = MagicMock() mock_resp.status_code = 200 mock_resp.json.return_value = {} mock_resp.raise_for_status.return_value = None mock_get.return_value = mock_resp _call_spotify_api("tok", "/me/top/artists", {"limit": 50}) assert mock_get.call_args[1]["params"] == {"limit": 50} @patch("handler.time.sleep") @patch("handler.requests.get") def test_rate_limit_above_threshold_raises_immediately( self, mock_get, mock_sleep ): """Retry-After > 10s raises SpotifyRateLimitError without sleeping.""" rate_resp = MagicMock() rate_resp.status_code = 429 rate_resp.headers = {"Retry-After": "15"} mock_get.return_value = rate_resp with pytest.raises(SpotifyRateLimitError, match="exceeds threshold"): _call_spotify_api("tok", "/me/top/artists") mock_sleep.assert_not_called() # Only one request — raised immediately on first 429 assert mock_get.call_count == 1 @patch("handler.time.sleep") @patch("handler.requests.get") def test_rate_limit_below_threshold_retries_normally( self, mock_get, mock_sleep ): """Retry-After <= 10s sleeps and retries as before.""" rate_resp = MagicMock() rate_resp.status_code = 429 rate_resp.headers = {"Retry-After": "5"} ok_resp = MagicMock() ok_resp.status_code = 200 ok_resp.json.return_value = {"items": []} ok_resp.raise_for_status.return_value = None mock_get.side_effect = [rate_resp, ok_resp] result = _call_spotify_api("tok", "/me/top/artists") assert result == {"items": []} assert mock_get.call_count == 2 mock_sleep.assert_called_once() sleep_time = mock_sleep.call_args[0][0] assert 5.0 <= sleep_time <= 6.0 # --------------------------------------------------------------------------- # TestProduceToKafka # --------------------------------------------------------------------------- class TestProduceToKafka: def test_produce_serializes_json(self): mock_producer = MagicMock() message = {"spotify_user_id": "user1", "data": "test"} _produce_to_kafka(mock_producer, TOPIC, "user1", message) mock_producer.produce.assert_called_once_with( TOPIC, value=json.dumps(message).encode("utf-8"), key=b"user1", on_delivery=_delivery_report, ) mock_producer.poll.assert_called_once_with(0) def test_delivery_report_logs_error(self): mock_msg = MagicMock() mock_msg.key.return_value = b"user1" mock_error = MagicMock() mock_error.__str__ = MagicMock(return_value="broker down") # Should not raise — just logs _delivery_report(mock_error, mock_msg) def test_delivery_report_no_error(self): mock_msg = MagicMock() mock_msg.key.return_value = b"user1" # Should not raise on success _delivery_report(None, mock_msg) # --------------------------------------------------------------------------- # TestCreateKafkaProducer # --------------------------------------------------------------------------- class TestCreateKafkaProducer: @patch.dict(os.environ, {"KAFKA_BOOTSTRAP_SERVERS": ""}) def test_returns_none_when_not_configured(self): result = _create_kafka_producer() assert result is None @patch.dict(os.environ, {}, clear=False) def test_returns_none_when_env_missing(self): env = { k: v for k, v in os.environ.items() if k != "KAFKA_BOOTSTRAP_SERVERS" } with patch.dict(os.environ, env, clear=True): result = _create_kafka_producer() assert result is None @patch.dict(os.environ, { "KAFKA_BOOTSTRAP_SERVERS": "broker:9092", "KAFKA_SECURITY_PROTOCOL": "SSL", }) def test_creates_producer_when_configured(self): import types mock_module = types.ModuleType("confluent_kafka") mock_producer_instance = MagicMock() mock_module.Producer = MagicMock(return_value=mock_producer_instance) with patch.dict("sys.modules", {"confluent_kafka": mock_module}): result = _create_kafka_producer() assert result is mock_producer_instance mock_module.Producer.assert_called_once_with({ "bootstrap.servers": "broker:9092", "security.protocol": "SSL", }) @patch.dict(os.environ, { "KAFKA_BOOTSTRAP_SERVERS": "broker:9092", }) def test_returns_none_when_import_fails(self): with patch.dict("sys.modules", {"confluent_kafka": None}): result = _create_kafka_producer() assert result is None # --------------------------------------------------------------------------- # TestProcessFan # --------------------------------------------------------------------------- class TestProcessFan: @patch("handler._produce_to_kafka") @patch("handler._call_spotify_api") @patch("handler._refresh_token") def test_success_with_kafka( self, mock_refresh, mock_api, mock_produce ): mock_refresh.return_value = ("access-tok", None) mock_api.side_effect = [ TOP_ARTISTS_RESPONSE, RECENTLY_PLAYED_RESPONSE, ] mock_producer = MagicMock() result = _process_fan( DESERIALIZED_PRESAVE, CLIENT_ID, CLIENT_SECRET, mock_producer, TOPIC, ) assert result["success"] is True mock_refresh.assert_called_once_with( "refresh-tok-abc", CLIENT_ID, CLIENT_SECRET ) assert mock_api.call_count == 2 mock_produce.assert_called_once() msg = mock_produce.call_args[0][3] assert msg["spotify_user_id"] == "spotify-user-123" assert msg["refresh_token"] == "refresh-tok-abc" assert msg["new_refresh_token"] is None assert "top_artists" in msg["endpoints"] assert "recently_played" in msg["endpoints"] assert "collected_at" in msg @patch("handler._call_spotify_api") @patch("handler._refresh_token") def test_success_without_kafka(self, mock_refresh, mock_api): mock_refresh.return_value = ("access-tok", None) mock_api.side_effect = [ TOP_ARTISTS_RESPONSE, RECENTLY_PLAYED_RESPONSE, ] result = _process_fan( DESERIALIZED_PRESAVE, CLIENT_ID, CLIENT_SECRET, None, TOPIC, ) assert result["success"] is True def test_no_refresh_token(self): fan = {"spotifyUserId": "user1"} result = _process_fan(fan, CLIENT_ID, CLIENT_SECRET, None, TOPIC) assert result["success"] is False @patch("handler._refresh_token") def test_token_refresh_failure(self, mock_refresh): mock_refresh.side_effect = TokenRefreshError("revoked") result = _process_fan( DESERIALIZED_PRESAVE, CLIENT_ID, CLIENT_SECRET, None, TOPIC ) assert result["success"] is False @patch("handler._call_spotify_api") @patch("handler._refresh_token") def test_spotify_api_failure(self, mock_refresh, mock_api): mock_refresh.return_value = ("tok", None) mock_api.side_effect = SpotifyApiError("429 exhausted") result = _process_fan( DESERIALIZED_PRESAVE, CLIENT_ID, CLIENT_SECRET, None, TOPIC ) assert result["success"] is False @patch("handler._refresh_token") def test_unexpected_error(self, mock_refresh): mock_refresh.side_effect = RuntimeError("boom") result = _process_fan( DESERIALIZED_PRESAVE, CLIENT_ID, CLIENT_SECRET, None, TOPIC ) assert result["success"] is False @patch("handler._produce_to_kafka") @patch("handler._call_spotify_api") @patch("handler._refresh_token") def test_rotated_token_includes_new_token_in_message( self, mock_refresh, mock_api, mock_produce ): mock_refresh.return_value = ("access-tok", "new-rotated-refresh") mock_api.side_effect = [ TOP_ARTISTS_RESPONSE, RECENTLY_PLAYED_RESPONSE, ] mock_producer = MagicMock() result = _process_fan( DESERIALIZED_PRESAVE, CLIENT_ID, CLIENT_SECRET, mock_producer, TOPIC, ) assert result["success"] is True mock_produce.assert_called_once() msg = mock_produce.call_args[0][3] assert msg["refresh_token"] == "refresh-tok-abc" assert msg["new_refresh_token"] == "new-rotated-refresh" @patch("handler._produce_to_kafka") @patch("handler._call_spotify_api") @patch("handler._refresh_token") def test_same_refresh_token_returned_sets_new_to_none( self, mock_refresh, mock_api, mock_produce ): """When Spotify returns the same refresh token, new_refresh_token is None.""" mock_refresh.return_value = ("access-tok", "refresh-tok-abc") mock_api.side_effect = [ TOP_ARTISTS_RESPONSE, RECENTLY_PLAYED_RESPONSE, ] mock_producer = MagicMock() result = _process_fan( DESERIALIZED_PRESAVE, CLIENT_ID, CLIENT_SECRET, mock_producer, TOPIC, ) assert result["success"] is True mock_produce.assert_called_once() msg = mock_produce.call_args[0][3] assert msg["refresh_token"] == "refresh-tok-abc" assert msg["new_refresh_token"] is None @patch("handler._call_spotify_api") @patch("handler._refresh_token") def test_rate_limit_error_propagates(self, mock_refresh, mock_api): """SpotifyRateLimitError is NOT caught — propagates to caller.""" mock_refresh.return_value = ("tok", None) mock_api.side_effect = SpotifyRateLimitError("severe throttle") with pytest.raises(SpotifyRateLimitError): _process_fan( DESERIALIZED_PRESAVE, CLIENT_ID, CLIENT_SECRET, None, TOPIC, ) # --------------------------------------------------------------------------- # TestProcessFanBatch # --------------------------------------------------------------------------- class TestProcessFanBatch: @patch("handler._process_fan", return_value={"success": True, "timings": {}}) def test_processes_all_fans(self, mock_fan): """All fans succeed.""" fans = [ {"spotifyUserId": "u1", "refreshToken": "t1", "partitionKey": "pk1", "sortKey": "sk1"}, {"spotifyUserId": "u2", "refreshToken": "t2", "partitionKey": "pk2", "sortKey": "sk2"}, {"spotifyUserId": "u3", "refreshToken": "t3", "partitionKey": "pk3", "sortKey": "sk3"}, ] ctx = _make_context() result = _process_fan_batch( fans, CLIENT_ID, CLIENT_SECRET, None, TOPIC, ctx ) assert result == { "processed": 3, "errors": 0, "rate_limited": 0, "timed_out": False, } assert mock_fan.call_count == 3 @patch("handler._process_fan") def test_mixed_success_and_failure(self, mock_fan): """Some fans fail, others succeed.""" mock_fan.side_effect = [ {"success": True, "timings": {}}, {"success": False, "timings": {}}, {"success": True, "timings": {}}, ] fans = [ {"spotifyUserId": "u1", "refreshToken": "t1"}, {"spotifyUserId": "u2", "refreshToken": "t2"}, {"spotifyUserId": "u3", "refreshToken": "t3"}, ] ctx = _make_context() result = _process_fan_batch( fans, CLIENT_ID, CLIENT_SECRET, None, TOPIC, ctx ) assert result["processed"] == 2 assert result["errors"] == 1 assert result["rate_limited"] == 0 @patch("handler._process_fan") def test_circuit_breaker_trips(self, mock_fan): """3 consecutive SpotifyRateLimitError raises the exception.""" mock_fan.side_effect = SpotifyRateLimitError("severe") fans = [ {"spotifyUserId": f"u{i}", "refreshToken": f"t{i}"} for i in range(5) ] ctx = _make_context() with pytest.raises(SpotifyRateLimitError): _process_fan_batch( fans, CLIENT_ID, CLIENT_SECRET, None, TOPIC, ctx ) # Should have processed exactly 3 fans before tripping assert mock_fan.call_count == 3 @patch("handler._process_fan") def test_circuit_breaker_resets_on_success(self, mock_fan): """Rate limit, then success, then rate limit does not trip.""" mock_fan.side_effect = [ SpotifyRateLimitError("throttle"), # consecutive=1 {"success": True, "timings": {}}, # reset to 0 SpotifyRateLimitError("throttle"), # consecutive=1 {"success": True, "timings": {}}, # reset to 0 ] fans = [ {"spotifyUserId": f"u{i}", "refreshToken": f"t{i}"} for i in range(4) ] ctx = _make_context() result = _process_fan_batch( fans, CLIENT_ID, CLIENT_SECRET, None, TOPIC, ctx ) assert result["processed"] == 2 assert result["rate_limited"] == 2 assert result["errors"] == 0 @patch("handler._process_fan", return_value={"success": True, "timings": {}}) def test_timeout_detection(self, mock_fan): """Stops processing when remaining time is low.""" fans = [ {"spotifyUserId": "u1", "refreshToken": "t1"}, {"spotifyUserId": "u2", "refreshToken": "t2"}, {"spotifyUserId": "u3", "refreshToken": "t3"}, ] ctx = MagicMock() # First call: plenty of time; second+third calls: below threshold # (second call is the check, third is in the log message) ctx.get_remaining_time_in_millis.side_effect = [300000, 50000, 50000] result = _process_fan_batch( fans, CLIENT_ID, CLIENT_SECRET, None, TOPIC, ctx ) assert result["timed_out"] is True assert result["processed"] == 1 assert mock_fan.call_count == 1 @patch("handler._process_fan", return_value={"success": True, "timings": {}}) @patch("handler._normalize_fan_record") def test_normalizes_snowflake_format( self, mock_normalize, mock_fan ): """Verifies _normalize_fan_record is called for each fan.""" mock_normalize.return_value = { "spotifyUserId": "u1", "refreshToken": "t1", "partitionKey": "pk1", "sortKey": "sk1", } fans = [ {"SPOTIFY_USER_ID": "u1", "REFRESH_TOKEN": "t1", "PARTITION_KEY": "pk1", "SORT_KEY": "sk1"}, ] ctx = _make_context() result = _process_fan_batch( fans, CLIENT_ID, CLIENT_SECRET, None, TOPIC, ctx ) mock_normalize.assert_called_once_with(fans[0]) assert result["processed"] == 1 # --------------------------------------------------------------------------- # TestProcessFile # --------------------------------------------------------------------------- class TestProcessFile: def _setup_s3(self, mock_s3, records): body = _make_gzip_body(records) mock_s3.get_object.return_value = _mock_s3_response(body) @patch("handler._process_fan", return_value={"success": True, "timings": {}}) def test_processes_presave_records(self, mock_fan): mock_s3 = MagicMock() self._setup_s3(mock_s3, [PRESAVE_RECORD_DDB]) ctx = _make_context() result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) assert result["processed"] == 1 assert result["errors"] == 0 mock_fan.assert_called_once() fan_arg = mock_fan.call_args[0][0] assert fan_arg["spotifyUserId"] == "spotify-user-123" @patch("handler._process_fan") def test_skips_non_presave_records(self, mock_fan): mock_s3 = MagicMock() self._setup_s3(mock_s3, [ EMAIL_RECORD_DDB, GROUP_RECORD_DDB, APPLE_RECORD_DDB, ]) ctx = _make_context() result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) assert result["skipped"] == 3 assert result["processed"] == 0 mock_fan.assert_not_called() @patch("handler._process_fan", return_value={"success": True, "timings": {}}) def test_handles_mixed_record_types(self, mock_fan): mock_s3 = MagicMock() self._setup_s3(mock_s3, [ EMAIL_RECORD_DDB, PRESAVE_RECORD_DDB, GROUP_RECORD_DDB, PRESAVE_RECORD_DDB, APPLE_RECORD_DDB, ]) ctx = _make_context() result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) assert result["processed"] == 2 assert result["skipped"] == 3 assert mock_fan.call_count == 2 @patch("handler._process_fan", return_value={"success": True, "timings": {}}) def test_timeout_detection(self, mock_fan): mock_s3 = MagicMock() # Two presave records, but timeout after first self._setup_s3(mock_s3, [PRESAVE_RECORD_DDB, PRESAVE_RECORD_DDB]) ctx = MagicMock() # First call: plenty of time; second call: below 90s threshold ctx.get_remaining_time_in_millis.side_effect = [300000, 50000] result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) assert result["timed_out"] is True assert result["processed"] == 1 assert mock_fan.call_count == 1 @patch("handler._process_fan", return_value={"success": True, "timings": {}}) def test_timeout_boundary_exact_buffer_does_not_stop(self, mock_fan): """remaining == buffer (90000) should NOT stop (strict less-than).""" mock_s3 = MagicMock() self._setup_s3(mock_s3, [PRESAVE_RECORD_DDB, PRESAVE_RECORD_DDB]) ctx = MagicMock() # Both calls return exactly the buffer value — should NOT timeout ctx.get_remaining_time_in_millis.return_value = 90000 result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) assert result["timed_out"] is False assert result["processed"] == 2 assert mock_fan.call_count == 2 @patch("handler._process_fan") def test_invalid_json_line(self, mock_fan): # Build gzip with one bad line and one good record lines = "not-json\n" + json.dumps(PRESAVE_RECORD_DDB) buf = BytesIO() with gzip.GzipFile(fileobj=buf, mode="wb") as gz: gz.write(lines.encode("utf-8")) mock_s3 = MagicMock() mock_s3.get_object.return_value = _mock_s3_response(buf.getvalue()) mock_fan.return_value = {"success": True, "timings": {}} ctx = _make_context() result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) assert result["skipped"] == 1 # bad JSON line assert result["processed"] == 1 mock_fan.assert_called_once() @patch("handler._process_fan") def test_empty_file(self, mock_fan): mock_s3 = MagicMock() self._setup_s3(mock_s3, []) ctx = _make_context() result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) assert result == { "processed": 0, "skipped": 0, "errors": 0, "timed_out": False, } mock_fan.assert_not_called() @patch("handler._process_fan") def test_fan_error_doesnt_fail_file(self, mock_fan): mock_s3 = MagicMock() self._setup_s3(mock_s3, [ PRESAVE_RECORD_DDB, PRESAVE_RECORD_DDB, PRESAVE_RECORD_DDB, ]) # First fails, second succeeds, third fails mock_fan.side_effect = [ {"success": False, "timings": {}}, {"success": True, "timings": {}}, {"success": False, "timings": {}}, ] ctx = _make_context() result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) assert result["processed"] == 1 assert result["errors"] == 2 assert result["timed_out"] is False @patch("handler._process_fan", return_value={"success": True, "timings": {}}) def test_no_context_skips_timeout_check(self, mock_fan): mock_s3 = MagicMock() self._setup_s3(mock_s3, [PRESAVE_RECORD_DDB]) result = _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, None, ) assert result["processed"] == 1 assert result["timed_out"] is False def test_s3_get_object_failure_propagates(self): mock_s3 = MagicMock() mock_s3.get_object.side_effect = Exception("NoSuchKey") ctx = _make_context() with pytest.raises(Exception, match="NoSuchKey"): _process_file( mock_s3, BUCKET, FILE_KEY, None, TOPIC, CLIENT_ID, CLIENT_SECRET, ctx, ) # --------------------------------------------------------------------------- # TestHandler # --------------------------------------------------------------------------- ENV_VARS = { "SPOTIFY_CLIENT_ID": CLIENT_ID, "SPOTIFY_CLIENT_SECRET": CLIENT_SECRET, "KAFKA_TOPIC": TOPIC, "ENVIRONMENT": "dev", "KAFKA_BOOTSTRAP_SERVERS": "", } class TestHandler: @patch.dict(os.environ, ENV_VARS) @patch("handler._process_file") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_processes_sqs_records( self, mock_boto3, mock_kafka, mock_process ): mock_process.return_value = { "processed": 10, "skipped": 5, "errors": 0, "timed_out": False, } event = _make_sqs_event([ {"bucket": BUCKET, "key": "data/f1.json.gz", "item_count": 15}, ]) result = handler(event, _make_context()) assert result == {"batchItemFailures": []} mock_process.assert_called_once() @patch.dict(os.environ, ENV_VARS) @patch("handler._process_file") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_partial_batch_failure( self, mock_boto3, mock_kafka, mock_process ): mock_process.side_effect = [ { "processed": 10, "skipped": 0, "errors": 0, "timed_out": False, }, Exception("S3 error"), ] event = _make_sqs_event([ {"bucket": BUCKET, "key": "data/f1.json.gz", "item_count": 10}, {"bucket": BUCKET, "key": "data/f2.json.gz", "item_count": 10}, ]) result = handler(event, _make_context()) assert len(result["batchItemFailures"]) == 1 assert result["batchItemFailures"][0]["itemIdentifier"] == "msg-1" @patch.dict(os.environ, ENV_VARS) @patch("handler._process_file") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_all_records_fail( self, mock_boto3, mock_kafka, mock_process ): mock_process.side_effect = Exception("boom") event = _make_sqs_event([ {"bucket": BUCKET, "key": "data/f1.json.gz", "item_count": 10}, {"bucket": BUCKET, "key": "data/f2.json.gz", "item_count": 10}, ]) result = handler(event, _make_context()) assert len(result["batchItemFailures"]) == 2 @patch.dict(os.environ, ENV_VARS) @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_empty_records(self, mock_boto3, mock_kafka): result = handler({"Records": []}, _make_context()) assert result == {"batchItemFailures": []} @patch.dict(os.environ, ENV_VARS) @patch("handler._process_file") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_timeout_reported_as_failure( self, mock_boto3, mock_kafka, mock_process ): mock_process.return_value = { "processed": 5, "skipped": 0, "errors": 0, "timed_out": True, } event = _make_sqs_event([ {"bucket": BUCKET, "key": "data/f1.json.gz", "item_count": 100}, ]) result = handler(event, _make_context()) assert len(result["batchItemFailures"]) == 1 assert result["batchItemFailures"][0]["itemIdentifier"] == "msg-0" @patch.dict(os.environ, ENV_VARS) @patch("handler._process_file") @patch("handler._create_kafka_producer") @patch("handler.boto3") def test_kafka_flush_called( self, mock_boto3, mock_kafka_create, mock_process ): mock_producer = MagicMock() mock_producer.flush.return_value = 0 mock_kafka_create.return_value = mock_producer mock_process.return_value = { "processed": 1, "skipped": 0, "errors": 0, "timed_out": False, } event = _make_sqs_event([ {"bucket": BUCKET, "key": "data/f1.json.gz", "item_count": 1}, ]) handler(event, _make_context()) mock_producer.flush.assert_called_once() @patch.dict(os.environ, ENV_VARS) @patch("handler._process_file") @patch("handler._create_kafka_producer") @patch("handler.boto3") def test_kafka_flush_undelivered_fails_batch( self, mock_boto3, mock_kafka_create, mock_process ): mock_producer = MagicMock() mock_producer.flush.return_value = 3 mock_kafka_create.return_value = mock_producer mock_process.return_value = { "processed": 1, "skipped": 0, "errors": 0, "timed_out": False, } event = _make_sqs_event([ {"bucket": BUCKET, "key": "data/f1.json.gz", "item_count": 1}, ]) result = handler(event, _make_context()) # Undelivered messages should cause all SQS messages to fail assert len(result["batchItemFailures"]) == 1 assert result["batchItemFailures"][0]["itemIdentifier"] == "msg-0" mock_producer.flush.assert_called_once() @patch.dict(os.environ, ENV_VARS) @patch("handler._process_file") @patch("handler._create_kafka_producer") @patch("handler.boto3") def test_kafka_flush_undelivered_deduplicates_failures( self, mock_boto3, mock_kafka_create, mock_process ): """Messages already in batchItemFailures are not duplicated.""" mock_producer = MagicMock() mock_producer.flush.return_value = 2 mock_kafka_create.return_value = mock_producer mock_process.side_effect = [ Exception("S3 error"), # msg-0 already fails { "processed": 1, "skipped": 0, "errors": 0, "timed_out": False, }, ] event = _make_sqs_event([ {"bucket": BUCKET, "key": "data/f1.json.gz", "item_count": 1}, {"bucket": BUCKET, "key": "data/f2.json.gz", "item_count": 1}, ]) result = handler(event, _make_context()) # Both should fail, but msg-0 should not be duplicated ids = [f["itemIdentifier"] for f in result["batchItemFailures"]] assert ids.count("msg-0") == 1 assert ids.count("msg-1") == 1 assert len(ids) == 2 @patch.dict(os.environ, ENV_VARS) @patch("handler._process_file") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_multiple_sqs_records_all_succeed( self, mock_boto3, mock_kafka, mock_process ): mock_process.return_value = { "processed": 10, "skipped": 0, "errors": 0, "timed_out": False, } event = _make_sqs_event([ {"bucket": BUCKET, "key": "data/f1.json.gz", "item_count": 10}, {"bucket": BUCKET, "key": "data/f2.json.gz", "item_count": 10}, {"bucket": BUCKET, "key": "data/f3.json.gz", "item_count": 10}, ]) result = handler(event, _make_context()) assert result == {"batchItemFailures": []} assert mock_process.call_count == 3 @patch.dict(os.environ, ENV_VARS) @patch("handler._process_fan_batch") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_fan_batch_message( self, mock_boto3, mock_kafka, mock_batch ): """Handler processes fan batch messages correctly.""" mock_batch.return_value = { "processed": 2, "errors": 0, "rate_limited": 0, "timed_out": False, } event = _make_sqs_event([{ "source": "backfill", "fans": [ {"spotifyUserId": "user1", "refreshToken": "tok1", "partitionKey": "pk1", "sortKey": "sk1"}, {"spotifyUserId": "user2", "refreshToken": "tok2", "partitionKey": "pk2", "sortKey": "sk2"}, ], "fan_count": 2, }]) result = handler(event, _make_context()) assert result == {"batchItemFailures": []} mock_batch.assert_called_once() # Verify the fans list was passed through call_args = mock_batch.call_args assert len(call_args[0][0]) == 2 # fans list assert call_args[0][0][0]["spotifyUserId"] == "user1" @patch.dict(os.environ, ENV_VARS) @patch("handler._process_fan_batch") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_circuit_breaker_fails_sqs_message( self, mock_boto3, mock_kafka, mock_batch ): """SpotifyRateLimitError causes batch item failure.""" mock_batch.side_effect = SpotifyRateLimitError("severe") event = _make_sqs_event([{ "source": "backfill", "fans": [ {"spotifyUserId": "user1", "refreshToken": "tok1", "partitionKey": "pk1", "sortKey": "sk1"}, ], "fan_count": 1, }]) result = handler(event, _make_context()) assert len(result["batchItemFailures"]) == 1 assert result["batchItemFailures"][0]["itemIdentifier"] == "msg-0" @patch.dict(os.environ, ENV_VARS) @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_unknown_message_format_skipped( self, mock_boto3, mock_kafka ): """Message with neither 'fans' nor 'bucket' is skipped.""" event = _make_sqs_event([ {"something": "unexpected", "data": 123}, ]) result = handler(event, _make_context()) # Unknown format is skipped, not reported as failure assert result == {"batchItemFailures": []} @patch.dict(os.environ, ENV_VARS) @patch("handler._process_fan_batch") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_fan_batch_timeout_reported_as_failure( self, mock_boto3, mock_kafka, mock_batch ): mock_batch.return_value = { "processed": 5, "errors": 0, "rate_limited": 0, "timed_out": True, } event = _make_sqs_event([{ "source": "backfill", "fans": [{"spotifyUserId": "u1", "refreshToken": "t1", "partitionKey": "pk", "sortKey": "sk"}], "fan_count": 1, }]) result = handler(event, _make_context()) assert len(result["batchItemFailures"]) == 1 assert result["batchItemFailures"][0]["itemIdentifier"] == "msg-0" @patch.dict(os.environ, ENV_VARS) @patch("handler._process_fan_batch") @patch("handler._create_kafka_producer", return_value=None) @patch("handler.boto3") def test_fan_batch_with_snowflake_format( self, mock_boto3, mock_kafka, mock_batch ): mock_batch.return_value = { "processed": 1, "errors": 0, "rate_limited": 0, "timed_out": False, } event = _make_sqs_event([{ "source": "recollection", "fans": [{"SPOTIFY_USER_ID": "u1", "REFRESH_TOKEN": "t1", "PARTITION_KEY": "pk", "SORT_KEY": "sk"}], "fan_count": 1, }]) result = handler(event, _make_context()) assert result == {"batchItemFailures": []} mock_batch.assert_called_once() fans_arg = mock_batch.call_args[0][0] assert fans_arg[0]["SPOTIFY_USER_ID"] == "u1"