61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
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()
|