249 lines
9.7 KiB
Python
249 lines
9.7 KiB
Python
"""Tests for media downloads and Joplin resource rewriting."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.blocks import (
|
|
make_file_placeholder,
|
|
make_image_placeholder,
|
|
render_blocks_to_markdown,
|
|
)
|
|
from src.loss_report import LossReport
|
|
from src.media import resolve_media, resolve_media_policy
|
|
from src.providers.base import ProviderError
|
|
from src.providers.chatgpt import parse_asset_file_id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Asset reference parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestParseAssetFileId:
|
|
def test_plain_sediment(self):
|
|
assert parse_asset_file_id("sediment://file_00000000245c71fda5") == "file_00000000245c71fda5"
|
|
|
|
def test_generated_image_with_hash_and_page(self):
|
|
ref = "sediment://8456107fc383a53#file_00000000979c71f685#p_6.png"
|
|
assert parse_asset_file_id(ref) == "file_00000000979c71f685"
|
|
|
|
def test_file_service_scheme(self):
|
|
assert parse_asset_file_id("file-service://file-AbCdEf") == "file-AbCdEf"
|
|
|
|
def test_unrecognised(self):
|
|
assert parse_asset_file_id("https://example.com/x.png") is None
|
|
assert parse_asset_file_id("") is None
|
|
assert parse_asset_file_id(None) is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Media policy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMediaPolicy:
|
|
def test_default(self, monkeypatch):
|
|
monkeypatch.delenv("EXPORTER_DOWNLOAD_MEDIA", raising=False)
|
|
assert resolve_media_policy() == "images"
|
|
|
|
def test_valid(self, monkeypatch):
|
|
monkeypatch.setenv("EXPORTER_DOWNLOAD_MEDIA", "all")
|
|
assert resolve_media_policy() == "all"
|
|
|
|
def test_invalid_falls_back(self, monkeypatch, caplog):
|
|
monkeypatch.setenv("EXPORTER_DOWNLOAD_MEDIA", "bogus")
|
|
assert resolve_media_policy() == "images"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resolve_media
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeProvider:
|
|
"""Minimal provider exposing download_asset + the ref parser."""
|
|
|
|
def __init__(self, assets=None, fail_refs=None):
|
|
self._assets = assets or {}
|
|
self._fail_refs = fail_refs or {}
|
|
self.calls = []
|
|
|
|
def parse_asset_file_id(self, ref):
|
|
return parse_asset_file_id(ref)
|
|
|
|
def download_asset(self, ref):
|
|
self.calls.append(ref)
|
|
if ref in self._fail_refs:
|
|
raise ProviderError("chatgpt", "download_asset", self._fail_refs[ref])
|
|
return self._assets[ref] # (content, mime, file_name)
|
|
|
|
|
|
def _conv_with(blocks):
|
|
return {
|
|
"id": "conv-1",
|
|
"title": "Has Media",
|
|
"provider": "chatgpt",
|
|
"project": None,
|
|
"created_at": "2026-05-20T00:00:00+00:00",
|
|
"messages": [{"role": "user", "blocks": blocks}],
|
|
}
|
|
|
|
|
|
class TestResolveMedia:
|
|
def test_downloads_image_and_inlines(self, tmp_path):
|
|
ref = "sediment://file_img1"
|
|
provider = _FakeProvider({ref: (b"\x89PNG\r\n", "image/png", "x.png")})
|
|
block = make_image_placeholder(ref=ref, source="user_upload")
|
|
conv = _conv_with([block])
|
|
report = LossReport()
|
|
|
|
n = resolve_media(conv, provider, tmp_path, "provider/project/year", "images", report)
|
|
|
|
assert n == 1
|
|
assert report.media_downloaded == 1
|
|
assert block["local_path"] == "media/file_img1.png"
|
|
rendered = render_blocks_to_markdown([block])
|
|
assert rendered == ""
|
|
# File written under the conversation's media/ dir
|
|
written = list(tmp_path.rglob("media/file_img1.png"))
|
|
assert written and written[0].read_bytes() == b"\x89PNG\r\n"
|
|
|
|
def test_images_policy_skips_files(self, tmp_path):
|
|
ref = "sediment://file_audio1"
|
|
provider = _FakeProvider({ref: (b"RIFF", "audio/wav", "a.wav")})
|
|
block = make_file_placeholder(ref=ref, mime="audio/wav", size_bytes=1000)
|
|
conv = _conv_with([block])
|
|
report = LossReport()
|
|
|
|
n = resolve_media(conv, provider, tmp_path, "provider/project/year", "images", report)
|
|
assert n == 0
|
|
assert "local_path" not in block
|
|
assert provider.calls == []
|
|
|
|
def test_all_policy_downloads_files(self, tmp_path):
|
|
ref = "sediment://file_audio1"
|
|
provider = _FakeProvider({ref: (b"RIFFdata", "audio/wav", "a.wav")})
|
|
block = make_file_placeholder(ref=ref, mime="audio/wav", size_bytes=8)
|
|
conv = _conv_with([block])
|
|
report = LossReport()
|
|
|
|
n = resolve_media(conv, provider, tmp_path, "provider/project/year", "all", report)
|
|
assert n == 1
|
|
assert block["local_path"] == "media/file_audio1.wav"
|
|
rendered = render_blocks_to_markdown([block])
|
|
assert "media/file_audio1.wav" in rendered and rendered.startswith("> 📎")
|
|
|
|
def test_off_policy_noop(self, tmp_path):
|
|
provider = _FakeProvider({"sediment://file_x": (b"x", "image/png", None)})
|
|
block = make_image_placeholder(ref="sediment://file_x", source="user_upload")
|
|
conv = _conv_with([block])
|
|
report = LossReport()
|
|
assert resolve_media(conv, provider, tmp_path, "provider/project/year", "off", report) == 0
|
|
assert provider.calls == []
|
|
|
|
def test_idempotent_uses_disk(self, tmp_path):
|
|
ref = "sediment://file_img1"
|
|
provider = _FakeProvider({ref: (b"\x89PNG", "image/png", "x.png")})
|
|
conv = _conv_with([make_image_placeholder(ref=ref, source="user_upload")])
|
|
report = LossReport()
|
|
resolve_media(conv, provider, tmp_path, "provider/project/year", "images", report)
|
|
assert len(provider.calls) == 1
|
|
|
|
# Second run, fresh blocks: file already on disk → no new API call.
|
|
conv2 = _conv_with([make_image_placeholder(ref=ref, source="user_upload")])
|
|
resolve_media(conv2, provider, tmp_path, "provider/project/year", "images", LossReport())
|
|
assert len(provider.calls) == 1 # unchanged
|
|
assert conv2["messages"][0]["blocks"][0]["local_path"] == "media/file_img1.png"
|
|
|
|
def test_failure_keeps_placeholder_and_counts(self, tmp_path):
|
|
ref = "sediment://file_gone"
|
|
provider = _FakeProvider(fail_refs={ref: RuntimeError("Signed URL returned HTTP 404")})
|
|
block = make_image_placeholder(ref=ref, source="model_generated")
|
|
conv = _conv_with([block])
|
|
report = LossReport()
|
|
|
|
n = resolve_media(conv, provider, tmp_path, "provider/project/year", "images", report)
|
|
assert n == 0
|
|
assert "local_path" not in block
|
|
assert report.media_failed["expired-or-missing"] == 1
|
|
# Still renders as a placeholder, not a broken image link
|
|
assert render_blocks_to_markdown([block]).startswith("> 🖼️")
|
|
|
|
def test_provider_without_download_asset(self, tmp_path):
|
|
"""claude-code has no remote assets — resolve_media must no-op."""
|
|
class NoDownload:
|
|
pass
|
|
block = make_image_placeholder(ref="sediment://file_x", source="user_upload")
|
|
conv = _conv_with([block])
|
|
assert resolve_media(conv, NoDownload(), tmp_path, "provider/project/year", "images", LossReport()) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Joplin media rewriting
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeJoplin:
|
|
def __init__(self):
|
|
self.uploaded = []
|
|
self._n = 0
|
|
|
|
def create_resource(self, file_path, title=None):
|
|
self.uploaded.append(Path(file_path).name)
|
|
self._n += 1
|
|
return f"res{self._n}"
|
|
|
|
|
|
class TestUploadMediaAndRewrite:
|
|
def _note(self, tmp_path, body):
|
|
media = tmp_path / "media"
|
|
media.mkdir()
|
|
(media / "file_img1.png").write_bytes(b"\x89PNG")
|
|
(media / "clip.wav").write_bytes(b"RIFF")
|
|
return body
|
|
|
|
def test_rewrites_image_and_file_links(self, tmp_path):
|
|
from src.joplin import upload_media_and_rewrite
|
|
body = self._note(
|
|
tmp_path,
|
|
"Look: \n"
|
|
"> 📎 **File attached** — [clip.wav](media/clip.wav) (audio/wav)",
|
|
)
|
|
client = _FakeJoplin()
|
|
new_body, res_map = upload_media_and_rewrite(body, tmp_path, client, {})
|
|
|
|
assert "" in new_body
|
|
assert "(:/res2)" in new_body
|
|
assert "media/" not in new_body
|
|
assert res_map == {"media/file_img1.png": "res1", "media/clip.wav": "res2"}
|
|
assert len(client.uploaded) == 2
|
|
|
|
def test_reuses_known_resource_ids(self, tmp_path):
|
|
from src.joplin import upload_media_and_rewrite
|
|
body = self._note(tmp_path, "")
|
|
client = _FakeJoplin()
|
|
existing = {"media/file_img1.png": "existing-res"}
|
|
new_body, res_map = upload_media_and_rewrite(body, tmp_path, client, existing)
|
|
assert "(:/existing-res)" in new_body
|
|
assert client.uploaded == [] # no re-upload
|
|
assert res_map == existing
|
|
|
|
def test_missing_file_left_as_link(self, tmp_path):
|
|
from src.joplin import upload_media_and_rewrite
|
|
(tmp_path / "media").mkdir()
|
|
body = ""
|
|
client = _FakeJoplin()
|
|
new_body, res_map = upload_media_and_rewrite(body, tmp_path, client, {})
|
|
assert new_body == body
|
|
assert res_map == {}
|
|
assert client.uploaded == []
|
|
|
|
def test_no_media_links_untouched(self, tmp_path):
|
|
from src.joplin import upload_media_and_rewrite
|
|
body = "Just text with [a link](https://example.com) and `media/foo` in code."
|
|
client = _FakeJoplin()
|
|
new_body, res_map = upload_media_and_rewrite(body, tmp_path, client, {})
|
|
assert new_body == body
|
|
assert res_map == {}
|