#!/usr/bin/env python3 """ Tests for the request-budget logic in gdl-sync.py. python3 -m unittest discover -s scripts -p 'test_*.py' Deliberately stdlib-only, so it runs anywhere the sync itself runs. What is covered here is the part that decides whether to spend a request — the part whose absence got the archive's Instagram account suspended. """ import datetime as dt import importlib.util import json import sys import tempfile import unittest from pathlib import Path _spec = importlib.util.spec_from_file_location( "gdl_sync", Path(__file__).with_name("gdl-sync.py")) gdl = importlib.util.module_from_spec(_spec) sys.modules["gdl_sync"] = gdl _spec.loader.exec_module(gdl) NOW = dt.datetime(2026, 8, 18, 12, 0, tzinfo=dt.timezone.utc) NOW_TS = NOW.timestamp() def ago(hours: float) -> str: return (NOW - dt.timedelta(hours=hours)).isoformat() class SourceSelection(unittest.TestCase): def test_only_stories_is_a_single_cheap_source(self): srcs = gdl.Profile("u").sources(gdl.STORIES_ONLY) self.assertEqual([s.kind for s in srcs], ["stories"]) self.assertEqual(srcs[0].directory, "story - u") def test_full_sync_covers_every_surface(self): srcs = gdl.Profile("u").sources(set(gdl.ALL_KINDS)) self.assertEqual([s.kind for s in srcs], list(gdl.ALL_KINDS)) def test_reels_and_stories_go_to_their_own_directories(self): by_kind = {s.kind: s for s in gdl.Profile("u").sources(set(gdl.ALL_KINDS))} self.assertEqual(by_kind["posts"].directory, "u") self.assertEqual(by_kind["reels"].directory, "u - reels") # Highlights derive their directory from the title mid-extraction. self.assertEqual(by_kind["highlights"].directory, "") class PlanSource(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.state = gdl.SyncState(Path(self.tmp.name) / "state.json") self.posts = gdl.Profile("u").sources({"posts"})[0] self.stories = gdl.Profile("u").sources({"stories"})[0] def tearDown(self): self.tmp.cleanup() def test_first_run_seeds(self): fetch, seed, _ = gdl.plan_source(self.posts, self.state, NOW_TS, 20) self.assertTrue(fetch) self.assertTrue(seed) def test_seeding_happens_only_once(self): self.state.mark_seeded(self.posts.url, ago(720)) fetch, seed, reason = gdl.plan_source(self.posts, self.state, NOW_TS, 20) self.assertTrue(fetch) self.assertFalse(seed, "a seeded source must never be re-probed") self.assertIn("already seeded", reason) def test_stories_never_seed(self): # A story cannot be in the archive before it is fetched, so probing # would double the cost of the cheapest surface for no benefit. _, seed, reason = gdl.plan_source(self.stories, self.state, NOW_TS, 20) self.assertFalse(seed) self.assertIn("no seed", reason) def test_recent_fetch_is_refused(self): self.state.mark_fetched(self.posts.url, ago(3)) fetch, _, reason = gdl.plan_source(self.posts, self.state, NOW_TS, 20) self.assertFalse(fetch) self.assertIn("under the", reason) def test_an_old_fetch_is_allowed_again(self): self.state.mark_fetched(self.posts.url, ago(30)) fetch, _, _ = gdl.plan_source(self.posts, self.state, NOW_TS, 20) self.assertTrue(fetch) def test_daily_stories_pass_a_20h_floor(self): # The cadence this is built for: once a day, every day. self.state.mark_fetched(self.stories.url, ago(24)) fetch, _, _ = gdl.plan_source(self.stories, self.state, NOW_TS, 20) self.assertTrue(fetch) def test_force_disables_the_floor(self): self.state.mark_fetched(self.posts.url, ago(1)) fetch, _, _ = gdl.plan_source(self.posts, self.state, NOW_TS, 0.0) self.assertTrue(fetch) def test_the_aborted_run_scenario(self): """ Yesterday's failure: a run died mid-way and the restart re-enumerated every profile. Seeded-but-not-fetched must not re-probe. """ self.state.mark_seeded(self.posts.url, ago(0.5)) fetch, seed, _ = gdl.plan_source(self.posts, self.state, NOW_TS, 20) self.assertTrue(fetch, "the fetch still needs to happen") self.assertFalse(seed, "but the listing pass must not be paid for twice") class StatePersistence(unittest.TestCase): def test_state_survives_a_reload(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "state.json" a = gdl.SyncState(path) a.mark_seeded("https://x/", ago(1)) a.mark_fetched("https://x/", ago(1)) a.save() b = gdl.SyncState(path) self.assertFalse(b.needs_seed("https://x/")) self.assertEqual(b.last_fetch("https://x/"), ago(1)) def test_a_corrupt_state_file_never_blocks_a_sync(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "state.json" path.write_text("{ not json") self.assertTrue(gdl.SyncState(path).needs_seed("https://x/")) class ProbeCaching(unittest.TestCase): def test_fresh_entries_are_reused_and_stale_ones_are_not(self): with tempfile.TemporaryDirectory() as d: cache = gdl.ProbeCache(Path(d) / "p.json", ttl_hours=24) cache.put("https://x/", [{"shortcode": "A", "post_shortcode": "A", "num": 1, "media_id": "1"}], ago(1)) self.assertEqual(len(cache.get("https://x/", NOW_TS)), 1) cache.put("https://y/", [{"shortcode": "B", "post_shortcode": "B", "num": 1, "media_id": "2"}], ago(48)) self.assertIsNone(cache.get("https://y/", NOW_TS)) def test_cache_keeps_only_the_fields_seeding_needs(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "p.json" cache = gdl.ProbeCache(path, ttl_hours=24) cache.put("https://x/", [{"shortcode": "A", "post_shortcode": "A", "num": 1, "media_id": "1", "description": "x" * 5000}], ago(0)) cache.save() self.assertNotIn("description", path.read_text()) def test_a_miss_is_reported_rather_than_guessed(self): with tempfile.TemporaryDirectory() as d: cache = gdl.ProbeCache(Path(d) / "p.json", ttl_hours=24) self.assertIsNone(cache.get("https://never-seen/", NOW_TS)) class Seeding(unittest.TestCase): """The bug that seeded 5 of 2275: matching the wrong shortcode field.""" def test_posts_are_keyed_by_post_shortcode(self): item = {"shortcode": "childcode", "post_shortcode": "POSTCODE", "num": 2, "media_id": "9"} self.assertEqual(gdl.live_key(item, "posts"), ("POSTCODE", 2)) def test_stories_are_keyed_by_the_per_item_shortcode(self): item = {"shortcode": "ITEMCODE", "post_shortcode": "reelid", "num": 3, "media_id": "9"} self.assertEqual(gdl.live_key(item, "stories"), ("ITEMCODE", 1)) self.assertEqual(gdl.live_key(item, "highlights"), ("ITEMCODE", 1)) def test_index_existing_normalises_a_missing_index_to_one(self): held = gdl.index_existing([ "u/2023-04-19_u - ABC.mp4", "u/2023-04-12_u - DEF - 3.jpg", "u/2023-04-12_u - DEF.txt", # sidecars are not media "u/2023-04-12_u - DEF.json", ]) self.assertEqual(held, {("ABC", 1), ("DEF", 3)}) def test_seeding_marks_only_what_is_already_held(self): with tempfile.TemporaryDirectory() as d: db = Path(d) / "a.db" live = [ {"post_shortcode": "HELD", "shortcode": "x", "num": 1, "media_id": "11"}, {"post_shortcode": "NEW", "shortcode": "y", "num": 1, "media_id": "22"}, ] n = gdl.seed_archive_db(db, {("HELD", 1)}, live, "posts") self.assertEqual(n, 1) import sqlite3 rows = {r[0] for r in sqlite3.connect(db).execute( "SELECT entry FROM archive")} self.assertEqual(rows, {"instagram11"}) class Publishing(unittest.TestCase): def test_publish_only_ever_adds(self): cmd = gdl.rsync_command(Path("/stage"), "host:/archives", dry_run=False) self.assertIn("--ignore-existing", cmd) self.assertNotIn("--delete", cmd) def test_tooling_files_are_excluded_from_the_archive(self): cmd = " ".join(gdl.rsync_command(Path("/stage"), "/dest", dry_run=True)) for pattern in ("gdl-sync*.json", "*.db"): self.assertIn(pattern, cmd) self.assertIn("--dry-run", cmd) class UrlsFile(unittest.TestCase): def test_reads_every_form_a_person_might_paste(self): with tempfile.TemporaryDirectory() as d: p = Path(d) / "urls.txt" p.write_text( "# comment\n" "https://www.instagram.com/a/\n" "https://instagram.com/b\n" "www.instagram.com/c/\n" "d\n" " e # trailing\n" "\n" "https://www.instagram.com/a/\n" # duplicate "https://www.instagram.com/p/ABC123/\n" # a post, not a profile "not a username\n") self.assertEqual(gdl.read_urls_file(p), ["a", "b", "c", "d", "e"]) if __name__ == "__main__": unittest.main(verbosity=2)