396 lines
14 KiB
Python
396 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from twitter.media_manifest import (
|
|
atomic_write_json,
|
|
canonical_manifest_bytes,
|
|
find_location,
|
|
hash_file,
|
|
merge_item,
|
|
merge_manifest_items,
|
|
new_original_item,
|
|
payload_inventory,
|
|
selected_url,
|
|
validate_manifest,
|
|
_from_wire_dict,
|
|
)
|
|
from twitter.models import MediaItem, MediaLocation, MediaManifest, Publication
|
|
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,
|
|
)
|
|
local_sha = find_location(old, "local").sha256
|
|
fallback = MediaLocation(
|
|
id="fallback:catbox:fixture",
|
|
kind="fallback",
|
|
provider="catbox",
|
|
url="https://files.catbox.moe/fallback.png",
|
|
sha256=local_sha,
|
|
uploaded_at=NOW,
|
|
availability="unknown",
|
|
)
|
|
old_with_fallback = MediaItem(
|
|
post_id=old.post_id,
|
|
media_id=old.media_id,
|
|
kind=old.kind,
|
|
role=old.role,
|
|
handle=old.handle,
|
|
embed=old.embed,
|
|
locations=old.locations + (fallback,),
|
|
publication=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_with_fallback, 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 = MediaItem(
|
|
post_id="100",
|
|
media_id="AAA",
|
|
kind="ocr",
|
|
role="ocr",
|
|
handle="example",
|
|
locations=(),
|
|
publication=None,
|
|
)
|
|
result = merge_manifest_items((), (derived,))
|
|
self.assertEqual(result, (derived,))
|
|
|
|
def test_validator_rejects_local_publication(self) -> None:
|
|
item = MediaItem(
|
|
post_id="100",
|
|
media_id="AAA",
|
|
kind="orig",
|
|
role="orig",
|
|
handle="example",
|
|
embed=True,
|
|
locations=(
|
|
MediaLocation(
|
|
id="local",
|
|
kind="local",
|
|
local_path=Path("100_AAA_orig.png"),
|
|
sha256=None,
|
|
bytes=None,
|
|
media_type="image/png",
|
|
integrity="missing",
|
|
verified_at=NOW,
|
|
),
|
|
),
|
|
publication=Publication(
|
|
selected_location_id="local",
|
|
selected_at=NOW,
|
|
reason="manual",
|
|
),
|
|
)
|
|
manifest = MediaManifest(
|
|
schema_version=2,
|
|
root_post_id="100",
|
|
items=(item,),
|
|
)
|
|
self.assertIn(
|
|
"items[0] selects non-HTTPS location local", validate_manifest(manifest)
|
|
)
|
|
|
|
|
|
class FromWireDictTests(unittest.TestCase):
|
|
"""Tests for the _from_wire_dict boundary parser."""
|
|
|
|
def test_full_wire_payload_round_trips(self) -> None:
|
|
raw: dict = {
|
|
"schema_version": 2,
|
|
"root_post_id": "100",
|
|
"captured_at": NOW,
|
|
"items": [
|
|
{
|
|
"post_id": "100",
|
|
"media_id": "AAA",
|
|
"role": "orig",
|
|
"embed": True,
|
|
"locations": [
|
|
{
|
|
"id": "origin:x",
|
|
"kind": "origin",
|
|
"provider": "x",
|
|
"url": "https://pbs.twimg.com/media/AAA",
|
|
"availability": "unknown",
|
|
"checked_at": None,
|
|
},
|
|
{
|
|
"id": "local",
|
|
"kind": "local",
|
|
"path": "100_AAA_orig.png",
|
|
"sha256": "deadbeef",
|
|
"bytes": 12345,
|
|
"media_type": "image/png",
|
|
"integrity": "present",
|
|
"verified_at": NOW,
|
|
},
|
|
],
|
|
"publication": {
|
|
"selected_location_id": "origin:x",
|
|
"selected_at": NOW,
|
|
"reason": "default",
|
|
},
|
|
}
|
|
],
|
|
"mirrors": [],
|
|
}
|
|
manifest = _from_wire_dict(raw)
|
|
self.assertEqual(manifest.schema_version, 2)
|
|
self.assertEqual(manifest.root_post_id, "100")
|
|
self.assertEqual(manifest.captured_at, NOW)
|
|
self.assertEqual(len(manifest.items), 1)
|
|
item = manifest.items[0]
|
|
self.assertEqual(item.post_id, "100")
|
|
self.assertEqual(item.media_id, "AAA")
|
|
self.assertEqual(item.kind, "orig")
|
|
self.assertEqual(item.embed, True)
|
|
self.assertEqual(len(item.locations), 2)
|
|
loc0, loc1 = item.locations
|
|
self.assertEqual(loc0.id, "origin:x")
|
|
self.assertEqual(loc0.kind, "origin")
|
|
self.assertEqual(loc0.url, "https://pbs.twimg.com/media/AAA")
|
|
self.assertEqual(loc0.provider, "x")
|
|
self.assertEqual(loc0.availability, "unknown")
|
|
self.assertEqual(loc1.id, "local")
|
|
self.assertEqual(loc1.kind, "local")
|
|
self.assertEqual(loc1.integrity, "present")
|
|
self.assertEqual(loc1.sha256, "deadbeef")
|
|
self.assertEqual(loc1.bytes, 12345)
|
|
self.assertIsNotNone(item.publication)
|
|
assert item.publication is not None
|
|
self.assertEqual(item.publication.selected_location_id, "origin:x")
|
|
self.assertEqual(item.publication.reason, "default")
|
|
|
|
def test_minimal_wire_payload(self) -> None:
|
|
raw: dict = {
|
|
"schema_version": 2,
|
|
"root_post_id": "100",
|
|
"items": [
|
|
{
|
|
"post_id": "100",
|
|
"media_id": "AAA",
|
|
"role": "orig",
|
|
"locations": [],
|
|
}
|
|
],
|
|
}
|
|
manifest = _from_wire_dict(raw)
|
|
self.assertEqual(manifest.root_post_id, "100")
|
|
self.assertEqual(len(manifest.items), 1)
|
|
self.assertEqual(manifest.items[0].kind, "orig")
|
|
|
|
def test_wire_payload_with_derived_from(self) -> None:
|
|
raw: dict = {
|
|
"schema_version": 2,
|
|
"root_post_id": "100",
|
|
"items": [
|
|
{
|
|
"post_id": "100",
|
|
"media_id": "AAA",
|
|
"role": "ocr",
|
|
"derived_from": {
|
|
"post_id": "100",
|
|
"media_id": "AAA",
|
|
"role": "orig",
|
|
},
|
|
"embed": False,
|
|
"locations": [],
|
|
"publication": None,
|
|
}
|
|
],
|
|
}
|
|
manifest = _from_wire_dict(raw)
|
|
item = manifest.items[0]
|
|
self.assertEqual(item.kind, "ocr")
|
|
self.assertIsNotNone(item.derived_from)
|
|
assert item.derived_from is not None
|
|
self.assertEqual(item.derived_from.role, "orig")
|
|
self.assertIsNone(item.publication)
|
|
|
|
|
|
class CanonicalManifestBytesTests(unittest.TestCase):
|
|
"""Tests for canonical_manifest_bytes with typed MediaManifest."""
|
|
|
|
def test_canonical_manifest_bytes_is_stable(self) -> None:
|
|
manifest = MediaManifest(
|
|
schema_version=2,
|
|
root_post_id="100",
|
|
items=(
|
|
MediaItem(
|
|
post_id="100",
|
|
media_id="AAA",
|
|
kind="orig",
|
|
role="orig",
|
|
locations=(
|
|
MediaLocation(
|
|
id="origin:x",
|
|
kind="origin",
|
|
url="https://pbs.twimg.com/media/AAA",
|
|
availability="unknown",
|
|
),
|
|
),
|
|
publication=Publication(
|
|
selected_location_id="origin:x",
|
|
selected_at=NOW,
|
|
reason="default",
|
|
),
|
|
),
|
|
),
|
|
)
|
|
first = canonical_manifest_bytes(manifest)
|
|
second = canonical_manifest_bytes(manifest)
|
|
self.assertEqual(first, second)
|
|
decoded = json.loads(first.decode("utf-8"))
|
|
self.assertEqual(decoded["schema_version"], 2)
|
|
self.assertEqual(decoded["root_post_id"], "100")
|
|
self.assertEqual(decoded["mirrors"], [])
|
|
self.assertEqual(len(decoded["items"]), 1)
|
|
self.assertEqual(decoded["items"][0]["media_id"], "AAA")
|
|
|
|
def test_canonical_manifest_bytes_omits_mirrors_and_canonicalizes(self) -> None:
|
|
manifest = MediaManifest(
|
|
schema_version=2,
|
|
root_post_id="100",
|
|
items=(
|
|
MediaItem(
|
|
post_id="100",
|
|
media_id="AAA",
|
|
kind="orig",
|
|
role="orig",
|
|
locations=(
|
|
MediaLocation(
|
|
id="origin:x",
|
|
kind="origin",
|
|
url="https://x",
|
|
availability="unknown",
|
|
),
|
|
),
|
|
publication=Publication(
|
|
selected_location_id="origin:x",
|
|
selected_at=NOW,
|
|
reason="default",
|
|
),
|
|
),
|
|
),
|
|
)
|
|
encoded = canonical_manifest_bytes(manifest)
|
|
self.assertIn(b'"mirrors":[]', encoded)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|