64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
import importlib
|
|
from pathlib import Path
|
|
from threading import Event
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from great_ai.deploy.reload_event_handler import ReloadEventHandler
|
|
from watchdog.events import FileClosedEvent, FileModifiedEvent, FileMovedEvent
|
|
|
|
|
|
def test_reload_debounces_edit_and_replacement_events() -> None:
|
|
completed = Event()
|
|
calls = []
|
|
|
|
def restart() -> None:
|
|
calls.append("restart")
|
|
completed.set()
|
|
|
|
handler = ReloadEventHandler(restart, debounce_seconds=0.02)
|
|
try:
|
|
handler.dispatch(FileModifiedEvent("service.py"))
|
|
handler.dispatch(FileClosedEvent("service.py"))
|
|
handler.dispatch(FileMovedEvent("temporary", "service.py"))
|
|
assert completed.wait(timeout=2)
|
|
assert calls == ["restart"]
|
|
handler.dispatch(FileModifiedEvent("__service__.py"))
|
|
finally:
|
|
handler.close()
|
|
handler.dispatch(FileModifiedEvent("service.py"))
|
|
assert calls == ["restart"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"module_name, operation",
|
|
[
|
|
("great_ai.__main__", "serve"),
|
|
("great_ai.large_file.__main__", "handle_command"),
|
|
],
|
|
)
|
|
def test_cli_operational_errors_exit_nonzero(module_name: str, operation: str) -> None:
|
|
module = importlib.import_module(module_name)
|
|
with patch.object(module, operation, side_effect=ValueError("failed")):
|
|
with pytest.raises(SystemExit) as error:
|
|
module.main()
|
|
assert error.value.code == 1
|
|
|
|
|
|
def test_cli_resolves_nested_scripts_and_notebooks(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from great_ai.__main__ import get_script_name, get_script_name_of_notebook
|
|
|
|
monkeypatch.syspath_prepend(str(tmp_path))
|
|
package = tmp_path / "nested"
|
|
package.mkdir()
|
|
(package / "__init__.py").write_text("")
|
|
(package / "service.py").write_text("value = 7\n")
|
|
monkeypatch.chdir(tmp_path)
|
|
assert get_script_name("nested/service.py") == "nested.service"
|
|
assert importlib.import_module("nested.service").value == 7
|
|
assert get_script_name(str(package / "service.py")) == "nested.service"
|
|
assert (
|
|
get_script_name_of_notebook("nested/service.ipynb") == "nested/__service__.py"
|
|
)
|