Derive asset preview URLs from the file path (#15509)

preview_url was assembled from a /api/view link whose type was chosen by
matching the asset's tags against "input" then "output". Anything written
anywhere else - temp above all, where preview nodes put their images - fell
off the end of that chain and came back with no preview at all. Tags are
user-editable, so removing one also silently destroyed the URL.

Derive the URL from where the file actually sits instead. That covers every
root /api/view serves, temp included, and no longer depends on tags or on a
filename in user_metadata. A file outside those roots, or content no client
can render from its own bytes, gets no preview URL rather than one that
cannot work.

Nominated previews are resolved a page at a time rather than per row, so a
list costs one extra query however long it is. A preview that is soft-deleted
or not visible to the caller drops out of that lookup and is no longer
advertised.

Co-authored-by: guill <jacob.e.segal@gmail.com>
This commit is contained in:
Simon Pinfold
2026-08-18 21:00:28 -07:00
committed by GitHub
parent a3e5cce506
commit 3aba3daef3
8 changed files with 607 additions and 36 deletions

View File

@@ -2,6 +2,7 @@ import asyncio
import functools
import json
import logging
import mimetypes
import os
import urllib.parse
import uuid
@@ -32,6 +33,7 @@ from app.assets.services import (
create_from_hash,
delete_asset_reference,
get_asset_detail,
get_preview_file_paths,
list_assets_page,
list_tags,
remove_tags,
@@ -40,7 +42,7 @@ from app.assets.services import (
upload_from_temp_path,
)
from app.assets.services.cursor import InvalidCursorError
from app.assets.services.path_utils import compute_display_name
from app.assets.services.path_utils import compute_asset_response_paths
from app.assets.services.tagging import list_tag_histogram
ROUTES = web.RouteTableDef()
@@ -207,44 +209,62 @@ def _validate_sort_field(requested: str | None) -> str:
return "created_at"
def _build_preview_url_from_view(tags: list[str], user_metadata: dict[str, Any] | None) -> str | None:
"""Build a /api/view preview URL from asset tags and user_metadata filename."""
if not user_metadata:
# What a client can render from the bytes themselves; anything else needs a nominated preview.
PREVIEWABLE_MIME_PREFIXES = ("image/", "video/", "audio/", "text/")
# models is deliberately absent: /api/view has no directory type for it.
VIEWABLE_NAMESPACES = frozenset({"input", "output", "temp"})
def _has_previewable_content(asset: schemas.AssetData | None, file_path: str | None) -> bool:
if asset is None:
return False
# Resolved from the path, not the caller-editable name, so a rename cannot change what previews.
raw = asset.mime_type or mimetypes.guess_type(file_path or "")[0] or ""
return raw.split(";", 1)[0].strip().lower().startswith(PREVIEWABLE_MIME_PREFIXES)
def _build_view_url(file_path: str | None) -> str | None:
# /api/view is a FileResponse: byte-range seeking, no user header, no access write.
if not file_path:
return None
filename = user_metadata.get("filename")
if not filename:
paths = compute_asset_response_paths(file_path)
if not paths:
return None
logical_path, relative_path = paths
namespace = logical_path.split("/", 1)[0]
if namespace not in VIEWABLE_NAMESPACES or not relative_path:
return None
if "input" in tags:
view_type = "input"
elif "output" in tags:
view_type = "output"
else:
return None
subfolder = ""
if "/" in filename:
subfolder, filename = filename.rsplit("/", 1)
encoded_filename = urllib.parse.quote(filename, safe="")
url = f"/api/view?type={view_type}&filename={encoded_filename}"
subfolder, _, filename = relative_path.rpartition("/")
url = f"/api/view?type={namespace}&filename={urllib.parse.quote(filename, safe='')}"
if subfolder:
url += f"&subfolder={urllib.parse.quote(subfolder, safe='')}"
return url
def _build_asset_response(result: schemas.AssetDetailResult | schemas.UploadResult) -> schemas_out.Asset:
"""Build an Asset response from a service result."""
def _resolve_preview_paths(
results: "list[schemas.AssetDetailResult] | list[schemas.AssetSummaryData]",
) -> dict[str, str]:
# A miss means no live preview - that is what keeps a soft-deleted one quiet.
preview_ids = {r.ref.preview_id for r in results if r.ref.preview_id}
return get_preview_file_paths(sorted(preview_ids))
def _build_asset_response(
result: schemas.AssetDetailResult | schemas.UploadResult,
preview_paths: dict[str, str],
) -> schemas_out.Asset:
if result.ref.preview_id:
preview_detail = get_asset_detail(result.ref.preview_id)
if preview_detail:
preview_url = _build_preview_url_from_view(preview_detail.tags, preview_detail.ref.user_metadata)
else:
preview_url = None
# A nominated preview is one whatever it holds, so no media check here.
preview_url = _build_view_url(preview_paths.get(result.ref.preview_id))
elif _has_previewable_content(result.asset, result.ref.file_path):
preview_url = _build_view_url(result.ref.file_path)
else:
preview_url = _build_preview_url_from_view(result.tags, result.ref.user_metadata)
preview_url = None
if result.ref.file_path:
display_name = compute_display_name(result.ref.file_path)
paths = compute_asset_response_paths(result.ref.file_path)
display_name = paths[1] if paths else None
# In-root loader path (model category dropped): what model loaders consume.
loader_path = result.ref.loader_path
else:
@@ -324,7 +344,8 @@ async def list_assets_route(request: web.Request) -> web.Response:
except InvalidCursorError as e:
return _build_error_response(400, "INVALID_CURSOR", str(e))
summaries = [_build_asset_response(item) for item in result.items]
preview_paths = _resolve_preview_paths(result.items)
summaries = [_build_asset_response(item, preview_paths) for item in result.items]
# has_more semantics differ by mode:
# - cursor mode: a non-empty next_cursor means there are more results.
@@ -363,7 +384,7 @@ async def get_asset_route(request: web.Request) -> web.Response:
{"id": reference_id},
)
payload = _build_asset_response(result)
payload = _build_asset_response(result, _resolve_preview_paths([result]))
except ValueError as e:
return _build_error_response(
404, "ASSET_NOT_FOUND", str(e), {"id": reference_id}
@@ -494,7 +515,7 @@ async def create_asset_from_hash_route(request: web.Request) -> web.Response:
404, "ASSET_NOT_FOUND", f"Asset content {body.hash} does not exist"
)
asset = _build_asset_response(result)
asset = _build_asset_response(result, _resolve_preview_paths([result]))
payload_out = schemas_out.AssetCreated(
**asset.model_dump(),
created_new=result.created_new,
@@ -585,7 +606,7 @@ async def upload_asset(request: web.Request) -> web.Response:
logging.exception("upload_asset failed for owner_id=%s", owner_id)
return _build_error_response(500, "INTERNAL", "Unexpected server error.")
asset = _build_asset_response(result)
asset = _build_asset_response(result, _resolve_preview_paths([result]))
payload_out = schemas_out.AssetCreated(
**asset.model_dump(),
created_new=result.created_new,
@@ -615,7 +636,7 @@ async def update_asset_route(request: web.Request) -> web.Response:
owner_id=USER_MANAGER.get_request_user_id(request),
preview_id=body.preview_id,
)
payload = _build_asset_response(result)
payload = _build_asset_response(result, _resolve_preview_paths([result]))
except PermissionError as pe:
return _build_error_response(403, "FORBIDDEN", str(pe), {"id": reference_id})
except ValueError as ve:

View File

@@ -28,6 +28,7 @@ from app.assets.database.queries.asset_reference import (
get_reference_by_id,
get_reference_with_owner_check,
get_reference_ids_by_ids,
get_reference_paths_by_ids,
get_references_by_paths_and_asset_ids,
get_references_for_prefixes,
get_unenriched_references,
@@ -101,6 +102,7 @@ __all__ = [
"get_reference_by_id",
"get_reference_with_owner_check",
"get_reference_ids_by_ids",
"get_reference_paths_by_ids",
"get_reference_tags",
"get_references_by_paths_and_asset_ids",
"get_references_for_prefixes",

View File

@@ -1064,6 +1064,27 @@ def get_references_by_paths_and_asset_ids(
return winners
def get_reference_paths_by_ids(
session: Session,
reference_ids: list[str],
) -> dict[str, str]:
"""Map reference id -> file_path for live, file-backed references."""
if not reference_ids:
return {}
paths: dict[str, str] = {}
for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS):
rows = session.execute(
select(AssetReference.id, AssetReference.file_path).where(
AssetReference.id.in_(chunk),
AssetReference.file_path.is_not(None),
AssetReference.deleted_at.is_(None),
)
)
paths.update({rid: fp for rid, fp in rows})
return paths
def get_reference_ids_by_ids(
session: Session,
reference_ids: list[str],

View File

@@ -4,6 +4,7 @@ from app.assets.services.asset_management import (
get_asset_by_hash,
get_asset_detail,
list_assets_page,
get_preview_file_paths,
resolve_asset_for_download,
set_asset_preview,
update_asset_metadata,
@@ -83,6 +84,7 @@ __all__ = [
"list_tags",
"cleanup_unreferenced_assets",
"remove_tags",
"get_preview_file_paths",
"resolve_asset_for_download",
"set_asset_preview",
"update_asset_metadata",

View File

@@ -21,6 +21,7 @@ from app.assets.database.queries import (
reference_exists_for_asset_id,
delete_reference_by_id,
fetch_reference_and_asset,
get_reference_paths_by_ids,
soft_delete_reference_by_id,
fetch_reference_asset_and_tags,
get_asset_by_hash as queries_get_asset_by_hash,
@@ -424,6 +425,14 @@ def resolve_hash_to_path(
)
def get_preview_file_paths(preview_ids: list[str]) -> dict[str, str]:
"""Map preview reference id -> file_path, in one query for the whole page."""
if not preview_ids:
return {}
with create_session() as session:
return get_reference_paths_by_ids(session, reference_ids=preview_ids)
def resolve_asset_for_download(
reference_id: str,
owner_id: str = "",

View File

@@ -43,7 +43,7 @@ def test_uses_persisted_loader_path_without_recomputing():
loader_path="SENTINEL/stored.safetensors",
)
resp = _build_asset_response(result)
resp = _build_asset_response(result, {})
assert resp.loader_path == "SENTINEL/stored.safetensors"
@@ -67,7 +67,7 @@ def test_null_stored_loader_path_is_served_as_null(tmp_path: Path):
mock_fp.models_dir = str(models)
result = _make_result(file_path=str(f), loader_path=None)
resp = _build_asset_response(result)
resp = _build_asset_response(result, {})
assert resp.loader_path is None
assert resp.display_name == "checkpoints/bar.safetensors"
@@ -77,7 +77,7 @@ def test_all_path_fields_null_without_file_path():
"""API-created / hash-only references (no file_path) expose no paths."""
result = _make_result(file_path=None, loader_path=None)
resp = _build_asset_response(result)
resp = _build_asset_response(result, {})
assert resp.loader_path is None
assert resp.display_name is None

View File

@@ -0,0 +1,300 @@
from datetime import datetime
from pathlib import Path
from unittest.mock import patch
import pytest
from app.assets.api.routes import _build_asset_response
from app.assets.services.schemas import AssetData, AssetDetailResult, ReferenceData
_TS = datetime(2024, 1, 1, 0, 0, 0)
@pytest.fixture
def sandboxed_comfy_roots(tmp_path: Path):
with patch("app.assets.services.path_utils.folder_paths") as fp:
fp.get_input_directory.return_value = str(tmp_path / "input")
fp.get_output_directory.return_value = str(tmp_path / "output")
fp.get_temp_directory.return_value = str(tmp_path / "temp")
fp.models_dir = str(tmp_path / "models")
yield tmp_path
def _make_result(
*,
ref_id: str = "ref-1",
name: str = "ComfyUI_temp_abcde_00001_.png",
file_path: str | None = None,
mime_type: str | None = "image/png",
preview_id: str | None = None,
tags: list[str] | None = None,
user_metadata: dict | None = None,
with_asset: bool = True,
) -> AssetDetailResult:
ref = ReferenceData(
id=ref_id,
name=name,
file_path=file_path,
loader_path=None,
user_metadata=user_metadata,
preview_id=preview_id,
created_at=_TS,
updated_at=_TS,
last_access_time=_TS,
)
asset = (
AssetData(hash="blake3:abc", size_bytes=1024, mime_type=mime_type)
if with_asset
else None
)
return AssetDetailResult(ref=ref, asset=asset, tags=tags or [])
@pytest.mark.parametrize(
("root", "relative"),
[
("temp", "ComfyUI_temp_abcde_00001_.png"),
("output", "ComfyUI_00001_.png"),
("input", "example.png"),
],
)
def test_every_view_root_gets_a_preview_url(
sandboxed_comfy_roots: Path, root: str, relative: str
):
resp = _build_asset_response(
_make_result(name=relative, file_path=str(sandboxed_comfy_roots / root / relative)),
{},
)
assert resp.preview_url == f"/api/view?type={root}&filename={relative}", (
f"a file in {root} must get a preview URL; temp is the one the old "
f"tag chain fell off the end of"
)
@pytest.mark.parametrize(
"tags",
[[], ["input"], ["output"], ["models", "model_type:checkpoints"]],
)
def test_preview_url_does_not_depend_on_tags(
sandboxed_comfy_roots: Path, tags: list[str]
):
resp = _build_asset_response(
_make_result(file_path=str(sandboxed_comfy_roots / "temp" / "a.png"), tags=tags), {}
)
assert resp.preview_url == "/api/view?type=temp&filename=a.png", (
"tags are user-editable; removing one must not destroy the preview"
)
def test_preview_url_does_not_depend_on_the_metadata_filename(
sandboxed_comfy_roots: Path,
):
resp = _build_asset_response(
_make_result(
file_path=str(sandboxed_comfy_roots / "output" / "a.png"), user_metadata=None
),
{},
)
assert resp.preview_url == "/api/view?type=output&filename=a.png", (
"a reference carrying no metadata filename must still get a preview"
)
def test_subfolder_is_split_out_and_both_halves_encoded(sandboxed_comfy_roots: Path):
resp = _build_asset_response(
_make_result(
name="my shot.png",
file_path=str(sandboxed_comfy_roots / "output" / "runs & takes" / "my shot.png"),
),
{},
)
assert resp.preview_url == (
"/api/view?type=output&filename=my%20shot.png&subfolder=runs%20%26%20takes"
), "an unencoded & or space in the path would break the query string"
def test_preview_id_resolves_through_the_page_lookup(sandboxed_comfy_roots: Path):
result = _make_result(
file_path=str(sandboxed_comfy_roots / "models" / "checkpoints" / "m.safetensors"),
mime_type="application/safetensors",
preview_id="preview-ref",
)
resp = _build_asset_response(
result, {"preview-ref": str(sandboxed_comfy_roots / "output" / "thumb.png")}
)
assert resp.preview_url == "/api/view?type=output&filename=thumb.png", (
"a nominated preview stands in for content with no visual form"
)
assert resp.preview_id == "preview-ref"
def test_unresolvable_preview_id_yields_no_url(sandboxed_comfy_roots: Path):
result = _make_result(
file_path=str(sandboxed_comfy_roots / "output" / "a.png"), preview_id="gone"
)
resp = _build_asset_response(result, {})
assert resp.preview_url is None, (
"a preview absent from the lookup is soft-deleted, invisible or "
"path-less, so advertising it would promise a URL that 404s - and the "
"asset's own bytes are a different picture, not a degraded one"
)
@pytest.mark.parametrize(
("name", "mime_type"),
[
("notes.txt", "text/plain"),
("notes.md", "text/markdown"),
("rows.csv", "text/csv"),
("page.html", "text/html"),
],
)
def test_text_is_previewable(sandboxed_comfy_roots: Path, name: str, mime_type: str):
resp = _build_asset_response(
_make_result(
name=name,
file_path=str(sandboxed_comfy_roots / "output" / name),
mime_type=mime_type,
),
{},
)
assert resp.preview_url == f"/api/view?type=output&filename={name}", (
"text assets are rendered as a snippet from preview_url, so withholding "
"it leaves that with nothing to fetch; the dangerous members stay safe "
"because /api/view forces them to download, not because they get no URL"
)
@pytest.mark.parametrize(
"mime_type",
["application/safetensors", "application/gguf", "application/octet-stream"],
)
def test_no_preview_url_for_content_a_browser_cannot_render(
sandboxed_comfy_roots: Path, mime_type: str
):
resp = _build_asset_response(
_make_result(
name="model.safetensors",
file_path=str(sandboxed_comfy_roots / "input" / "model.safetensors"),
mime_type=mime_type,
),
{},
)
assert resp.preview_url is None, (
"content a browser cannot render must not advertise itself as a preview"
)
@pytest.mark.parametrize(
("name", "expected"),
[
("shot.png", "/api/view?type=temp&filename=shot.png"),
("clip.mp4", "/api/view?type=temp&filename=clip.mp4"),
("model.safetensors", None),
],
)
def test_missing_mime_type_falls_back_to_the_path(
sandboxed_comfy_roots: Path, name: str, expected: str | None
):
resp = _build_asset_response(
_make_result(
name=name, file_path=str(sandboxed_comfy_roots / "temp" / name), mime_type=None
),
{},
)
assert resp.preview_url == expected, (
"a previewable file must not lose its preview just because the scan "
"that found it recorded no mime type"
)
@pytest.mark.parametrize(
("name", "stored_filename", "expected"),
[
("untitled", "shot.png", "/api/view?type=temp&filename=shot.png"),
("shot.png", "weights.safetensors", None),
],
)
def test_previewability_follows_the_path_not_the_editable_name(
sandboxed_comfy_roots: Path, name: str, stored_filename: str, expected: str | None
):
resp = _build_asset_response(
_make_result(
name=name,
file_path=str(sandboxed_comfy_roots / "temp" / stored_filename),
mime_type=None,
),
{},
)
assert resp.preview_url == expected, (
"name is editable through PUT /api/assets/{id}, so deriving "
"previewability from it would let a rename create or destroy a preview "
"without the bytes changing"
)
def test_mime_type_parameters_do_not_defeat_the_media_check(
sandboxed_comfy_roots: Path,
):
resp = _build_asset_response(
_make_result(
file_path=str(sandboxed_comfy_roots / "temp" / "a.png"),
mime_type="IMAGE/PNG; charset=binary",
),
{},
)
assert resp.preview_url == "/api/view?type=temp&filename=a.png"
def test_no_preview_url_for_a_model(sandboxed_comfy_roots: Path):
resp = _build_asset_response(
_make_result(
name="m.png",
file_path=str(sandboxed_comfy_roots / "models" / "checkpoints" / "m.png"),
),
{},
)
assert resp.preview_url is None, (
"models is not a root /api/view can address, whatever the file is"
)
def test_no_preview_url_for_a_path_outside_every_root(sandboxed_comfy_roots: Path):
resp = _build_asset_response(_make_result(file_path="/elsewhere/a.png"), {})
assert resp.preview_url is None, (
"/api/view cannot address a file outside the roots it serves"
)
def test_no_preview_url_without_a_file_path(sandboxed_comfy_roots: Path):
resp = _build_asset_response(_make_result(file_path=None), {})
assert resp.preview_url is None, (
"an API-created reference has no path, so no view URL can be derived"
)
def test_no_preview_url_without_content(sandboxed_comfy_roots: Path):
resp = _build_asset_response(
_make_result(
file_path=str(sandboxed_comfy_roots / "temp" / "a.png"), with_asset=False
),
{},
)
assert resp.preview_url is None, "no asset row means there is nothing to preview"

View File

@@ -0,0 +1,216 @@
import contextlib
import json
import re
import uuid
import requests
def test_preview_url_serves_the_asset(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-url-{uuid.uuid4().hex[:6]}"
name = f"{scope}.png"
data = make_asset_bytes(name, 2048)
body = asset_factory(name, ["output", "unit-tests", scope], {}, data)
assert re.fullmatch(r"/api/view\?type=output&filename=[^&]+", body["preview_url"]), (
f"unexpected preview URL shape: {body['preview_url']!r}"
)
r = http.get(api_base + body["preview_url"], timeout=120)
assert r.status_code == 200, r.text
assert r.content == data, "the preview URL must serve the asset's own bytes"
def test_preview_url_honours_range_requests(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-range-{uuid.uuid4().hex[:6]}"
name = f"{scope}.png"
data = make_asset_bytes(name, 2048)
body = asset_factory(name, ["output", "unit-tests", scope], {}, data)
r = http.get(
api_base + body["preview_url"], headers={"Range": "bytes=10-109"}, timeout=120
)
assert r.status_code == 206, (
f"expected a partial response, got {r.status_code}: native <video>/<audio> "
f"seek by byte range and will not play a source that ignores it"
)
assert r.content == data[10:110], "the served range must be the requested one"
def test_preview_url_needs_no_user_header(
api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-anon-{uuid.uuid4().hex[:6]}"
name = f"{scope}.png"
data = make_asset_bytes(name, 1024)
body = asset_factory(name, ["input", "unit-tests", scope], {}, data)
with requests.Session() as bare:
r = bare.get(api_base + body["preview_url"], timeout=120)
assert r.status_code == 200, (
f"a browser fetching <img src> cannot attach a Comfy-User header, so the "
f"preview URL must resolve without one; got {r.status_code}: {r.text}"
)
assert r.content == data
def test_preview_url_survives_tag_removal(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-tags-{uuid.uuid4().hex[:6]}"
name = f"{scope}.png"
data = make_asset_bytes(name, 2048)
body = asset_factory(name, ["input", "unit-tests", scope], {}, data)
aid = body["id"]
preview_url = body["preview_url"]
assert preview_url, "an uploaded image starts out with a preview"
r = http.delete(
f"{api_base}/api/assets/{aid}/tags", json={"tags": ["input"]}, timeout=120
)
assert r.status_code == 200, r.text
after = http.get(f"{api_base}/api/assets/{aid}", timeout=120).json()
assert "input" not in after["tags"]
assert after["preview_url"] == preview_url, (
"dropping the tag that used to select the view type must not take the "
"preview with it"
)
assert http.get(api_base + preview_url, timeout=120).status_code == 200
def test_preview_url_is_the_nominated_preview_when_one_is_set(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-id-{uuid.uuid4().hex[:6]}"
thumb_name = f"{scope}_thumb.png"
thumb_data = make_asset_bytes(thumb_name, 1024)
thumb = asset_factory(thumb_name, ["input", "unit-tests", scope], {}, thumb_data)
model_name = f"{scope}.safetensors"
files = {"file": (model_name, make_asset_bytes(model_name, 2048), "application/octet-stream")}
form_data = {
"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", scope]),
"name": model_name,
"preview_id": thumb["id"],
}
r = http.post(api_base + "/api/assets", files=files, data=form_data, timeout=120)
model = r.json()
assert r.status_code in (200, 201), model
try:
assert model["preview_id"] == thumb["id"]
assert model["preview_url"] == thumb["preview_url"], (
"a nominated preview stands in for content with no visual form"
)
got = http.get(api_base + model["preview_url"], timeout=120)
assert got.status_code == 200, got.text
assert got.content == thumb_data
finally:
with contextlib.suppress(Exception):
http.delete(f"{api_base}/api/assets/{model['id']}", timeout=30)
def test_soft_deleted_preview_is_not_advertised(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-gone-{uuid.uuid4().hex[:6]}"
thumb_name = f"{scope}_thumb.png"
thumb = asset_factory(
thumb_name, ["input", "unit-tests", scope], {}, make_asset_bytes(thumb_name, 1024)
)
parent_name = f"{scope}_parent.png"
parent = asset_factory(
parent_name, ["input", "unit-tests", scope], {}, make_asset_bytes(parent_name, 1024)
)
r = http.put(
f"{api_base}/api/assets/{parent['id']}",
json={"preview_id": thumb["id"]},
timeout=120,
)
assert r.status_code == 200, r.text
assert r.json()["preview_url"] == thumb["preview_url"]
assert http.delete(f"{api_base}/api/assets/{thumb['id']}", timeout=30).status_code in (200, 204)
after = http.get(f"{api_base}/api/assets/{parent['id']}", timeout=120).json()
assert after["preview_id"] == thumb["id"]
assert after.get("preview_url") is None, (
"soft delete leaves the parent's preview_id pointing at it, so the "
"response has to drop the URL rather than promise one that 404s"
)
def test_no_preview_url_for_content_a_browser_cannot_render(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-model-{uuid.uuid4().hex[:6]}"
name = f"{scope}.safetensors"
body = asset_factory(
name,
["models", "model_type:checkpoints", "unit-tests", scope],
{},
make_asset_bytes(name, 2048),
)
assert body.get("preview_url") is None, (
"model weights have no preview of their own"
)
listed = http.get(
api_base + "/api/assets", params={"include_tags": scope}, timeout=120
).json()["assets"]
assert [a.get("preview_url") for a in listed] == [None], (
"the list route must withhold it too, not just the detail route"
)
def test_text_asset_gets_a_preview_url_that_serves_its_content(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-text-{uuid.uuid4().hex[:6]}"
name = f"{scope}.txt"
data = b"line one\nline two\n" + make_asset_bytes(name, 256)
body = asset_factory(name, ["output", "unit-tests", scope], {}, data)
assert body.get("preview_url"), (
"text assets are rendered as a snippet fetched from preview_url"
)
r = http.get(api_base + body["preview_url"], timeout=120)
assert r.status_code == 200, r.text
assert r.content == data
def test_dangerous_text_preview_is_still_forced_to_download(
http: requests.Session, api_base: str, asset_factory
):
scope = f"preview-html-{uuid.uuid4().hex[:6]}"
name = f"{scope}.html"
data = f"<html><script>alert('{scope}')</script></html>".encode()
body = asset_factory(name, ["output", "unit-tests", scope], {}, data)
assert body.get("preview_url"), "text/html matches the previewable prefix"
r = http.get(api_base + body["preview_url"], timeout=120)
r.content
assert r.status_code == 200
ct = r.headers.get("Content-Type", "").lower()
cd = r.headers.get("Content-Disposition", "").lower()
assert ct.startswith("application/octet-stream"), (
f"admitting text/ to the previewable set must not let HTML render "
f"inline in the app origin; got {ct!r}"
)
assert "attachment" in cd, f"expected a forced download, got {cd!r}"