mirror of
https://github.com/rendercv/rendercv.git
synced 2026-04-18 05:52:54 -04:00
ec884975d4
- Use Path.open() instead of open() in pdf_png.py (PTH123) - Chain KeyError with 'from e' in pdf_png.py (B904) - Fix comment placement in section.py to avoid false unused-ignore - Update watcher test to mock observer.join() instead of time.sleep()
141 lines
4.3 KiB
Python
141 lines
4.3 KiB
Python
import threading
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import typer
|
|
import watchdog.events
|
|
|
|
from rendercv.cli.render_command import watcher
|
|
from rendercv.cli.render_command.watcher import EventHandler
|
|
|
|
|
|
class TestRunFunctionIfFilesChange:
|
|
def test_runs_function_immediately_on_start(self, tmp_path):
|
|
file_path = tmp_path / "test.yaml"
|
|
file_path.touch()
|
|
mock_function = MagicMock()
|
|
|
|
mock_observer_class = MagicMock()
|
|
mock_observer_class.return_value.join.side_effect = KeyboardInterrupt
|
|
with patch.object(watcher.watchdog.observers, "Observer", mock_observer_class):
|
|
watcher.run_function_if_files_change([file_path], mock_function)
|
|
|
|
mock_function.assert_called_once()
|
|
|
|
def test_reruns_function_when_file_is_modified(self, tmp_path):
|
|
watched_file = tmp_path / "test.yaml"
|
|
watched_file.write_text("initial", encoding="utf-8")
|
|
|
|
call_count = 0
|
|
|
|
def tracked_function():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
|
|
watcher_thread = threading.Thread(
|
|
target=watcher.run_function_if_files_change,
|
|
args=([watched_file], tracked_function),
|
|
daemon=True,
|
|
)
|
|
watcher_thread.start()
|
|
|
|
time.sleep(0.2)
|
|
initial_count = call_count
|
|
|
|
watched_file.write_text("first edit", encoding="utf-8")
|
|
time.sleep(0.2)
|
|
|
|
assert call_count > initial_count
|
|
|
|
def test_reruns_function_when_secondary_file_is_modified(self, tmp_path):
|
|
main_file = tmp_path / "cv.yaml"
|
|
main_file.write_text("main content", encoding="utf-8")
|
|
design_file = tmp_path / "design.yaml"
|
|
design_file.write_text("initial design", encoding="utf-8")
|
|
|
|
call_count = 0
|
|
|
|
def tracked_function():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
|
|
watcher_thread = threading.Thread(
|
|
target=watcher.run_function_if_files_change,
|
|
args=([main_file, design_file], tracked_function),
|
|
daemon=True,
|
|
)
|
|
watcher_thread.start()
|
|
|
|
time.sleep(0.2)
|
|
count_before_edit = call_count
|
|
|
|
# Edit the design file (not the main file)
|
|
design_file.write_text("updated design", encoding="utf-8")
|
|
time.sleep(0.2)
|
|
|
|
assert call_count > count_before_edit
|
|
|
|
def test_continues_watching_after_typer_exit(self, tmp_path):
|
|
watched_file = tmp_path / "test.yaml"
|
|
watched_file.write_text("initial", encoding="utf-8")
|
|
|
|
call_count = 0
|
|
should_raise = False
|
|
|
|
def tracked_function():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if should_raise:
|
|
raise typer.Exit(code=1)
|
|
|
|
watcher_thread = threading.Thread(
|
|
target=watcher.run_function_if_files_change,
|
|
args=([watched_file], tracked_function),
|
|
daemon=True,
|
|
)
|
|
watcher_thread.start()
|
|
|
|
time.sleep(0.2)
|
|
should_raise = True
|
|
count_before_exit = call_count
|
|
watched_file.write_text("edit that raises exit", encoding="utf-8")
|
|
time.sleep(0.2)
|
|
|
|
assert call_count > count_before_exit
|
|
|
|
should_raise = False
|
|
count_after_exit = call_count
|
|
watched_file.write_text("edit after exit", encoding="utf-8")
|
|
time.sleep(0.2)
|
|
|
|
assert call_count > count_after_exit
|
|
|
|
|
|
class TestEventHandler:
|
|
def test_ignores_events_for_unwatched_files(self):
|
|
mock_fn = MagicMock()
|
|
handler = EventHandler(mock_fn, watched_files={"/watched/file.yaml"})
|
|
|
|
event = watchdog.events.FileModifiedEvent("/other/file.yaml")
|
|
handler.on_modified(event)
|
|
|
|
mock_fn.assert_not_called()
|
|
|
|
def test_calls_function_for_watched_file(self):
|
|
mock_fn = MagicMock()
|
|
handler = EventHandler(mock_fn, watched_files={"/watched/file.yaml"})
|
|
|
|
event = watchdog.events.FileModifiedEvent("/watched/file.yaml")
|
|
handler.on_modified(event)
|
|
|
|
mock_fn.assert_called_once()
|
|
|
|
def test_suppresses_typer_exit(self):
|
|
mock_fn = MagicMock(side_effect=typer.Exit(code=1))
|
|
handler = EventHandler(mock_fn, watched_files={"/watched/file.yaml"})
|
|
|
|
event = watchdog.events.FileModifiedEvent("/watched/file.yaml")
|
|
handler.on_modified(event)
|
|
|
|
mock_fn.assert_called_once()
|