test(aggregators): add comprehensive tests for Reddit Highlights aggregator

+1

aggregators/reddit-highlights/tests/__init__.py

··· 1 + """Tests for Reddit Highlights Aggregator."""

+352

aggregators/reddit-highlights/tests/test_config.py

··· 1 + """ 2 + Tests for config module. 3 + """ 4 + import os 5 + import pytest 6 + from pathlib import Path 7 + 8 + from src.config import ConfigLoader, ConfigError 9 + from src.models import LogLevel 10 + 11 + 12 + class TestConfigLoaderInit: 13 + """Tests for ConfigLoader initialization.""" 14 + 15 + def test_stores_config_path(self, tmp_path): 16 + """Test that config path is stored.""" 17 + config_path = tmp_path / "config.yaml" 18 + loader = ConfigLoader(config_path) 19 + assert loader.config_path == config_path 20 + 21 + 22 + class TestConfigLoaderLoad: 23 + """Tests for ConfigLoader.load method.""" 24 + 25 + def test_raises_if_file_not_found(self, tmp_path): 26 + """Test that ConfigError is raised if file doesn't exist.""" 27 + config_path = tmp_path / "nonexistent.yaml" 28 + loader = ConfigLoader(config_path) 29 + 30 + with pytest.raises(ConfigError, match="not found"): 31 + loader.load() 32 + 33 + def test_raises_on_invalid_yaml(self, tmp_path): 34 + """Test that ConfigError is raised for invalid YAML.""" 35 + config_path = tmp_path / "config.yaml" 36 + config_path.write_text("invalid: yaml: ::::") 37 + loader = ConfigLoader(config_path) 38 + 39 + with pytest.raises(ConfigError, match="Failed to parse YAML"): 40 + loader.load() 41 + 42 + def test_raises_on_empty_file(self, tmp_path): 43 + """Test that ConfigError is raised for empty file.""" 44 + config_path = tmp_path / "config.yaml" 45 + config_path.write_text("") 46 + loader = ConfigLoader(config_path) 47 + 48 + with pytest.raises(ConfigError, match="empty"): 49 + loader.load() 50 + 51 + def test_raises_if_coves_api_url_missing(self, tmp_path): 52 + """Test that ConfigError is raised if coves_api_url is missing.""" 53 + config_path = tmp_path / "config.yaml" 54 + config_path.write_text(""" 55 + subreddits: 56 + - name: nba 57 + community_handle: nba.coves.social 58 + """) 59 + loader = ConfigLoader(config_path) 60 + 61 + with pytest.raises(ConfigError, match="coves_api_url"): 62 + loader.load() 63 + 64 + def test_raises_on_invalid_url(self, tmp_path): 65 + """Test that ConfigError is raised for invalid URL.""" 66 + config_path = tmp_path / "config.yaml" 67 + config_path.write_text(""" 68 + coves_api_url: "not-a-valid-url" 69 + subreddits: 70 + - name: nba 71 + community_handle: nba.coves.social 72 + """) 73 + loader = ConfigLoader(config_path) 74 + 75 + with pytest.raises(ConfigError, match="Invalid URL"): 76 + loader.load() 77 + 78 + def test_raises_on_file_url_scheme(self, tmp_path): 79 + """Test that ConfigError is raised for file:// URL scheme.""" 80 + config_path = tmp_path / "config.yaml" 81 + config_path.write_text(""" 82 + coves_api_url: "file:///etc/passwd" 83 + subreddits: 84 + - name: nba 85 + community_handle: nba.coves.social 86 + """) 87 + loader = ConfigLoader(config_path) 88 + 89 + with pytest.raises(ConfigError, match="Invalid URL"): 90 + loader.load() 91 + 92 + def test_raises_if_no_subreddits(self, tmp_path): 93 + """Test that ConfigError is raised if no subreddits defined.""" 94 + config_path = tmp_path / "config.yaml" 95 + config_path.write_text(""" 96 + coves_api_url: "https://coves.social" 97 + subreddits: [] 98 + """) 99 + loader = ConfigLoader(config_path) 100 + 101 + with pytest.raises(ConfigError, match="at least one subreddit"): 102 + loader.load() 103 + 104 + def test_loads_valid_config(self, tmp_path): 105 + """Test successful loading of valid config.""" 106 + config_path = tmp_path / "config.yaml" 107 + config_path.write_text(""" 108 + coves_api_url: "https://coves.social" 109 + subreddits: 110 + - name: nba 111 + community_handle: nba.coves.social 112 + enabled: true 113 + - name: soccer 114 + community_handle: soccer.coves.social 115 + enabled: false 116 + allowed_domains: 117 + - streamable.com 118 + - gfycat.com 119 + log_level: debug 120 + """) 121 + loader = ConfigLoader(config_path) 122 + config = loader.load() 123 + 124 + assert config.coves_api_url == "https://coves.social" 125 + assert len(config.subreddits) == 2 126 + assert config.subreddits[0].name == "nba" 127 + assert config.subreddits[0].community_handle == "nba.coves.social" 128 + assert config.subreddits[0].enabled is True 129 + assert config.subreddits[1].name == "soccer" 130 + assert config.subreddits[1].enabled is False 131 + assert "streamable.com" in config.allowed_domains 132 + assert "gfycat.com" in config.allowed_domains 133 + assert config.log_level == LogLevel.DEBUG 134 + 135 + def test_uses_default_allowed_domains(self, tmp_path): 136 + """Test that default allowed_domains is used when not specified.""" 137 + config_path = tmp_path / "config.yaml" 138 + config_path.write_text(""" 139 + coves_api_url: "https://coves.social" 140 + subreddits: 141 + - name: nba 142 + community_handle: nba.coves.social 143 + """) 144 + loader = ConfigLoader(config_path) 145 + config = loader.load() 146 + 147 + assert "streamable.com" in config.allowed_domains 148 + 149 + def test_uses_default_log_level(self, tmp_path): 150 + """Test that default log_level is used when not specified.""" 151 + config_path = tmp_path / "config.yaml" 152 + config_path.write_text(""" 153 + coves_api_url: "https://coves.social" 154 + subreddits: 155 + - name: nba 156 + community_handle: nba.coves.social 157 + """) 158 + loader = ConfigLoader(config_path) 159 + config = loader.load() 160 + 161 + assert config.log_level == LogLevel.INFO 162 + 163 + def test_invalid_log_level_raises(self, tmp_path): 164 + """Test that invalid log_level raises ConfigError.""" 165 + config_path = tmp_path / "config.yaml" 166 + config_path.write_text(""" 167 + coves_api_url: "https://coves.social" 168 + subreddits: 169 + - name: nba 170 + community_handle: nba.coves.social 171 + log_level: invalid_level 172 + """) 173 + loader = ConfigLoader(config_path) 174 + 175 + with pytest.raises(ConfigError, match="Invalid log_level"): 176 + loader.load() 177 + 178 + def test_environment_variable_override(self, tmp_path, monkeypatch): 179 + """Test that COVES_API_URL env var overrides config file.""" 180 + config_path = tmp_path / "config.yaml" 181 + config_path.write_text(""" 182 + coves_api_url: "https://coves.social" 183 + subreddits: 184 + - name: nba 185 + community_handle: nba.coves.social 186 + """) 187 + monkeypatch.setenv("COVES_API_URL", "https://custom.coves.social") 188 + loader = ConfigLoader(config_path) 189 + config = loader.load() 190 + 191 + assert config.coves_api_url == "https://custom.coves.social" 192 + 193 + 194 + class TestSubredditParsing: 195 + """Tests for subreddit configuration parsing.""" 196 + 197 + def test_missing_name_raises(self, tmp_path): 198 + """Test that missing subreddit name raises ConfigError.""" 199 + config_path = tmp_path / "config.yaml" 200 + config_path.write_text(""" 201 + coves_api_url: "https://coves.social" 202 + subreddits: 203 + - community_handle: nba.coves.social 204 + """) 205 + loader = ConfigLoader(config_path) 206 + 207 + with pytest.raises(ConfigError, match="name"): 208 + loader.load() 209 + 210 + def test_missing_community_handle_raises(self, tmp_path): 211 + """Test that missing community_handle raises ConfigError.""" 212 + config_path = tmp_path / "config.yaml" 213 + config_path.write_text(""" 214 + coves_api_url: "https://coves.social" 215 + subreddits: 216 + - name: nba 217 + """) 218 + loader = ConfigLoader(config_path) 219 + 220 + with pytest.raises(ConfigError, match="community_handle"): 221 + loader.load() 222 + 223 + def test_empty_name_raises(self, tmp_path): 224 + """Test that empty subreddit name raises ConfigError.""" 225 + config_path = tmp_path / "config.yaml" 226 + config_path.write_text(""" 227 + coves_api_url: "https://coves.social" 228 + subreddits: 229 + - name: "" 230 + community_handle: nba.coves.social 231 + """) 232 + loader = ConfigLoader(config_path) 233 + 234 + with pytest.raises(ConfigError, match="cannot be empty"): 235 + loader.load() 236 + 237 + def test_empty_community_handle_raises(self, tmp_path): 238 + """Test that empty community_handle raises ConfigError.""" 239 + config_path = tmp_path / "config.yaml" 240 + config_path.write_text(""" 241 + coves_api_url: "https://coves.social" 242 + subreddits: 243 + - name: nba 244 + community_handle: "" 245 + """) 246 + loader = ConfigLoader(config_path) 247 + 248 + with pytest.raises(ConfigError, match="cannot be empty"): 249 + loader.load() 250 + 251 + def test_defaults_enabled_to_true(self, tmp_path): 252 + """Test that enabled defaults to True.""" 253 + config_path = tmp_path / "config.yaml" 254 + config_path.write_text(""" 255 + coves_api_url: "https://coves.social" 256 + subreddits: 257 + - name: nba 258 + community_handle: nba.coves.social 259 + """) 260 + loader = ConfigLoader(config_path) 261 + config = loader.load() 262 + 263 + assert config.subreddits[0].enabled is True 264 + 265 + def test_normalizes_name_to_lowercase(self, tmp_path): 266 + """Test that subreddit name is normalized to lowercase.""" 267 + config_path = tmp_path / "config.yaml" 268 + config_path.write_text(""" 269 + coves_api_url: "https://coves.social" 270 + subreddits: 271 + - name: NBA 272 + community_handle: nba.coves.social 273 + """) 274 + loader = ConfigLoader(config_path) 275 + config = loader.load() 276 + 277 + assert config.subreddits[0].name == "nba" 278 + 279 + def test_strips_whitespace(self, tmp_path): 280 + """Test that whitespace is stripped from names.""" 281 + config_path = tmp_path / "config.yaml" 282 + config_path.write_text(""" 283 + coves_api_url: "https://coves.social" 284 + subreddits: 285 + - name: " nba " 286 + community_handle: " nba.coves.social " 287 + """) 288 + loader = ConfigLoader(config_path) 289 + config = loader.load() 290 + 291 + assert config.subreddits[0].name == "nba" 292 + assert config.subreddits[0].community_handle == "nba.coves.social" 293 + 294 + 295 + class TestUrlValidation: 296 + """Tests for URL validation.""" 297 + 298 + def test_accepts_https_url(self, tmp_path): 299 + """Test that HTTPS URLs are accepted.""" 300 + config_path = tmp_path / "config.yaml" 301 + config_path.write_text(""" 302 + coves_api_url: "https://coves.social" 303 + subreddits: 304 + - name: nba 305 + community_handle: nba.coves.social 306 + """) 307 + loader = ConfigLoader(config_path) 308 + config = loader.load() 309 + 310 + assert config.coves_api_url == "https://coves.social" 311 + 312 + def test_accepts_http_url(self, tmp_path): 313 + """Test that HTTP URLs are accepted (for local dev).""" 314 + config_path = tmp_path / "config.yaml" 315 + config_path.write_text(""" 316 + coves_api_url: "http://localhost:8080" 317 + subreddits: 318 + - name: nba 319 + community_handle: nba.coves.social 320 + """) 321 + loader = ConfigLoader(config_path) 322 + config = loader.load() 323 + 324 + assert config.coves_api_url == "http://localhost:8080" 325 + 326 + def test_rejects_javascript_url(self, tmp_path): 327 + """Test that javascript: URLs are rejected.""" 328 + config_path = tmp_path / "config.yaml" 329 + config_path.write_text(""" 330 + coves_api_url: "javascript:alert(1)" 331 + subreddits: 332 + - name: nba 333 + community_handle: nba.coves.social 334 + """) 335 + loader = ConfigLoader(config_path) 336 + 337 + with pytest.raises(ConfigError, match="Invalid URL"): 338 + loader.load() 339 + 340 + def test_rejects_data_url(self, tmp_path): 341 + """Test that data: URLs are rejected.""" 342 + config_path = tmp_path / "config.yaml" 343 + config_path.write_text(""" 344 + coves_api_url: "data:text/html,<script>alert(1)</script>" 345 + subreddits: 346 + - name: nba 347 + community_handle: nba.coves.social 348 + """) 349 + loader = ConfigLoader(config_path) 350 + 351 + with pytest.raises(ConfigError, match="Invalid URL"): 352 + loader.load()

