1382 lines
59 KiB
Python
1382 lines
59 KiB
Python
"""Physical-file prompt inspection is bounded, conservative and read-only."""
|
|
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
from PIL import Image, PngImagePlugin
|
|
|
|
try:
|
|
from test_api_images import ImageAPITestCase, make_png
|
|
import inspection_parity as parity
|
|
except ImportError:
|
|
from tests.test_api_images import ImageAPITestCase, make_png
|
|
from tests import inspection_parity as parity
|
|
|
|
from utils.pair_execution import EXECUTION_KEY, record_pair_execution
|
|
from database.pair_ingestion import ingest_resolved_pair
|
|
from utils.prompt_pair_resolver import resolve_prompt_pair
|
|
|
|
|
|
def graph(positive="portrait", negative="blurry"):
|
|
return {
|
|
"1": {"class_type": "CLIPTextEncode", "inputs": {"text": positive}},
|
|
"2": {"class_type": "CLIPTextEncode", "inputs": {"text": negative}},
|
|
"3": {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["1", 0], "negative": ["2", 0]},
|
|
},
|
|
}
|
|
|
|
|
|
def png(path, **metadata):
|
|
info = PngImagePlugin.PngInfo()
|
|
for key, value in metadata.items():
|
|
info.add_text(key, value if isinstance(value, str) else json.dumps(value))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with Image.new("RGB", (4, 4)) as image:
|
|
image.save(path, pnginfo=info)
|
|
return path
|
|
|
|
|
|
class TestPromptInspector(ImageAPITestCase):
|
|
async def inspect(self, path="image.png", root="0"):
|
|
params = {} if root is None else {"root": root}
|
|
response = await self.client.get(
|
|
f"/prompt_manager/images/inspect/{path}", params=params
|
|
)
|
|
return response, await response.json()
|
|
|
|
def side(self, text, source="graph-inferred"):
|
|
return {"status": "resolved", "text": text, "source": source}
|
|
|
|
async def test_unindexed_graph_pair_is_read_without_indexing_or_decoding(self):
|
|
path = png(self.output_dir / "image.png", prompt=graph(negative=""))
|
|
before = path.read_bytes()
|
|
with self.api.db.model.get_connection() as connection:
|
|
rows = connection.execute("SELECT total_changes()").fetchone()[0]
|
|
with patch.object(Image.Image, "load", side_effect=AssertionError("decode")):
|
|
response, result = await self.inspect()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(
|
|
result["image"],
|
|
{"root_index": 0, "path": "image.png", "filename": "image.png"},
|
|
)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
self.assertEqual(result["prompts"]["negative"], self.side(""))
|
|
self.assertEqual(path.read_bytes(), before)
|
|
with self.api.db.model.get_connection() as connection:
|
|
self.assertEqual(
|
|
connection.execute("SELECT total_changes()").fetchone()[0], rows
|
|
)
|
|
self.assertEqual(
|
|
connection.execute("SELECT COUNT(*) FROM prompts").fetchone()[0], 0
|
|
)
|
|
self.assertNotIn(str(self.comfy_dir), json.dumps(result))
|
|
|
|
async def test_root_identity_collisions_are_not_filename_fallbacks(self):
|
|
other = self.comfy_dir / "input"
|
|
other.mkdir()
|
|
first = png(self.output_dir / "image.png", prompt=graph("output"))
|
|
png(other / "image.png", prompt=graph("input"))
|
|
self._link_image(first, "Saved output")
|
|
self.api._get_all_output_dirs = lambda: [self.output_dir, other]
|
|
_, result = await self.inspect(root="1")
|
|
self.assertEqual(result["image"]["root_index"], 1)
|
|
self.assertEqual(result["prompts"]["positive"]["text"], "input")
|
|
make_png(other / "plain.png")
|
|
_, result = await self.inspect("plain.png", root="1")
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "absent")
|
|
|
|
async def test_requires_valid_root_safe_relative_path_existing_media(self):
|
|
make_png(self.output_dir / "image.png")
|
|
for root in (None, "-1", "1", "x", "0.0"):
|
|
response, result = await self.inspect(root=root)
|
|
self.assertEqual(response.status, 400, root)
|
|
self.assertNotIn(str(self.comfy_dir), json.dumps(result))
|
|
for path in (
|
|
"%2Fetc%2Fpasswd.png",
|
|
"..%2Fsecret.png",
|
|
"%5Csecret.png",
|
|
"bad.txt",
|
|
"null%00.png",
|
|
"C:%5Csecret.png",
|
|
):
|
|
response, result = await self.inspect(path)
|
|
self.assertEqual(response.status, 403, path)
|
|
response, result = await self.inspect("missing.png")
|
|
self.assertEqual(response.status, 404)
|
|
outside = png(self.comfy_dir / "outside.png", prompt=graph("private"))
|
|
self._make_symlink(outside, self.output_dir / "escape.png")
|
|
response, result = await self.inspect("escape.png")
|
|
self.assertEqual(response.status, 403)
|
|
self.assertNotIn("private", json.dumps(result))
|
|
|
|
async def test_missing_negative_is_distinct_from_explicit_empty(self):
|
|
source = graph()
|
|
del source["3"]["inputs"]["negative"]
|
|
png(self.output_dir / "image.png", prompt=source)
|
|
_, result = await self.inspect()
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
self.assertEqual(result["prompts"]["negative"]["status"], "absent")
|
|
self.assertIsNone(result["prompts"]["negative"]["text"])
|
|
|
|
async def test_graph_ambiguity_and_unknown_semantics_do_not_use_saved_fallback(
|
|
self,
|
|
):
|
|
path = self.output_dir / "image.png"
|
|
self._link_image(png(path), "Legacy saved")
|
|
for expected in ("ambiguous", "unresolved"):
|
|
source = graph()
|
|
if expected == "ambiguous":
|
|
source["4"] = source["3"]
|
|
else:
|
|
source["1"]["class_type"] = "UnknownEncoder"
|
|
png(
|
|
path,
|
|
prompt=source,
|
|
parameters=(
|
|
"Fallback\nNegative prompt: noise\n" "Steps: 20, Sampler: Euler"
|
|
),
|
|
)
|
|
_, result = await self.inspect()
|
|
for side in result["prompts"].values():
|
|
self.assertEqual(side["status"], expected)
|
|
self.assertIsNone(side["text"])
|
|
|
|
async def test_parameters_and_partial_resolution_are_conservative(self):
|
|
path = self.output_dir / "image.png"
|
|
png(path, parameters="A mountain\nNegative prompt: \nSteps: 20, Sampler: Euler")
|
|
_, result = await self.inspect()
|
|
self.assertEqual(
|
|
result["prompts"]["positive"],
|
|
self.side("A mountain", "embedded-parameters"),
|
|
)
|
|
self.assertEqual(
|
|
result["prompts"]["negative"], self.side("", "embedded-parameters")
|
|
)
|
|
source = graph()
|
|
source["2"]["class_type"] = "UnknownEncoder"
|
|
png(path, prompt=source)
|
|
_, result = await self.inspect()
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
self.assertEqual(result["prompts"]["negative"]["status"], "unresolved")
|
|
|
|
async def test_execution_evidence_wins_over_graph_and_saved_recipe(self):
|
|
source = {
|
|
"1": {
|
|
"class_type": "PromptManagerPair",
|
|
"inputs": {"text": "portrait", "negative_text": "blurry"},
|
|
},
|
|
"3": {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["1", 0], "negative": ["1", 1]},
|
|
},
|
|
}
|
|
evidence = {}
|
|
record_pair_execution(
|
|
evidence,
|
|
"1",
|
|
text="portrait",
|
|
negative_text="blurry",
|
|
positive_text="portrait LoRA trigger",
|
|
negative_effective_text="blurry",
|
|
)
|
|
path = png(self.output_dir / "image.png", prompt=source, **evidence)
|
|
self._link_image(path, "Saved recipe")
|
|
_, result = await self.inspect()
|
|
self.assertEqual(
|
|
result["prompts"]["positive"],
|
|
self.side("portrait LoRA trigger", "execution"),
|
|
)
|
|
self.assertEqual(
|
|
result["prompts"]["negative"], self.side("blurry", "execution")
|
|
)
|
|
png(path, prompt=source, **{EXECUTION_KEY: "broken"})
|
|
_, result = await self.inspect()
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
|
|
async def test_saved_associations_remain_unverified_without_file_evidence(self):
|
|
path = make_png(self.output_dir / "image.png")
|
|
self._link_image(path, "Saved positive")
|
|
_, result = await self.inspect()
|
|
self.assertEqual(
|
|
result["prompts"]["positive"],
|
|
{
|
|
"status": "unresolved",
|
|
"text": None,
|
|
"source": "saved-association",
|
|
"message": (
|
|
"A saved library association exists, but it cannot be "
|
|
"verified against this file."
|
|
),
|
|
},
|
|
)
|
|
self.assertEqual(result["prompts"]["negative"]["status"], "absent")
|
|
other = make_png(self.output_dir / "paired.png")
|
|
metadata = {"prompt": graph("Effective positive", "Recorded negative")}
|
|
resolution = resolve_prompt_pair(metadata)
|
|
ingest_resolved_pair(self.api.db, str(other), metadata, resolution)
|
|
_, result = await self.inspect("paired.png")
|
|
self.assertEqual(
|
|
result["prompts"]["positive"],
|
|
{
|
|
"status": "unresolved",
|
|
"text": None,
|
|
"source": "saved-association",
|
|
"message": (
|
|
"A saved library association exists, but it cannot be "
|
|
"verified against this file."
|
|
),
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
result["prompts"]["negative"],
|
|
{
|
|
"status": "unresolved",
|
|
"text": None,
|
|
"source": "saved-association",
|
|
"message": (
|
|
"A saved library association exists, but it cannot be "
|
|
"verified against this file."
|
|
),
|
|
},
|
|
)
|
|
|
|
async def test_corrupt_malformed_and_no_metadata_do_not_invent_text(self):
|
|
path = self.output_dir / "image.png"
|
|
for content, expected, message in (
|
|
(b"broken", "unresolved", "Metadata in this file could not be read."),
|
|
(None, "absent", None),
|
|
("bad-json", "unresolved", "Prompt metadata in this file is malformed."),
|
|
):
|
|
if isinstance(content, bytes):
|
|
path.write_bytes(content)
|
|
else:
|
|
png(path, **({"prompt": content} if content else {}))
|
|
_, result = await self.inspect()
|
|
self.assertEqual(result["prompts"]["positive"]["status"], expected)
|
|
self.assertIsNone(result["prompts"]["positive"]["text"])
|
|
self.assertEqual(result["prompts"]["negative"].get("message"), message)
|
|
|
|
async def test_compressed_international_and_post_pixel_metadata(self):
|
|
import struct
|
|
import zlib
|
|
|
|
path = self.output_dir / "image.png"
|
|
for kind in (
|
|
"compressed",
|
|
"international",
|
|
"compressed-international",
|
|
"post-idat",
|
|
):
|
|
info = PngImagePlugin.PngInfo()
|
|
value = json.dumps(graph("猫"), ensure_ascii=False)
|
|
if kind == "compressed":
|
|
info.add_text("prompt", json.dumps(graph()), zip=True)
|
|
elif kind in ("international", "compressed-international"):
|
|
info.add_itxt(
|
|
"prompt",
|
|
value,
|
|
lang="en",
|
|
tkey="Prompt",
|
|
zip=kind.startswith("compressed"),
|
|
)
|
|
else:
|
|
payload = b"prompt\0" + json.dumps(graph()).encode()
|
|
info.add_text("unrelated", "header text")
|
|
with Image.new("RGB", (4, 4)) as image:
|
|
image.save(path, pnginfo=info)
|
|
if kind == "post-idat":
|
|
content = path.read_bytes()
|
|
chunk = (
|
|
struct.pack(">I", len(payload))
|
|
+ b"tEXt"
|
|
+ payload
|
|
+ struct.pack(">I", zlib.crc32(b"tEXt" + payload))
|
|
)
|
|
path.write_bytes(content[:-12] + chunk + content[-12:])
|
|
with patch.object(
|
|
Image.Image, "load", side_effect=AssertionError("pixels")
|
|
):
|
|
_, result = await self.inspect()
|
|
self.assertEqual(
|
|
result["prompts"]["positive"]["text"],
|
|
"猫" if "international" in kind else "portrait",
|
|
)
|
|
|
|
async def test_corrupt_truncated_duplicate_and_expanding_metadata_fail_closed(self):
|
|
import struct
|
|
import zlib
|
|
from utils.prompt_inspector import MAX_METADATA_BYTES
|
|
|
|
path = self.output_dir / "image.png"
|
|
base = png(path, prompt=graph()).read_bytes()
|
|
cases = [base[:-1], base[:30]]
|
|
index = base.index(b"tEXt")
|
|
broken = bytearray(base)
|
|
# Corrupt the retained record's body: its CRC is checked. (A corrupt
|
|
# keyword names an unretained chunk, which is skipped unchecked.)
|
|
broken[index + 4 + len(b"prompt\0") + 1] ^= 1
|
|
cases.append(bytes(broken))
|
|
for kind, payload in (
|
|
(b"tEXt", b"prompt\0{}"),
|
|
(b"zTXt", b"prompt\0\0" + zlib.compress(b"x" * (MAX_METADATA_BYTES + 1))),
|
|
(b"tEXt", b"prompt\0" + b"x" * (MAX_METADATA_BYTES + 1)),
|
|
(b"tEXt", b"bad-key-without-separator"),
|
|
(b"zTXt", b"prompt\0\1bad-compression"),
|
|
(b"iTXt", b"prompt\0\2\0\0\0text"),
|
|
(b"iTXt", b"prompt\0\1\1\0\0text"),
|
|
(b"iTXt", b"prompt\0\0\0\0\0\xff"),
|
|
):
|
|
chunk = (
|
|
struct.pack(">I", len(payload))
|
|
+ kind
|
|
+ payload
|
|
+ struct.pack(">I", zlib.crc32(kind + payload))
|
|
)
|
|
cases.append(base[:-12] + chunk + base[-12:])
|
|
for index, content in enumerate(cases):
|
|
path.write_bytes(content)
|
|
_, result = await self.inspect()
|
|
self.assertEqual(
|
|
result["prompts"]["positive"]["status"], "unresolved", index
|
|
)
|
|
self.assertIsNone(result["prompts"]["positive"]["text"])
|
|
|
|
async def test_jpeg_header_only_no_metadata_and_read_failure_are_safe(self):
|
|
path = self.output_dir / "image.jpg"
|
|
with Image.new("RGB", (4, 4)) as image:
|
|
image.save(path)
|
|
with patch.object(Image.Image, "load", side_effect=AssertionError("pixels")):
|
|
response, result = await self.inspect("image.jpg")
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "absent")
|
|
with patch(
|
|
"py.api.inspection_service.read_inspection_file",
|
|
side_effect=RuntimeError(str(self.comfy_dir)),
|
|
):
|
|
response, result = await self.inspect("image.jpg")
|
|
self.assertEqual(response.status, 500)
|
|
self.assertNotIn(str(self.comfy_dir), json.dumps(result))
|
|
|
|
async def test_invalid_json_depth_and_saved_filename_collision_do_not_escape(self):
|
|
path = png(self.output_dir / "image.png", prompt="[" * 2000)
|
|
_, result = await self.inspect()
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
self._link_image(path, "Saved first")
|
|
other = self.comfy_dir / "input"
|
|
other.mkdir()
|
|
make_png(other / "image.png")
|
|
self.api._get_all_output_dirs = lambda: [self.output_dir, other]
|
|
with patch.object(
|
|
self.api.db,
|
|
"get_image_prompt_info",
|
|
wraps=self.api.db.get_image_prompt_info,
|
|
) as lookup:
|
|
_, result = await self.inspect(root="1")
|
|
lookup.assert_called_once_with(
|
|
str(other / "image.png"), False, report_ambiguity=True
|
|
)
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "absent")
|
|
|
|
async def test_video_inspection_uses_bounded_container_evidence(self):
|
|
import io
|
|
from unittest.mock import MagicMock
|
|
from utils.prompt_inspector import MAX_METADATA_BYTES
|
|
|
|
(self.output_dir / "clip.mp4").write_bytes(b"video fixture")
|
|
metadata = {"prompt": graph("Video portrait", "")}
|
|
payload = json.dumps(
|
|
{"format": {"tags": {"comment": json.dumps(metadata)}}}
|
|
).encode()
|
|
process = MagicMock()
|
|
process.stdout = io.BytesIO(payload)
|
|
process.poll.return_value = 0
|
|
process.returncode = 0
|
|
with (
|
|
patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
|
|
patch(
|
|
"utils.prompt_inspector.subprocess.Popen", return_value=process
|
|
) as spawn,
|
|
):
|
|
response, result = await self.inspect("clip.mp4")
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("Video portrait"))
|
|
self.assertEqual(result["prompts"]["negative"], self.side(""))
|
|
self.assertTrue(process.stdout.closed)
|
|
process.wait.assert_called()
|
|
self.assertNotIn("shell", spawn.call_args.kwargs)
|
|
self.assertIn(str(self.output_dir / "clip.mp4"), spawn.call_args.args[0])
|
|
process.stdout = io.BytesIO(b"x" * (MAX_METADATA_BYTES + 1))
|
|
with (
|
|
patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
|
|
patch("utils.prompt_inspector.subprocess.Popen", return_value=process),
|
|
):
|
|
_, result = await self.inspect("clip.mp4")
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
self.assertTrue(process.stdout.closed)
|
|
|
|
async def test_video_missing_probe_and_timeout_preserve_uncertainty(self):
|
|
import io
|
|
from unittest.mock import MagicMock
|
|
|
|
(self.output_dir / "clip.webm").write_bytes(b"fixture")
|
|
with patch("utils.prompt_inspector.ffprobe_path", return_value=None):
|
|
_, result = await self.inspect("clip.webm")
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
process = MagicMock()
|
|
process.stdout = io.BytesIO(b"")
|
|
process.poll.return_value = None
|
|
process.returncode = -9
|
|
with (
|
|
patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
|
|
patch("utils.prompt_inspector.subprocess.Popen", return_value=process),
|
|
):
|
|
_, result = await self.inspect("clip.webm")
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
process.kill.assert_called_once()
|
|
process.wait.assert_called()
|
|
self.assertTrue(process.stdout.closed)
|
|
|
|
async def test_video_description_execution_evidence_and_invalid_envelope(self):
|
|
import io
|
|
from unittest.mock import MagicMock
|
|
|
|
path = self.output_dir / "clip.mp4"
|
|
path.write_bytes(b"fixture")
|
|
source = {
|
|
"1": {
|
|
"class_type": "PromptManagerPair",
|
|
"inputs": {"text": "portrait", "negative_text": "bad"},
|
|
},
|
|
"3": {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["1", 0], "negative": ["1", 1]},
|
|
},
|
|
}
|
|
evidence = {}
|
|
record_pair_execution(
|
|
evidence,
|
|
"1",
|
|
text="portrait",
|
|
negative_text="bad",
|
|
positive_text="portrait trigger",
|
|
negative_effective_text="bad",
|
|
)
|
|
for value in (
|
|
{
|
|
"format": {
|
|
"tags": {
|
|
"comment": "not-json",
|
|
"description": json.dumps({"prompt": source, **evidence}),
|
|
}
|
|
}
|
|
},
|
|
[],
|
|
{"format": {"tags": []}},
|
|
{"format": {"tags": {}}},
|
|
{
|
|
"format": {
|
|
"tags": {
|
|
"comment": json.dumps(
|
|
{"prompt": source, EXECUTION_KEY: "bad-json"}
|
|
)
|
|
}
|
|
}
|
|
},
|
|
):
|
|
process = MagicMock()
|
|
process.stdout = io.BytesIO(json.dumps(value).encode())
|
|
process.poll.return_value = 0
|
|
process.returncode = 0
|
|
with (
|
|
patch(
|
|
"utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"
|
|
),
|
|
patch("utils.prompt_inspector.subprocess.Popen", return_value=process),
|
|
):
|
|
_, result = await self.inspect("clip.mp4")
|
|
positive = result["prompts"]["positive"]
|
|
if (
|
|
isinstance(value, list)
|
|
or not isinstance(value["format"]["tags"], dict)
|
|
or "bad-json" in json.dumps(value)
|
|
):
|
|
self.assertEqual(positive["status"], "unresolved")
|
|
elif not value["format"]["tags"]:
|
|
self.assertEqual(positive["status"], "absent")
|
|
else:
|
|
self.assertEqual(positive, self.side("portrait trigger", "execution"))
|
|
|
|
async def test_large_sparse_pixels_are_skipped_and_chunk_count_is_bounded(self):
|
|
import struct
|
|
import zlib
|
|
from utils.prompt_inspector import read_inspection_metadata, MetadataReadError
|
|
|
|
path = self.output_dir / "large.png"
|
|
header = struct.pack(">IIBBBBB", 1024, 1024, 8, 2, 0, 0, 0)
|
|
|
|
def chunk(kind, data):
|
|
return (
|
|
struct.pack(">I", len(data))
|
|
+ kind
|
|
+ data
|
|
+ struct.pack(">I", zlib.crc32(kind + data))
|
|
)
|
|
|
|
with path.open("wb") as stream:
|
|
stream.write(b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", header))
|
|
stream.write(struct.pack(">I4s", 64 * 1024 * 1024, b"IDAT"))
|
|
stream.seek(64 * 1024 * 1024, 1)
|
|
stream.write(b"\0\0\0\0")
|
|
stream.write(chunk(b"tEXt", b"prompt\0" + json.dumps(graph()).encode()))
|
|
stream.write(chunk(b"IEND", b""))
|
|
original_open = type(path).open
|
|
reads = []
|
|
|
|
def tracked_open(item, *args, **kwargs):
|
|
stream = original_open(item, *args, **kwargs)
|
|
original_read = stream.read
|
|
|
|
def read(size=-1):
|
|
value = original_read(size)
|
|
reads.append(len(value))
|
|
return value
|
|
|
|
stream.read = read
|
|
return stream
|
|
|
|
with patch.object(type(path), "open", tracked_open):
|
|
metadata = read_inspection_metadata(path)
|
|
self.assertEqual(json.loads(metadata["prompt"]), graph())
|
|
self.assertLess(sum(reads), 4096)
|
|
with patch("utils.prompt_inspector.MAX_PNG_CHUNKS", 1):
|
|
with self.assertRaises(MetadataReadError):
|
|
read_inspection_metadata(path)
|
|
|
|
async def test_large_jpeg_uses_bounded_header_without_rejecting_valid_file(self):
|
|
from utils.prompt_inspector import MAX_HEADER_BYTES
|
|
|
|
path = self.output_dir / "large.jpg"
|
|
with Image.new("RGB", (4, 4)) as image:
|
|
image.save(path)
|
|
with path.open("ab") as stream:
|
|
stream.write(b"x" * (MAX_HEADER_BYTES + 1))
|
|
self._link_image(path, "Saved large JPEG")
|
|
with patch.object(Image.Image, "load", side_effect=AssertionError("pixels")):
|
|
_, result = await self.inspect("large.jpg")
|
|
self.assertEqual(
|
|
result["prompts"]["positive"],
|
|
{
|
|
"status": "unresolved",
|
|
"text": None,
|
|
"source": "saved-association",
|
|
"message": (
|
|
"A saved library association exists, but it cannot be "
|
|
"verified against this file."
|
|
),
|
|
},
|
|
)
|
|
|
|
async def test_invalid_or_execution_only_video_evidence_never_uses_saved_fallback(
|
|
self,
|
|
):
|
|
import io
|
|
from unittest.mock import MagicMock
|
|
|
|
path = self.output_dir / "clip.mp4"
|
|
path.write_bytes(b"fixture")
|
|
self._link_image(path, "Saved recipe must not replace unknown evidence")
|
|
for envelope in (
|
|
{"prompt": None},
|
|
{"parameters": None},
|
|
{EXECUTION_KEY: {"version": 1, "nodes": {}}},
|
|
):
|
|
process = MagicMock()
|
|
process.stdout = io.BytesIO(
|
|
json.dumps(
|
|
{
|
|
"format": {
|
|
"tags": {
|
|
"comment": json.dumps(envelope),
|
|
"description": json.dumps(
|
|
{"prompt": graph("Other comment")}
|
|
),
|
|
}
|
|
}
|
|
}
|
|
).encode()
|
|
)
|
|
process.poll.return_value = 0
|
|
process.returncode = 0
|
|
with (
|
|
patch(
|
|
"utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"
|
|
),
|
|
patch("utils.prompt_inspector.subprocess.Popen", return_value=process),
|
|
):
|
|
_, result = await self.inspect("clip.mp4")
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
self.assertIsNone(result["prompts"]["positive"]["text"])
|
|
|
|
async def test_same_path_same_size_replacement_does_not_resolve_stale_saved_prompts(
|
|
self,
|
|
):
|
|
import io
|
|
|
|
metadata = {"prompt": graph("Original portrait", "Original negative")}
|
|
path = png(self.output_dir / "image.png", **metadata)
|
|
ingest_resolved_pair(
|
|
self.api.db, str(path), metadata, resolve_prompt_pair(metadata)
|
|
)
|
|
_, original = await self.inspect()
|
|
self.assertEqual(
|
|
original["prompts"]["positive"], self.side("Original portrait")
|
|
)
|
|
size = path.stat().st_size
|
|
replacement = io.BytesIO()
|
|
with Image.new("RGB", (4, 4), (0, 0, 255)) as image:
|
|
image.save(replacement, "PNG")
|
|
content = replacement.getvalue()
|
|
self.assertLessEqual(len(content), size)
|
|
path.write_bytes(content.ljust(size, b"\0"))
|
|
self.assertEqual(path.stat().st_size, size)
|
|
_, replaced = await self.inspect()
|
|
for record in replaced["prompts"].values():
|
|
self.assertEqual(record["status"], "unresolved")
|
|
self.assertIsNone(record["text"])
|
|
self.assertEqual(record["source"], "saved-association")
|
|
self.assertIn("cannot be verified", record["message"])
|
|
self.assertNotIn("Original portrait", json.dumps(replaced))
|
|
|
|
async def test_full_inspection_envelope_preserves_existing_roles_and_identity(self):
|
|
from pathlib import Path
|
|
|
|
source = graph(negative="")
|
|
source["3"]["inputs"].update(seed=9007199254740993, cfg=6.5, steps=28)
|
|
raw = json.dumps(source)
|
|
workflow = '{"nodes":[],"seed":9007199254740993}'
|
|
path = png(self.output_dir / "image.png", prompt=raw, workflow=workflow)
|
|
self._link_image(path, "Saved text must not replace file roles")
|
|
original_open = Path.open
|
|
reads = []
|
|
|
|
def opened(candidate, *args, **kwargs):
|
|
if candidate == path:
|
|
reads.append(candidate)
|
|
return original_open(candidate, *args, **kwargs)
|
|
|
|
with (
|
|
patch.object(Path, "open", opened),
|
|
patch.object(Image.Image, "load", side_effect=AssertionError("pixels")),
|
|
patch.object(
|
|
self.api.db,
|
|
"get_image_prompt_info",
|
|
wraps=self.api.db.get_image_prompt_info,
|
|
) as lookup,
|
|
):
|
|
response, result = await self.inspect()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result.get("inspection_version"), 1)
|
|
self.assertEqual(
|
|
result["image"],
|
|
{
|
|
"root_index": 0,
|
|
"path": "image.png",
|
|
"filename": "image.png",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
result["prompts"],
|
|
{
|
|
"positive": self.side("portrait"),
|
|
"negative": self.side(""),
|
|
},
|
|
)
|
|
self.assertRegex(result["root_scope"], r"^[0-9a-f]{64}$")
|
|
self.assertEqual(
|
|
result["metadata"]["raw"], {"prompt": raw, "workflow": workflow}
|
|
)
|
|
self.assertEqual(
|
|
result["metadata"]["record_formats"],
|
|
{
|
|
"prompt": "original-text",
|
|
"workflow": "original-text",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
result["metadata"]["prompt_graph"]["value"]["3"]["inputs"]["seed"],
|
|
"9007199254740993",
|
|
)
|
|
self.assertEqual(result["file"]["format"], "PNG")
|
|
self.assertEqual((result["file"]["width"], result["file"]["height"]), (4, 4))
|
|
self.assertEqual(result["file"]["observation"], "current-file")
|
|
self.assertEqual(result["resolution"]["sampler_id"], "3")
|
|
self.assertEqual(result["library"]["status"], "linked")
|
|
self.assertEqual(reads, [path])
|
|
lookup.assert_called_once_with(str(path), False, report_ambiguity=True)
|
|
self.assertLessEqual(len(await response.read()), 8 * 1024 * 1024)
|
|
self.assertNotIn(str(self.comfy_dir), json.dumps(result))
|
|
|
|
async def test_workflow_has_its_own_budget_and_never_hides_prompts(self):
|
|
from utils.prompt_inspector import ROLE_METADATA_BYTES, WORKFLOW_METADATA_BYTES
|
|
|
|
path = self.output_dir / "image.png"
|
|
# Over the role budget, within the workflow budget: fully readable.
|
|
workflow = "x" * ROLE_METADATA_BYTES
|
|
png(path, prompt=json.dumps(graph()), workflow=workflow)
|
|
response, result = await self.inspect()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result["metadata"]["status"], "available")
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
self.assertEqual(result["metadata"]["raw"]["workflow"], workflow)
|
|
self.assertEqual(result["metadata"]["workflow_graph"]["status"], "malformed")
|
|
# Over the workflow budget: only the workflow is limited.
|
|
png(
|
|
path,
|
|
prompt=json.dumps(graph()),
|
|
workflow="x" * (WORKFLOW_METADATA_BYTES + 1),
|
|
)
|
|
response, result = await self.inspect()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result["metadata"]["status"], "available")
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
self.assertEqual(result["prompts"]["negative"], self.side("blurry"))
|
|
self.assertEqual(
|
|
result["metadata"]["record_formats"],
|
|
{"prompt": "original-text", "workflow": "limited"},
|
|
)
|
|
self.assertEqual(set(result["metadata"]["raw"]), {"prompt"})
|
|
self.assertEqual(
|
|
result["metadata"]["workflow_graph"], {"status": "limited", "value": None}
|
|
)
|
|
self.assertEqual(result["metadata"]["prompt_graph"]["status"], "parsed")
|
|
|
|
def parity_fixture(self, name):
|
|
path = self.output_dir / f"{name}.png"
|
|
parity.write_png(path, parity.build(name))
|
|
return path
|
|
|
|
async def test_inspect_route_resolves_506_node_graph(self):
|
|
# f03: 506 executable nodes. The default (ingestion) profile's 500-node
|
|
# gate reports "Resolution limit reached"; the inspection profile bounds
|
|
# work instead of graph size, so the route resolves the pair.
|
|
self.parity_fixture("f03")
|
|
response, result = await self.inspect("f03.png")
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(
|
|
result["prompts"],
|
|
{"positive": self.side(parity.POS), "negative": self.side(parity.NEG)},
|
|
)
|
|
self.assertEqual(result["resolution"]["state"], "resolved")
|
|
self.assertIsNone(result["resolution"]["reason_code"])
|
|
self.assertIsNone(result["resolution"]["node"])
|
|
self.assertEqual(result["metadata"]["prompt_graph"]["status"], "parsed")
|
|
|
|
async def test_inspect_route_reports_no_sampler_instead_of_absent(self):
|
|
# f17 with a sampler class the inspector does not know: a prompt graph
|
|
# is a record, never "not recorded".
|
|
built = parity.build("f17")
|
|
built["prompt"]["13"]["class_type"] = "FutureSampler"
|
|
parity.write_png(self.output_dir / "f17.png", built)
|
|
response, result = await self.inspect("f17.png")
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(
|
|
{role: side["status"] for role, side in result["prompts"].items()},
|
|
{"positive": "unresolved", "negative": "unresolved"},
|
|
)
|
|
self.assertEqual(result["resolution"]["state"], "unresolved")
|
|
self.assertEqual(result["resolution"]["reason_code"], "no-sampler")
|
|
self.assertEqual(
|
|
result["resolution"]["reason"],
|
|
"No supported sampler pair found (saw FutureSampler)",
|
|
)
|
|
self.assertEqual(
|
|
result["prompts"]["positive"]["message"], "No supported sampler was found."
|
|
)
|
|
|
|
async def test_inspect_route_exposes_reason_code_and_blocking_node(self):
|
|
# f06a's encoder is supported since Task 2.4; an unknown encoder is not.
|
|
built = parity.build("f06a")
|
|
built["prompt"]["6"]["class_type"] = "FutureTextEncode"
|
|
parity.write_png(self.output_dir / "f06a.png", built)
|
|
_, result = await self.inspect("f06a.png")
|
|
resolution = result["resolution"]
|
|
self.assertEqual(resolution["state"], "unresolved")
|
|
self.assertEqual(resolution["reason_code"], "unsupported-node")
|
|
self.assertEqual(
|
|
resolution["node"], {"id": "6", "class_type": "FutureTextEncode"}
|
|
)
|
|
self.assertEqual(
|
|
{role: side.get("message") for role, side in result["prompts"].items()},
|
|
dict.fromkeys(
|
|
("positive", "negative"),
|
|
"Prompt uses a node the inspector does not support "
|
|
"(FutureTextEncode).",
|
|
),
|
|
)
|
|
png(self.output_dir / "image.png", prompt="{broken")
|
|
_, result = await self.inspect()
|
|
self.assertEqual(result["resolution"]["reason_code"], "malformed")
|
|
make_png(self.output_dir / "image.png")
|
|
_, result = await self.inspect()
|
|
self.assertEqual(result["resolution"]["state"], "absent")
|
|
self.assertIsNone(result["resolution"]["reason_code"])
|
|
self.assertIsNone(result["resolution"]["node"])
|
|
|
|
async def test_workflow_only_keeps_existing_role_uncertainty(self):
|
|
path = png(self.output_dir / "image.png", workflow={"nodes": []})
|
|
self._link_image(path, "Saved uncertainty")
|
|
_, result = await self.inspect()
|
|
self.assertEqual(result["prompts"]["positive"]["source"], "saved-association")
|
|
self.assertEqual(
|
|
result.get("metadata", {}).get("workflow_graph", {}).get("status"), "parsed"
|
|
)
|
|
self.assertIsNone(result["prompts"]["positive"]["text"])
|
|
self.assertEqual(result["prompts"]["negative"]["status"], "absent")
|
|
|
|
async def test_library_failure_does_not_erase_file_evidence(self):
|
|
png(self.output_dir / "image.png", prompt=graph("File portrait"))
|
|
with patch.object(
|
|
self.api.db,
|
|
"get_image_prompt_info",
|
|
side_effect=OSError("private DB detail"),
|
|
):
|
|
response, result = await self.inspect()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("File portrait"))
|
|
self.assertEqual(result.get("library", {}).get("status"), "unavailable")
|
|
self.assertEqual(result["metadata"]["prompt_graph"]["status"], "parsed")
|
|
self.assertNotIn("private DB detail", json.dumps(result))
|
|
|
|
async def test_file_change_is_retryable_without_saved_text_fallback(self):
|
|
from pathlib import Path
|
|
|
|
path = png(self.output_dir / "image.png", prompt=graph())
|
|
self._link_image(path, "Saved recipe")
|
|
original_open = Path.open
|
|
|
|
def opened(candidate, *args, **kwargs):
|
|
stream = original_open(candidate, *args, **kwargs)
|
|
if candidate == path and args and args[0] == "rb":
|
|
with original_open(path, "ab") as output:
|
|
output.write(b"changed")
|
|
return stream
|
|
|
|
with patch.object(Path, "open", opened):
|
|
response, result = await self.inspect()
|
|
self.assertEqual(response.status, 409)
|
|
self.assertFalse(result["success"])
|
|
self.assertNotIn("code", result)
|
|
self.assertNotIn("Saved recipe", json.dumps(result))
|
|
self.assertEqual(response.headers.get("Cache-Control"), "no-store")
|
|
|
|
async def test_surrogate_and_ordinary_unicode_round_trip_without_replacement(self):
|
|
source = graph("猫 😀 \ud800", "\udfff")
|
|
raw = json.dumps(source)
|
|
png(
|
|
self.output_dir / "image.png",
|
|
prompt=raw,
|
|
workflow='{"nodes":[],"label":"\\ud800 猫 😀"}',
|
|
)
|
|
response, result = await self.inspect()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("猫 😀 \ud800"))
|
|
self.assertEqual(result["prompts"]["negative"], self.side("\udfff"))
|
|
self.assertEqual(result["metadata"]["raw"]["prompt"], raw)
|
|
body = await response.read()
|
|
self.assertIn("猫 😀".encode(), body)
|
|
self.assertIn(b"\\ud800", body)
|
|
self.assertNotIn("\ufffd".encode(), body)
|
|
self.assertLessEqual(len(body), 8 * 1024 * 1024)
|
|
|
|
async def video_inspect(self, comment=None, description=None):
|
|
import io
|
|
import threading
|
|
from unittest.mock import MagicMock
|
|
from utils.prompt_inspector import MAX_METADATA_BYTES
|
|
|
|
path = self.output_dir / "clip.mp4"
|
|
path.write_bytes(b"fixture")
|
|
tags = {
|
|
key: value
|
|
for key, value in (("comment", comment), ("description", description))
|
|
if value is not None
|
|
}
|
|
payload = json.dumps({"format": {"tags": tags}}).encode()
|
|
|
|
class ProbeOutput(io.BytesIO):
|
|
def read(self, size=-1):
|
|
self.requested_size = size
|
|
return super().read(size)
|
|
|
|
process = MagicMock()
|
|
process.stdout = ProbeOutput(payload)
|
|
process.poll.return_value = 0
|
|
process.returncode = 0
|
|
timers = []
|
|
timer_type = threading.Timer
|
|
|
|
def timer(*args):
|
|
instance = timer_type(*args)
|
|
timers.append(instance)
|
|
return instance
|
|
|
|
with (
|
|
patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
|
|
patch(
|
|
"utils.prompt_inspector.subprocess.Popen", return_value=process
|
|
) as spawn,
|
|
patch("utils.prompt_inspector.threading.Timer", side_effect=timer),
|
|
):
|
|
response, result = await self.inspect("clip.mp4")
|
|
self.assertEqual(spawn.call_count, 1)
|
|
self.assertTrue(process.stdout.closed)
|
|
self.assertEqual(process.stdout.requested_size, MAX_METADATA_BYTES + 1)
|
|
self.assertEqual(len(timers), 1)
|
|
self.assertTrue(timers[0].finished.is_set())
|
|
return response, result
|
|
|
|
async def test_video_role_description_wins_over_workflow_only_comment(self):
|
|
evidence = {}
|
|
source = graph("chosen", "")
|
|
# CLIP execution is unnecessary here: use the established pair-node carrier.
|
|
source["1"] = {
|
|
"class_type": "PromptManagerPair",
|
|
"inputs": {"text": "chosen", "negative_text": ""},
|
|
}
|
|
source["3"]["inputs"]["negative"] = ["1", 1]
|
|
record_pair_execution(
|
|
evidence,
|
|
"1",
|
|
text="chosen",
|
|
negative_text="",
|
|
positive_text="chosen trigger",
|
|
negative_effective_text="",
|
|
)
|
|
_, result = await self.video_inspect(
|
|
json.dumps({"workflow": {"nodes": [{"id": "ignored"}]}}),
|
|
json.dumps(
|
|
{
|
|
"prompt": source,
|
|
"workflow": {"nodes": [{"id": "chosen"}]},
|
|
**evidence,
|
|
}
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
result["prompts"]["positive"], self.side("chosen trigger", "execution")
|
|
)
|
|
self.assertEqual(result["prompts"]["negative"], self.side("", "execution"))
|
|
self.assertEqual(
|
|
result.get("metadata", {}).get("workflow_graph", {}).get("value"),
|
|
{"nodes": [{"id": "chosen"}]},
|
|
)
|
|
self.assertNotIn("ignored", json.dumps(result))
|
|
|
|
async def test_video_present_malformed_role_markers_block_other_carriers(self):
|
|
for marker in (
|
|
{"prompt": None},
|
|
{"prompt": "not-json"},
|
|
{"parameters": None},
|
|
{EXECUTION_KEY: "malformed"},
|
|
):
|
|
for comment, description in (
|
|
(
|
|
json.dumps({"workflow": {"nodes": [{"id": "ignored"}]}}),
|
|
json.dumps(marker),
|
|
),
|
|
(json.dumps(marker), json.dumps({"prompt": graph("must not win")})),
|
|
):
|
|
_, result = await self.video_inspect(comment, description)
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
self.assertIsNone(result["prompts"]["positive"]["text"])
|
|
self.assertEqual(
|
|
result.get("metadata", {}).get("workflow_graph", {}).get("status"),
|
|
"absent",
|
|
)
|
|
self.assertNotIn("must not win", json.dumps(result))
|
|
|
|
async def test_video_workflow_carrier_fallback_never_merges_tags(self):
|
|
path = self.output_dir / "clip.mp4"
|
|
path.write_bytes(b"fixture")
|
|
self._link_image(path, "Saved workflow association")
|
|
_, result = await self.video_inspect(
|
|
'{"workflow":{"nodes":[{"id":"first"}]}}',
|
|
'{"workflow":{"nodes":[{"id":"second"}]},"unrelated":"unselected"}',
|
|
)
|
|
self.assertEqual(result["prompts"]["positive"]["source"], "saved-association")
|
|
self.assertEqual(
|
|
result.get("metadata", {}).get("workflow_graph", {}).get("value"),
|
|
{"nodes": [{"id": "first"}]},
|
|
)
|
|
self.assertEqual(
|
|
result["metadata"]["raw"], {"workflow": '{"nodes": [{"id": "first"}]}'}
|
|
)
|
|
self.assertNotIn("second", json.dumps(result))
|
|
self.assertNotIn("unselected", json.dumps(result))
|
|
|
|
async def test_video_object_graph_decimal_and_unsafe_int_projection_is_exact(self):
|
|
raw = json.dumps(graph("video", ""))
|
|
raw = raw.replace(
|
|
'"positive": ["1", 0]',
|
|
'"seed":9007199254740993,"cfg":6.5000000000000000001,"positive": ["1", 0]',
|
|
)
|
|
for prompt in (raw, json.dumps(raw)):
|
|
carrier = (
|
|
'{"prompt":'
|
|
+ prompt
|
|
+ ',"workflow":{"nodes":[],"cfg":6.5000000000000000001,'
|
|
'"seed":9007199254740993}}'
|
|
)
|
|
_, result = await self.video_inspect(carrier)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("video"))
|
|
inputs = (
|
|
result.get("metadata", {})
|
|
.get("prompt_graph", {})
|
|
.get("value", {})
|
|
.get("3", {})
|
|
.get("inputs", {})
|
|
)
|
|
self.assertEqual(inputs.get("cfg"), "6.5000000000000000001")
|
|
self.assertEqual(inputs["seed"], "9007199254740993")
|
|
self.assertEqual(inputs["positive"], ["1", 0])
|
|
self.assertEqual(
|
|
result["metadata"]["workflow_graph"]["value"]["cfg"],
|
|
"6.5000000000000000001",
|
|
)
|
|
self.assertNotIn("video_carrier_text", result)
|
|
self.assertNotIn("carrier_text", json.dumps(result))
|
|
|
|
async def test_video_original_string_and_reserialized_object_formats_are_distinct(
|
|
self,
|
|
):
|
|
raw = json.dumps(graph())
|
|
for record, expected in (
|
|
(json.dumps(raw), "original-text"),
|
|
(raw, "reserialized-json"),
|
|
):
|
|
_, result = await self.video_inspect(
|
|
'{"prompt":' + record + ',"workflow":' + record + "}"
|
|
)
|
|
self.assertEqual(
|
|
result.get("metadata", {}).get("record_formats"),
|
|
{"prompt": expected, "workflow": expected},
|
|
)
|
|
self.assertEqual(result["metadata"]["raw"]["prompt"], raw)
|
|
|
|
async def test_video_null_role_marker_raises_before_successful_read(self):
|
|
for key in ("prompt", "parameters", EXECUTION_KEY):
|
|
_, result = await self.video_inspect(json.dumps({key: None}))
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
self.assertEqual(result.get("metadata", {}).get("status"), "unreadable")
|
|
|
|
async def test_video_null_workflow_keeps_valid_legacy_roles(self):
|
|
_, result = await self.video_inspect(
|
|
json.dumps({"prompt": graph(), "workflow": None})
|
|
)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
self.assertEqual(
|
|
result.get("metadata", {}).get("workflow_graph"),
|
|
{"status": "malformed", "value": None},
|
|
)
|
|
self.assertEqual(result["metadata"]["record_formats"]["workflow"], "invalid")
|
|
self.assertNotIn("workflow", result["metadata"]["raw"])
|
|
|
|
async def test_video_strict_projection_failure_preserves_legacy_roles(self):
|
|
source = json.dumps(graph())
|
|
for suffix, expected in (
|
|
(',"duplicate":1,"duplicate":2', "malformed"),
|
|
(',"depth":' + "[" * 64 + "0" + "]" * 64, "limited"),
|
|
):
|
|
_, result = await self.video_inspect(
|
|
'{"prompt":' + source + ',"workflow":{"nodes":[]}' + suffix + "}"
|
|
)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
self.assertEqual(
|
|
result.get("metadata", {}).get("prompt_graph", {}).get("status"),
|
|
expected,
|
|
)
|
|
self.assertIsNone(result["metadata"]["prompt_graph"]["value"])
|
|
self.assertEqual(result["metadata"]["workflow_graph"]["status"], expected)
|
|
|
|
async def test_video_carrier_nonfinite_outside_graph_inputs_is_tolerated(self):
|
|
# ComfyUI quirks: NaN in an unrelated carrier field, in ``is_changed`` and
|
|
# in ``_meta`` are validated and dropped; they never reach the projection.
|
|
source = graph()
|
|
source["3"]["is_changed"] = ["NAN_MARKER"]
|
|
source["3"]["_meta"] = {"title": "K", "weight": "NAN_MARKER"}
|
|
text = json.dumps(source).replace('"NAN_MARKER"', "NaN")
|
|
_, result = await self.video_inspect(
|
|
'{"prompt":' + text + ',"workflow":{"nodes":[]},"nonfinite":NaN}'
|
|
)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
prompt_graph = result["metadata"]["prompt_graph"]
|
|
self.assertEqual(prompt_graph["status"], "parsed")
|
|
self.assertEqual(set(prompt_graph["value"]["3"]), {"class_type", "inputs"})
|
|
self.assertEqual(
|
|
result["metadata"]["workflow_graph"],
|
|
{"status": "parsed", "value": {"nodes": []}},
|
|
)
|
|
|
|
async def test_video_numeric_role_text_never_becomes_sendable(self):
|
|
source = graph(positive=123)
|
|
_, result = await self.video_inspect(json.dumps({"prompt": source}))
|
|
self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
|
|
self.assertIsNone(result["prompts"]["positive"]["text"])
|
|
self.assertEqual(
|
|
result.get("metadata", {}).get("prompt_graph", {}).get("status"), "parsed"
|
|
)
|
|
|
|
async def scoped_descriptor(self, root_index=0):
|
|
response = await self.client.get("/prompt_manager/images/output?mode=workspace")
|
|
entries = (await response.json())["images"]
|
|
return next(entry for entry in entries if entry["root_index"] == root_index)
|
|
|
|
async def assert_scope_skips_access(self, scope, root="0", status=409):
|
|
from pathlib import Path
|
|
|
|
original_stat = Path.stat
|
|
|
|
def stat(candidate, *args, **kwargs):
|
|
if candidate.name == "image.png":
|
|
raise AssertionError("candidate stat before scope guard")
|
|
return original_stat(candidate, *args, **kwargs)
|
|
|
|
params = {"root_scope": scope}
|
|
if root is not None:
|
|
params["root"] = root
|
|
with (
|
|
patch.object(Path, "stat", stat),
|
|
patch(
|
|
"utils.prompt_inspector.subprocess.Popen",
|
|
side_effect=AssertionError("probe"),
|
|
),
|
|
patch.object(
|
|
self.api.db,
|
|
"get_image_prompt_info",
|
|
side_effect=AssertionError("saved lookup"),
|
|
),
|
|
patch.object(
|
|
self.api, "_file_response", side_effect=AssertionError("file response")
|
|
),
|
|
patch(
|
|
"py.api.inspection_service.read_inspection_file",
|
|
side_effect=AssertionError("metadata read"),
|
|
),
|
|
):
|
|
response = await self.client.get(
|
|
"/prompt_manager/images/inspect/image.png", params=params
|
|
)
|
|
self.assertEqual(response.status, status)
|
|
result = await response.json()
|
|
self.assertFalse(result["success"])
|
|
if status == 409:
|
|
self.assertEqual(result["code"], "source-changed")
|
|
self.assertEqual(response.headers.get("Cache-Control"), "no-store")
|
|
self.assertNotIn(str(self.comfy_dir), json.dumps(result))
|
|
|
|
async def test_root_scope_equal_label_reorder_or_replacement_never_reads_other_file(
|
|
self,
|
|
):
|
|
from py.config import GalleryConfig
|
|
|
|
roots = [self.comfy_dir / name / "output" for name in ("a", "b", "c")]
|
|
for index, root in enumerate(roots):
|
|
png(root / "image.png", prompt=graph(str(index)))
|
|
with patch.object(
|
|
GalleryConfig,
|
|
"path_anchors",
|
|
classmethod(lambda cls: [str(root.parent) for root in roots]),
|
|
):
|
|
self.api._get_all_output_dirs = lambda: roots[:2]
|
|
descriptor = await self.scoped_descriptor()
|
|
other = await self.scoped_descriptor(1)
|
|
self.assertEqual(descriptor["root_dir"], other["root_dir"])
|
|
for changed in ([roots[1], roots[0]], [roots[2], roots[1]]):
|
|
self.api._get_all_output_dirs = lambda: changed
|
|
await self.assert_scope_skips_access(
|
|
descriptor.get("root_scope", "0" * 64)
|
|
)
|
|
|
|
async def test_root_scope_symlink_retarget_requires_refresh(self):
|
|
first, second = self.comfy_dir / "a", self.comfy_dir / "b"
|
|
png(first / "image.png", prompt=graph("first"))
|
|
png(second / "image.png", prompt=graph("second"))
|
|
link = self.comfy_dir / "linked"
|
|
self._make_symlink(first, link)
|
|
self.api._get_all_output_dirs = lambda: [link]
|
|
descriptor = await self.scoped_descriptor()
|
|
link.unlink()
|
|
self._make_symlink(second, link)
|
|
await self.assert_scope_skips_access(descriptor.get("root_scope", "0" * 64))
|
|
|
|
async def test_root_scope_rotated_key_and_removed_index_skip_reader_and_db(self):
|
|
png(self.output_dir / "image.png", prompt=graph())
|
|
descriptor = await self.scoped_descriptor()
|
|
import py.api.images as images_module
|
|
|
|
with patch.object(images_module, "_WORKSPACE_CURSOR_SECRET", b"new-key"):
|
|
await self.assert_scope_skips_access(descriptor.get("root_scope", "0" * 64))
|
|
self.api._get_all_output_dirs = lambda: []
|
|
await self.assert_scope_skips_access(descriptor.get("root_scope", "0" * 64))
|
|
|
|
async def test_root_scope_malformed_request_skips_reader(self):
|
|
png(self.output_dir / "image.png", prompt=graph())
|
|
for token in ("", "z" * 64, "a" * 63, "a" * 65, "a" * 4097):
|
|
await self.assert_scope_skips_access(token, status=400)
|
|
for root in (None, "-1", "x", "0.0", "+0"):
|
|
await self.assert_scope_skips_access("a" * 64, root=root, status=400)
|
|
|
|
async def test_root_scope_echo_preserves_legacy_image_and_roles(self):
|
|
png(self.output_dir / "image.png", prompt=graph())
|
|
descriptor = await self.scoped_descriptor()
|
|
token = descriptor.get("root_scope", "0" * 64)
|
|
_, original = await self.inspect()
|
|
response = await self.client.get(
|
|
"/prompt_manager/images/inspect/image.png",
|
|
params={"root": "0", "root_scope": token},
|
|
)
|
|
result = await response.json()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result.get("root_scope"), token)
|
|
self.assertEqual(
|
|
result["image"],
|
|
{"root_index": 0, "path": "image.png", "filename": "image.png"},
|
|
)
|
|
self.assertEqual(result["prompts"], original["prompts"])
|
|
outside = png(self.comfy_dir / "outside.png", prompt=graph("private"))
|
|
self._make_symlink(outside, self.output_dir / "escape.png")
|
|
for path in ("..%2Foutside.png", "escape.png"):
|
|
denied = await self.client.get(
|
|
f"/prompt_manager/images/inspect/{path}",
|
|
params={"root": "0", "root_scope": token},
|
|
)
|
|
self.assertEqual(denied.status, 403)
|
|
|
|
async def test_root_scope_source_change_returns_refresh_instead_of_other_root(self):
|
|
png(self.output_dir / "image.png", prompt=graph("old source"))
|
|
other = self.comfy_dir / "input"
|
|
png(other / "image.png", prompt=graph("different source"))
|
|
self.api._get_all_output_dirs = lambda: [self.output_dir, other]
|
|
descriptor = await self.scoped_descriptor()
|
|
self.api._get_all_output_dirs = lambda: [other, self.output_dir]
|
|
response = await self.client.get(
|
|
"/prompt_manager/images/inspect/image.png",
|
|
params={"root": 0, "root_scope": descriptor.get("root_scope", "0" * 64)},
|
|
)
|
|
self.assertEqual(response.status, 409)
|
|
result = await response.json()
|
|
self.assertEqual(result["code"], "source-changed")
|
|
self.assertNotIn("different source", json.dumps(result))
|
|
|
|
async def test_video_unrepresentable_decimal_preserves_file_roles_and_raw(self):
|
|
carrier = json.dumps({"prompt": graph()}).replace(
|
|
'"positive": ["1", 0]',
|
|
'"cfg":1e1000000000000000000000000,"positive": ["1", 0]',
|
|
)
|
|
response, result = await self.video_inspect(carrier)
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result["prompts"]["positive"], self.side("portrait"))
|
|
self.assertEqual(
|
|
result["metadata"]["prompt_graph"], {"status": "malformed", "value": None}
|
|
)
|
|
self.assertEqual(
|
|
result["metadata"]["record_formats"]["prompt"], "reserialized-json"
|
|
)
|
|
self.assertIn("Infinity", result["metadata"]["raw"]["prompt"])
|
|
|
|
async def test_root_scope_guard_and_evidence_use_one_current_roots_snapshot(self):
|
|
png(self.output_dir / "image.png", prompt=graph())
|
|
descriptor = await self.scoped_descriptor()
|
|
with patch.object(
|
|
self.api, "_get_all_output_dirs", side_effect=[[self.output_dir]]
|
|
) as roots:
|
|
response = await self.client.get(
|
|
"/prompt_manager/images/inspect/image.png",
|
|
params={"root": 0, "root_scope": descriptor["root_scope"]},
|
|
)
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(
|
|
(await response.json())["root_scope"], descriptor["root_scope"]
|
|
)
|
|
roots.assert_called_once_with()
|
|
|
|
async def test_exact_saved_association_ambiguity_preserves_physical_file_evidence(
|
|
self,
|
|
):
|
|
for readable in (False, True):
|
|
with self.subTest(readable=readable):
|
|
filename = "readable.png" if readable else "empty.png"
|
|
metadata = (
|
|
{"prompt": json.dumps(graph("Physical portrait", ""))}
|
|
if readable
|
|
else {}
|
|
)
|
|
path = png(self.output_dir / filename, **metadata)
|
|
before = path.read_bytes()
|
|
results, lookups = [], []
|
|
for count, expected in ((0, "none"), (1, "linked"), (2, "ambiguous")):
|
|
if count:
|
|
positive = self._save_prompt(
|
|
f"Saved owner {readable} {count}",
|
|
category=f"Category {count}",
|
|
tags=[f"owner-{count}"],
|
|
)
|
|
self.api.db.link_image_to_prompt(positive, str(path))
|
|
with patch.object(
|
|
self.api.db,
|
|
"get_image_prompt_info",
|
|
wraps=self.api.db.get_image_prompt_info,
|
|
) as lookup:
|
|
response, result = await self.inspect(filename)
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(result["library"]["status"], expected)
|
|
self.assertEqual(lookup.call_count, 1)
|
|
lookups.append(lookup)
|
|
results.append(result)
|
|
self.assertEqual(result["metadata"]["raw"], metadata)
|
|
self.assertEqual(path.read_bytes(), before)
|
|
if count == 1:
|
|
self.assertEqual(result["library"]["positive_id"], positive)
|
|
self.assertEqual(
|
|
result["library"]["metadata_scope"], "positive"
|
|
)
|
|
self.assertEqual(result["library"]["category"], "Category 1")
|
|
self.assertEqual(result["library"]["tags"], ["owner-1"])
|
|
if not readable:
|
|
self.assertEqual(
|
|
result["prompts"]["positive"]["source"],
|
|
"saved-association",
|
|
)
|
|
if readable or count != 1:
|
|
self.assertEqual(result["prompts"], results[0]["prompts"])
|
|
self.assertEqual(
|
|
results[2]["library"],
|
|
{
|
|
"status": "ambiguous",
|
|
"scope": "saved-library",
|
|
"positive_id": None,
|
|
"pair_id": None,
|
|
"title": None,
|
|
"category": None,
|
|
"tags": [],
|
|
"metadata_scope": None,
|
|
},
|
|
)
|
|
for field in (
|
|
"image",
|
|
"file",
|
|
"metadata",
|
|
"prompts",
|
|
"resolution",
|
|
"root_scope",
|
|
):
|
|
self.assertEqual(results[2][field], results[0][field])
|
|
self.assertNotIn("Saved owner", json.dumps(results[2]))
|
|
self.assertNotIn("association_status", json.dumps(results[2]))
|
|
for lookup in lookups:
|
|
lookup.assert_called_once_with(
|
|
str(path), False, report_ambiguity=True
|
|
)
|