digital-garbage
3931677507
Pieces B and C: Scene gains source_image (aspect-ratio reference only -- resolution itself stays a project.video setting, ImageTransformKJ's "total_pixels" keep_proportion fits the picture's aspect into that budget) and references (media ids, in order). Inspector's Scene tab gets pick/clear/ remove controls; picking routes through the existing Media bin via a small pending-pick handshake in boot.js (one bin, one onPick, so "which field asked" has to be tracked between the click that asked and the click that answered). References wire into MiniMaxH3ReferenceToVideo's ref_images autogrow input dynamically, per run, only for what a scene actually has -- an unwired slot, not an empty image, is what "not in use" means to that node. The target is data on the role itself (wireTo), not a slot id boot.js would otherwise have to know, keeping the "core never names an implementation" contract intact one layer up into the app. Verifying this exposed a second real core gap (after the MatchType one): core/graph.py had no idea what an Autogrow (V3's io.Autogrow) input even is -- "ref_images" reported as one opaque COMFY_AUTOGROW_V3 slot, its real per-instance sockets (ref_image_0, ref_image_1, ...) invisible to declared()/build(), so a wire naming one was refused as an input the node does not have. Fixed generically in comfy_types.py (autogrow_instances()) and graph.py, confirmed against ComfyUI's own _io.py (finalize_prefix's dotted join, build_nested_inputs's runtime un-nesting) -- not guessed, and proven end to end with a real override()+build() test using the exact edit boot.js makes.
504 lines
20 KiB
Python
504 lines
20 KiB
Python
"""The project store, and the same store over HTTP.
|
|
|
|
A project is the one thing in the app that has to survive a reload, so the tests
|
|
that matter here are about what happens to it when the input is not what the app
|
|
would have sent: a hand-edited file, a body that is not an object, an id that is
|
|
really a path.
|
|
"""
|
|
|
|
import json
|
|
import socket
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from core import config, projects
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(config, "PROJECTS_DIR", tmp_path / "projects")
|
|
return tmp_path / "projects"
|
|
|
|
|
|
# ── the model ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_a_new_project_can_be_typed_into(store):
|
|
"""One empty scene, not none: a timeline with nothing on it has nowhere to
|
|
put a prompt, so an empty new project would need an Add before it was usable."""
|
|
made = projects.create("First")
|
|
assert made.name == "First"
|
|
assert len(made.scenes) == 1
|
|
assert made.scenes[0].text == ""
|
|
|
|
|
|
def test_a_project_survives_a_round_trip(store):
|
|
made = projects.create("Round trip")
|
|
made.scenes.append(projects.Scene(text="a wolf on a ridge"))
|
|
projects.save(made)
|
|
|
|
read = projects.get(made.id)
|
|
assert [s.text for s in read.scenes] == ["", "a wolf on a ridge"]
|
|
assert read.name == "Round trip"
|
|
|
|
|
|
def test_scenes_keep_their_order(store):
|
|
made = projects.create("Ordered")
|
|
made.scenes = [projects.Scene(text=str(i)) for i in range(6)]
|
|
projects.save(made)
|
|
assert [s.text for s in projects.get(made.id).scenes] == [str(i) for i in range(6)]
|
|
|
|
|
|
def test_a_hand_edited_file_does_not_crash_the_app(store):
|
|
"""The store reads files a person can edit. Every one of these used to be an
|
|
attribute error on a dict that was a list, or a str that was None."""
|
|
made = projects.create("Edited")
|
|
path = store / f"{made.id}.json"
|
|
|
|
for body in ('{"scenes": "not a list"}', '{"scenes": [null, 3, "x"]}',
|
|
'{"name": 42}', '[]', '"just a string"', '{}'):
|
|
path.write_text(body, encoding="utf-8")
|
|
read = projects.get(made.id)
|
|
assert read is not None, body
|
|
assert isinstance(read.scenes, list)
|
|
assert isinstance(read.name, str)
|
|
|
|
|
|
def test_unreadable_json_is_not_a_project(store):
|
|
made = projects.create("Broken")
|
|
(store / f"{made.id}.json").write_text("{oh no", encoding="utf-8")
|
|
assert projects.get(made.id) is None
|
|
assert projects.listing() == [] # and it does not take the list down with it
|
|
|
|
|
|
def test_a_scene_without_an_id_gets_one(store):
|
|
"""Ids address a scene from the client. Two scenes sharing one, or a scene
|
|
with none, is a selection that lands on the wrong row."""
|
|
p = projects.Project.from_dict({"scenes": [{"text": "a"}, {"text": "b"}]})
|
|
ids = [s.id for s in p.scenes]
|
|
assert all(projects.is_id(i) for i in ids)
|
|
assert len(set(ids)) == 2
|
|
|
|
|
|
def test_an_id_from_a_request_cannot_be_a_path(store):
|
|
"""The filename is a generated id or the call fails. A sanitiser that
|
|
repairs "../../x" writes somewhere nobody asked for and says it saved."""
|
|
for bad in ("../escape", "a/b", "", "..", "x" * 64, None, 5, "ABCDEF123456"):
|
|
assert projects.is_id(bad) is False
|
|
assert projects.get(bad) is None
|
|
assert projects.delete(bad) is False
|
|
|
|
|
|
def test_the_listing_is_newest_first(store):
|
|
a = projects.create("A")
|
|
b = projects.create("B")
|
|
a.name = "A again"
|
|
projects.save(a) # touches updated_at
|
|
assert [p["id"] for p in projects.listing()] == [a.id, b.id]
|
|
assert [p["name"] for p in projects.listing()][0] == "A again"
|
|
|
|
|
|
def test_a_long_name_is_cut_not_refused(store):
|
|
made = projects.create("x" * 500)
|
|
assert len(made.name) == projects.MAX_NAME
|
|
|
|
|
|
def test_delete_removes_it(store):
|
|
made = projects.create("Doomed")
|
|
assert projects.delete(made.id) is True
|
|
assert projects.get(made.id) is None
|
|
assert projects.delete(made.id) is False
|
|
|
|
|
|
def test_a_save_is_atomic(store):
|
|
"""A reader must never see half a project. The temp file is what makes that
|
|
true, and a leftover .tmp in the directory would be read as one."""
|
|
made = projects.create("Atomic")
|
|
projects.save(made)
|
|
assert list(store.glob("*.tmp")) == []
|
|
assert len(projects.listing()) == 1
|
|
|
|
|
|
# ── over HTTP ────────────────────────────────────────────────────────────
|
|
|
|
pytest.importorskip("aiohttp")
|
|
|
|
|
|
@pytest.fixture
|
|
def server(store):
|
|
from aiohttp import web as aioweb
|
|
from core import routes
|
|
|
|
app = aioweb.Application()
|
|
table = aioweb.RouteTableDef()
|
|
routes.register(table, prefix="/funpack")
|
|
app.add_routes(table)
|
|
|
|
holder = {"ready": threading.Event()}
|
|
|
|
def run():
|
|
import asyncio
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
runner = aioweb.AppRunner(app)
|
|
loop.run_until_complete(runner.setup())
|
|
site = aioweb.TCPSite(runner, "127.0.0.1", 0)
|
|
loop.run_until_complete(site.start())
|
|
holder["port"] = site._server.sockets[0].getsockname()[1]
|
|
holder["loop"] = loop
|
|
holder["ready"].set()
|
|
loop.run_forever()
|
|
|
|
thread = threading.Thread(target=run, daemon=True)
|
|
thread.start()
|
|
assert holder["ready"].wait(20), "the test server did not start"
|
|
yield holder["port"]
|
|
holder["loop"].call_soon_threadsafe(holder["loop"].stop)
|
|
|
|
|
|
def _request(port, method, path, body=None):
|
|
payload = b"" if body is None else json.dumps(body).encode()
|
|
head = (f"{method} {path} HTTP/1.1\r\nHost: localhost\r\n"
|
|
f"Content-Type: application/json\r\n"
|
|
f"Content-Length: {len(payload)}\r\nConnection: close\r\n\r\n").encode()
|
|
with socket.create_connection(("127.0.0.1", port), timeout=30) as sock:
|
|
sock.sendall(head + payload)
|
|
chunks = []
|
|
while True:
|
|
got = sock.recv(65536)
|
|
if not got:
|
|
break
|
|
chunks.append(got)
|
|
header, _, rest = b"".join(chunks).partition(b"\r\n\r\n")
|
|
status = int(header.split()[1])
|
|
try:
|
|
return status, json.loads(rest.decode() or "{}")
|
|
except ValueError:
|
|
return status, {"raw": rest.decode(errors="replace")}
|
|
|
|
|
|
def test_create_list_read_save_delete(server):
|
|
status, made = _request(server, "POST", "/funpack/api/projects", {"name": "HTTP"})
|
|
assert status == 200 and made["name"] == "HTTP"
|
|
pid = made["id"]
|
|
|
|
status, listed = _request(server, "GET", "/funpack/api/projects")
|
|
assert status == 200 and [p["id"] for p in listed["projects"]] == [pid]
|
|
assert listed["projects"][0]["scenes"] == 1
|
|
|
|
made["scenes"].append({"text": "a second scene"})
|
|
status, saved = _request(server, "PUT", f"/funpack/api/projects/{pid}", made)
|
|
assert status == 200 and len(saved["scenes"]) == 2
|
|
|
|
status, read = _request(server, "GET", f"/funpack/api/projects/{pid}")
|
|
assert status == 200 and read["scenes"][1]["text"] == "a second scene"
|
|
|
|
assert _request(server, "DELETE", f"/funpack/api/projects/{pid}")[0] == 200
|
|
assert _request(server, "GET", f"/funpack/api/projects/{pid}")[0] == 404
|
|
|
|
|
|
def test_a_put_cannot_rename_its_own_target(server):
|
|
"""The id comes from the URL. A body that could name a different project is
|
|
a write over someone else's work, reported as a success."""
|
|
_, first = _request(server, "POST", "/funpack/api/projects", {"name": "First"})
|
|
_, second = _request(server, "POST", "/funpack/api/projects", {"name": "Second"})
|
|
|
|
_request(server, "PUT", f"/funpack/api/projects/{first['id']}",
|
|
{**first, "id": second["id"], "name": "Overwritten"})
|
|
|
|
_, untouched = _request(server, "GET", f"/funpack/api/projects/{second['id']}")
|
|
assert untouched["name"] == "Second"
|
|
_, target = _request(server, "GET", f"/funpack/api/projects/{first['id']}")
|
|
assert target["name"] == "Overwritten"
|
|
|
|
|
|
def test_a_body_that_is_not_an_object_is_refused(server):
|
|
_, made = _request(server, "POST", "/funpack/api/projects", {"name": "Shape"})
|
|
for body in ([], "text", 7):
|
|
status, _ = _request(server, "PUT", f"/funpack/api/projects/{made['id']}", body)
|
|
assert status == 400, body
|
|
|
|
|
|
def test_saving_something_that_is_not_there_is_a_404(server):
|
|
status, _ = _request(server, "PUT", "/funpack/api/projects/0123456789ab", {"name": "Ghost"})
|
|
assert status == 404
|
|
|
|
|
|
def test_a_path_shaped_id_is_refused_not_served(server):
|
|
for pid in ("..", "%2e%2e", "a/b"):
|
|
status, _ = _request(server, "GET", f"/funpack/api/projects/{pid}")
|
|
assert status in (404, 400), pid
|
|
|
|
|
|
# --- moving a project between machines ----------------------------------------
|
|
|
|
def test_a_downloaded_project_imports_back_as_a_new_one(server):
|
|
"""The round trip the feature exists for: save it, move the file
|
|
somewhere else, load it there."""
|
|
_, made = _request(server, "POST", "/funpack/api/projects", {"name": "Portable"})
|
|
made["scenes"].append({"text": "a second scene"})
|
|
_request(server, "PUT", f"/funpack/api/projects/{made['id']}", made)
|
|
|
|
status, downloaded = _request(server, "GET", f"/funpack/api/projects/{made['id']}/download")
|
|
assert status == 200
|
|
assert downloaded["name"] == "Portable"
|
|
assert len(downloaded["scenes"]) == 2
|
|
|
|
status, imported = _request(server, "POST", "/funpack/api/projects/import", downloaded)
|
|
assert status == 200
|
|
assert imported["name"] == "Portable"
|
|
assert [s["text"] for s in imported["scenes"]] == ["", "a second scene"]
|
|
|
|
status, listed = _request(server, "GET", "/funpack/api/projects")
|
|
assert sorted(p["id"] for p in listed["projects"]) == sorted([made["id"], imported["id"]]), \
|
|
"the import did not land as a project of its own"
|
|
|
|
|
|
def test_importing_a_project_never_reuses_its_old_id(server):
|
|
"""Importing the SAME file twice -- or importing it back on the machine it
|
|
came from -- must not silently overwrite whatever already has that id."""
|
|
_, made = _request(server, "POST", "/funpack/api/projects", {"name": "Original"})
|
|
status, downloaded = _request(server, "GET", f"/funpack/api/projects/{made['id']}/download")
|
|
assert downloaded["id"] == made["id"]
|
|
|
|
status, imported = _request(server, "POST", "/funpack/api/projects/import", downloaded)
|
|
assert status == 200
|
|
assert imported["id"] != made["id"]
|
|
|
|
# The original is untouched -- an id collision would have overwritten it.
|
|
_, original = _request(server, "GET", f"/funpack/api/projects/{made['id']}")
|
|
assert original["name"] == "Original"
|
|
|
|
|
|
def test_an_id_offered_in_the_body_does_not_win_either(server):
|
|
"""Not just the downloaded shape -- ANY id in an imported body is a
|
|
liability, including a made-up one that could collide with something
|
|
imported later."""
|
|
status, imported = _request(server, "POST", "/funpack/api/projects/import",
|
|
{"id": "aaaaaaaaaaaa", "name": "Sneaky", "scenes": [{"text": ""}]})
|
|
assert status == 200
|
|
assert imported["id"] != "aaaaaaaaaaaa"
|
|
|
|
|
|
def test_something_that_does_not_look_like_a_project_is_refused(server):
|
|
for body in ({"name": "not a project"}, {}):
|
|
status, resp = _request(server, "POST", "/funpack/api/projects/import", body)
|
|
assert status == 400, body
|
|
assert "problems" in resp
|
|
|
|
|
|
def test_a_scenes_field_of_the_wrong_shape_degrades_to_empty_rather_than_erroring(server):
|
|
"""`scenes` present but not a list is the same "never trust a field" rule
|
|
every other project load already follows (Project.from_dict), not a
|
|
special case this route invents -- an import is not the one place a
|
|
malformed file gets to crash instead of degrading."""
|
|
status, imported = _request(server, "POST", "/funpack/api/projects/import",
|
|
{"name": "Odd", "scenes": "not a list"})
|
|
assert status == 200, imported
|
|
assert imported["scenes"] == []
|
|
|
|
|
|
def test_downloading_a_project_that_does_not_exist_is_a_404(server):
|
|
status, _ = _request(server, "GET", "/funpack/api/projects/0123456789ab/download")
|
|
assert status == 404
|
|
|
|
|
|
def test_a_disk_failure_during_import_is_not_blamed_on_the_file(server, monkeypatch):
|
|
"""from_dict() is lenient by design -- it cannot actually raise on a
|
|
well-formed body. The only thing that can fail here is save()'s own I/O,
|
|
which is not the file's fault and should not be reported as though it
|
|
were."""
|
|
from core import projects
|
|
|
|
def broken_save(_project):
|
|
raise OSError("No space left on device")
|
|
monkeypatch.setattr(projects, "save", broken_save)
|
|
|
|
status, resp = _request(server, "POST", "/funpack/api/projects/import",
|
|
{"name": "Fine", "scenes": [{"text": ""}]})
|
|
assert status == 500, resp
|
|
assert "No space left on device" in resp["problems"][0]
|
|
assert "invalid" not in resp["problems"][0].lower()
|
|
|
|
|
|
# --- what the project is generated at ----------------------------------------
|
|
|
|
def test_video_settings_survive_a_round_trip(store):
|
|
made = projects.create("With settings")
|
|
made.video = {"width": 832, "height": 480, "length": 97}
|
|
projects.save(made)
|
|
|
|
back = projects.get(made.id)
|
|
assert back.video == {"width": 832, "height": 480, "length": 97}
|
|
|
|
|
|
def test_a_project_file_is_never_trusted_for_a_setting():
|
|
"""The one thing in this app that outlives the code that wrote it."""
|
|
project = projects.Project.from_dict({"video": {
|
|
"width": "832", # a number that arrived as text
|
|
"height": 480,
|
|
"length": 0, # out of range
|
|
"batch": -4, # out of range
|
|
"huge": 999_999, # out of range
|
|
"flag": True, # bool is an int in Python
|
|
"name": "wide", # not a number at all
|
|
7: 512, # not a name
|
|
}})
|
|
assert project.video == {"width": 832, "height": 480}
|
|
|
|
|
|
def test_video_is_an_object_or_it_is_nothing():
|
|
for junk in ([512, 512], "832x480", 7, None):
|
|
assert projects.Project.from_dict({"video": junk}).video == {}
|
|
|
|
|
|
def test_a_project_with_no_settings_has_an_empty_one(store):
|
|
assert projects.Project.from_dict({}).video == {}
|
|
assert projects.create("Fresh").video == {}
|
|
|
|
|
|
def test_a_scene_keeps_a_crop_and_a_rating(store):
|
|
made = projects.create("Rated")
|
|
made.scenes = [projects.Scene(text="a cat", length=48, rating="liked")]
|
|
projects.save(made)
|
|
|
|
back = projects.get(made.id).scenes[0]
|
|
assert back.length == 48
|
|
assert back.rating == "liked"
|
|
|
|
|
|
def test_a_scene_field_read_back_is_never_trusted():
|
|
scene = projects.Scene.from_dict({
|
|
"length": "48", # a number that arrived as text
|
|
"rating": "perfect", # a word an older scale had, and this one has not
|
|
})
|
|
assert scene.length == 48
|
|
assert scene.rating is None
|
|
|
|
for junk in (0, -4, True, None, "soon", 999_999):
|
|
assert projects.Scene.from_dict({"length": junk}).length is None, junk
|
|
|
|
|
|
def test_a_scenes_source_image_and_references_survive_a_round_trip(store):
|
|
made = projects.create("Referenced")
|
|
made.scenes = [projects.Scene(text="a fox", source_image="abcdef012345",
|
|
references=["111111111111", "222222222222"])]
|
|
projects.save(made)
|
|
|
|
back = projects.get(made.id).scenes[0]
|
|
assert back.source_image == "abcdef012345"
|
|
assert back.references == ["111111111111", "222222222222"]
|
|
|
|
|
|
def test_a_scenes_media_ids_that_are_not_shaped_like_media_ids_are_dropped():
|
|
scene = projects.Scene.from_dict({
|
|
"source_image": "not a real id",
|
|
"references": ["111111111111", "also not one", 4, None],
|
|
})
|
|
assert scene.source_image is None
|
|
assert scene.references == ["111111111111"]
|
|
|
|
|
|
def test_a_scenes_references_field_of_the_wrong_shape_degrades_to_empty():
|
|
for junk in ("111111111111", 4, None, {"a": 1}):
|
|
assert projects.Scene.from_dict({"references": junk}).references == [], junk
|
|
|
|
|
|
# --- the project-level prompt fields: negative, anchor, postfix --------------
|
|
|
|
def test_negative_survives_a_round_trip(store):
|
|
made = projects.create("Negative")
|
|
made.negative = "blurry, low quality"
|
|
projects.save(made)
|
|
assert projects.get(made.id).negative == "blurry, low quality"
|
|
|
|
|
|
def test_anchor_survives_a_round_trip(store):
|
|
made = projects.create("Anchor")
|
|
made.anchor = "a lone rider on a frost-glider"
|
|
projects.save(made)
|
|
assert projects.get(made.id).anchor == "a lone rider on a frost-glider"
|
|
|
|
|
|
def test_postfix_and_its_toggle_survive_a_round_trip(store):
|
|
made = projects.create("Postfix")
|
|
made.postfix = "cinematic, 35mm"
|
|
made.postfix_enabled = False
|
|
projects.save(made)
|
|
|
|
back = projects.get(made.id)
|
|
assert back.postfix == "cinematic, 35mm"
|
|
assert back.postfix_enabled is False
|
|
|
|
|
|
def test_a_fresh_project_has_empty_text_fields_and_postfix_on(store):
|
|
made = projects.create("Fresh")
|
|
assert made.negative == made.anchor == made.postfix == ""
|
|
assert made.postfix_enabled is True
|
|
|
|
|
|
def test_a_project_file_from_before_these_fields_existed_reads_as_the_v4_default():
|
|
"""postfix_enabled defaults ON, matching v4 -- a project saved before this
|
|
field existed must not silently read as "postfix off" just because
|
|
nothing was there to say "on"."""
|
|
project = projects.Project.from_dict({"name": "Old"})
|
|
assert project.postfix_enabled is True
|
|
assert project.negative == project.anchor == project.postfix == ""
|
|
|
|
|
|
def test_the_text_fields_read_back_are_never_trusted():
|
|
for junk in (0, -4, True, None, ["a", "list"], {"not": "a string"}):
|
|
project = projects.Project.from_dict(
|
|
{"negative": junk, "anchor": junk, "postfix": junk})
|
|
assert project.negative == project.anchor == project.postfix == "", junk
|
|
|
|
|
|
def test_postfix_enabled_read_back_is_never_trusted():
|
|
for junk in ("yes", 1, 0, None, [], "false"):
|
|
# Anything that is not a real bool degrades to the v4 default (on),
|
|
# the same rule every other field here follows: a file cannot say
|
|
# something with a value core does not recognise as that thing.
|
|
assert projects.Project.from_dict({"postfix_enabled": junk}).postfix_enabled is True, junk
|
|
assert projects.Project.from_dict({"postfix_enabled": False}).postfix_enabled is False
|
|
|
|
|
|
def test_variables_survive_a_round_trip(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(config, "PROJECTS_DIR", tmp_path)
|
|
p = projects.create("v")
|
|
p.variables = [{"name": "subject", "value": "a fox"}]
|
|
projects.save(p)
|
|
back = projects.get(p.id)
|
|
assert back.variables == [{"name": "subject", "value": "a fox"}]
|
|
|
|
|
|
def test_a_fresh_project_has_no_variables():
|
|
assert projects.Project().variables == []
|
|
|
|
|
|
def test_a_variable_with_no_name_is_dropped():
|
|
p = projects.Project.from_dict({"variables": [
|
|
{"name": "", "value": "x"}, {"name": " ", "value": "y"}, {"name": "ok", "value": "z"},
|
|
]})
|
|
assert p.variables == [{"name": "ok", "value": "z"}]
|
|
|
|
|
|
def test_a_leading_dollar_in_a_saved_variable_name_is_stripped():
|
|
p = projects.Project.from_dict({"variables": [{"name": "$subject", "value": "fox"}]})
|
|
assert p.variables == [{"name": "subject", "value": "fox"}]
|
|
|
|
|
|
def test_a_variables_value_that_is_not_a_string_reads_back_empty():
|
|
p = projects.Project.from_dict({"variables": [{"name": "a", "value": 5}, {"name": "b"}]})
|
|
assert p.variables == [{"name": "a", "value": ""}, {"name": "b", "value": ""}]
|
|
|
|
|
|
def test_variables_that_are_not_a_list_read_back_empty():
|
|
for junk in ("nope", {"a": "b"}, 5, None):
|
|
assert projects.Project.from_dict({"variables": junk}).variables == []
|
|
|
|
|
|
def test_entries_in_variables_that_are_not_objects_are_dropped():
|
|
p = projects.Project.from_dict({"variables": ["nope", 5, {"name": "ok", "value": "1"}]})
|
|
assert p.variables == [{"name": "ok", "value": "1"}]
|