+286

aggregators/reddit-highlights/tests/test_coves_client.py

··· 1 + """ 2 + Tests for coves_client module. 3 + """ 4 + import pytest 5 + from unittest.mock import MagicMock, patch 6 + import requests 7 + 8 + from src.coves_client import ( 9 + CovesClient, 10 + CovesAPIError, 11 + CovesAuthenticationError, 12 + CovesForbiddenError, 13 + CovesNotFoundError, 14 + CovesRateLimitError, 15 + ) 16 + 17 + 18 + class TestCovesClientInit: 19 + """Tests for CovesClient initialization.""" 20 + 21 + def test_valid_api_key(self): 22 + """Test initialization with valid API key.""" 23 + # Generate a valid 70-character API key 24 + api_key = "ckapi_" + "a" * 64 25 + client = CovesClient("https://coves.social", api_key) 26 + assert client.api_key == api_key 27 + assert client.api_url == "https://coves.social" 28 + 29 + def test_strips_trailing_slash_from_url(self): 30 + """Test that trailing slash is stripped from API URL.""" 31 + api_key = "ckapi_" + "a" * 64 32 + client = CovesClient("https://coves.social/", api_key) 33 + assert client.api_url == "https://coves.social" 34 + 35 + def test_empty_api_key_raises(self): 36 + """Test that empty API key raises ValueError.""" 37 + with pytest.raises(ValueError, match="cannot be empty"): 38 + CovesClient("https://coves.social", "") 39 + 40 + def test_wrong_prefix_raises(self): 41 + """Test that API key with wrong prefix raises ValueError.""" 42 + with pytest.raises(ValueError, match="must start with"): 43 + CovesClient("https://coves.social", "invalid_" + "a" * 63) 44 + 45 + def test_wrong_length_raises(self): 46 + """Test that API key with wrong length raises ValueError.""" 47 + with pytest.raises(ValueError, match="must be 70 characters"): 48 + CovesClient("https://coves.social", "ckapi_tooshort") 49 + 50 + def test_session_headers_set(self): 51 + """Test that session headers are properly set.""" 52 + api_key = "ckapi_" + "b" * 64 53 + client = CovesClient("https://coves.social", api_key) 54 + assert client.session.headers["Authorization"] == f"Bearer {api_key}" 55 + assert client.session.headers["Content-Type"] == "application/json" 56 + 57 + 58 + class TestCovesClientAuthenticate: 59 + """Tests for authenticate method.""" 60 + 61 + def test_authenticate_is_noop(self): 62 + """Test that authenticate is a no-op for API key auth.""" 63 + api_key = "ckapi_" + "c" * 64 64 + client = CovesClient("https://coves.social", api_key) 65 + # Should not raise any exceptions 66 + client.authenticate() 67 + 68 + 69 + class TestCovesClientCreatePost: 70 + """Tests for create_post method.""" 71 + 72 + @pytest.fixture 73 + def client(self): 74 + api_key = "ckapi_" + "d" * 64 75 + return CovesClient("https://coves.social", api_key) 76 + 77 + def test_successful_post_creation(self, client): 78 + """Test successful post creation.""" 79 + mock_response = MagicMock() 80 + mock_response.ok = True 81 + mock_response.status_code = 200 82 + mock_response.json.return_value = {"uri": "at://did:plc:test/social.coves.post/abc123"} 83 + 84 + with patch.object(client.session, "post", return_value=mock_response): 85 + uri = client.create_post( 86 + community_handle="test.coves.social", 87 + content="Test content", 88 + facets=[], 89 + title="Test Title", 90 + ) 91 + assert uri == "at://did:plc:test/social.coves.post/abc123" 92 + 93 + def test_post_with_embed(self, client): 94 + """Test post creation with embed.""" 95 + mock_response = MagicMock() 96 + mock_response.ok = True 97 + mock_response.json.return_value = {"uri": "at://did:plc:test/social.coves.post/xyz"} 98 + 99 + with patch.object(client.session, "post", return_value=mock_response) as mock_post: 100 + embed = {"$type": "social.coves.embed.external", "external": {"uri": "https://example.com"}} 101 + client.create_post( 102 + community_handle="test.coves.social", 103 + content="", 104 + facets=[], 105 + embed=embed, 106 + ) 107 + # Verify embed was included in request 108 + call_args = mock_post.call_args 109 + assert call_args[1]["json"]["embed"] == embed 110 + 111 + def test_post_with_thumbnail_url(self, client): 112 + """Test post creation with thumbnail URL.""" 113 + mock_response = MagicMock() 114 + mock_response.ok = True 115 + mock_response.json.return_value = {"uri": "at://did:plc:test/social.coves.post/thumb"} 116 + 117 + with patch.object(client.session, "post", return_value=mock_response) as mock_post: 118 + client.create_post( 119 + community_handle="test.coves.social", 120 + content="", 121 + facets=[], 122 + thumbnail_url="https://example.com/thumb.jpg", 123 + ) 124 + call_args = mock_post.call_args 125 + assert call_args[1]["json"]["thumbnailUrl"] == "https://example.com/thumb.jpg" 126 + 127 + def test_401_raises_authentication_error(self, client): 128 + """Test that 401 response raises CovesAuthenticationError.""" 129 + mock_response = MagicMock() 130 + mock_response.ok = False 131 + mock_response.status_code = 401 132 + mock_response.text = "Unauthorized" 133 + 134 + with patch.object(client.session, "post", return_value=mock_response): 135 + with pytest.raises(CovesAuthenticationError): 136 + client.create_post("test.coves.social", "", []) 137 + 138 + def test_403_raises_forbidden_error(self, client): 139 + """Test that 403 response raises CovesForbiddenError.""" 140 + mock_response = MagicMock() 141 + mock_response.ok = False 142 + mock_response.status_code = 403 143 + mock_response.text = "Forbidden" 144 + 145 + with patch.object(client.session, "post", return_value=mock_response): 146 + with pytest.raises(CovesForbiddenError): 147 + client.create_post("test.coves.social", "", []) 148 + 149 + def test_404_raises_not_found_error(self, client): 150 + """Test that 404 response raises CovesNotFoundError.""" 151 + mock_response = MagicMock() 152 + mock_response.ok = False 153 + mock_response.status_code = 404 154 + mock_response.text = "Not Found" 155 + 156 + with patch.object(client.session, "post", return_value=mock_response): 157 + with pytest.raises(CovesNotFoundError): 158 + client.create_post("test.coves.social", "", []) 159 + 160 + def test_429_raises_rate_limit_error(self, client): 161 + """Test that 429 response raises CovesRateLimitError.""" 162 + mock_response = MagicMock() 163 + mock_response.ok = False 164 + mock_response.status_code = 429 165 + mock_response.text = "Rate Limited" 166 + 167 + with patch.object(client.session, "post", return_value=mock_response): 168 + with pytest.raises(CovesRateLimitError): 169 + client.create_post("test.coves.social", "", []) 170 + 171 + def test_500_raises_api_error(self, client): 172 + """Test that 500 response raises CovesAPIError.""" 173 + mock_response = MagicMock() 174 + mock_response.ok = False 175 + mock_response.status_code = 500 176 + mock_response.text = "Internal Server Error" 177 + 178 + with patch.object(client.session, "post", return_value=mock_response): 179 + with pytest.raises(CovesAPIError): 180 + client.create_post("test.coves.social", "", []) 181 + 182 + def test_invalid_json_response_raises_api_error(self, client): 183 + """Test that invalid JSON response raises CovesAPIError.""" 184 + mock_response = MagicMock() 185 + mock_response.ok = True 186 + mock_response.status_code = 200 187 + mock_response.text = "not json" 188 + mock_response.json.side_effect = ValueError("Invalid JSON") 189 + 190 + with patch.object(client.session, "post", return_value=mock_response): 191 + with pytest.raises(CovesAPIError, match="Invalid response"): 192 + client.create_post("test.coves.social", "", []) 193 + 194 + def test_missing_uri_in_response_raises_api_error(self, client): 195 + """Test that missing uri in response raises CovesAPIError.""" 196 + mock_response = MagicMock() 197 + mock_response.ok = True 198 + mock_response.status_code = 200 199 + mock_response.text = '{"cid": "abc"}' 200 + mock_response.json.return_value = {"cid": "abc"} # No uri field 201 + 202 + with patch.object(client.session, "post", return_value=mock_response): 203 + with pytest.raises(CovesAPIError, match="Invalid response"): 204 + client.create_post("test.coves.social", "", []) 205 + 206 + def test_network_error_propagates(self, client): 207 + """Test that network errors propagate.""" 208 + with patch.object(client.session, "post", side_effect=requests.ConnectionError("Network error")): 209 + with pytest.raises(requests.RequestException): 210 + client.create_post("test.coves.social", "", []) 211 + 212 + 213 + class TestCreateExternalEmbed: 214 + """Tests for create_external_embed method.""" 215 + 216 + @pytest.fixture 217 + def client(self): 218 + api_key = "ckapi_" + "e" * 64 219 + return CovesClient("https://coves.social", api_key) 220 + 221 + def test_basic_embed(self, client): 222 + """Test basic external embed creation.""" 223 + embed = client.create_external_embed( 224 + uri="https://streamable.com/abc123", 225 + title="Test Video", 226 + description="A test video", 227 + ) 228 + assert embed["$type"] == "social.coves.embed.external" 229 + assert embed["external"]["uri"] == "https://streamable.com/abc123" 230 + assert embed["external"]["title"] == "Test Video" 231 + assert embed["external"]["description"] == "A test video" 232 + 233 + def test_embed_with_sources(self, client): 234 + """Test embed with sources.""" 235 + sources = [{"uri": "https://reddit.com/r/test", "title": "r/test", "domain": "reddit.com"}] 236 + embed = client.create_external_embed( 237 + uri="https://streamable.com/abc123", 238 + title="Test", 239 + description="Test", 240 + sources=sources, 241 + ) 242 + assert embed["external"]["sources"] == sources 243 + 244 + def test_embed_with_video_metadata(self, client): 245 + """Test embed with video metadata.""" 246 + embed = client.create_external_embed( 247 + uri="https://streamable.com/abc123", 248 + title="Test", 249 + description="Test", 250 + embed_type="video", 251 + provider="streamable", 252 + domain="streamable.com", 253 + ) 254 + assert embed["external"]["embedType"] == "video" 255 + assert embed["external"]["provider"] == "streamable" 256 + assert embed["external"]["domain"] == "streamable.com" 257 + 258 + def test_optional_fields_not_included_when_none(self, client): 259 + """Test that optional fields are not included when None.""" 260 + embed = client.create_external_embed( 261 + uri="https://example.com", 262 + title="Test", 263 + description="Test", 264 + ) 265 + assert "sources" not in embed["external"] 266 + assert "embedType" not in embed["external"] 267 + assert "provider" not in embed["external"] 268 + assert "domain" not in embed["external"] 269 + 270 + 271 + class TestGetTimestamp: 272 + """Tests for _get_timestamp method.""" 273 + 274 + @pytest.fixture 275 + def client(self): 276 + api_key = "ckapi_" + "f" * 64 277 + return CovesClient("https://coves.social", api_key) 278 + 279 + def test_returns_iso_format(self, client): 280 + """Test that timestamp is in ISO 8601 format.""" 281 + timestamp = client._get_timestamp() 282 + # Should end with Z (UTC) 283 + assert timestamp.endswith("Z") 284 + # Should be parseable as ISO format 285 + from datetime import datetime 286 + datetime.fromisoformat(timestamp.replace("Z", "+00:00"))

