diff --git a/scripts/twitter/tests/__init__.py b/scripts/twitter/tests/__init__.py
new file mode 100644
index 0000000..ff2b71a
--- /dev/null
+++ b/scripts/twitter/tests/__init__.py
@@ -0,0 +1 @@
+"""Isolated tests for Threadwell Twitter tooling."""
\ No newline at end of file
diff --git a/scripts/twitter/tests/helpers.py b/scripts/twitter/tests/helpers.py
new file mode 100644
index 0000000..21ec8a9
--- /dev/null
+++ b/scripts/twitter/tests/helpers.py
@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+
+NOW = "2026-08-24T12:00:00Z"
+
+
+def write_json(path: Path, data: dict) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(
+ json.dumps(data, indent=2, ensure_ascii=False) + "\n",
+ encoding="utf-8",
+ newline="\n",
+ )
+
+
+def read_json(path: Path) -> dict:
+ return json.loads(path.read_text(encoding="utf-8"))
+
+
+def legacy_thread(asset_dir: Path, note_dir: Path) -> tuple[Path, Path]:
+ asset_dir.mkdir(parents=True, exist_ok=True)
+ note_dir.mkdir(parents=True, exist_ok=True)
+ original = asset_dir / "100_AAA_orig.png"
+ original.write_bytes(b"png fixture bytes")
+ write_json(
+ asset_dir / "thread_data.json",
+ {
+ "root_post_id": "100",
+ "source_url": "https://x.com/example/status/100",
+ "posts": [
+ {
+ "post_id": "100",
+ "author": "Example",
+ "handle": "example",
+ "text": "Fixture post",
+ "timestamp": "2026-08-24 12:00:00",
+ "media_urls": [
+ "https://pbs.twimg.com/media/AAA?format=png&name=orig"
+ ],
+ "reply_to_id": None,
+ "quote_of_id": None,
+ "metrics": {
+ "reply_count": 0,
+ "repost_count": 0,
+ "like_count": 0,
+ "view_count": 1,
+ },
+ }
+ ],
+ },
+ )
+ write_json(
+ asset_dir / "media.json",
+ {
+ "root_post_id": "100",
+ "items": [
+ {
+ "post_id": "100",
+ "media_id": "AAA",
+ "handle": "example",
+ "role": "orig",
+ "filename": original.name,
+ "publish": True,
+ "url": "https://files.catbox.moe/fallback.png",
+ "embed": True,
+ }
+ ],
+ },
+ )
+ (note_dir / "index.md").write_text(
+ "---\npost_id: \"100\"\ndraft: false\n---\n\n"
+ "**1/**\n\nFixture post\n\n"
+ "\n",
+ encoding="utf-8",
+ newline="\n",
+ )
+ return asset_dir, note_dir
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_backup_assets.py b/scripts/twitter/tests/test_backup_assets.py
new file mode 100644
index 0000000..1107b22
--- /dev/null
+++ b/scripts/twitter/tests/test_backup_assets.py
@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.backup_assets import backup_thread
+from twitter.media_audit import audit_thread
+from twitter.media_manifest import (
+ canonical_manifest_bytes,
+ hash_file,
+ inventory_digest,
+ payload_inventory,
+)
+from twitter.media_migrate import migrate_legacy_thread
+from twitter.tests.helpers import NOW, legacy_thread, read_json
+
+
+class BackupTests(unittest.TestCase):
+ def test_backup_copies_and_verifies_one_thread(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(
+ root / "assets" / "threads" / "example" / "fixture",
+ root / "notes",
+ )
+ migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ destination = root / "backup"
+ result = backup_thread(
+ asset_dir,
+ assets_root=root / "assets",
+ destination_root=destination,
+ destination_id="cozy",
+ now=NOW,
+ )
+ self.assertEqual(result.state, "synced")
+ copied = destination / "threads" / "example" / "fixture"
+ self.assertTrue((copied / "100_AAA_orig.png").is_file())
+ manifest = read_json(asset_dir / "media.json")
+ self.assertEqual(manifest["mirrors"][0]["state"], "synced")
+
+ def test_missing_destination_parent_does_not_change_publication(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(root / "assets" / "thread", root / "notes")
+ migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ before = read_json(asset_dir / "media.json")["items"][0]["publication"]
+ result = backup_thread(
+ asset_dir,
+ assets_root=root / "assets",
+ destination_root=root / "absent" / "share",
+ destination_id="cozy",
+ now=NOW,
+ require_destination_root=True,
+ )
+ self.assertEqual(result.state, "error")
+ after = read_json(asset_dir / "media.json")["items"][0]["publication"]
+ self.assertEqual(after, before)
+
+ def test_inventory_change_marks_mirror_stale(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(
+ root / "assets" / "threads" / "example" / "fixture",
+ root / "notes",
+ )
+ migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ destination = root / "backup"
+ backup_thread(
+ asset_dir,
+ assets_root=root / "assets",
+ destination_root=destination,
+ destination_id="cozy",
+ now=NOW,
+ )
+ manifest = read_json(asset_dir / "media.json")
+ inventory = payload_inventory(asset_dir, manifest)
+ self.assertEqual(
+ manifest["mirrors"][0]["inventory_digest"],
+ inventory_digest(inventory),
+ )
+ backup_inventory_before = inventory_digest(payload_inventory(asset_dir, manifest))
+ (asset_dir / "100_AAA_orig.png").write_bytes(b"modified")
+ new_inventory = payload_inventory(asset_dir, manifest)
+ self.assertNotEqual(
+ inventory_digest(new_inventory),
+ backup_inventory_before,
+ )
+ report = audit_thread(asset_dir, note_dir, set())
+ self.assertTrue(any("mirror stale" in issue for issue in report.issues))
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_emit_media.py b/scripts/twitter/tests/test_emit_media.py
new file mode 100644
index 0000000..2f039d8
--- /dev/null
+++ b/scripts/twitter/tests/test_emit_media.py
@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.emit_archive import missing_local_media, preserve_review_state
+from twitter.media_manifest import new_original_item
+from twitter.render import render_media
+
+
+class EmitMediaTests(unittest.TestCase):
+ def test_render_media_uses_image_and_video_markup(self) -> None:
+ self.assertEqual(
+ render_media("https://pbs.twimg.com/media/A?format=png&name=orig"),
+ "",
+ )
+ self.assertEqual(
+ render_media("https://video.twimg.com/ext_tw_video/A.mp4"),
+ '',
+ )
+
+ def test_refresh_preserves_draft_and_reviewed_tags(self) -> None:
+ old = (
+ "---\n"
+ "title: Old\n"
+ "draft: false\n"
+ "tags:\n"
+ " - archive\n"
+ " - twitter\n"
+ " - example\n"
+ " - vulkan\n"
+ "---\n\nOld body\n"
+ )
+ fresh = (
+ "---\n"
+ "title: Fresh\n"
+ "draft: true\n"
+ "tags:\n"
+ " - archive\n"
+ " - twitter\n"
+ " - example\n"
+ "---\n\nFresh body\n"
+ )
+ merged = preserve_review_state(old, fresh, mechanical_tags={"archive", "twitter", "example"})
+ self.assertIn("draft: false", merged)
+ self.assertIn(" - vulkan", merged)
+ self.assertIn("Fresh body", merged)
+ self.assertNotIn("Old body", merged)
+
+ def test_missing_local_media_is_reported(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ path = Path(raw) / "missing.png"
+ item = new_original_item(
+ post_id="100",
+ media_id="AAA",
+ handle="example",
+ origin_url="https://pbs.twimg.com/media/AAA?format=png&name=orig",
+ filename=path.name,
+ local_path=path,
+ now="2026-08-24T12:00:00Z",
+ )
+ self.assertEqual(missing_local_media([item]), ["100/AAA/orig"])
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_fallback_media.py b/scripts/twitter/tests/test_fallback_media.py
new file mode 100644
index 0000000..32fc280
--- /dev/null
+++ b/scripts/twitter/tests/test_fallback_media.py
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.fallback_media import activate_fallback, restore_origin
+from twitter.media_manifest import selected_url
+from twitter.media_migrate import migrate_legacy_thread
+from twitter.tests.helpers import NOW, legacy_thread, read_json
+
+
+class FallbackTests(unittest.TestCase):
+ def test_existing_matching_fallback_does_not_upload(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(root / "assets", root / "notes")
+ migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ calls: list[Path] = []
+
+ def upload(path: Path) -> str:
+ calls.append(path)
+ return "https://files.catbox.moe/new.png"
+
+ result = activate_fallback(
+ asset_dir,
+ note_dir,
+ media_id="AAA",
+ role="orig",
+ provider="catbox",
+ confirm_origin_unavailable=True,
+ now=NOW,
+ upload=upload,
+ )
+ self.assertEqual(calls, [])
+ self.assertEqual(result.url, "https://files.catbox.moe/fallback.png")
+ manifest = read_json(asset_dir / "media.json")
+ self.assertEqual(selected_url(manifest["items"][0]), result.url)
+
+ def test_note_rewrite_failure_keeps_origin_selected(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(root / "assets", root / "notes")
+ migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ (note_dir / "index.md").write_text("no media reference\n", encoding="utf-8")
+ with self.assertRaisesRegex(ValueError, "selected origin reference not found exactly once"):
+ activate_fallback(
+ asset_dir,
+ note_dir,
+ media_id="AAA",
+ role="orig",
+ provider="catbox",
+ confirm_origin_unavailable=True,
+ now=NOW,
+ upload=lambda _path: "https://files.catbox.moe/fallback.png",
+ )
+ manifest = read_json(asset_dir / "media.json")
+ self.assertTrue(selected_url(manifest["items"][0]).startswith("https://pbs.twimg.com/"))
+
+ def test_restore_origin_retains_fallback(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(root / "assets", root / "notes")
+ migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ activate_fallback(
+ asset_dir,
+ note_dir,
+ media_id="AAA",
+ role="orig",
+ provider="catbox",
+ confirm_origin_unavailable=True,
+ now=NOW,
+ upload=lambda _path: "https://files.catbox.moe/fallback.png",
+ )
+ restore_origin(asset_dir, note_dir, media_id="AAA", now=NOW)
+ item = read_json(asset_dir / "media.json")["items"][0]
+ self.assertTrue(selected_url(item).startswith("https://pbs.twimg.com/"))
+ self.assertTrue(any(loc.get("kind") == "fallback" for loc in item["locations"]))
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_frozen.py b/scripts/twitter/tests/test_frozen.py
new file mode 100644
index 0000000..7086512
--- /dev/null
+++ b/scripts/twitter/tests/test_frozen.py
@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.frozen import frozen_match, load_frozen_ids
+from twitter.tests.helpers import write_json
+
+
+class FrozenTests(unittest.TestCase):
+ def test_descendant_id_freezes_entire_thread(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ frozen_file = root / "do_not_refetch.txt"
+ frozen_file.write_text("# comment\n200\n", encoding="utf-8")
+ asset_dir = root / "thread"
+ write_json(
+ asset_dir / "thread_data.json",
+ {
+ "root_post_id": "100",
+ "posts": [
+ {"post_id": "100"},
+ {"post_id": "200"},
+ ],
+ },
+ )
+ write_json(asset_dir / "media.json", {"root_post_id": "100", "items": []})
+ self.assertEqual(load_frozen_ids(frozen_file), {"200"})
+ self.assertEqual(frozen_match(asset_dir, {"200"}), "200")
+
+ def test_unrelated_thread_is_writable(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ asset_dir = Path(raw) / "thread"
+ write_json(
+ asset_dir / "thread_data.json",
+ {"root_post_id": "100", "posts": [{"post_id": "100"}]},
+ )
+ write_json(asset_dir / "media.json", {"root_post_id": "100", "items": []})
+ self.assertIsNone(frozen_match(asset_dir, {"200"}))
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_media_audit.py b/scripts/twitter/tests/test_media_audit.py
new file mode 100644
index 0000000..2724e89
--- /dev/null
+++ b/scripts/twitter/tests/test_media_audit.py
@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.media_audit import (
+ AuditReport,
+ audit_local_item,
+ audit_thread,
+ classify_origin_response,
+)
+from twitter.media_manifest import (
+ atomic_write_json,
+ find_location,
+ hash_file,
+ new_original_item,
+)
+from twitter.media_migrate import migrate_legacy_thread
+from twitter.tests.helpers import NOW, legacy_thread, read_json
+
+
+class MediaAuditTests(unittest.TestCase):
+ def test_changed_local_bytes_are_mismatch(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ path = Path(raw) / "asset.png"
+ path.write_bytes(b"first")
+ item = new_original_item(
+ post_id="100",
+ media_id="AAA",
+ handle="example",
+ origin_url="https://pbs.twimg.com/media/AAA?format=png&name=orig",
+ filename=path.name,
+ local_path=path,
+ now=NOW,
+ )
+ path.write_bytes(b"changed")
+ issue = audit_local_item(item, path.parent)
+ self.assertEqual(issue, "100/AAA/orig local mismatch")
+
+ def test_rate_limit_does_not_mean_unavailable(self) -> None:
+ outcome = classify_origin_response(429, "rate limited", NOW)
+ self.assertEqual(outcome["result"], "error")
+ self.assertFalse(outcome["confirms_unavailable"])
+
+ def test_frozen_thread_is_reported(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, _note_dir = legacy_thread(root / "assets", root / "notes")
+ migrate_legacy_thread(asset_dir, root / "notes", now=NOW, apply=True)
+ report = audit_thread(asset_dir, root / "notes", {"100"})
+ self.assertEqual(report.frozen, True)
+ self.assertTrue(any("frozen: skipped" in issue for issue in report.issues))
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_media_manifest.py b/scripts/twitter/tests/test_media_manifest.py
new file mode 100644
index 0000000..63145f4
--- /dev/null
+++ b/scripts/twitter/tests/test_media_manifest.py
@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.media_manifest import (
+ atomic_write_json,
+ find_location,
+ hash_file,
+ merge_item,
+ merge_manifest_items,
+ new_original_item,
+ selected_url,
+ validate_manifest,
+)
+from twitter.tests.helpers import NOW
+
+
+class ManifestIoTests(unittest.TestCase):
+ def test_hash_file_is_sha256(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ path = Path(raw) / "asset.bin"
+ path.write_bytes(b"abc")
+ self.assertEqual(
+ hash_file(path),
+ hashlib.sha256(b"abc").hexdigest(),
+ )
+
+ def test_atomic_write_json_replaces_complete_document(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ path = Path(raw) / "media.json"
+ path.write_text('{"old": true}\n', encoding="utf-8")
+ atomic_write_json(path, {"schema_version": 2, "items": []})
+ self.assertEqual(
+ json.loads(path.read_text(encoding="utf-8")),
+ {"schema_version": 2, "items": []},
+ )
+ self.assertEqual(list(path.parent.glob(".media.json.*.tmp")), [])
+
+
+class CanonicalManifestTests(unittest.TestCase):
+ def test_original_has_independent_origin_and_local_locations(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ local = Path(raw) / "100_AAA_orig.png"
+ local.write_bytes(b"png fixture bytes")
+ item = new_original_item(
+ post_id="100",
+ media_id="AAA",
+ handle="example",
+ origin_url="https://pbs.twimg.com/media/AAA?format=png&name=orig",
+ filename=local.name,
+ local_path=local,
+ now=NOW,
+ )
+ self.assertEqual(selected_url(item), item["locations"][0]["url"])
+ self.assertEqual(find_location(item, "local")["integrity"], "present")
+ self.assertEqual(find_location(item, "local")["sha256"], hash_file(local))
+
+ def test_missing_local_is_explicit(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ local = Path(raw) / "missing.png"
+ item = new_original_item(
+ post_id="100",
+ media_id="AAA",
+ handle="example",
+ origin_url="https://pbs.twimg.com/media/AAA?format=png&name=orig",
+ filename=local.name,
+ local_path=local,
+ now=NOW,
+ )
+ self.assertEqual(find_location(item, "local")["integrity"], "missing")
+ self.assertIsNone(find_location(item, "local")["sha256"])
+
+ def test_merge_preserves_fallback_and_publication(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ local = Path(raw) / "100_AAA_orig.png"
+ local.write_bytes(b"png fixture bytes")
+ old = new_original_item(
+ post_id="100",
+ media_id="AAA",
+ handle="example",
+ origin_url="https://pbs.twimg.com/media/AAA?format=png&name=orig",
+ filename=local.name,
+ local_path=local,
+ now=NOW,
+ )
+ old["locations"].append(
+ {
+ "id": "fallback:catbox:fixture",
+ "kind": "fallback",
+ "provider": "catbox",
+ "url": "https://files.catbox.moe/fallback.png",
+ "sha256": find_location(old, "local")["sha256"],
+ "uploaded_at": NOW,
+ "availability": "unknown",
+ "checked_at": None,
+ "check": None,
+ }
+ )
+ old["publication"] = {
+ "selected_location_id": "fallback:catbox:fixture",
+ "selected_at": NOW,
+ "reason": "origin-unavailable",
+ }
+ fresh = new_original_item(
+ post_id="100",
+ media_id="AAA",
+ handle="example",
+ origin_url="https://pbs.twimg.com/media/AAA?format=png&name=orig",
+ filename=local.name,
+ local_path=local,
+ now="2026-08-25T12:00:00Z",
+ )
+ merged = merge_item(old, fresh)
+ self.assertEqual(selected_url(merged), "https://files.catbox.moe/fallback.png")
+ self.assertIsNotNone(find_location(merged, "fallback:catbox:fixture"))
+ self.assertEqual(merge_item(merged, fresh), merged)
+
+ def test_merge_refuses_origin_identity_change(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ local = Path(raw) / "100_AAA_orig.png"
+ local.write_bytes(b"png fixture bytes")
+ old = new_original_item(
+ post_id="100",
+ media_id="AAA",
+ handle="example",
+ origin_url="https://pbs.twimg.com/media/AAA?format=png&name=orig",
+ filename=local.name,
+ local_path=local,
+ now=NOW,
+ )
+ fresh = new_original_item(
+ post_id="100",
+ media_id="AAA",
+ handle="example",
+ origin_url="https://pbs.twimg.com/media/DIFFERENT?format=png&name=orig",
+ filename=local.name,
+ local_path=local,
+ now=NOW,
+ )
+ with self.assertRaisesRegex(ValueError, "origin URL changed"):
+ merge_item(old, fresh)
+
+ def test_unmatched_derived_item_survives_manifest_merge(self) -> None:
+ derived = {
+ "post_id": "100",
+ "media_id": "AAA",
+ "handle": "example",
+ "role": "ocr",
+ "derived_from": {"post_id": "100", "media_id": "AAA", "role": "orig"},
+ "embed": False,
+ "locations": [],
+ "publication": None,
+ }
+ self.assertEqual(merge_manifest_items([derived], []), [derived])
+
+ def test_validator_rejects_local_publication(self) -> None:
+ manifest = {
+ "schema_version": 2,
+ "root_post_id": "100",
+ "items": [
+ {
+ "post_id": "100",
+ "media_id": "AAA",
+ "handle": "example",
+ "role": "orig",
+ "derived_from": None,
+ "embed": True,
+ "locations": [
+ {
+ "id": "local",
+ "kind": "local",
+ "path": "100_AAA_orig.png",
+ "sha256": None,
+ "bytes": None,
+ "media_type": "image/png",
+ "integrity": "missing",
+ "verified_at": NOW,
+ }
+ ],
+ "publication": {
+ "selected_location_id": "local",
+ "selected_at": NOW,
+ "reason": "manual",
+ },
+ }
+ ],
+ "mirrors": [],
+ }
+ self.assertIn("items[0] selects non-HTTPS location local", validate_manifest(manifest))
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_media_migrate.py b/scripts/twitter/tests/test_media_migrate.py
new file mode 100644
index 0000000..37bfc00
--- /dev/null
+++ b/scripts/twitter/tests/test_media_migrate.py
@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.media_manifest import find_location, selected_url, validate_manifest
+from twitter.media_migrate import migrate_legacy_thread
+from twitter.tests.helpers import NOW, legacy_thread, read_json
+
+
+class LegacyMigrationTests(unittest.TestCase):
+ def test_origin_local_and_catbox_survive_migration(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(root / "assets", root / "notes")
+ result = migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=False)
+ self.assertEqual(result.state, "migratable")
+ self.assertFalse(result.changed)
+ legacy = read_json(asset_dir / "media.json")
+ self.assertNotIn("schema_version", legacy)
+
+ applied = migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ self.assertTrue(applied.changed)
+ manifest = read_json(asset_dir / "media.json")
+ self.assertEqual(validate_manifest(manifest), [])
+ item = manifest["items"][0]
+ self.assertEqual(
+ find_location(item, "origin:x")["url"],
+ "https://pbs.twimg.com/media/AAA?format=png&name=orig",
+ )
+ fallback = next(
+ location
+ for location in item["locations"]
+ if location.get("kind") == "fallback"
+ )
+ self.assertEqual(fallback["url"], "https://files.catbox.moe/fallback.png")
+ self.assertEqual(selected_url(item), find_location(item, "origin:x")["url"])
+
+ def test_second_migration_is_idempotent(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(root / "assets", root / "notes")
+ migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ before = (asset_dir / "media.json").read_bytes()
+ result = migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ self.assertFalse(result.changed)
+ self.assertEqual((asset_dir / "media.json").read_bytes(), before)
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_media_refs.py b/scripts/twitter/tests/test_media_refs.py
new file mode 100644
index 0000000..db26bef
--- /dev/null
+++ b/scripts/twitter/tests/test_media_refs.py
@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.media_migrate import migrate_legacy_thread
+from twitter.media_refs import plan_thread_rewrites
+from twitter.tests.helpers import NOW, legacy_thread
+
+
+class MediaReferenceTests(unittest.TestCase):
+ def test_catbox_reference_becomes_origin_without_losing_fallback(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ asset_dir, note_dir = legacy_thread(root / "assets", root / "notes")
+ result = migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True)
+ self.assertTrue(result.changed)
+ text = (note_dir / "index.md").read_text(encoding="utf-8")
+ self.assertIn(
+ "",
+ text,
+ )
+ self.assertNotIn("files.catbox.moe", text)
+ self.assertIn("**1/**\n\nFixture post", text)
+ self.assertIn('post_id: "100"', text)
+ self.assertIn("draft: false", text)
+
+ def test_filename_line_with_two_images_becomes_two_embeds(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ note_dir = Path(raw)
+ note = note_dir / "index.md"
+ note.write_text(
+ "before\nMedia (not lifted): `a.png` `b.jpg`\nafter\n",
+ encoding="utf-8",
+ )
+ plan = plan_thread_rewrites(
+ note_dir,
+ filename_origins={
+ "a.png": "https://pbs.twimg.com/media/A?format=png&name=orig",
+ "b.jpg": "https://pbs.twimg.com/media/B?format=jpg&name=orig",
+ },
+ fallback_origins={},
+ )
+ self.assertEqual(plan.issues, ())
+ self.assertIn("", plan.files[0].after)
+ self.assertIn("", plan.files[0].after)
+ self.assertNotIn("Media (not lifted)", plan.files[0].after)
+
+ def test_video_uses_video_element(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ note_dir = Path(raw)
+ note = note_dir / "index.md"
+ note.write_text("Media (not lifted): `clip.mp4`\n", encoding="utf-8")
+ plan = plan_thread_rewrites(
+ note_dir,
+ filename_origins={"clip.mp4": "https://video.twimg.com/ext_tw_video/clip.mp4"},
+ fallback_origins={},
+ )
+ self.assertIn(
+ '',
+ plan.files[0].after,
+ )
+
+ def test_untracked_catbox_url_blocks_rewrite(self) -> None:
+ with tempfile.TemporaryDirectory() as raw:
+ note_dir = Path(raw)
+ note = note_dir / "index.md"
+ note.write_text("\n", encoding="utf-8")
+ plan = plan_thread_rewrites(
+ note_dir,
+ filename_origins={},
+ fallback_origins={},
+ )
+ self.assertEqual(plan.files, ())
+ self.assertEqual(
+ plan.issues,
+ ("untracked fallback reference: https://files.catbox.moe/unknown.png",),
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/scripts/twitter/tests/test_tw_media_commands.py b/scripts/twitter/tests/test_tw_media_commands.py
new file mode 100644
index 0000000..0e9cd19
--- /dev/null
+++ b/scripts/twitter/tests/test_tw_media_commands.py
@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import unittest
+
+from twitter.tw import build_parser
+
+
+class TwMediaCommandTests(unittest.TestCase):
+ def test_migrate_defaults_to_dry_run(self) -> None:
+ args = build_parser().parse_args(["migrate-media", "--id", "100"])
+ self.assertEqual(args.cmd, "migrate-media")
+ self.assertFalse(args.apply)
+
+ def test_fallback_requires_exact_media(self) -> None:
+ args = build_parser().parse_args(
+ [
+ "fallback",
+ "--id",
+ "100",
+ "--media-id",
+ "AAA",
+ "--role",
+ "orig",
+ "--confirm-origin-unavailable",
+ ]
+ )
+ self.assertEqual(args.media_id, "AAA")
+ self.assertTrue(args.confirm_origin_unavailable)
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file