app/backend/tests/flow/test_modules.py

"""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 fluksio.flow import modules
from fluksio.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("")


# ---------------------------------------------------------------------------
# Adopting the venv Fluksio was installed into
#
# The workflow this exists for: make a venv, install what you work with, then
# `pip install fluksio` into the same one. Building a second environment beside
# it would leave node code unable to import the packages that are the point.
# ---------------------------------------------------------------------------


@pytest.fixture
def their_venv(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    """Pretend the engine is running from a venv somebody else made."""
    directory = tmp_path / "research"
    (directory / "bin").mkdir(parents=True)
    (directory / "bin" / "python").write_text("")
    monkeypatch.setattr(modules.sys, "prefix", str(directory))
    monkeypatch.setattr(modules.sys, "base_prefix", str(tmp_path / "usr"))
    # Undoes the suite-wide "managed" in tests/__init__.py.
    monkeypatch.setattr(modules.settings, "NODE_VENV", "auto")
    return directory


def test_a_venv_of_your_own_is_adopted(venv: Path, their_venv: Path):
    assert modules.adopted() == their_venv
    assert modules.venv_python() == str(their_venv / "bin" / "python")


def test_an_adopted_venv_is_never_synced(venv: Path, their_venv: Path):
    """The whole hazard: sync means "hold exactly this", so it would empty it."""
    ok, output = modules.sync("numpy>=2")

    assert ok is False
    assert "not Fluksio's to install" in output
    # Nothing was built beside it either.
    assert not venv.exists()


def test_reconcile_leaves_an_adopted_venv_alone(
    venv: Path, their_venv: Path, tmp_path: Path
):
    store = FlowStore(tmp_path / "flows")
    store.write_requirements("numpy>=2\n")

    modules.reconcile(store)

    assert not venv.exists()
    assert modules.venv_python() == str(their_venv / "bin" / "python")


def test_a_venv_the_engine_already_owns_is_kept(venv: Path, their_venv: Path):
    """An upgrade must not take away packages somebody installed on purpose."""
    (venv / "bin").mkdir(parents=True)
    (venv / "bin" / "python").write_text("")

    assert modules.adopted() is None
    assert modules.venv_python() == str(venv / "bin" / "python")


def test_managed_is_what_a_container_asks_for(
    venv: Path, their_venv: Path, monkeypatch: pytest.MonkeyPatch
):
    monkeypatch.setattr(modules.settings, "NODE_VENV", "managed")

    assert modules.adopted() is None


def test_an_interpreter_can_be_named_outright(
    venv: Path, their_venv: Path, monkeypatch: pytest.MonkeyPatch
):
    monkeypatch.setattr(modules.settings, "NODE_VENV", "/opt/conda/bin/python")

    assert modules.venv_python() == "/opt/conda/bin/python"


def test_info_describes_an_adopted_venv(venv: Path, their_venv: Path, tmp_path: Path):
    store = FlowStore(tmp_path / "flows")
    store.write_requirements("numpy>=2\n")

    information = modules.info(store)

    assert information.adopted is True
    assert information.venv_path == str(their_venv)
    assert information.python_version
    # The manifest does not describe it, so it cannot be out of step with it.
    assert information.applied is True