+261

aggregators/reddit-highlights/tests/test_link_extractor.py

··· 1 + """ 2 + Tests for link_extractor module. 3 + """ 4 + import pytest 5 + from unittest.mock import MagicMock, patch 6 + import requests 7 + from src.link_extractor import LinkExtractor 8 + 9 + 10 + class TestLinkExtractor: 11 + """Tests for LinkExtractor class.""" 12 + 13 + def test_init_default_domains(self): 14 + """Test default allowed domains.""" 15 + extractor = LinkExtractor() 16 + assert "streamable.com" in extractor.allowed_domains 17 + 18 + def test_init_custom_domains(self): 19 + """Test custom allowed domains.""" 20 + extractor = LinkExtractor(allowed_domains=["example.com", "test.org"]) 21 + assert "example.com" in extractor.allowed_domains 22 + assert "test.org" in extractor.allowed_domains 23 + assert "streamable.com" not in extractor.allowed_domains 24 + 25 + 26 + class TestIsAllowedUrl: 27 + """Tests for is_allowed_url method.""" 28 + 29 + @pytest.fixture 30 + def extractor(self): 31 + return LinkExtractor() 32 + 33 + def test_streamable_url(self, extractor): 34 + """Test streamable.com URL detection.""" 35 + assert extractor.is_allowed_url("https://streamable.com/abc123") 36 + assert extractor.is_allowed_url("http://streamable.com/xyz789") 37 + assert extractor.is_allowed_url("https://www.streamable.com/test") 38 + 39 + def test_non_streamable_url(self, extractor): 40 + """Test rejection of non-streamable URLs.""" 41 + assert not extractor.is_allowed_url("https://youtube.com/watch?v=123") 42 + assert not extractor.is_allowed_url("https://reddit.com/r/nba") 43 + assert not extractor.is_allowed_url("https://twitter.com/video") 44 + 45 + def test_empty_url(self, extractor): 46 + """Test empty URL handling.""" 47 + assert not extractor.is_allowed_url("") 48 + assert not extractor.is_allowed_url(None) 49 + 50 + def test_invalid_url(self, extractor): 51 + """Test invalid URL handling.""" 52 + assert not extractor.is_allowed_url("not a url") 53 + assert not extractor.is_allowed_url("streamable.com/abc") # Missing scheme 54 + 55 + 56 + class TestExtractVideoUrl: 57 + """Tests for extract_video_url method.""" 58 + 59 + @pytest.fixture 60 + def extractor(self): 61 + return LinkExtractor() 62 + 63 + def test_direct_link(self, extractor): 64 + """Test extraction from direct link.""" 65 + entry = MagicMock() 66 + entry.link = "https://streamable.com/abc123" 67 + entry.content = None 68 + entry.description = None 69 + entry.summary = None 70 + 71 + result = extractor.extract_video_url(entry) 72 + assert result == "https://streamable.com/abc123" 73 + 74 + def test_link_in_content(self, extractor): 75 + """Test extraction from entry content.""" 76 + entry = MagicMock() 77 + entry.link = "https://reddit.com/r/nba/comments/123" 78 + 79 + content_item = MagicMock() 80 + content_item.value = 'Check out this play: <a href="https://streamable.com/xyz789">video</a>' 81 + entry.content = [content_item] 82 + entry.description = None 83 + entry.summary = None 84 + 85 + result = extractor.extract_video_url(entry) 86 + assert result == "https://streamable.com/xyz789" 87 + 88 + def test_link_in_description(self, extractor): 89 + """Test extraction from entry description.""" 90 + entry = MagicMock() 91 + entry.link = "https://reddit.com/r/nba/comments/123" 92 + entry.content = None 93 + entry.description = "Amazing dunk! https://streamable.com/dunk99" 94 + entry.summary = None 95 + 96 + result = extractor.extract_video_url(entry) 97 + assert result == "https://streamable.com/dunk99" 98 + 99 + def test_link_in_summary(self, extractor): 100 + """Test extraction from entry summary.""" 101 + entry = MagicMock() 102 + entry.link = "https://reddit.com/r/nba/comments/123" 103 + entry.content = None 104 + entry.description = None 105 + entry.summary = "Game winner: https://streamable.com/winner1" 106 + 107 + result = extractor.extract_video_url(entry) 108 + assert result == "https://streamable.com/winner1" 109 + 110 + def test_no_video_link(self, extractor): 111 + """Test when no video link is present.""" 112 + entry = MagicMock() 113 + entry.link = "https://reddit.com/r/nba/comments/123" 114 + entry.content = None 115 + entry.description = "Just a text post about basketball" 116 + entry.summary = None 117 + 118 + result = extractor.extract_video_url(entry) 119 + assert result is None 120 + 121 + def test_multiple_links_returns_first(self, extractor): 122 + """Test that first video link is returned when multiple present.""" 123 + entry = MagicMock() 124 + entry.link = "https://reddit.com/r/nba/comments/123" 125 + entry.content = None 126 + entry.description = ( 127 + "First: https://streamable.com/first " 128 + "Second: https://streamable.com/second" 129 + ) 130 + entry.summary = None 131 + 132 + result = extractor.extract_video_url(entry) 133 + assert result == "https://streamable.com/first" 134 + 135 + 136 + class TestNormalizeUrl: 137 + """Tests for URL normalization.""" 138 + 139 + @pytest.fixture 140 + def extractor(self): 141 + return LinkExtractor() 142 + 143 + def test_removes_trailing_slash(self, extractor): 144 + """Test trailing slash removal.""" 145 + url = extractor._normalize_url("https://streamable.com/abc123/") 146 + assert url == "https://streamable.com/abc123" 147 + 148 + def test_upgrades_http_to_https(self, extractor): 149 + """Test HTTP to HTTPS upgrade.""" 150 + url = extractor._normalize_url("http://streamable.com/abc123") 151 + assert url == "https://streamable.com/abc123" 152 + 153 + 154 + class TestGetVideoId: 155 + """Tests for get_video_id method.""" 156 + 157 + @pytest.fixture 158 + def extractor(self): 159 + return LinkExtractor() 160 + 161 + def test_extracts_video_id(self, extractor): 162 + """Test video ID extraction.""" 163 + video_id = extractor.get_video_id("https://streamable.com/abc123") 164 + assert video_id == "abc123" 165 + 166 + def test_handles_www_prefix(self, extractor): 167 + """Test www prefix handling.""" 168 + video_id = extractor.get_video_id("https://www.streamable.com/xyz789") 169 + assert video_id == "xyz789" 170 + 171 + def test_empty_url(self, extractor): 172 + """Test empty URL handling.""" 173 + assert extractor.get_video_id("") is None 174 + assert extractor.get_video_id(None) is None 175 + 176 + def test_url_with_query_params(self, extractor): 177 + """Test URL with query parameters.""" 178 + video_id = extractor.get_video_id("https://streamable.com/test123?foo=bar") 179 + assert video_id == "test123" 180 + 181 + 182 + class TestGetThumbnailUrl: 183 + """Tests for get_thumbnail_url method.""" 184 + 185 + @pytest.fixture 186 + def extractor(self): 187 + return LinkExtractor() 188 + 189 + def test_returns_thumbnail_on_success(self, extractor): 190 + """Test successful thumbnail fetch.""" 191 + mock_response = MagicMock() 192 + mock_response.status_code = 200 193 + mock_response.json.return_value = { 194 + "thumbnail_url": "https://cdn.streamable.com/image/abc123.jpg", 195 + "type": "video", 196 + } 197 + 198 + with patch("requests.get", return_value=mock_response): 199 + result = extractor.get_thumbnail_url("https://streamable.com/abc123") 200 + assert result == "https://cdn.streamable.com/image/abc123.jpg" 201 + 202 + def test_returns_none_on_network_error(self, extractor): 203 + """Test handling of network errors.""" 204 + with patch("requests.get", side_effect=requests.ConnectionError("Network error")): 205 + result = extractor.get_thumbnail_url("https://streamable.com/abc123") 206 + assert result is None 207 + 208 + def test_returns_none_on_http_error(self, extractor): 209 + """Test handling of HTTP errors.""" 210 + mock_response = MagicMock() 211 + mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found") 212 + 213 + with patch("requests.get", return_value=mock_response): 214 + result = extractor.get_thumbnail_url("https://streamable.com/abc123") 215 + assert result is None 216 + 217 + def test_returns_none_on_invalid_json(self, extractor): 218 + """Test handling of invalid JSON response.""" 219 + mock_response = MagicMock() 220 + mock_response.status_code = 200 221 + mock_response.json.side_effect = ValueError("Invalid JSON") 222 + 223 + with patch("requests.get", return_value=mock_response): 224 + result = extractor.get_thumbnail_url("https://streamable.com/abc123") 225 + assert result is None 226 + 227 + def test_returns_none_for_empty_url(self, extractor): 228 + """Test empty URL handling.""" 229 + assert extractor.get_thumbnail_url("") is None 230 + assert extractor.get_thumbnail_url(None) is None 231 + 232 + def test_returns_none_for_non_allowed_url(self, extractor): 233 + """Test rejection of non-allowed URLs.""" 234 + # Should not even make a request for non-allowed domains 235 + with patch("requests.get") as mock_get: 236 + result = extractor.get_thumbnail_url("https://youtube.com/watch?v=123") 237 + assert result is None 238 + mock_get.assert_not_called() 239 + 240 + def test_returns_none_when_thumbnail_missing(self, extractor): 241 + """Test handling when thumbnail_url is missing from response.""" 242 + mock_response = MagicMock() 243 + mock_response.status_code = 200 244 + mock_response.json.return_value = {"type": "video", "title": "Test"} 245 + 246 + with patch("requests.get", return_value=mock_response): 247 + result = extractor.get_thumbnail_url("https://streamable.com/abc123") 248 + assert result is None 249 + 250 + def test_uses_correct_oembed_url(self, extractor): 251 + """Test that correct oembed URL is constructed.""" 252 + mock_response = MagicMock() 253 + mock_response.status_code = 200 254 + mock_response.json.return_value = {"thumbnail_url": "https://example.com/thumb.jpg"} 255 + 256 + with patch("requests.get", return_value=mock_response) as mock_get: 257 + extractor.get_thumbnail_url("https://streamable.com/xyz789") 258 + mock_get.assert_called_once() 259 + call_url = mock_get.call_args[0][0] 260 + assert "api.streamable.com/oembed" in call_url 261 + assert "url=https://streamable.com/xyz789" in call_url

