stroblme
71628abcbc
The worker script is handed to the interpreter by path, so app/flow was sys.path[0] for every node: `import queue` got the engine's. It now drops its own directory before anything else imports, and runs with the deployment's credentials scrubbed out of its environment. Also: reload builds off the event loop, the pool wakes what is blocked on it when it stops, a refused metrics flush is kept for the next one rather than dropped, and the cascade events are paired through failures.
78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
"""The venv node code imports from is built and kept in step with a manifest."""
|
|
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.flow import modules
|
|
from app.flow.store import FlowStore
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
shutil.which("uv") is None, reason="module management needs uv"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def venv(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
directory = tmp_path / "user-venv"
|
|
monkeypatch.setattr(modules, "VENV_DIR", directory)
|
|
return directory
|
|
|
|
|
|
def test_an_empty_manifest_gives_a_venv_of_its_own(venv: Path, tmp_path: Path):
|
|
store = FlowStore(tmp_path / "flows")
|
|
store.write_requirements("")
|
|
modules.sync("")
|
|
|
|
assert Path(modules.venv_python()).exists()
|
|
assert modules.venv_python().startswith(str(venv))
|
|
|
|
information = modules.info(store)
|
|
assert information.applied is True
|
|
assert information.python_version
|
|
|
|
|
|
def test_a_manifest_already_applied_is_not_installed_again(
|
|
venv: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
store = FlowStore(tmp_path / "flows")
|
|
store.write_requirements("")
|
|
modules.sync("")
|
|
marker = (venv / ".applied").read_text()
|
|
|
|
# Recorded rather than refused: reconcile swallows what it fails on, so an
|
|
# exception raised in here would never reach the test.
|
|
calls: list[object] = []
|
|
monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: calls.append(args))
|
|
modules.reconcile(store)
|
|
|
|
assert calls == []
|
|
assert (venv / ".applied").read_text() == marker
|
|
|
|
|
|
def test_a_missing_uv_is_a_failed_apply_rather_than_a_crash(
|
|
venv: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
def no_uv(*args: object, **kwargs: object) -> None:
|
|
raise FileNotFoundError(2, "No such file or directory: 'uv'")
|
|
|
|
monkeypatch.setattr(subprocess, "run", no_uv)
|
|
ok, output = modules.sync("")
|
|
|
|
# The route turns this into a 400 with the output; an exception would be a
|
|
# 500 with nothing in it.
|
|
assert ok is False
|
|
assert "uv could not run" in output
|
|
|
|
|
|
def test_a_manifest_that_does_not_resolve_leaves_the_venv_alone(venv: Path):
|
|
modules.sync("")
|
|
ok, output = modules.sync("fluksio-no-such-package-anywhere==9.9.9")
|
|
|
|
assert ok is False
|
|
assert output
|
|
# The marker still describes the manifest that actually installed.
|
|
assert (venv / ".applied").read_text() == modules._digest("")
|