| 1 | import json |
| 2 | import tempfile |
| 3 | import unittest |
| 4 | from argparse import Namespace |
| 5 | from datetime import datetime |
| 6 | from pathlib import Path |
| 7 | from unittest import mock |
| 8 | |
| 9 | import scripts.crawl as crawl |
| 10 | |
| 11 | |
| 12 | class CrawlTests(unittest.TestCase): |
| 13 | def test_significance_skip_reason_allows_common_description_terms(self) -> None: |
| 14 | repo = { |
| 15 | "name": "awesome-tool", |
| 16 | "description": "Includes sample data and a live demo for deployment.", |
| 17 | "topics": ["ai", "demo"], |
| 18 | "fork": False, |
| 19 | "is_template": False, |
| 20 | } |
| 21 | |
| 22 | self.assertIsNone(crawl.significance_skip_reason(repo)) |
| 23 | |
| 24 | def test_significance_skip_reason_still_flags_name_tokens(self) -> None: |
| 25 | repo = { |
| 26 | "name": "starter-kit", |
| 27 | "description": "Production-ready auth service.", |
| 28 | "topics": ["ai"], |
| 29 | "fork": False, |
| 30 | "is_template": False, |
| 31 | } |
| 32 | |
| 33 | self.assertEqual(crawl.significance_skip_reason(repo), "low_signal_keyword") |
| 34 | |
| 35 | def test_get_json_preserves_payload_contract(self) -> None: |
| 36 | client = crawl.GitHubClient("token") |
| 37 | entry = crawl.CacheEntry( |
| 38 | status=200, payload={"ok": True}, headers={}, fetched_at=crawl.utc_now() |
| 39 | ) |
| 40 | |
| 41 | with mock.patch.object(client, "get_json_entry", return_value=entry): |
| 42 | self.assertEqual(client.get_json("https://example.com"), {"ok": True}) |
| 43 | |
| 44 | def test_load_previous_star_snapshot_checks_all_raw_dirs_and_logs_failures(self) -> None: |
| 45 | tests_root = Path(__file__).resolve().parent |
| 46 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 47 | base = Path(tmpdir) |
| 48 | snapshot_dir = base / "snapshots" |
| 49 | raw_default_dir = base / "raw-default" |
| 50 | custom_output_dir = base / "custom-output" |
| 51 | snapshot_dir.mkdir() |
| 52 | raw_default_dir.mkdir() |
| 53 | custom_output_dir.mkdir() |
| 54 | |
| 55 | (snapshot_dir / "2026-W21-stars.json").write_text( |
| 56 | '{"week": "2026-W21", "stars": {"owner/current": 1}}\n', encoding="utf-8" |
| 57 | ) |
| 58 | (raw_default_dir / "2026-W20.json").write_text("{not-json}\n", encoding="utf-8") |
| 59 | (raw_default_dir / "2026-W19.json").write_text( |
| 60 | '{"week": "2026-W19", "stars": {"owner/older": 42}}\n', encoding="utf-8" |
| 61 | ) |
| 62 | |
| 63 | with mock.patch.object(crawl, "log") as log_mock: |
| 64 | stars = crawl.load_previous_star_snapshot( |
| 65 | snapshot_dir, "2026-W21", custom_output_dir, raw_default_dir |
| 66 | ) |
| 67 | |
| 68 | self.assertEqual(stars, {"owner/older": 42}) |
| 69 | logged = "\n".join(call.args[0] for call in log_mock.call_args_list) |
| 70 | self.assertIn("invalid JSON", logged) |
| 71 | |
| 72 | def test_validate_payload_accepts_rate_limit_metadata_schema(self) -> None: |
| 73 | payload = { |
| 74 | "week": "2026-W21", |
| 75 | "crawled_at": "2026-05-18T10:00:00Z", |
| 76 | "new_repos": [], |
| 77 | "trending_repos": [], |
| 78 | "signals": {"top_topics": []}, |
| 79 | "metadata": { |
| 80 | "api_calls_used": 1, |
| 81 | "cache_hits": 2, |
| 82 | "stale_cache_hits": 3, |
| 83 | "rate_limit_limit": 5000, |
| 84 | "rate_limit_remaining": 4990, |
| 85 | "rate_limit_reset": 1747562400, |
| 86 | "rate_limit_resource": "search", |
| 87 | "partial_failures": [], |
| 88 | "snapshot_path": "data/snapshots/2026-W21-stars.json", |
| 89 | }, |
| 90 | } |
| 91 | |
| 92 | crawl.validate_payload(payload) |
| 93 | |
| 94 | def test_pause_for_rate_limit_cools_down_without_reset_hint(self) -> None: |
| 95 | client = crawl.GitHubClient("token") |
| 96 | client.rate_limit_limit = 5000 |
| 97 | client.rate_limit_remaining = 0 |
| 98 | client.rate_limit_resource = "core" |
| 99 | |
| 100 | with mock.patch("scripts.crawl.time.sleep") as sleep_mock: |
| 101 | client._pause_for_rate_limit("https://example.com") |
| 102 | |
| 103 | sleep_mock.assert_called_once() |
| 104 | |
| 105 | def test_main_uses_open_ended_queries_for_live_runs(self) -> None: |
| 106 | queries: list[str] = [] |
| 107 | |
| 108 | class FakeClient: |
| 109 | def __init__(self, token: str, **kwargs) -> None: |
| 110 | self.token = token |
| 111 | self.api_calls_used = 0 |
| 112 | self.cache_hits = 0 |
| 113 | self.stale_cache_hits = 0 |
| 114 | self.rate_limit_limit = None |
| 115 | self.rate_limit_remaining = None |
| 116 | self.rate_limit_reset = None |
| 117 | self.rate_limit_resource = None |
| 118 | self.errors = [] |
| 119 | |
| 120 | def search_repositories(self, query: str, *, max_results: int = 1000): |
| 121 | queries.append(query) |
| 122 | return [] |
| 123 | |
| 124 | def has_readme(self, full_name: str) -> bool: |
| 125 | return True |
| 126 | |
| 127 | args = Namespace( |
| 128 | since="2026-05-11", |
| 129 | as_of=None, |
| 130 | max_results=25, |
| 131 | output="data/raw/test-live.json", |
| 132 | topic=None, |
| 133 | config=None, |
| 134 | ) |
| 135 | with ( |
| 136 | mock.patch.object(crawl, "parse_args", return_value=args), |
| 137 | mock.patch.dict("os.environ", {"GITHUB_TOKEN": "token"}, clear=False), |
| 138 | mock.patch.object(crawl, "GitHubClient", FakeClient), |
| 139 | mock.patch.object(crawl, "load_previous_star_snapshot", return_value={}), |
| 140 | mock.patch.object(crawl, "write_payload"), |
| 141 | mock.patch.object(crawl, "print"), |
| 142 | ): |
| 143 | exit_code = crawl.main() |
| 144 | |
| 145 | self.assertEqual(exit_code, 0) |
| 146 | self.assertEqual(queries, ["created:>2026-05-11 stars:>50", "pushed:>2026-05-11 stars:>50"]) |
| 147 | |
| 148 | def test_main_uses_bounded_queries_for_backfills_and_fails_on_partial_errors(self) -> None: |
| 149 | queries: list[str] = [] |
| 150 | |
| 151 | class FakeClient: |
| 152 | def __init__(self, token: str, **kwargs) -> None: |
| 153 | self.token = token |
| 154 | self.api_calls_used = 0 |
| 155 | self.cache_hits = 0 |
| 156 | self.stale_cache_hits = 0 |
| 157 | self.rate_limit_limit = None |
| 158 | self.rate_limit_remaining = None |
| 159 | self.rate_limit_reset = None |
| 160 | self.rate_limit_resource = None |
| 161 | self.errors = ["README lookup failed"] |
| 162 | |
| 163 | def search_repositories(self, query: str, *, max_results: int = 1000): |
| 164 | queries.append(query) |
| 165 | return [] |
| 166 | |
| 167 | def has_readme(self, full_name: str) -> bool: |
| 168 | return True |
| 169 | |
| 170 | args = Namespace( |
| 171 | since="2026-05-11", |
| 172 | as_of="2026-05-18", |
| 173 | max_results=25, |
| 174 | output="data/raw/test-backfill.json", |
| 175 | topic=None, |
| 176 | config=None, |
| 177 | ) |
| 178 | with ( |
| 179 | mock.patch.object(crawl, "parse_args", return_value=args), |
| 180 | mock.patch.dict("os.environ", {"GITHUB_TOKEN": "token"}, clear=False), |
| 181 | mock.patch.object(crawl, "GitHubClient", FakeClient), |
| 182 | mock.patch.object(crawl, "load_previous_star_snapshot", return_value={}), |
| 183 | mock.patch.object(crawl, "write_payload"), |
| 184 | mock.patch.object(crawl, "print"), |
| 185 | ): |
| 186 | exit_code = crawl.main() |
| 187 | |
| 188 | self.assertEqual(exit_code, 1) |
| 189 | self.assertEqual( |
| 190 | queries, |
| 191 | [ |
| 192 | "created:2026-05-11..2026-05-18 stars:>50", |
| 193 | "pushed:2026-05-11..2026-05-18 stars:>50", |
| 194 | ], |
| 195 | ) |
| 196 | |
| 197 | def test_main_emits_observability_ledger(self) -> None: |
| 198 | class FakeClient: |
| 199 | def __init__(self, token: str, **kwargs) -> None: |
| 200 | self.token = token |
| 201 | self.api_calls_used = 4 |
| 202 | self.cache_hits = 3 |
| 203 | self.cache_misses = 4 |
| 204 | self.stale_cache_hits = 1 |
| 205 | self.rate_limit_events = 2 |
| 206 | self.secondary_rate_limit_hit = True |
| 207 | self.rate_limit_limit = 5000 |
| 208 | self.rate_limit_remaining = 4988 |
| 209 | self.rate_limit_reset = 1747562400 |
| 210 | self.rate_limit_resource = "search" |
| 211 | self.errors = [] |
| 212 | |
| 213 | def search_repositories(self, query: str, *, max_results: int = 1000): |
| 214 | return [] |
| 215 | |
| 216 | def has_readme(self, full_name: str) -> bool: |
| 217 | return True |
| 218 | |
| 219 | args = Namespace( |
| 220 | since="2026-05-11", |
| 221 | as_of="2026-05-18", |
| 222 | max_results=25, |
| 223 | output="data/raw/test-observability.json", |
| 224 | topic="general", |
| 225 | config=None, |
| 226 | ) |
| 227 | with ( |
| 228 | mock.patch.object(crawl, "parse_args", return_value=args), |
| 229 | mock.patch.dict( |
| 230 | "os.environ", {"GITHUB_TOKEN": "token", "GITHUB_RUN_ID": "123"}, clear=False |
| 231 | ), |
| 232 | mock.patch.object(crawl, "GitHubClient", FakeClient), |
| 233 | mock.patch.object(crawl, "load_previous_star_snapshot", return_value={}), |
| 234 | mock.patch.object(crawl, "write_payload"), |
| 235 | mock.patch.object(crawl, "emit_ledger") as emit_mock, |
| 236 | mock.patch.object(crawl, "print"), |
| 237 | ): |
| 238 | exit_code = crawl.main() |
| 239 | |
| 240 | self.assertEqual(exit_code, 0) |
| 241 | ledger = emit_mock.call_args.args[0] |
| 242 | output_path = emit_mock.call_args.args[1] |
| 243 | self.assertEqual(ledger.schema_version, "observability_v1") |
| 244 | self.assertEqual(ledger.run_id, "123") |
| 245 | self.assertEqual(ledger.crawl_metrics[0].source_type, "github") |
| 246 | self.assertEqual(ledger.crawl_metrics[0].cache_misses, 4) |
| 247 | self.assertTrue(ledger.crawl_metrics[0].secondary_rate_limit_hit) |
| 248 | self.assertTrue(output_path.as_posix().endswith("-github-crawl.json")) |
| 249 | |
| 250 | def test_load_reusable_github_payload_accepts_same_day_matching_artifact(self) -> None: |
| 251 | tests_root = Path(__file__).resolve().parent |
| 252 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 253 | base = Path(tmpdir) |
| 254 | output = base / "data/raw/2026-W21.json" |
| 255 | args = Namespace( |
| 256 | since="2026-05-12", |
| 257 | as_of="2026-05-19", |
| 258 | max_results=25, |
| 259 | output=str(output), |
| 260 | topic=None, |
| 261 | config=None, |
| 262 | ) |
| 263 | since = datetime(2026, 5, 12, tzinfo=crawl.UTC) |
| 264 | window_end = datetime(2026, 5, 19, tzinfo=crawl.UTC) |
| 265 | crawled_at = datetime(2026, 5, 19, 8, 0, tzinfo=crawl.UTC) |
| 266 | checksum = crawl.github_crawl_config_checksum(args, since, window_end, 25) |
| 267 | payload = { |
| 268 | "week": "2026-W21", |
| 269 | "crawled_at": crawl.iso_timestamp(crawled_at), |
| 270 | "new_repos": [], |
| 271 | "trending_repos": [], |
| 272 | "signals": {"top_topics": []}, |
| 273 | "metadata": { |
| 274 | "api_calls_used": 1, |
| 275 | "cache_hits": 0, |
| 276 | "stale_cache_hits": 0, |
| 277 | "rate_limit_limit": None, |
| 278 | "rate_limit_remaining": None, |
| 279 | "rate_limit_reset": None, |
| 280 | "rate_limit_resource": None, |
| 281 | "partial_failures": [], |
| 282 | "snapshot_path": "data/snapshots/2026-W21-stars.json", |
| 283 | "crawl_window": {"since": "2026-05-12", "until": "2026-05-19"}, |
| 284 | "crawl_config_checksum": checksum, |
| 285 | "schema_checksum": crawl.github_schema_checksum(), |
| 286 | "same_day_reuse": {"status": "not_reused", "source": "github"}, |
| 287 | }, |
| 288 | } |
| 289 | payload["metadata"]["artifact_checksum"] = crawl.github_artifact_checksum(payload) |
| 290 | crawl.write_payload(output, payload) |
| 291 | |
| 292 | reused = crawl.load_reusable_github_payload( |
| 293 | output, |
| 294 | week="2026-W21", |
| 295 | crawled_at=datetime(2026, 5, 19, 10, 0, tzinfo=crawl.UTC), |
| 296 | since=since, |
| 297 | window_end=window_end, |
| 298 | config_checksum=checksum, |
| 299 | ) |
| 300 | |
| 301 | self.assertIsNotNone(reused) |
| 302 | self.assertEqual(reused["metadata"]["same_day_reuse"]["status"], "reused") |
| 303 | self.assertEqual(reused["metadata"]["same_day_reuse"]["source_id"], "github-search") |
| 304 | |
| 305 | def test_main_reuses_valid_same_day_raw_artifact_without_github_token(self) -> None: |
| 306 | tests_root = Path(__file__).resolve().parent |
| 307 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 308 | base = Path(tmpdir) |
| 309 | existing = base / "reuse/raw-data/2026-W21.json" |
| 310 | output = base / "data/raw/2026-W21.json" |
| 311 | args = Namespace( |
| 312 | since="2026-05-12", |
| 313 | as_of="2026-05-19", |
| 314 | max_results=25, |
| 315 | output=str(output), |
| 316 | topic=None, |
| 317 | config=None, |
| 318 | reuse_artifact=str(existing), |
| 319 | source_refresh_policy="reuse-same-day", |
| 320 | run_started_at="2026-05-19T10:00:00Z", |
| 321 | current_code_sha="sha", |
| 322 | ) |
| 323 | since = datetime(2026, 5, 12, tzinfo=crawl.UTC) |
| 324 | window_end = datetime(2026, 5, 19, tzinfo=crawl.UTC) |
| 325 | checksum = crawl.github_crawl_config_checksum(args, since, window_end, 25) |
| 326 | payload = { |
| 327 | "week": "2026-W21", |
| 328 | "crawled_at": "2026-05-19T08:00:00Z", |
| 329 | "new_repos": [], |
| 330 | "trending_repos": [], |
| 331 | "signals": {"top_topics": []}, |
| 332 | "metadata": { |
| 333 | "api_calls_used": 1, |
| 334 | "cache_hits": 0, |
| 335 | "stale_cache_hits": 0, |
| 336 | "rate_limit_limit": None, |
| 337 | "rate_limit_remaining": None, |
| 338 | "rate_limit_reset": None, |
| 339 | "rate_limit_resource": None, |
| 340 | "partial_failures": [], |
| 341 | "run_id": "111111", |
| 342 | "snapshot_path": "data/snapshots/2026-W21-stars.json", |
| 343 | "crawl_window": {"since": "2026-05-12", "until": "2026-05-19"}, |
| 344 | "crawl_config_checksum": checksum, |
| 345 | "schema_checksum": crawl.github_schema_checksum(), |
| 346 | "same_day_reuse": { |
| 347 | "status": "not_reused", |
| 348 | "source": "github", |
| 349 | "source_id": crawl.GITHUB_SOURCE_ID, |
| 350 | }, |
| 351 | "crawler_code_sha": "sha", |
| 352 | }, |
| 353 | } |
| 354 | payload["metadata"]["artifact_checksum"] = crawl.github_artifact_checksum(payload) |
| 355 | crawl.write_payload(existing, payload) |
| 356 | |
| 357 | with ( |
| 358 | mock.patch.object(crawl, "parse_args", return_value=args), |
| 359 | mock.patch.dict("os.environ", {}, clear=True), |
| 360 | mock.patch.object( |
| 361 | crawl, "utc_now", return_value=datetime(2026, 5, 19, 10, 0, tzinfo=crawl.UTC) |
| 362 | ), |
| 363 | ): |
| 364 | exit_code = crawl.main() |
| 365 | |
| 366 | self.assertEqual(exit_code, 0) |
| 367 | reused = json.loads(output.read_text(encoding="utf-8")) |
| 368 | self.assertEqual(reused["metadata"]["same_day_reuse"]["status"], "reused") |
| 369 | self.assertEqual(reused["metadata"]["source_refresh_policy"], "reuse-same-day") |
| 370 | |
| 371 | def test_main_emits_github_source_id_in_same_day_reuse_metadata(self) -> None: |
| 372 | class FakeClient: |
| 373 | def __init__(self, token: str, **kwargs) -> None: |
| 374 | self.token = token |
| 375 | self.api_calls_used = 0 |
| 376 | self.cache_hits = 0 |
| 377 | self.stale_cache_hits = 0 |
| 378 | self.rate_limit_limit = None |
| 379 | self.rate_limit_remaining = None |
| 380 | self.rate_limit_reset = None |
| 381 | self.rate_limit_resource = None |
| 382 | self.errors = [] |
| 383 | |
| 384 | def search_repositories(self, query: str, *, max_results: int = 1000): |
| 385 | return [] |
| 386 | |
| 387 | def has_readme(self, full_name: str) -> bool: |
| 388 | return True |
| 389 | |
| 390 | tests_root = Path(__file__).resolve().parent |
| 391 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 392 | output = Path(tmpdir) / "data/raw/2026-W21.json" |
| 393 | args = Namespace( |
| 394 | since="2026-05-12", |
| 395 | as_of="2026-05-19", |
| 396 | max_results=25, |
| 397 | output=str(output), |
| 398 | topic=None, |
| 399 | config=None, |
| 400 | force_refresh=True, |
| 401 | ) |
| 402 | |
| 403 | with ( |
| 404 | mock.patch.object(crawl, "parse_args", return_value=args), |
| 405 | mock.patch.dict("os.environ", {"GITHUB_TOKEN": "token"}, clear=False), |
| 406 | mock.patch.object(crawl, "GitHubClient", FakeClient), |
| 407 | mock.patch.object(crawl, "load_previous_star_snapshot", return_value={}), |
| 408 | mock.patch.object( |
| 409 | crawl, "snapshots_dir", return_value=Path(tmpdir) / "data/snapshots" |
| 410 | ), |
| 411 | mock.patch.object( |
| 412 | crawl, "utc_now", return_value=datetime(2026, 5, 19, 10, 0, tzinfo=crawl.UTC) |
| 413 | ), |
| 414 | ): |
| 415 | exit_code = crawl.main() |
| 416 | |
| 417 | self.assertEqual(exit_code, 0) |
| 418 | payload = json.loads(output.read_text(encoding="utf-8")) |
| 419 | self.assertEqual(payload["metadata"]["same_day_reuse"]["status"], "not_reused") |
| 420 | self.assertEqual(payload["metadata"]["same_day_reuse"]["source"], "github") |
| 421 | self.assertEqual(payload["metadata"]["same_day_reuse"]["source_id"], "github-search") |
| 422 | |
| 423 | def test_load_reusable_github_payload_rejects_config_mismatch(self) -> None: |
| 424 | tests_root = Path(__file__).resolve().parent |
| 425 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 426 | base = Path(tmpdir) |
| 427 | output = base / "data/raw/2026-W21.json" |
| 428 | payload = { |
| 429 | "week": "2026-W21", |
| 430 | "crawled_at": "2026-05-19T08:00:00Z", |
| 431 | "new_repos": [], |
| 432 | "trending_repos": [], |
| 433 | "signals": {"top_topics": []}, |
| 434 | "metadata": { |
| 435 | "api_calls_used": 1, |
| 436 | "cache_hits": 0, |
| 437 | "stale_cache_hits": 0, |
| 438 | "rate_limit_limit": None, |
| 439 | "rate_limit_remaining": None, |
| 440 | "rate_limit_reset": None, |
| 441 | "rate_limit_resource": None, |
| 442 | "partial_failures": [], |
| 443 | "snapshot_path": "data/snapshots/2026-W21-stars.json", |
| 444 | "crawl_window": {"since": "2026-05-12", "until": "2026-05-19"}, |
| 445 | "crawl_config_checksum": "old", |
| 446 | "schema_checksum": crawl.github_schema_checksum(), |
| 447 | "same_day_reuse": {"status": "not_reused", "source": "github"}, |
| 448 | }, |
| 449 | } |
| 450 | payload["metadata"]["artifact_checksum"] = crawl.github_artifact_checksum(payload) |
| 451 | crawl.write_payload(output, payload) |
| 452 | |
| 453 | reused = crawl.load_reusable_github_payload( |
| 454 | output, |
| 455 | week="2026-W21", |
| 456 | crawled_at=datetime(2026, 5, 19, 10, 0, tzinfo=crawl.UTC), |
| 457 | since=datetime(2026, 5, 12, tzinfo=crawl.UTC), |
| 458 | window_end=datetime(2026, 5, 19, tzinfo=crawl.UTC), |
| 459 | config_checksum="new", |
| 460 | ) |
| 461 | |
| 462 | self.assertIsNone(reused) |
| 463 | |
| 464 | def test_load_reusable_github_payload_rejects_missing_code_fingerprint_when_required( |
| 465 | self, |
| 466 | ) -> None: |
| 467 | tests_root = Path(__file__).resolve().parent |
| 468 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 469 | base = Path(tmpdir) |
| 470 | output = base / "data/raw/2026-W21.json" |
| 471 | args = Namespace( |
| 472 | since="2026-05-12", |
| 473 | as_of="2026-05-19", |
| 474 | max_results=25, |
| 475 | output=str(output), |
| 476 | topic=None, |
| 477 | config=None, |
| 478 | ) |
| 479 | since = datetime(2026, 5, 12, tzinfo=crawl.UTC) |
| 480 | window_end = datetime(2026, 5, 19, tzinfo=crawl.UTC) |
| 481 | checksum = crawl.github_crawl_config_checksum(args, since, window_end, 25) |
| 482 | payload = { |
| 483 | "week": "2026-W21", |
| 484 | "crawled_at": "2026-05-19T08:00:00Z", |
| 485 | "new_repos": [], |
| 486 | "trending_repos": [], |
| 487 | "signals": {"top_topics": []}, |
| 488 | "metadata": { |
| 489 | "api_calls_used": 1, |
| 490 | "cache_hits": 0, |
| 491 | "stale_cache_hits": 0, |
| 492 | "rate_limit_limit": None, |
| 493 | "rate_limit_remaining": None, |
| 494 | "rate_limit_reset": None, |
| 495 | "rate_limit_resource": None, |
| 496 | "partial_failures": [], |
| 497 | "snapshot_path": "data/snapshots/2026-W21-stars.json", |
| 498 | "crawl_window": {"since": "2026-05-12", "until": "2026-05-19"}, |
| 499 | "crawl_config_checksum": checksum, |
| 500 | "schema_checksum": crawl.github_schema_checksum(), |
| 501 | "same_day_reuse": {"status": "not_reused", "source": "github"}, |
| 502 | }, |
| 503 | } |
| 504 | payload["metadata"]["artifact_checksum"] = crawl.github_artifact_checksum(payload) |
| 505 | crawl.write_payload(output, payload) |
| 506 | |
| 507 | reused = crawl.load_reusable_github_payload( |
| 508 | output, |
| 509 | week="2026-W21", |
| 510 | crawled_at=datetime(2026, 5, 19, 10, 0, tzinfo=crawl.UTC), |
| 511 | since=since, |
| 512 | window_end=window_end, |
| 513 | config_checksum=checksum, |
| 514 | current_code_sha="sha", |
| 515 | ) |
| 516 | |
| 517 | self.assertIsNone(reused) |
| 518 | |
| 519 | def test_restore_reused_snapshot_rejects_unsafe_metadata_path(self) -> None: |
| 520 | tests_root = Path(__file__).resolve().parent |
| 521 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 522 | base = Path(tmpdir) |
| 523 | reuse_path = base / "reuse/raw/2026-W21.json" |
| 524 | source_snapshot = base / "reuse/snapshots/2026-W21-stars.json" |
| 525 | source_snapshot.parent.mkdir(parents=True) |
| 526 | source_snapshot.write_text('{"stars": {"owner/repo": 1}}\n', encoding="utf-8") |
| 527 | |
| 528 | for unsafe_path in ( |
| 529 | "/home/azureuser/source/SquadScope/data/snapshots/2026-W21-stars.json", |
| 530 | "data/snapshots/../raw/evil.json", |
| 531 | ): |
| 532 | with mock.patch.object(crawl, "write_payload") as write_mock: |
| 533 | crawl.restore_reused_snapshot(reuse_path, {"snapshot_path": unsafe_path}) |
| 534 | |
| 535 | write_mock.assert_not_called() |
| 536 | |
| 537 | |
| 538 | if __name__ == "__main__": |
| 539 | unittest.main() |