+273

aggregators/reddit-highlights/tests/test_models.py

··· 1 + """ 2 + Tests for models module. 3 + """ 4 + import pytest 5 + from datetime import datetime 6 + 7 + from src.models import RedditPost, SubredditConfig, AggregatorConfig, LogLevel 8 + 9 + 10 + class TestLogLevel: 11 + """Tests for LogLevel enum.""" 12 + 13 + def test_all_values(self): 14 + """Test all log level values.""" 15 + assert LogLevel.DEBUG.value == "debug" 16 + assert LogLevel.INFO.value == "info" 17 + assert LogLevel.WARNING.value == "warning" 18 + assert LogLevel.ERROR.value == "error" 19 + assert LogLevel.CRITICAL.value == "critical" 20 + 21 + def test_from_string(self): 22 + """Test creating LogLevel from string.""" 23 + assert LogLevel("debug") == LogLevel.DEBUG 24 + assert LogLevel("info") == LogLevel.INFO 25 + assert LogLevel("warning") == LogLevel.WARNING 26 + 27 + def test_invalid_value_raises(self): 28 + """Test that invalid value raises ValueError.""" 29 + with pytest.raises(ValueError): 30 + LogLevel("invalid") 31 + 32 + 33 + class TestRedditPost: 34 + """Tests for RedditPost dataclass.""" 35 + 36 + def test_valid_post(self): 37 + """Test creating a valid RedditPost.""" 38 + post = RedditPost( 39 + id="t3_abc123", 40 + title="Test Post", 41 + link="https://streamable.com/xyz", 42 + reddit_url="https://reddit.com/r/nba/comments/abc123", 43 + subreddit="nba", 44 + author="testuser", 45 + ) 46 + assert post.id == "t3_abc123" 47 + assert post.title == "Test Post" 48 + assert post.subreddit == "nba" 49 + 50 + def test_optional_fields(self): 51 + """Test optional fields have correct defaults.""" 52 + post = RedditPost( 53 + id="t3_abc123", 54 + title="Test Post", 55 + link="https://example.com", 56 + reddit_url="https://reddit.com/r/nba", 57 + subreddit="nba", 58 + author="testuser", 59 + ) 60 + assert post.published is None 61 + assert post.streamable_url is None 62 + 63 + def test_with_optional_fields(self): 64 + """Test creating post with optional fields.""" 65 + now = datetime.now() 66 + post = RedditPost( 67 + id="t3_abc123", 68 + title="Test Post", 69 + link="https://example.com", 70 + reddit_url="https://reddit.com/r/nba", 71 + subreddit="nba", 72 + author="testuser", 73 + published=now, 74 + streamable_url="https://streamable.com/xyz", 75 + ) 76 + assert post.published == now 77 + assert post.streamable_url == "https://streamable.com/xyz" 78 + 79 + def test_empty_id_raises(self): 80 + """Test that empty id raises ValueError.""" 81 + with pytest.raises(ValueError, match="id cannot be empty"): 82 + RedditPost( 83 + id="", 84 + title="Test", 85 + link="https://example.com", 86 + reddit_url="https://reddit.com", 87 + subreddit="nba", 88 + author="test", 89 + ) 90 + 91 + def test_empty_title_raises(self): 92 + """Test that empty title raises ValueError.""" 93 + with pytest.raises(ValueError, match="title cannot be empty"): 94 + RedditPost( 95 + id="t3_abc", 96 + title="", 97 + link="https://example.com", 98 + reddit_url="https://reddit.com", 99 + subreddit="nba", 100 + author="test", 101 + ) 102 + 103 + def test_empty_subreddit_raises(self): 104 + """Test that empty subreddit raises ValueError.""" 105 + with pytest.raises(ValueError, match="subreddit cannot be empty"): 106 + RedditPost( 107 + id="t3_abc", 108 + title="Test", 109 + link="https://example.com", 110 + reddit_url="https://reddit.com", 111 + subreddit="", 112 + author="test", 113 + ) 114 + 115 + 116 + class TestSubredditConfig: 117 + """Tests for SubredditConfig dataclass.""" 118 + 119 + def test_valid_config(self): 120 + """Test creating valid SubredditConfig.""" 121 + config = SubredditConfig( 122 + name="nba", 123 + community_handle="nba.coves.social", 124 + ) 125 + assert config.name == "nba" 126 + assert config.community_handle == "nba.coves.social" 127 + assert config.enabled is True # Default 128 + 129 + def test_disabled_config(self): 130 + """Test creating disabled SubredditConfig.""" 131 + config = SubredditConfig( 132 + name="nba", 133 + community_handle="nba.coves.social", 134 + enabled=False, 135 + ) 136 + assert config.enabled is False 137 + 138 + def test_empty_name_raises(self): 139 + """Test that empty name raises ValueError.""" 140 + with pytest.raises(ValueError, match="name cannot be empty"): 141 + SubredditConfig( 142 + name="", 143 + community_handle="nba.coves.social", 144 + ) 145 + 146 + def test_whitespace_name_raises(self): 147 + """Test that whitespace-only name raises ValueError.""" 148 + with pytest.raises(ValueError, match="name cannot be empty"): 149 + SubredditConfig( 150 + name=" ", 151 + community_handle="nba.coves.social", 152 + ) 153 + 154 + def test_empty_community_handle_raises(self): 155 + """Test that empty community_handle raises ValueError.""" 156 + with pytest.raises(ValueError, match="community_handle cannot be empty"): 157 + SubredditConfig( 158 + name="nba", 159 + community_handle="", 160 + ) 161 + 162 + def test_invalid_subreddit_name_format_raises(self): 163 + """Test that invalid subreddit name format raises ValueError.""" 164 + with pytest.raises(ValueError, match="Invalid subreddit name format"): 165 + SubredditConfig( 166 + name="nba/../../../etc/passwd", 167 + community_handle="nba.coves.social", 168 + ) 169 + 170 + def test_special_chars_in_name_raises(self): 171 + """Test that special characters in name raise ValueError.""" 172 + with pytest.raises(ValueError, match="Invalid subreddit name format"): 173 + SubredditConfig( 174 + name="nba<script>", 175 + community_handle="nba.coves.social", 176 + ) 177 + 178 + def test_valid_name_with_underscore(self): 179 + """Test that underscores in name are allowed.""" 180 + config = SubredditConfig( 181 + name="nba_discussion", 182 + community_handle="nba.coves.social", 183 + ) 184 + assert config.name == "nba_discussion" 185 + 186 + def test_valid_name_with_hyphen(self): 187 + """Test that hyphens in name are allowed.""" 188 + config = SubredditConfig( 189 + name="nba-highlights", 190 + community_handle="nba.coves.social", 191 + ) 192 + assert config.name == "nba-highlights" 193 + 194 + def test_is_frozen(self): 195 + """Test that SubredditConfig is immutable.""" 196 + config = SubredditConfig( 197 + name="nba", 198 + community_handle="nba.coves.social", 199 + ) 200 + with pytest.raises(AttributeError): 201 + config.name = "soccer" 202 + 203 + 204 + class TestAggregatorConfig: 205 + """Tests for AggregatorConfig dataclass.""" 206 + 207 + def test_valid_config(self): 208 + """Test creating valid AggregatorConfig.""" 209 + subreddit = SubredditConfig(name="nba", community_handle="nba.coves.social") 210 + config = AggregatorConfig( 211 + coves_api_url="https://coves.social", 212 + subreddits=(subreddit,), 213 + ) 214 + assert config.coves_api_url == "https://coves.social" 215 + assert len(config.subreddits) == 1 216 + assert config.log_level == LogLevel.INFO # Default 217 + 218 + def test_custom_log_level(self): 219 + """Test custom log level.""" 220 + subreddit = SubredditConfig(name="nba", community_handle="nba.coves.social") 221 + config = AggregatorConfig( 222 + coves_api_url="https://coves.social", 223 + subreddits=(subreddit,), 224 + log_level=LogLevel.DEBUG, 225 + ) 226 + assert config.log_level == LogLevel.DEBUG 227 + 228 + def test_custom_allowed_domains(self): 229 + """Test custom allowed domains.""" 230 + subreddit = SubredditConfig(name="nba", community_handle="nba.coves.social") 231 + config = AggregatorConfig( 232 + coves_api_url="https://coves.social", 233 + subreddits=(subreddit,), 234 + allowed_domains=("streamable.com", "gfycat.com"), 235 + ) 236 + assert "streamable.com" in config.allowed_domains 237 + assert "gfycat.com" in config.allowed_domains 238 + 239 + def test_empty_coves_api_url_raises(self): 240 + """Test that empty coves_api_url raises ValueError.""" 241 + subreddit = SubredditConfig(name="nba", community_handle="nba.coves.social") 242 + with pytest.raises(ValueError, match="coves_api_url cannot be empty"): 243 + AggregatorConfig( 244 + coves_api_url="", 245 + subreddits=(subreddit,), 246 + ) 247 + 248 + def test_empty_subreddits_raises(self): 249 + """Test that empty subreddits raises ValueError.""" 250 + with pytest.raises(ValueError, match="subreddits cannot be empty"): 251 + AggregatorConfig( 252 + coves_api_url="https://coves.social", 253 + subreddits=(), 254 + ) 255 + 256 + def test_is_frozen(self): 257 + """Test that AggregatorConfig is immutable.""" 258 + subreddit = SubredditConfig(name="nba", community_handle="nba.coves.social") 259 + config = AggregatorConfig( 260 + coves_api_url="https://coves.social", 261 + subreddits=(subreddit,), 262 + ) 263 + with pytest.raises(AttributeError): 264 + config.coves_api_url = "https://other.com" 265 + 266 + def test_default_allowed_domains(self): 267 + """Test default allowed domains.""" 268 + subreddit = SubredditConfig(name="nba", community_handle="nba.coves.social") 269 + config = AggregatorConfig( 270 + coves_api_url="https://coves.social", 271 + subreddits=(subreddit,), 272 + ) 273 + assert "streamable.com" in config.allowed_domains

