"""Tests for marketing_intelligence.storage.s3.""" from __future__ import annotations import asyncio import json from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import pytest from botocore.exceptions import ClientError from marketing_intelligence.core.models import ( ArtistSnapshotRecord, CampaignConfigRecord, CampaignSentiment, RunRecord, SoundRecord, TrendSignal, ) from marketing_intelligence.storage.s3 import S3Backend, _is_not_found _MOD = "marketing_intelligence.storage.s3" # ──────────────────────────── helpers ───────────────────────────────────────── def _client_error(code: str) -> ClientError: return ClientError({"Error": {"Code": code, "Message": "err"}}, "GetObject") def _make_backend() -> tuple[S3Backend, MagicMock]: with patch(f"{_MOD}.settings") as s: s.s3_bucket = "test-bucket" s.aws_region = "us-east-1" with patch(f"{_MOD}.boto3") as mock_boto3: client = MagicMock() mock_boto3.client.return_value = client backend = S3Backend() return backend, client def _run_record() -> RunRecord: return RunRecord( run_id="artist_1", run_type="discovery", campaign_key="artist_track", started_at="2024-01-01T00:00:00Z", ) def _campaign_config() -> CampaignConfigRecord: return CampaignConfigRecord( campaign_key="artist_track", artist_key="artist", post_id="post1", content_tier=1, ) def _sentiment() -> CampaignSentiment: return CampaignSentiment( campaign_config_key="artist_track", run_id="run1", observed_at="2024-01-01T00:00:00Z", sentiment="positive", confidence=0.9, summary="Great", reasoning="Very positive", how_sound_is_used="background", confidence_rationale="High", trend_signals=[ TrendSignal( signal="viral", evidence="lots of views", supporting_video_ids=["v1"] ) ], ) # ──────────────────────────── _is_not_found ─────────────────────────────────── class TestIsNotFound: def test_no_such_key(self) -> None: assert _is_not_found(_client_error("NoSuchKey")) def test_404(self) -> None: assert _is_not_found(_client_error("404")) def test_other_error(self) -> None: assert not _is_not_found(_client_error("AccessDenied")) # ──────────────────────────── S3Backend init ────────────────────────────────── class TestS3BackendInit: def test_requires_bucket(self) -> None: with patch(f"{_MOD}.settings") as s: s.s3_bucket = "" with pytest.raises(ValueError, match="S3_BUCKET"): S3Backend() def test_creates_boto3_client(self) -> None: with patch(f"{_MOD}.settings") as s: s.s3_bucket = "bucket" s.aws_region = "eu-west-1" with patch(f"{_MOD}.boto3") as mock_boto3: mock_boto3.client.return_value = MagicMock() S3Backend() mock_boto3.client.assert_called_once_with("s3", region_name="eu-west-1") # ──────────────────────────── _get_json / _put_json ────────────────────────── class TestGetPutJson: def test_get_json_returns_parsed(self) -> None: backend, client = _make_backend() body = MagicMock() body.read.return_value = json.dumps({"k": "v"}).encode() client.get_object.return_value = {"Body": body} with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_thread: mock_thread.side_effect = lambda fn, *a, **kw: fn() result = asyncio.run(backend._get_json("some/key.json")) assert result == {"k": "v"} def test_get_json_returns_none_on_not_found(self) -> None: backend, client = _make_backend() client.get_object.side_effect = _client_error("NoSuchKey") with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_thread: mock_thread.side_effect = lambda fn, *a, **kw: fn() result = asyncio.run(backend._get_json("missing.json")) assert result is None def test_get_json_reraises_other_errors(self) -> None: backend, client = _make_backend() client.get_object.side_effect = _client_error("AccessDenied") with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_thread: mock_thread.side_effect = lambda fn, *a, **kw: fn() with pytest.raises(ClientError): asyncio.run(backend._get_json("key.json")) def test_put_json_returns_s3_uri(self) -> None: backend, client = _make_backend() with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_thread: mock_thread.return_value = None result = asyncio.run(backend._put_json("a/b.json", {"x": 1})) assert result == "s3://test-bucket/a/b.json" # ──────────────────────────── write methods ─────────────────────────────────── class TestWriteMethods: def test_write_run_record(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/runs/artist_1.json" asyncio.run(backend.write_run_record(_run_record())) mock_put.assert_called_once() assert "artist_1" in mock_put.call_args[0][0] def test_write_campaign_post_with_sound(self) -> None: backend, _ = _make_backend() sound = SoundRecord(sound_id="s1", first_seen_at="2024-01-01T00:00:00Z") with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/posts/c/r/post1.json" asyncio.run( backend.write_campaign_post(_campaign_config(), sound, "artist", "run1") ) data = mock_put.call_args[0][1] assert "sound" in data def test_write_campaign_post_without_sound(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/posts/c/r/post1.json" asyncio.run( backend.write_campaign_post(_campaign_config(), None, "artist", "run1") ) data = mock_put.call_args[0][1] assert "sound" not in data def test_write_post_metrics_skips_empty(self) -> None: backend, _ = _make_backend() result = asyncio.run(backend.write_post_metrics([], "key", "run1")) assert result is None def test_write_post_metrics_with_items(self) -> None: backend, _ = _make_backend() items = [{"video_id": "v1", "views": 100}] with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/metrics/key/run1.json" result = asyncio.run(backend.write_post_metrics(items, "key", "run1")) assert result is not None def test_write_post_metrics_with_reasoning(self) -> None: backend, _ = _make_backend() items = [{"video_id": "v1", "views": 100}] with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/result" asyncio.run( backend.write_post_metrics(items, "key", "run1", reasoning="good") ) assert mock_put.call_count == 2 def test_write_artist_snapshot(self) -> None: backend, _ = _make_backend() record = ArtistSnapshotRecord( artist_key="artist", run_id="run1", observed_at="2024-01-01T00:00:00Z", ) with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/snapshots" asyncio.run(backend.write_artist_snapshot(record, "artist")) mock_put.assert_called_once() def test_write_sentiment(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/sentiment" asyncio.run(backend.write_sentiment(_sentiment())) mock_put.assert_called_once() def test_read_sentiment(self) -> None: backend, _ = _make_backend() expected: dict[str, Any] = {"sentiment": "positive"} with patch.object(backend, "_get_json", new_callable=AsyncMock) as mock_get: mock_get.return_value = expected result = asyncio.run(backend.read_sentiment("artist_track", "run1")) assert result == expected def test_write_pdf_report(self) -> None: backend, client = _make_backend() with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_thread: mock_thread.return_value = None result = asyncio.run( backend.write_pdf_report("artist_track", "run1", "/tmp/report.pdf") ) assert result == "s3://test-bucket/sentiment/artist_track/run1.pdf" # ──────────────────────────── video comments ────────────────────────────────── class TestVideoComments: def test_write_video_comments(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/comments/c/r/v1.json" result = asyncio.run( backend.write_video_comments("c", "r", "v1", {"comments": []}) ) assert result is not None def test_update_video_comments_sentiment_returns_false_if_missing(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_get_json", new_callable=AsyncMock) as mock_get: mock_get.return_value = None result = asyncio.run( backend.update_video_comments_sentiment( "c", "r", "v1", "positive", 0.9, "ok", [] ) ) assert result is False def test_update_video_comments_sentiment_updates(self) -> None: backend, _ = _make_backend() existing = {"video_id": "v1", "comments": []} with patch.object(backend, "_get_json", new_callable=AsyncMock) as mock_get: mock_get.return_value = existing with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/c" result = asyncio.run( backend.update_video_comments_sentiment( "c", "r", "v1", "positive", 0.9, "ok", ["tag"] ) ) assert result is True def test_read_all_video_comments_skips_sentinel_keys(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_list_keys", new_callable=AsyncMock) as mock_list: mock_list.return_value = [ "comments/c/r/manifest.json", "comments/c/r/request.json", "comments/c/r/v1.json", ] with patch.object(backend, "_get_json", new_callable=AsyncMock) as mock_get: mock_get.return_value = { "video_id": "v1", "comment_texts": [], "comment_count": 0, "hashtags": [], "is_original_sound": False, } results = asyncio.run(backend.read_all_video_comments("c", "r")) assert len(results) == 1 def test_read_all_video_comments_skips_already_analysed(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_list_keys", new_callable=AsyncMock) as mock_list: mock_list.return_value = ["comments/c/r/v1.json"] with patch.object(backend, "_get_json", new_callable=AsyncMock) as mock_get: mock_get.return_value = {"video_id": "v1", "sentiment": "positive"} results = asyncio.run(backend.read_all_video_comments("c", "r")) assert results == [] def test_read_all_video_comments_limit(self) -> None: backend, _ = _make_backend() keys = [f"comments/c/r/v{i}.json" for i in range(5)] with patch.object(backend, "_list_keys", new_callable=AsyncMock) as mock_list: mock_list.return_value = keys with patch.object(backend, "_get_json", new_callable=AsyncMock) as mock_get: mock_get.return_value = { "video_id": "v", "comment_texts": [], "comment_count": 0, "hashtags": [], "is_original_sound": False, } results = asyncio.run( backend.read_all_video_comments("c", "r", limit=2) ) assert len(results) == 2 def test_read_all_video_sentiments(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_list_keys", new_callable=AsyncMock) as mock_list: mock_list.return_value = ["comments/c/r/v1.json"] with patch.object(backend, "_get_json", new_callable=AsyncMock) as mock_get: mock_get.return_value = { "video_id": "v1", "sentiment": "positive", "confidence": 0.9, "summary": "ok", "key_themes": [], } results = asyncio.run(backend.read_all_video_sentiments("c", "r")) assert len(results) == 1 assert results[0]["sentiment"] == "positive" # ──────────────────────────── metrics snapshot ──────────────────────────────── class TestMetricsSnapshot: def test_read_latest_metrics_returns_empty_when_none(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_get_json", new_callable=AsyncMock) as mock_get: mock_get.return_value = None result = asyncio.run(backend.read_latest_metrics("key")) assert result == {} def test_write_latest_metrics(self) -> None: backend, _ = _make_backend() with patch.object(backend, "_put_json", new_callable=AsyncMock) as mock_put: mock_put.return_value = "s3://b/snapshots/key/latest.json" asyncio.run(backend.write_latest_metrics("key", {"v1": {}})) mock_put.assert_called_once()