diff --git a/tests/test_execution_output_database.py b/tests/test_execution_output_database.py index ee494a6a..080844be 100644 --- a/tests/test_execution_output_database.py +++ b/tests/test_execution_output_database.py @@ -17,9 +17,11 @@ except ImportError: import test_execution_history_database as history_tests -def bundle(filename="first.png", *, root="a" * 64, node="9", ordinal=0): +def bundle( + filename="first.png", *, root="a" * 64, node="9", ordinal=0, path_style="posix" +): return OutputBundle( - (OutputReference(node, ordinal, "output", "", filename, "posix", root),), + (OutputReference(node, ordinal, "output", "", filename, path_style, root),), "complete" if root else "partial", "reported" if root else "root_unavailable", ) diff --git a/tests/test_execution_output_gallery.py b/tests/test_execution_output_gallery.py index 661e3a8e..86a8f5c5 100644 --- a/tests/test_execution_output_gallery.py +++ b/tests/test_execution_output_gallery.py @@ -1,5 +1,7 @@ """Mutable metadata galleries never own or retarget frozen output references.""" +from pathlib import PurePosixPath, PureWindowsPath + from PIL import Image from PIL.PngImagePlugin import PngInfo @@ -18,11 +20,11 @@ class OutputGalleryTests(AdminAPITestCase): db = self.api.db start = self.start(db, item) db.write_execution_start(start) - key = snapshot_output_roots( - "posix", str(root or self.output_dir), None - ).output_key + output_root = root or self.output_dir + style = "windows" if isinstance(output_root, PureWindowsPath) else "posix" + key = snapshot_output_roots(style, str(output_root), None).output_key db.write_execution_outputs_for_finish( - self.finish(start), bundle(filename, root=key) + self.finish(start), bundle(filename, root=key, path_style=style) ) return start.run_id, db.get_execution_outputs(start.run_id) @@ -48,6 +50,20 @@ class OutputGalleryTests(AdminAPITestCase): ) } + def test_capture_matches_root_path_style_on_each_platform(self) -> None: + roots = ( + (PurePosixPath("/output"), "posix"), + (PureWindowsPath(r"D:\output"), "windows"), + ) + for item, (root, style) in enumerate(roots): + with self.subTest(style=style): + run, evidence = self.capture(item=item, root=root) + self.assertEqual(evidence["state"], "complete") + reference = evidence["references"][0] + self.assertEqual(reference["path_style"], style) + self.assertEqual(len(reference["root_key"]), 64) + self.assert_evidence(run, evidence) + def test_metadata_disabled_or_malformed_still_captures_without_gallery_rows(self): for item, metadata in enumerate(({}, {"prompt": "{malformed"})): with self.subTest(metadata=metadata): diff --git a/tests/test_execution_output_hooks.py b/tests/test_execution_output_hooks.py index a3a2a04a..cf27a8cc 100644 --- a/tests/test_execution_output_hooks.py +++ b/tests/test_execution_output_hooks.py @@ -1,6 +1,8 @@ """Exact successful host callbacks persist independently bounded output evidence.""" from dataclasses import replace +import os +from pathlib import Path import threading import sys from types import SimpleNamespace @@ -309,21 +311,53 @@ class OutputHooksTests(fixtures.unittest.TestCase): self.assertEqual(registration.writer.context_budget.charged_bytes, 0) def test_default_adapter_uses_configured_output_and_temp_namespaces(self): - folder_paths = SimpleNamespace( - get_output_directory=lambda: "/output", get_temp_directory=lambda: "/temp" + native_root = Path(self.directory.name) + cases = ( + ( + os.name, + str(native_root / "output"), + str(native_root / "temp"), + "windows" if os.name == "nt" else "posix", + ), + ("posix", "/configured", "/configured", "posix"), + ("nt", r"D:\configured", r"D:\configured", "windows"), ) - with patch.dict(sys.modules, {"folder_paths": folder_paths}): - registration = HistoryRegistration( - self.queue, fixtures.JOB, lambda: self.db + for item, (platform, output_root, temp_root, style) in enumerate(cases): + folder_paths = SimpleNamespace( + get_output_directory=lambda: output_root, + get_temp_directory=lambda: temp_root, ) - self.registrations.append(registration) - self.wait_for(lambda: registration.writer.status().accepting) - self.dequeue(0) - self.queue.task_done(0, history(), fixtures.status()) - self.drain(registration) - self.assertEqual(self.outputs(0)["state"], "complete") - self.assertEqual(len(self.outputs(0)["references"][0]["root_key"]), 64) - self.assertEqual(registration._roots, {}) + with ( + self.subTest(platform=platform, root=output_root), + patch.dict(sys.modules, {"folder_paths": folder_paths}), + patch("utils.execution_history.os", SimpleNamespace(name=platform)), + ): + registration = HistoryRegistration( + self.queue, fixtures.JOB, lambda: self.db + ) + self.registrations.append(registration) + self.wait_for(lambda: registration.writer.status().accepting) + self.dequeue(item) + payload = history() + payload["outputs"]["save"]["images"].append( + {"filename": "a.png", "subfolder": "", "type": "temp"} + ) + self.queue.task_done(item, payload, fixtures.status()) + self.drain(registration) + evidence = self.outputs(item) + self.assertEqual(evidence["state"], "complete") + references = evidence["references"] + self.assertEqual(len(references[0]["root_key"]), 64) + self.assertEqual(len(references[1]["root_key"]), 64) + self.assertEqual([r["path_style"] for r in references], [style] * 2) + self.assertEqual( + [r["folder_type"] for r in references], ["output", "temp"] + ) + self.assertNotEqual( + references[0]["root_key"], references[1]["root_key"] + ) + self.assertEqual(registration._roots, {}) + self.assertTrue(registration.dispose()) def test_sealed_context_is_discarded_before_output_reservation(self): pool = ContextBudget(1048576)