ergosteurandClaude Sonnet 5
0753395391
Lets an out-of-band link (shared by someone, not one of the tracked profiles) be pulled in directly by URL, filed under its owner's account like any other post. Bypasses profile planning, archive-db seeding, and the --min-interval floor entirely, since it's a single request rather than a recurring surface to budget against. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011qAds5qr7nZRq5R4yAuxUk
255 lines
10 KiB
Python
255 lines
10 KiB
Python
#!/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 PostUrl(unittest.TestCase):
|
|
"""--post-url: a one-off fetch outside the tracked profile list, whose
|
|
owning account is only known mid-extraction -- same reasoning as
|
|
highlights, so it must be exempted from the same forced-destination rule."""
|
|
|
|
def test_config_keys_a_username_directory(self):
|
|
config = gdl.build_config("1M", [6.0, 10.0], [3.0, 6.0])
|
|
for kind in ("post", "reel"):
|
|
self.assertEqual(
|
|
config["extractor"]["instagram"][kind]["directory"],
|
|
["{username}"])
|
|
|
|
def test_destination_is_not_forced_like_posts_and_reels(self):
|
|
staging = Path("/stage")
|
|
for subcategory in ("post", "reel"):
|
|
src = gdl.Source(subcategory, "https://www.instagram.com/p/ABC/",
|
|
"", subcategory)
|
|
cmd = gdl.gdl_command(src, staging, Path("/cfg.json"), "chrome:x",
|
|
None)
|
|
self.assertIn(str(staging), cmd)
|
|
self.assertNotIn(str(staging / subcategory), 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)
|