+178

aggregators/reddit-highlights/tests/test_rss_fetcher.py

··· 1 + """ 2 + Tests for rss_fetcher module. 3 + """ 4 + import pytest 5 + from unittest.mock import MagicMock, patch 6 + import requests 7 + 8 + from src.rss_fetcher import RSSFetcher 9 + 10 + 11 + class TestRSSFetcherInit: 12 + """Tests for RSSFetcher initialization.""" 13 + 14 + def test_default_timeout(self): 15 + """Test default timeout value.""" 16 + fetcher = RSSFetcher() 17 + assert fetcher.timeout == 30 18 + 19 + def test_default_max_retries(self): 20 + """Test default max_retries value.""" 21 + fetcher = RSSFetcher() 22 + assert fetcher.max_retries == 3 23 + 24 + def test_default_user_agent(self): 25 + """Test default user agent is set.""" 26 + fetcher = RSSFetcher() 27 + assert "Coves-Reddit-Aggregator" in fetcher.user_agent 28 + assert "coves.social" in fetcher.user_agent 29 + 30 + def test_custom_timeout(self): 31 + """Test custom timeout value.""" 32 + fetcher = RSSFetcher(timeout=60) 33 + assert fetcher.timeout == 60 34 + 35 + def test_custom_max_retries(self): 36 + """Test custom max_retries value.""" 37 + fetcher = RSSFetcher(max_retries=5) 38 + assert fetcher.max_retries == 5 39 + 40 + def test_custom_user_agent(self): 41 + """Test custom user agent.""" 42 + fetcher = RSSFetcher(user_agent="CustomBot/1.0") 43 + assert fetcher.user_agent == "CustomBot/1.0" 44 + 45 + 46 + class TestRSSFetcherFetchFeed: 47 + """Tests for fetch_feed method.""" 48 + 49 + @pytest.fixture 50 + def fetcher(self): 51 + return RSSFetcher(max_retries=2) 52 + 53 + def test_raises_on_empty_url(self, fetcher): 54 + """Test that ValueError is raised for empty URL.""" 55 + with pytest.raises(ValueError, match="URL cannot be empty"): 56 + fetcher.fetch_feed("") 57 + 58 + def test_successful_fetch(self, fetcher): 59 + """Test successful feed fetch.""" 60 + mock_response = MagicMock() 61 + mock_response.status_code = 200 62 + mock_response.content = b"""<?xml version="1.0" encoding="UTF-8"?> 63 + <rss version="2.0"> 64 + <channel> 65 + <title>Test Feed</title> 66 + <item> 67 + <title>Test Post</title> 68 + <link>https://reddit.com/r/test/post1</link> 69 + </item> 70 + </channel> 71 + </rss>""" 72 + mock_response.raise_for_status = MagicMock() 73 + 74 + with patch("requests.get", return_value=mock_response): 75 + feed = fetcher.fetch_feed("https://reddit.com/r/test/.rss") 76 + 77 + assert feed.feed.get("title") == "Test Feed" 78 + assert len(feed.entries) == 1 79 + assert feed.entries[0].title == "Test Post" 80 + 81 + def test_sends_user_agent_header(self, fetcher): 82 + """Test that User-Agent header is sent.""" 83 + mock_response = MagicMock() 84 + mock_response.status_code = 200 85 + mock_response.content = b"<rss><channel></channel></rss>" 86 + mock_response.raise_for_status = MagicMock() 87 + 88 + with patch("requests.get", return_value=mock_response) as mock_get: 89 + fetcher.fetch_feed("https://reddit.com/r/test/.rss") 90 + 91 + call_kwargs = mock_get.call_args[1] 92 + assert "User-Agent" in call_kwargs["headers"] 93 + assert call_kwargs["headers"]["User-Agent"] == fetcher.user_agent 94 + 95 + def test_retries_on_failure(self, fetcher): 96 + """Test that fetch is retried on failure.""" 97 + mock_response = MagicMock() 98 + mock_response.status_code = 200 99 + mock_response.content = b"<rss><channel></channel></rss>" 100 + mock_response.raise_for_status = MagicMock() 101 + 102 + # First call fails, second succeeds 103 + with patch("requests.get", side_effect=[ 104 + requests.ConnectionError("Connection failed"), 105 + mock_response 106 + ]) as mock_get: 107 + with patch("time.sleep"): # Don't actually sleep in tests 108 + feed = fetcher.fetch_feed("https://reddit.com/r/test/.rss") 109 + 110 + assert mock_get.call_count == 2 111 + assert feed is not None 112 + 113 + def test_raises_after_max_retries(self, fetcher): 114 + """Test that exception is raised after max retries exhausted.""" 115 + error = requests.ConnectionError("Connection failed") 116 + 117 + with patch("requests.get", side_effect=error): 118 + with patch("time.sleep"): 119 + with pytest.raises(requests.ConnectionError): 120 + fetcher.fetch_feed("https://reddit.com/r/test/.rss") 121 + 122 + def test_exponential_backoff(self, fetcher): 123 + """Test that exponential backoff is used between retries.""" 124 + mock_response = MagicMock() 125 + mock_response.status_code = 200 126 + mock_response.content = b"<rss><channel></channel></rss>" 127 + mock_response.raise_for_status = MagicMock() 128 + 129 + with patch("requests.get", side_effect=[ 130 + requests.Timeout("Timeout"), 131 + mock_response 132 + ]): 133 + with patch("time.sleep") as mock_sleep: 134 + fetcher.fetch_feed("https://reddit.com/r/test/.rss") 135 + 136 + # First retry should have 1 second delay (2^0) 137 + mock_sleep.assert_called_once_with(1) 138 + 139 + def test_uses_timeout(self, fetcher): 140 + """Test that timeout is passed to requests.""" 141 + mock_response = MagicMock() 142 + mock_response.status_code = 200 143 + mock_response.content = b"<rss><channel></channel></rss>" 144 + mock_response.raise_for_status = MagicMock() 145 + 146 + with patch("requests.get", return_value=mock_response) as mock_get: 147 + fetcher.fetch_feed("https://reddit.com/r/test/.rss") 148 + 149 + call_kwargs = mock_get.call_args[1] 150 + assert call_kwargs["timeout"] == fetcher.timeout 151 + 152 + def test_raises_for_http_error(self, fetcher): 153 + """Test that HTTP errors are propagated.""" 154 + mock_response = MagicMock() 155 + mock_response.status_code = 404 156 + mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found") 157 + 158 + with patch("requests.get", return_value=mock_response): 159 + with patch("time.sleep"): 160 + with pytest.raises(requests.HTTPError): 161 + fetcher.fetch_feed("https://reddit.com/r/nonexistent/.rss") 162 + 163 + def test_handles_rate_limit(self, fetcher): 164 + """Test that 429 Too Many Requests is retried.""" 165 + mock_response_429 = MagicMock() 166 + mock_response_429.status_code = 429 167 + mock_response_429.raise_for_status.side_effect = requests.HTTPError("429 Too Many Requests") 168 + 169 + mock_response_ok = MagicMock() 170 + mock_response_ok.status_code = 200 171 + mock_response_ok.content = b"<rss><channel><title>Success</title></channel></rss>" 172 + mock_response_ok.raise_for_status = MagicMock() 173 + 174 + with patch("requests.get", side_effect=[mock_response_429, mock_response_ok]): 175 + with patch("time.sleep"): 176 + feed = fetcher.fetch_feed("https://reddit.com/r/test/.rss") 177 + 178 + assert feed.feed.get("title") == "Success"

