stroblme
3b2c08bf51
Three things the one-folder-per-study layout ran into. **Discovery walks down.** A plain directory is now walked all the way, so `fluksio sync dev` finds `dev/s1_baseline/study.py` and naming each study is no longer the price of the layout. Hidden directories, `__pycache__`, `node_modules` and virtualenvs are left alone, and a package is taken whole. Two files that would import under one module name are refused, naming both: Python keeps one module per name, so the second would silently *be* the first — and a node's generated body imports by that name, so a worker would run the wrong study's code. The message says the fix, which is an `__init__.py` per study directory. A module that raises while importing is now a sentence naming the file rather than an importlib traceback. **`run` and `sweep` sync downwards too**, so the flow is found from the repository root without the sync-then-`--no-sync` two-step. A study that will not import is a warning rather than a stopped run, since a walk meets every study and a half-finished one two directories away is not this run's problem. The upload was already a no-op for a flow nothing changed in, so what the walk costs is import time — `--sync PATH` narrows it, and skipping unchanged subtrees would need a cache keyed on file state that is deliberately not here. **`serve` moves off a busy default port** — 8001, 8002, up to twenty — says which it took, and writes that one into `client.json`. A port given with `--port` still fails when it is taken, because naming one is asking for it.
467 lines
15 KiB
Python
467 lines
15 KiB
Python
"""The command line, and the one ordering it depends on."""
|
|
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from fluksio.cli import load_or_create_secret_key
|
|
|
|
|
|
def test_importing_the_cli_does_not_build_the_settings() -> None:
|
|
"""`_configure_environment` has to run before anything reads a setting.
|
|
|
|
The settings are built on the first import of `fluksio.core.config`, and
|
|
every engine module reaches it within an import or two. If importing the
|
|
CLI pulled it in, `DATA_DIR` would be fixed at whatever directory the
|
|
command was run from — which is how the database ends up in the cwd.
|
|
"""
|
|
leaked = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"import fluksio.cli, sys; print('fluksio.core.config' in sys.modules)",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
assert leaked.stdout.strip() == "False", leaked.stdout
|
|
|
|
|
|
def test_the_secret_key_is_kept_rather_than_regenerated(tmp_path: Path) -> None:
|
|
"""A new key each start would sign out every session and orphan secrets.enc."""
|
|
path = tmp_path / "secret_key"
|
|
first = load_or_create_secret_key(path)
|
|
assert load_or_create_secret_key(path) == first
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
|
|
|
|
def test_the_store_works_without_git(tmp_path: Path, monkeypatch) -> None:
|
|
"""A pip install on a locked-down host may have no git.
|
|
|
|
Flows are files, and that is what the store is for; the history is the part
|
|
that needs git. Losing it must not be a refusal to start.
|
|
"""
|
|
import subprocess as sp
|
|
|
|
from fluksio.flow.store import FlowStore
|
|
|
|
real_run = sp.run
|
|
|
|
def no_git(cmd, *args, **kwargs):
|
|
if cmd and cmd[0] == "git":
|
|
raise FileNotFoundError(2, "No such file or directory", "git")
|
|
return real_run(cmd, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(sp, "run", no_git)
|
|
monkeypatch.setattr(FlowStore, "_warned_no_git", False)
|
|
|
|
store = FlowStore(tmp_path / "flows")
|
|
assert store.head() == ""
|
|
store.write_requirements("numpy\n")
|
|
assert store.read_requirements() == "numpy\n"
|
|
|
|
|
|
def test_run_arguments_are_typed_by_the_flow_they_are_for() -> None:
|
|
"""`--lr 0.05` is a float because the flow says `lr` is one."""
|
|
from fluksio.sdk import SyncError
|
|
from fluksio.sdk.cli import _params
|
|
|
|
definition = {
|
|
"inputs": [
|
|
{"spec": {"name": "lr", "dtype": "float"}},
|
|
{"spec": {"name": "epochs", "dtype": "int"}},
|
|
{"spec": {"name": "resume", "dtype": "bool"}},
|
|
]
|
|
}
|
|
|
|
assert _params(definition, ["--lr", "0.05", "--epochs", "3", "--resume"]) == {
|
|
"lr": 0.05,
|
|
"epochs": 3,
|
|
"resume": True,
|
|
}
|
|
assert _params(definition, ["--lr=1e-4"]) == {"lr": 0.0001}
|
|
|
|
import pytest
|
|
|
|
with pytest.raises(SyncError, match="not an input of this flow"):
|
|
_params(definition, ["--nonesuch", "1"])
|
|
|
|
|
|
def test_serve_uses_the_installation_the_directory_belongs_to(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
"""A repository with its own venv gets its own engine, not the machine's."""
|
|
from fluksio import cli
|
|
|
|
monkeypatch.chdir(tmp_path)
|
|
monkeypatch.setattr(cli, "DEFAULT_HOME", tmp_path / "home" / ".fluksio")
|
|
|
|
# Nothing above it yet: one is made here rather than in the home directory.
|
|
made = cli._data_dir(None)
|
|
assert made == (tmp_path / ".fluksio").resolve()
|
|
assert (made / ".gitignore").exists()
|
|
|
|
# And is then found again from anywhere below it.
|
|
deep = tmp_path / "src" / "deep"
|
|
deep.mkdir(parents=True)
|
|
monkeypatch.chdir(deep)
|
|
assert cli._data_dir(None) == made
|
|
|
|
# `--global` asks for the shared one even so.
|
|
assert (
|
|
cli._data_dir(None, shared=True) == (tmp_path / "home" / ".fluksio").resolve()
|
|
)
|
|
|
|
# And `--data-dir` still names any directory outright.
|
|
named = cli._data_dir(str(tmp_path / "named"))
|
|
assert named == (tmp_path / "named").resolve()
|
|
|
|
|
|
def test_enrolling_needs_only_a_claim_code() -> None:
|
|
"""The default portal is what makes first-run one flag rather than two."""
|
|
from fluksio.cli import DEFAULT_PORTAL, _parser
|
|
|
|
args = _parser().parse_args(["enroll", "ABC-123"])
|
|
|
|
assert args.portal == DEFAULT_PORTAL
|
|
assert DEFAULT_PORTAL.startswith("https://")
|
|
|
|
# And a portal of your own still wins.
|
|
mine = _parser().parse_args(["enroll", "ABC-123", "--portal", "https://hub.me"])
|
|
assert mine.portal == "https://hub.me"
|
|
|
|
|
|
def test_run_syncs_by_default_and_can_be_told_not_to() -> None:
|
|
from fluksio.cli import _parser
|
|
|
|
assert _parser().parse_args(["run", "train"]).no_sync is False
|
|
assert _parser().parse_args(["run", "train", "--no-sync"]).no_sync is True
|
|
|
|
|
|
def test_a_sweep_is_the_product_of_the_parameters_given() -> None:
|
|
"""`--param lr=0.1,0.01 --param epochs=1,2` is four runs, typed by the flow."""
|
|
import pytest
|
|
|
|
from fluksio.sdk import SyncError
|
|
from fluksio.sdk.cli import _grid
|
|
|
|
definition = {
|
|
"inputs": [
|
|
{"spec": {"name": "lr", "dtype": "float"}},
|
|
{"spec": {"name": "epochs", "dtype": "int"}},
|
|
]
|
|
}
|
|
|
|
grid = _grid(definition, ["lr=0.1,0.01", "epochs=1,2"], seed=7)
|
|
assert [entry["params"] for entry in grid] == [
|
|
{"lr": 0.1, "epochs": 1},
|
|
{"lr": 0.1, "epochs": 2},
|
|
{"lr": 0.01, "epochs": 1},
|
|
{"lr": 0.01, "epochs": 2},
|
|
]
|
|
assert all(entry["seed"] == 7 for entry in grid)
|
|
|
|
with pytest.raises(SyncError, match="not an input of this flow"):
|
|
_grid(definition, ["nonesuch=1"], seed=None)
|
|
with pytest.raises(SyncError, match="name=value"):
|
|
_grid(definition, ["lr"], seed=None)
|
|
|
|
|
|
def test_the_local_engine_is_asked_for_rather_than_guessed() -> None:
|
|
from fluksio.cli import _parser
|
|
|
|
parser = _parser()
|
|
assert parser.parse_args(["run", "train"]).local is False
|
|
assert parser.parse_args(["run", "train", "--local"]).local is True
|
|
assert parser.parse_args(["runs", "--local"]).local is True
|
|
assert parser.parse_args(["sweep", "train", "--param", "lr=1"]).local is False
|
|
|
|
|
|
def test_a_local_run_always_waits(monkeypatch) -> None:
|
|
"""The engine is this process, so a run nobody waits for is thrown away."""
|
|
from contextlib import contextmanager
|
|
|
|
from fluksio.cli import _parser
|
|
from fluksio.sdk import cli
|
|
|
|
submitted: dict[str, object] = {}
|
|
|
|
class FakeHandle:
|
|
id = "run-1"
|
|
status = "ok"
|
|
result: dict[str, object] = {}
|
|
|
|
def wait(self, timeout: float = 0.0) -> "FakeHandle":
|
|
submitted["waited"] = True
|
|
return self
|
|
|
|
class FakeClient:
|
|
def get_flow(self, name: str) -> dict[str, object]:
|
|
return {"definition": {"inputs": []}}
|
|
|
|
def submit(self, flow, params, seed=None, no_cache=False, cause="sdk"):
|
|
submitted["flow"] = flow
|
|
submitted["no_cache"] = no_cache
|
|
submitted["cause"] = cause
|
|
return FakeHandle()
|
|
|
|
def run(self, run_id: str) -> dict[str, object]:
|
|
return {"nodes": [{"status": "cached"}, {"status": "ok"}]}
|
|
|
|
@contextmanager
|
|
def fake_engine():
|
|
yield FakeClient()
|
|
|
|
monkeypatch.setattr(cli, "_engine_client", fake_engine)
|
|
|
|
args = _parser().parse_args(["run", "train", "--local", "--no-sync", "--no-cache"])
|
|
assert cli.cmd_run(args, []) == 0
|
|
# "cli" rather than the SDK's default: the history says which asked.
|
|
assert submitted == {
|
|
"flow": "train",
|
|
"no_cache": True,
|
|
"waited": True,
|
|
"cause": "cli",
|
|
}
|
|
|
|
|
|
def test_ctrl_c_while_waiting_cancels_the_run(monkeypatch) -> None:
|
|
"""Interrupting means stop the run, not walk away leaving it going."""
|
|
from contextlib import contextmanager
|
|
|
|
from fluksio.cli import _parser
|
|
from fluksio.sdk import cli
|
|
|
|
cancelled: list[str] = []
|
|
|
|
class FakeHandle:
|
|
id = "run-1"
|
|
status = "running"
|
|
result: dict[str, object] = {}
|
|
|
|
def wait(self, timeout: float = 0.0):
|
|
raise KeyboardInterrupt
|
|
|
|
class FakeClient:
|
|
def get_flow(self, name: str) -> dict[str, object]:
|
|
return {"definition": {"inputs": []}}
|
|
|
|
def submit(self, flow, params, seed=None, no_cache=False, cause="sdk"):
|
|
return FakeHandle()
|
|
|
|
def cancel(self, run_id: str) -> None:
|
|
cancelled.append(run_id)
|
|
|
|
@contextmanager
|
|
def fake_engine():
|
|
yield FakeClient()
|
|
|
|
monkeypatch.setattr(cli, "_engine_client", fake_engine)
|
|
|
|
args = _parser().parse_args(["run", "train", "--local", "--no-sync"])
|
|
assert cli.cmd_run(args, []) == 130
|
|
assert cancelled == ["run-1"]
|
|
|
|
|
|
def test_an_artifact_input_may_be_named_rather_than_pasted() -> None:
|
|
"""The engine resolves either spelling; the CLI just stops mangling them."""
|
|
import json
|
|
|
|
from fluksio.sdk.cli import _coerce
|
|
|
|
assert _coerce("@run:123-abc.dataset", "artifact") == "@run:123-abc.dataset"
|
|
digest = "sha256:" + "a1" * 32
|
|
assert _coerce(digest, "artifact") == digest
|
|
# A reference a script already holds still arrives as the object it is.
|
|
reference = {"digest": digest, "size": 3}
|
|
assert _coerce(json.dumps(reference), "artifact") == reference
|
|
|
|
|
|
def test_the_run_prompt_keeps_declared_values_for_anything_left_blank() -> None:
|
|
"""Enter through the lot is what running with the defaults looks like."""
|
|
from unittest.mock import patch
|
|
|
|
from fluksio.sdk.cli import _ask_params
|
|
|
|
definition = {
|
|
"inputs": [
|
|
{"spec": {"name": "lr", "dtype": "float"}, "initial": 0.05},
|
|
{"spec": {"name": "epochs", "dtype": "int"}, "initial": 10},
|
|
]
|
|
}
|
|
|
|
with patch("builtins.input", side_effect=["", ""]):
|
|
# Nothing sent: an input the caller leaves out keeps what it declares,
|
|
# which is the same contract the browser's dialog has.
|
|
assert _ask_params(definition) == {}
|
|
|
|
with patch("builtins.input", side_effect=["0.01", "50"]):
|
|
assert _ask_params(definition) == {"lr": 0.01, "epochs": 50}
|
|
|
|
|
|
def test_the_run_prompt_names_an_answer_of_the_wrong_type() -> None:
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from fluksio.sdk import SyncError
|
|
from fluksio.sdk.cli import _ask_params
|
|
|
|
definition = {"inputs": [{"spec": {"name": "lr", "dtype": "float"}}]}
|
|
with patch("builtins.input", side_effect=["fast"]):
|
|
with pytest.raises(SyncError, match="lr"):
|
|
_ask_params(definition)
|
|
|
|
|
|
def test_an_export_is_parsed_with_its_selection() -> None:
|
|
"""`export` is a group of two tables, and both take the same filters."""
|
|
from fluksio.cli import _parser
|
|
from fluksio.sdk.cli import cmd_export_metrics, cmd_export_runs
|
|
|
|
parser = _parser()
|
|
metrics = parser.parse_args(
|
|
[
|
|
"export",
|
|
"metrics",
|
|
"--flow",
|
|
"train",
|
|
"--run",
|
|
"a",
|
|
"--run",
|
|
"b",
|
|
"--name",
|
|
"loss,val",
|
|
"--stride",
|
|
"5",
|
|
]
|
|
)
|
|
assert metrics.func is cmd_export_metrics
|
|
assert metrics.run == ["a", "b"]
|
|
assert metrics.stride == 5
|
|
assert metrics.format == "csv"
|
|
|
|
runs = parser.parse_args(["export", "runs", "--params", "lr", "--format", "jsonl"])
|
|
assert runs.func is cmd_export_runs
|
|
assert runs.params == "lr"
|
|
assert runs.format == "jsonl"
|
|
|
|
|
|
def test_the_metric_names_are_asked_for_rather_than_guessed() -> None:
|
|
"""A name is flow-qualified, so `--list` is what says what would match."""
|
|
from fluksio.cli import _parser
|
|
from fluksio.sdk.cli import _list_names
|
|
|
|
parser = _parser()
|
|
assert parser.parse_args(["export", "metrics"]).list_names is False
|
|
assert parser.parse_args(["export", "metrics", "--list"]).list_names is True
|
|
|
|
class Engine:
|
|
def runs(self, flow="", limit=0, **filters):
|
|
assert filters == {"status": "ok"}
|
|
return [{"id": "r-empty"}, {"id": "r-1"}]
|
|
|
|
def metrics(self, run_id, name="", stride=1):
|
|
# The newest run failed before it measured anything; the next one
|
|
# carries the vocabulary.
|
|
return [] if run_id == "r-empty" else [{"name": "train.train_loss"}]
|
|
|
|
args = parser.parse_args(
|
|
["export", "metrics", "--flow", "train", "--status", "ok", "--list"]
|
|
)
|
|
assert _list_names(Engine(), args) == 0
|
|
|
|
|
|
def test_an_engine_without_the_route_is_named_rather_than_404() -> None:
|
|
"""A client ships ahead of the engine; a flat 404 does not say so."""
|
|
from fluksio.sdk.cli import _too_old
|
|
|
|
class Old:
|
|
def summary(self):
|
|
return {"status": "ok", "version": "0.1.4"}
|
|
|
|
class Ancient:
|
|
def summary(self):
|
|
# Older than the field itself.
|
|
return {"status": "ok"}
|
|
|
|
assert "engine is 0.1.4" in _too_old(Old())
|
|
assert "engine is older" in _too_old(Ancient())
|
|
assert "pip install -U fluksio" in _too_old(Ancient())
|
|
|
|
|
|
def test_a_study_in_a_subfolder_is_found(tmp_path) -> None:
|
|
"""One directory per study is a layout; naming each is bookkeeping."""
|
|
from fluksio.sdk.cli import _below
|
|
|
|
(tmp_path / "dev" / "s1").mkdir(parents=True)
|
|
(tmp_path / "dev" / "s2" / "inner").mkdir(parents=True)
|
|
(tmp_path / "results").mkdir()
|
|
(tmp_path / "pkg").mkdir()
|
|
(tmp_path / ".hidden").mkdir()
|
|
(tmp_path / "dev" / "s1" / "study.py").write_text("")
|
|
(tmp_path / "dev" / "s2" / "inner" / "probe.py").write_text("")
|
|
(tmp_path / "results" / "notes.py").write_text("")
|
|
(tmp_path / "pkg" / "__init__.py").write_text("")
|
|
(tmp_path / "pkg" / "deep.py").write_text("")
|
|
(tmp_path / ".hidden" / "skip.py").write_text("")
|
|
|
|
found = {path.relative_to(tmp_path).as_posix() for path in _below(tmp_path)}
|
|
|
|
# However deep, plus packages whole — and nothing under a dot directory.
|
|
assert found == {
|
|
"dev/s1/study.py",
|
|
"dev/s2/inner/probe.py",
|
|
"results/notes.py",
|
|
"pkg",
|
|
}
|
|
|
|
|
|
def test_two_files_of_one_name_are_refused(tmp_path) -> None:
|
|
"""Python keeps one module per name, and a node's body imports by it."""
|
|
import pytest
|
|
|
|
from fluksio.sdk import SyncError
|
|
from fluksio.sdk.cli import _import, _module_of
|
|
|
|
for study in ("s1", "s2"):
|
|
(tmp_path / study).mkdir()
|
|
(tmp_path / study / "study.py").write_text("VALUE = 1\n")
|
|
|
|
first = tmp_path / "s1" / "study.py"
|
|
second = tmp_path / "s2" / "study.py"
|
|
_import(*_module_of(first), first)
|
|
with pytest.raises(SyncError, match="both import as 'study'"):
|
|
_import(*_module_of(second), second)
|
|
|
|
|
|
def test_run_and_sweep_take_what_to_sync() -> None:
|
|
from fluksio.cli import _parser
|
|
|
|
parser = _parser()
|
|
assert parser.parse_args(["sweep", "train", "--sync", "dev/s1"]).sync == ["dev/s1"]
|
|
args, _ = parser.parse_known_args(["run", "train", "--sync", "dev/s1"])
|
|
assert args.sync == ["dev/s1"]
|
|
# Nothing named means this directory, downwards.
|
|
assert parser.parse_args(["sweep", "train"]).sync == []
|
|
|
|
|
|
def test_serve_moves_off_a_port_that_is_taken() -> None:
|
|
"""A first start should not die on somebody else's dev server."""
|
|
import socket
|
|
|
|
from fluksio.cli import DEFAULT_PORT, _free_port, _parser
|
|
|
|
with socket.socket() as held:
|
|
held.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
held.bind(("127.0.0.1", 0))
|
|
held.listen(1)
|
|
taken = held.getsockname()[1]
|
|
|
|
assert _free_port("127.0.0.1", taken) == taken + 1
|
|
|
|
# A port that was asked for is not moved off: that is what asking means.
|
|
assert _parser().parse_args(["serve"]).port is None
|
|
assert _parser().parse_args(["serve", "--port", "9000"]).port == 9000
|
|
assert DEFAULT_PORT == 8000
|