stroblme
5c7a8773ad
The command line no longer imports a research module, so import-time GPU work ends with a short-lived child instead of holding a CUDA context for the length of a followed run. fluksio.sdk.declare starts the child, which writes each flow's document and shims as JSON; Declared stands in for Flow in sync(). Errors keep their type and message, a child's traceback reaches the user.
220 lines
7.6 KiB
Python
220 lines
7.6 KiB
Python
"""`fluksio sync` against a real engine: what it stores, and what it refuses."""
|
|
|
|
import sys
|
|
from collections.abc import Generator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fluksio.core.config import settings
|
|
from fluksio.sdk import FLOWS, MARKER, Declared, SyncError
|
|
from fluksio.sdk.client import ApiError, Client, sync
|
|
|
|
EXAMPLES = Path(__file__).parents[4] / "examples"
|
|
PREFIX = settings.API_V1_STR
|
|
ORIGIN = {
|
|
"kind": "python",
|
|
"repo": str(EXAMPLES),
|
|
"commit": "a1b2c3d4",
|
|
"dirty": False,
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def flows() -> Generator[dict, None, None]:
|
|
"""The example research package, imported the way `sync` imports it."""
|
|
sys.path.insert(0, str(EXAMPLES))
|
|
FLOWS.clear()
|
|
import myresearch.pipeline # noqa: F401
|
|
|
|
yield dict(FLOWS)
|
|
FLOWS.clear()
|
|
sys.path.remove(str(EXAMPLES))
|
|
|
|
|
|
@pytest.fixture
|
|
def api(client: TestClient, superuser_token_headers: dict[str, str]) -> Client:
|
|
return Client(http=client, token=superuser_token_headers["Authorization"][7:])
|
|
|
|
|
|
def commits() -> int:
|
|
"""How many commits the flow store has made."""
|
|
from fluksio.flow.store import FlowStore
|
|
|
|
store = FlowStore(settings.FLOWS_DIR)
|
|
result = store._git("rev-list", "--count", "HEAD")
|
|
return int(result.stdout.strip()) if result.returncode == 0 else 0
|
|
|
|
|
|
@pytest.mark.parametrize("repo", ["/", str(Path.home()), "examples"])
|
|
def test_sync_refuses_a_repo_every_run_would_walk_the_disk_for(api, flows, repo):
|
|
with pytest.raises(ApiError) as refused:
|
|
sync([flows["train"]], api, origin={**ORIGIN, "repo": repo})
|
|
|
|
assert refused.value.status == 422
|
|
assert "origin.repo" in str(refused.value.detail)
|
|
|
|
|
|
def test_sync_stores_a_runnable_flow_with_its_origin(api, flows):
|
|
reports = sync([flows["train"]], api, origin=ORIGIN)
|
|
|
|
assert [r.flow for r in reports] == ["train"]
|
|
assert reports[0].created and reports[0].published
|
|
|
|
stored = api.get_flow("train")
|
|
definition = stored["definition"]
|
|
assert definition["origin"]["commit"] == "a1b2c3d4"
|
|
assert [n["id"] for n in definition["nodes"]] == ["prepare", "fit", "evaluate"]
|
|
assert not stored["has_draft"]
|
|
# The store holds a complete definition, and the bodies say where the real
|
|
# code is rather than being a copy of it.
|
|
source = api.get_source("train", "fit")
|
|
assert source.startswith(MARKER)
|
|
assert "from myresearch.train import fit" in source
|
|
|
|
|
|
def test_a_flow_read_by_another_process_syncs_as_the_flow_does(api, flows):
|
|
"""`fluksio sync` imports in a child and uploads what it handed back."""
|
|
carried = Declared(flows["train"].declaration())
|
|
|
|
sync([carried], api, origin=ORIGIN)
|
|
stored = api.get_flow("train")["definition"]
|
|
assert stored["origin"]["declared_in"] == "myresearch/pipeline.py"
|
|
|
|
# Nothing differs from what the flow itself would have stored.
|
|
assert sync([flows["train"]], api, origin=ORIGIN)[0].unchanged
|
|
|
|
|
|
def test_a_node_carries_the_modules_its_code_reaches(api, flows):
|
|
"""The shim names one function; the cache has to key on what it calls."""
|
|
sync([flows["train"]], api, origin=ORIGIN)
|
|
|
|
nodes = {n["id"]: n for n in api.get_flow("train")["definition"]["nodes"]}
|
|
files = nodes["fit"]["code_files"]
|
|
|
|
# The module the function is in, and the package whose `__init__` runs on
|
|
# the way to it. Not the engine's own code, and not the standard library.
|
|
assert "myresearch.train" in files
|
|
assert all(path.endswith(".py") for path in files.values())
|
|
assert not any(name.startswith("fluksio") for name in files)
|
|
assert len(nodes["fit"]["code_digest"]) == 64
|
|
|
|
|
|
def test_a_node_whose_helper_moved_is_named_by_sync():
|
|
"""Editing a helper changes no shim, so nothing used to say it happened."""
|
|
from fluksio.sdk.client import _moved_code
|
|
|
|
stored = {
|
|
"definition": {
|
|
"nodes": [
|
|
{"id": "fit", "code_digest": "aaa"},
|
|
{"id": "prepare", "code_digest": "bbb"},
|
|
]
|
|
}
|
|
}
|
|
document = {
|
|
"nodes": [
|
|
{"id": "fit", "code_digest": "ccc"},
|
|
{"id": "prepare", "code_digest": "bbb"},
|
|
]
|
|
}
|
|
|
|
assert _moved_code(stored, document) == ["fit"]
|
|
# Nothing to compare against is not a change.
|
|
assert _moved_code(None, document) == []
|
|
|
|
|
|
def test_an_engine_that_records_no_digest_is_not_a_change_every_time():
|
|
"""It drops the field on parse, so comparing against it reports for ever.
|
|
|
|
Which is what an engine older than this client does, and what a node
|
|
drawn on the canvas looks like — neither is a helper somebody edited.
|
|
"""
|
|
from fluksio.sdk.client import _moved_code
|
|
|
|
older = {"definition": {"nodes": [{"id": "fit"}, {"id": "prepare"}]}}
|
|
document = {
|
|
"nodes": [
|
|
{"id": "fit", "code_digest": "ccc"},
|
|
{"id": "prepare", "code_digest": ""},
|
|
]
|
|
}
|
|
|
|
assert _moved_code(older, document) == []
|
|
|
|
|
|
def test_a_second_sync_changes_nothing_and_commits_nothing(api, flows):
|
|
sync([flows["train"]], api, origin=ORIGIN)
|
|
before = commits()
|
|
|
|
reports = sync([flows["train"]], api, origin=ORIGIN)
|
|
|
|
assert reports[0].unchanged
|
|
assert commits() == before
|
|
|
|
|
|
def test_sync_refuses_to_overwrite_a_canvas_edit(api, flows):
|
|
sync([flows["train"]], api, origin=ORIGIN)
|
|
api.put_source("train", "evaluate", "def process(weights):\n return {}\n")
|
|
|
|
with pytest.raises(SyncError, match="edited on the canvas"):
|
|
sync([flows["train"]], api, origin=ORIGIN)
|
|
|
|
reports = sync([flows["train"]], api, origin=ORIGIN, force=True)
|
|
assert "evaluate" in reports[0].changed
|
|
assert api.get_source("train", "evaluate").startswith(MARKER)
|
|
|
|
|
|
def test_an_interrupted_sync_finishes_on_the_next_one(api, flows):
|
|
"""A sync that died between the document and the bodies must be resumable.
|
|
|
|
That state — nodes named by a stored flow, with no source written for them
|
|
yet — used to read as a canvas edit, because the engine answers a new-node
|
|
template for a node with no body and the template carries no marker. The
|
|
advice was `--force`, for a canvas nobody had touched.
|
|
"""
|
|
target = flows["train"]
|
|
target.name = "resumed"
|
|
try:
|
|
# Exactly what the document PUT leaves behind before any body is written.
|
|
api.put_flow(target.document(ORIGIN) | {"version": 1})
|
|
assert api.get_source_entry("resumed", "fit")["missing"]
|
|
|
|
reports = sync([target], api, origin=ORIGIN)
|
|
|
|
assert "fit" in reports[0].changed
|
|
assert api.get_source("resumed", "fit").startswith(MARKER)
|
|
assert not api.get_source_entry("resumed", "fit")["missing"]
|
|
assert not api.get_flow("resumed")["has_draft"]
|
|
finally:
|
|
target.name = "train"
|
|
|
|
|
|
def test_sync_refuses_a_flow_it_did_not_create(api, flows):
|
|
api.put_flow({"name": "drawn", "version": 1, "nodes": [], "mode": "batch"})
|
|
api.publish("drawn", 1)
|
|
theirs = flows["train"]
|
|
theirs.name = "drawn"
|
|
|
|
try:
|
|
with pytest.raises(SyncError, match="not created by sync"):
|
|
sync([theirs], api, origin=ORIGIN)
|
|
finally:
|
|
theirs.name = "train"
|
|
|
|
|
|
def test_use_stores_the_rewiring_it_was_given(api, flows):
|
|
sync([flows["finetune"]], api, origin=ORIGIN)
|
|
|
|
nodes = {n["id"]: n for n in api.get_flow("finetune")["definition"]["nodes"]}
|
|
assert [p["name"] for p in nodes["fit"]["requires"]] == ["augmented", "lr"]
|
|
assert nodes["fit"]["params"] == {"epochs": 3}
|
|
|
|
|
|
def test_refresh_retires_the_workers(client, superuser_token_headers):
|
|
response = client.post(f"{PREFIX}/modules/refresh", headers=superuser_token_headers)
|
|
|
|
assert response.status_code == 200
|
|
assert "afresh" in response.json()["message"]
|