+299

aggregators/reddit-highlights/tests/test_state_manager.py

··· 1 + """ 2 + Tests for state_manager module. 3 + """ 4 + import json 5 + import pytest 6 + from datetime import datetime, timedelta 7 + from pathlib import Path 8 + from unittest.mock import patch, MagicMock 9 + 10 + from src.state_manager import StateManager 11 + 12 + 13 + class TestStateManagerInit: 14 + """Tests for StateManager initialization.""" 15 + 16 + def test_creates_new_state_file(self, tmp_path): 17 + """Test that new state file is created when it doesn't exist.""" 18 + state_file = tmp_path / "state.json" 19 + manager = StateManager(state_file) 20 + 21 + assert state_file.exists() 22 + assert manager.state == {"feeds": {}} 23 + 24 + def test_loads_existing_state_file(self, tmp_path): 25 + """Test that existing state file is loaded.""" 26 + state_file = tmp_path / "state.json" 27 + existing_state = { 28 + "feeds": { 29 + "nba": { 30 + "posted_guids": [{"guid": "test123", "post_uri": "at://...", "posted_at": "2024-01-01T00:00:00"}], 31 + "last_successful_run": "2024-01-01T00:00:00", 32 + } 33 + } 34 + } 35 + state_file.write_text(json.dumps(existing_state)) 36 + 37 + manager = StateManager(state_file) 38 + assert manager.state == existing_state 39 + 40 + def test_handles_corrupted_state_file(self, tmp_path): 41 + """Test that corrupted state file is handled gracefully.""" 42 + state_file = tmp_path / "state.json" 43 + state_file.write_text("not valid json {{{") 44 + 45 + manager = StateManager(state_file) 46 + 47 + # Should create new state and backup corrupted file 48 + assert manager.state == {"feeds": {}} 49 + backup_file = tmp_path / "state.json.corrupted" 50 + assert backup_file.exists() 51 + assert backup_file.read_text() == "not valid json {{{" 52 + 53 + def test_creates_parent_directories(self, tmp_path): 54 + """Test that parent directories are created if needed.""" 55 + state_file = tmp_path / "nested" / "deep" / "state.json" 56 + manager = StateManager(state_file) 57 + 58 + assert state_file.exists() 59 + assert manager.state == {"feeds": {}} 60 + 61 + 62 + class TestStateManagerIsPosted: 63 + """Tests for is_posted method.""" 64 + 65 + def test_returns_false_for_new_feed(self, tmp_path): 66 + """Test that is_posted returns False for new feed.""" 67 + state_file = tmp_path / "state.json" 68 + manager = StateManager(state_file) 69 + 70 + assert not manager.is_posted("nba", "newguid123") 71 + 72 + def test_returns_false_for_unposted_guid(self, tmp_path): 73 + """Test that is_posted returns False for unposted GUID.""" 74 + state_file = tmp_path / "state.json" 75 + manager = StateManager(state_file) 76 + manager.mark_posted("nba", "existingguid", "at://test") 77 + 78 + assert not manager.is_posted("nba", "differentguid") 79 + 80 + def test_returns_true_for_posted_guid(self, tmp_path): 81 + """Test that is_posted returns True for posted GUID.""" 82 + state_file = tmp_path / "state.json" 83 + manager = StateManager(state_file) 84 + manager.mark_posted("nba", "postedguid", "at://test") 85 + 86 + assert manager.is_posted("nba", "postedguid") 87 + 88 + 89 + class TestStateManagerMarkPosted: 90 + """Tests for mark_posted method.""" 91 + 92 + def test_marks_guid_as_posted(self, tmp_path): 93 + """Test that GUID is marked as posted.""" 94 + state_file = tmp_path / "state.json" 95 + manager = StateManager(state_file) 96 + 97 + manager.mark_posted("nba", "testguid", "at://did:plc:test/post/abc") 98 + 99 + assert manager.is_posted("nba", "testguid") 100 + 101 + def test_saves_state_to_file(self, tmp_path): 102 + """Test that state is persisted to file.""" 103 + state_file = tmp_path / "state.json" 104 + manager = StateManager(state_file) 105 + 106 + manager.mark_posted("nba", "testguid", "at://test") 107 + 108 + # Create new manager from same file 109 + manager2 = StateManager(state_file) 110 + assert manager2.is_posted("nba", "testguid") 111 + 112 + def test_stores_post_uri(self, tmp_path): 113 + """Test that post URI is stored.""" 114 + state_file = tmp_path / "state.json" 115 + manager = StateManager(state_file) 116 + 117 + manager.mark_posted("nba", "testguid", "at://did:plc:test/post/xyz") 118 + 119 + # Check the stored data 120 + posted_guids = manager.state["feeds"]["nba"]["posted_guids"] 121 + entry = next(e for e in posted_guids if e["guid"] == "testguid") 122 + assert entry["post_uri"] == "at://did:plc:test/post/xyz" 123 + 124 + def test_stores_timestamp(self, tmp_path): 125 + """Test that posted_at timestamp is stored.""" 126 + state_file = tmp_path / "state.json" 127 + manager = StateManager(state_file) 128 + 129 + manager.mark_posted("nba", "testguid", "at://test") 130 + 131 + posted_guids = manager.state["feeds"]["nba"]["posted_guids"] 132 + entry = next(e for e in posted_guids if e["guid"] == "testguid") 133 + # Should be a valid ISO timestamp 134 + datetime.fromisoformat(entry["posted_at"]) 135 + 136 + 137 + class TestStateManagerAtomicSave: 138 + """Tests for atomic save functionality.""" 139 + 140 + def test_atomic_write_uses_temp_file(self, tmp_path): 141 + """Test that atomic write uses temp file.""" 142 + state_file = tmp_path / "state.json" 143 + manager = StateManager(state_file) 144 + 145 + # The temp file should not exist after successful save 146 + temp_file = tmp_path / "state.json.tmp" 147 + manager.mark_posted("nba", "test", "at://test") 148 + assert not temp_file.exists() 149 + 150 + def test_write_error_cleans_up_temp_file(self, tmp_path): 151 + """Test that temp file is cleaned up on write error.""" 152 + state_file = tmp_path / "state.json" 153 + manager = StateManager(state_file) 154 + 155 + temp_file = tmp_path / "state.json.tmp" 156 + 157 + # Mock rename to fail 158 + with patch("pathlib.Path.rename", side_effect=OSError("Mock error")): 159 + with pytest.raises(OSError): 160 + manager._save_state({"feeds": {}}) 161 + 162 + # Temp file should be cleaned up 163 + assert not temp_file.exists() 164 + 165 + 166 + class TestStateManagerCleanup: 167 + """Tests for cleanup_old_entries method.""" 168 + 169 + def test_removes_entries_older_than_max_age(self, tmp_path): 170 + """Test that entries older than max_age_days are removed.""" 171 + state_file = tmp_path / "state.json" 172 + manager = StateManager(state_file, max_age_days=7) 173 + 174 + # Add old entry manually 175 + old_date = (datetime.now() - timedelta(days=10)).isoformat() 176 + manager.state["feeds"]["nba"] = { 177 + "posted_guids": [{"guid": "old", "post_uri": "at://old", "posted_at": old_date}], 178 + "last_successful_run": None, 179 + } 180 + 181 + # Add new entry 182 + manager.mark_posted("nba", "new", "at://new") 183 + 184 + # Old entry should be removed after cleanup (triggered by mark_posted) 185 + assert not manager.is_posted("nba", "old") 186 + assert manager.is_posted("nba", "new") 187 + 188 + def test_keeps_entries_within_max_guids(self, tmp_path): 189 + """Test that only max_guids_per_feed entries are kept.""" 190 + state_file = tmp_path / "state.json" 191 + manager = StateManager(state_file, max_guids_per_feed=3) 192 + 193 + # Add 5 entries 194 + for i in range(5): 195 + manager.mark_posted("nba", f"guid{i}", f"at://uri{i}") 196 + 197 + # Only most recent 3 should remain 198 + assert manager.get_posted_count("nba") == 3 199 + 200 + def test_handles_malformed_timestamps(self, tmp_path): 201 + """Test that malformed timestamps are handled gracefully.""" 202 + state_file = tmp_path / "state.json" 203 + manager = StateManager(state_file) 204 + 205 + # Add entry with malformed timestamp 206 + manager.state["feeds"]["nba"] = { 207 + "posted_guids": [ 208 + {"guid": "malformed", "post_uri": "at://test", "posted_at": "not-a-date"}, 209 + {"guid": "valid", "post_uri": "at://test2", "posted_at": datetime.now().isoformat()}, 210 + ], 211 + "last_successful_run": None, 212 + } 213 + 214 + # Cleanup should handle malformed entry without crashing 215 + manager.cleanup_old_entries("nba") 216 + 217 + # Malformed entry should be removed, valid should remain 218 + assert not manager.is_posted("nba", "malformed") 219 + assert manager.is_posted("nba", "valid") 220 + 221 + 222 + class TestStateManagerLastRun: 223 + """Tests for get_last_run and update_last_run methods.""" 224 + 225 + def test_get_last_run_returns_none_for_new_feed(self, tmp_path): 226 + """Test that get_last_run returns None for new feed.""" 227 + state_file = tmp_path / "state.json" 228 + manager = StateManager(state_file) 229 + 230 + assert manager.get_last_run("nba") is None 231 + 232 + def test_update_and_get_last_run(self, tmp_path): 233 + """Test updating and retrieving last run timestamp.""" 234 + state_file = tmp_path / "state.json" 235 + manager = StateManager(state_file) 236 + 237 + now = datetime.now() 238 + manager.update_last_run("nba", now) 239 + 240 + result = manager.get_last_run("nba") 241 + # Compare without microseconds (ISO format may lose precision) 242 + assert result.replace(microsecond=0) == now.replace(microsecond=0) 243 + 244 + def test_last_run_persisted_to_file(self, tmp_path): 245 + """Test that last run is persisted to file.""" 246 + state_file = tmp_path / "state.json" 247 + manager = StateManager(state_file) 248 + 249 + now = datetime.now() 250 + manager.update_last_run("nba", now) 251 + 252 + # Create new manager from same file 253 + manager2 = StateManager(state_file) 254 + result = manager2.get_last_run("nba") 255 + assert result.replace(microsecond=0) == now.replace(microsecond=0) 256 + 257 + 258 + class TestStateManagerGetPostedGuids: 259 + """Tests for get_all_posted_guids method.""" 260 + 261 + def test_returns_empty_list_for_new_feed(self, tmp_path): 262 + """Test that empty list is returned for new feed.""" 263 + state_file = tmp_path / "state.json" 264 + manager = StateManager(state_file) 265 + 266 + assert manager.get_all_posted_guids("nba") == [] 267 + 268 + def test_returns_all_guids(self, tmp_path): 269 + """Test that all posted GUIDs are returned.""" 270 + state_file = tmp_path / "state.json" 271 + manager = StateManager(state_file) 272 + 273 + manager.mark_posted("nba", "guid1", "at://1") 274 + manager.mark_posted("nba", "guid2", "at://2") 275 + manager.mark_posted("nba", "guid3", "at://3") 276 + 277 + guids = manager.get_all_posted_guids("nba") 278 + assert set(guids) == {"guid1", "guid2", "guid3"} 279 + 280 + 281 + class TestStateManagerGetPostedCount: 282 + """Tests for get_posted_count method.""" 283 + 284 + def test_returns_zero_for_new_feed(self, tmp_path): 285 + """Test that zero is returned for new feed.""" 286 + state_file = tmp_path / "state.json" 287 + manager = StateManager(state_file) 288 + 289 + assert manager.get_posted_count("nba") == 0 290 + 291 + def test_returns_correct_count(self, tmp_path): 292 + """Test that correct count is returned.""" 293 + state_file = tmp_path / "state.json" 294 + manager = StateManager(state_file) 295 + 296 + manager.mark_posted("nba", "guid1", "at://1") 297 + manager.mark_posted("nba", "guid2", "at://2") 298 + 299 + assert manager.get_posted_count("nba") == 2