workshop rework

tests/test_offline.py

@@ -0,0 +1,59 @@
+"""Plain Python checks, including real FFmpeg encoding when installed."""
+import binascii
+import os
+from pathlib import Path
+import shutil
+import struct
+import sys
+import tempfile
+import unittest
+import zlib
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "Content/Python"))
+from ue_workshop.paths import slug
+from ue_workshop.video import encode
+
+
+def png(path, rgb):
+    def chunk(kind, data):
+        return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", binascii.crc32(kind + data) & 0xffffffff)
+    path.write_bytes(b"\x89PNG\r\n\x1a\n" +
+                     chunk(b"IHDR", struct.pack(">IIBBBBB", 4, 4, 8, 2, 0, 0, 0)) +
+                     chunk(b"IDAT", zlib.compress((b"\x00" + bytes(rgb) * 4) * 4)) + chunk(b"IEND", b""))
+
+
+class OfflineTests(unittest.TestCase):
+    def test_generated_asset_name_cannot_escape_folder(self):
+        for value in ["../Outside", "A/B", "", "A.B", "0Shot", "A B"]:
+            with self.subTest(value=value), self.assertRaises(ValueError):
+                slug(value)
+        self.assertEqual(slug("MyShot_2"), "MyShot_2")
+
+    def test_missing_frames_are_rejected_before_encoding(self):
+        with tempfile.TemporaryDirectory() as directory:
+            folder = Path(directory)
+            png(folder / "frame_0001.png", (255, 0, 0))
+            with self.assertRaisesRegex(RuntimeError, "numeracji"):
+                encode(folder, 24)
+
+    def test_incomplete_render_is_not_encoded(self):
+        with tempfile.TemporaryDirectory() as directory:
+            folder = Path(directory)
+            png(folder / "frame_0000.png", (255, 0, 0))
+            with self.assertRaisesRegex(RuntimeError, "zamiast 2"):
+                encode(folder, 24, expected_frames=2)
+
+    @unittest.skipUnless(shutil.which("ffmpeg") or os.environ.get("WORKSHOP_FFMPEG"), "FFmpeg missing")
+    def test_real_mp4_encoding(self):
+        with tempfile.TemporaryDirectory() as directory:
+            folder = Path(directory) / "frames"
+            folder.mkdir()
+            png(folder / "frame_0000.png", (255, 0, 0))
+            png(folder / "frame_0001.png", (0, 0, 255))
+            movie = encode(folder, 24, expected_frames=2)
+            self.assertGreater(movie.stat().st_size, 100)
+            self.assertIn(b"ftyp", movie.read_bytes()[:32])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_progress.py

@@ -0,0 +1,60 @@
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
+from progress_display import ProgressDisplay
+
+
+class ProgressTests(unittest.TestCase):
+    def display(self, path):
+        self.now = 0
+        self.messages = []
+        return ProgressDisplay(path, clock=lambda: self.now,
+                               emit=lambda text, **kwargs: self.messages.append(text))
+
+    def test_partial_events_are_read_once_after_they_finish(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path = Path(folder) / "progress.jsonl"
+            display = self.display(path)
+            path.write_bytes(b'{"state":"prepar')
+            display.update()
+            self.assertEqual(len(self.messages), 1)
+            with path.open("ab") as stream:
+                stream.write(b'ing"}\n')
+            display.update()
+            display.update()
+            self.assertEqual(len(self.messages), 2)
+            self.assertIn("Przygotowywanie sceny", self.messages[-1])
+
+    def test_waiting_heartbeat_is_throttled(self):
+        with tempfile.TemporaryDirectory() as folder:
+            display = self.display(Path(folder) / "progress.jsonl")
+            self.now = 9
+            display.update()
+            self.assertEqual(len(self.messages), 1)
+            self.now = 10
+            display.update()
+            display.update()
+            self.assertEqual(len(self.messages), 2)
+            self.assertIn("Nadal pracuje", self.messages[-1])
+
+    def test_frame_progress_counts_actual_output_files(self):
+        with tempfile.TemporaryDirectory() as folder:
+            root = Path(folder)
+            (root / "frames").mkdir()
+            (root / "frames/frame_0000.png").touch()
+            (root / "frames/frame_0001.png").touch()
+            path = root / "progress.jsonl"
+            path.write_text(json.dumps(dict(state="rendering", output=folder, frames=4)) + "\n")
+            display = self.display(path)
+            display.update()
+            self.now = 2
+            display.update()
+            self.assertIn("2/4 klatek", self.messages[-1])
+
+
+if __name__ == "__main__":
+    unittest.main()