Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
304c90ea76b23e3ffc8d6366be97690d4e682a75
app/backend/tests/api/routes/test_runs.py
T
stroblmeandClaude Opus 5 4479eeb726
Docs / docs (push) Successful in 38s
Details
Playwright Tests / test-playwright (1, 2) (push) Successful in 2m56s
Details
Playwright Tests / test-playwright (2, 2) (push) Successful in 2m7s
Details
pre-commit / pre-commit (push) Failing after 2m17s
Details
Test Backend / test-backend (push) Successful in 2m54s
Details
Compose Smoke Test / test-compose (push) Successful in 44s
Details
Playwright Tests / merge-reports (push) Successful in 1m17s
Details
Follow a record into its fields, name the metrics, name the version 
Three things the first export pass got wrong for a real study.

**Dotted paths.** A node returns a record, not a scalar — the numbers arrive
inside `final_metrics` — so `--metrics final_metrics.train_loss` yielded an
empty column and `--metrics final_metrics` yielded the whole record in one
cell. Both sides of the wide table now take dotted paths, and the defaults
reach the same depth: every number a result carries is a column named by its
path, and inputs are compared leaf by leaf, so two configurations differing in
one field give that field as the axis rather than two blobs that are merely
not equal. Lists stay whole — a curve belongs in the long table.

**`--list`.** Metric names are flow-qualified, so `--name train_loss` matched
nothing and said only that. `fluksio export metrics --list` prints the names
the selection carries, and an empty export made with `--name` points at it.

**A version to compare.** The CLI ships ahead of the engine and a stale one
answered a flat 404 with nothing anywhere in the API to tell how old it was.
The engine reports `version` on `/observability/summary`, `fluksio status`
prints it, and a 404 from export now names both versions — or says "older"
when the field itself predates the engine. Bumped to 0.1.5, which is what
makes the number worth reading.

Also formats `flow/metrics.py`, which had been committed unformatted and was
the last `ruff format --check` failure.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A9Hdrmf2cwNABCnE5x9UJa
2026-08-27 20:42:19 +02:00

816 lines
28 KiB
Python
Raw Blame History

"""The stage cache, from the side that needs a database.
The pipeline half — what a hit restores and what a key is made of — is in
`tests/flow/test_runs.py`, which runs without one.
"""
import json
from datetime import UTC, datetime, timedelta
import pytest
from sqlmodel import Session, col, select
from fluksio.core.config import settings
from fluksio.core.db import engine as db_engine
from fluksio.flow.artifacts import ArtifactStore
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.pipeline import NodeOutcome
from fluksio.flow.runs import (
OUTPUT_CAP,
RunCache,
RunRejected,
RunService,
_cacheable,
new_run_id,
resolve_references,
seed_values,
)
from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef
from fluksio.models import Run, RunArtifact, RunMetric, RunNode
def test_a_run_cache_finds_what_an_earlier_run_recorded(tmp_path):
"""The run history is the cache; there is no second store to keep."""
store = ArtifactStore(tmp_path / "artifacts")
reference = store.put([b"payload"], name="data.bin")
plain, with_artifact, collected = "k-plain", "k-artifact", "k-collected"
with Session(db_engine) as session:
session.add(
RunNode(
run_id="cache-1",
node="study.a",
status="ok",
cache_key=plain,
outputs=json.dumps({"study.loss": 1.5}),
)
)
session.add(
RunNode(
run_id="cache-2",
node="study.b",
status="ok",
cache_key=with_artifact,
outputs=json.dumps({"study.data": reference}),
)
)
session.add(
RunNode(
run_id="cache-3",
node="study.c",
status="ok",
cache_key=collected,
outputs=json.dumps(
{"study.data": {**reference, "digest": "sha256:" + "1" * 64}}
),
)
)
session.commit()
cache = RunCache(store)
hit = cache.lookup(plain)
assert hit is not None and hit.outputs == {"study.loss": 1.5}
# Where its series is, so a run reusing it can read the curve back.
assert hit.metrics_run == "cache-1"
found = cache.lookup(with_artifact)
assert found is not None and found.outputs == {"study.data": reference}
# Its bytes have gone from the store, so the reference names nothing a
# restored run could open. That is a miss, not a broken run.
assert cache.lookup(collected) is None
assert cache.lookup("never-seen") is None
assert cache.lookup("") is None
def test_what_may_be_stored_as_a_cache_entry():
"""A row carries a key and its outputs together, or neither."""
ok = NodeOutcome(
node="study.a", ok=True, cache_key="k", output_values={"study.loss": 1.0}
)
assert _cacheable(ok) == '{"study.loss":1.0}'
# A node that published nothing is still an answer worth reusing.
assert _cacheable(ok.model_copy(update={"output_values": None})) == "null"
# Not cacheable: it failed, it has no key, or it returned too much.
assert _cacheable(ok.model_copy(update={"ok": False})) is None
assert _cacheable(ok.model_copy(update={"cache_key": ""})) is None
big = {"study.data": "x" * (OUTPUT_CAP + 1)}
assert _cacheable(ok.model_copy(update={"output_values": big})) is None
# -----------------------------------------------------------------------------
# Naming an artifact from outside the process that made it
#
# A python caller passes the reference it holds. A shell holds nothing, so the
# same input also takes `@run:<id>.<output>` or a bare digest, resolved here
# rather than in each client.
# -----------------------------------------------------------------------------
def artifact_flow() -> FlowDef:
"""A flow taking a dataset somebody else's run produced."""
dataset = MessageSpec(name="dataset", dtype=DType.ARTIFACT)
return FlowDef(
name="study",
mode="batch",
inputs=[FlowInput(spec=dataset)],
nodes=[NodeDef(id="train", requires=[dataset])],
)
@pytest.fixture
def made_artifact():
"""A finished run with one artifact, as a later run would find it."""
digest = "sha256:" + "a1" * 32
reference = {
"digest": digest,
"size": 12,
"media_type": "text/csv",
"name": "cities.csv",
}
run_id = new_run_id()
with Session(db_engine) as session:
session.add(
Run(
id=run_id,
flow="prepare",
status="ok",
result={"dataset": reference},
created_at=datetime.now(UTC),
)
)
session.add(
RunArtifact(
run_id=run_id,
name="prepare.dataset",
filename="cities.csv",
node="load",
digest=digest,
size=12,
)
)
session.commit()
yield run_id, reference
with Session(db_engine) as session:
session.delete(session.get(RunArtifact, (run_id, "prepare.dataset")))
session.delete(session.get(Run, run_id))
session.commit()
def test_a_run_reference_resolves_to_what_that_run_produced(made_artifact):
run_id, reference = made_artifact
resolved = resolve_references(
artifact_flow(), {"dataset": f"@run:{run_id}.dataset"}
)
# The producer's own reference, whatever type it made it — the row beside
# it is the fallback, not the first answer.
assert resolved["dataset"] == reference
def test_the_artifact_row_answers_with_the_name_the_node_gave_the_file(made_artifact):
"""Bytes the flow never declared as an output are reachable through the row."""
run_id, reference = made_artifact
with Session(db_engine) as session:
run = session.get(Run, run_id)
run.result = {}
session.add(run)
session.commit()
resolved = resolve_references(
artifact_flow(), {"dataset": f"@run:{run_id}.dataset"}
)
assert resolved["dataset"]["digest"] == reference["digest"]
# The file name, not the message it happened to leave on.
assert resolved["dataset"]["name"] == "cities.csv"
def test_a_bare_digest_resolves_to_the_bytes_under_it(made_artifact):
_run_id, reference = made_artifact
resolved = resolve_references(artifact_flow(), {"dataset": reference["digest"]})
assert resolved["dataset"]["digest"] == reference["digest"]
assert resolved["dataset"]["size"] == 12
def test_a_resolved_reference_passes_the_input_check(made_artifact):
run_id, _reference = made_artifact
flow = artifact_flow()
resolved = resolve_references(flow, {"dataset": f"@run:{run_id}.dataset"})
assert "study.dataset" in seed_values(flow, resolved)
def test_an_output_a_run_never_made_says_what_it_did(made_artifact):
run_id, _reference = made_artifact
with pytest.raises(RunRejected, match="prepare.dataset"):
resolve_references(artifact_flow(), {"dataset": f"@run:{run_id}.weights"})
def test_a_reference_to_no_run_at_all_is_refused():
with pytest.raises(RunRejected, match="no run"):
resolve_references(artifact_flow(), {"dataset": "@run:nothing.dataset"})
def test_an_unknown_digest_is_refused():
with pytest.raises(RunRejected, match="nothing here"):
resolve_references(artifact_flow(), {"dataset": "sha256:" + "b2" * 32})
def test_a_reference_passed_whole_is_left_alone(made_artifact):
"""A python caller already has the object, and hands it over as one."""
_run_id, reference = made_artifact
assert resolve_references(artifact_flow(), {"dataset": reference}) == {
"dataset": reference
}
def chaining_flow() -> FlowDef:
"""A flow taking a json config and a label another run worked out."""
meta = MessageSpec(name="meta", dtype=DType.JSON)
label = MessageSpec(name="label", dtype=DType.STR)
return FlowDef(
name="study",
mode="batch",
inputs=[FlowInput(spec=meta), FlowInput(spec=label)],
nodes=[NodeDef(id="train", requires=[meta, label])],
)
@pytest.fixture
def made_config():
"""A finished run whose result is an object, not bytes."""
meta = {"rows": 256, "source": "builtin"}
run_id = new_run_id()
with Session(db_engine) as session:
session.add(
Run(
id=run_id,
flow="generate",
status="ok",
result={"meta": meta, "label": "run-7"},
created_at=datetime.now(UTC),
)
)
session.commit()
yield run_id, meta
with Session(db_engine) as session:
session.delete(session.get(Run, run_id))
session.commit()
def test_a_json_input_may_name_a_run_s_output(made_config):
"""The gap this closes: chaining without pasting the object into a shell."""
run_id, meta = made_config
flow = chaining_flow()
resolved = resolve_references(flow, {"meta": f"@run:{run_id}.meta"})
assert resolved["meta"] == meta
# And it is the value's own type from here on, so the input check passes.
assert seed_values(flow, resolved)["study.meta"] == meta
def test_the_spelling_is_reserved_on_a_text_input_too(made_config):
run_id, _meta = made_config
resolved = resolve_references(chaining_flow(), {"label": f"@run:{run_id}.label"})
assert resolved["label"] == "run-7"
def test_text_that_names_nothing_is_still_left_alone(made_config):
"""Only the two spellings are read as names; everything else is a value."""
_run_id, _meta = made_config
params = {"label": "@run-of-the-mill", "meta": {"rows": 1}}
assert resolve_references(chaining_flow(), params) == params
def test_the_overview_counts_a_flow_the_list_page_would_not_reach(
client, superuser_token_headers
):
"""The list caps at 500 newest; the flow rail needs whole counts."""
with Session(db_engine) as session:
for index in range(3):
session.add(
Run(
id=f"ov-{index}",
flow="overviewed",
status="ok" if index else "running",
created_at=datetime(2026, 1, 1 + index, tzinfo=UTC),
)
)
session.commit()
rows = client.get(
f"{settings.API_V1_STR}/runs/overview", headers=superuser_token_headers
).json()
row = next(r for r in rows if r["flow"] == "overviewed")
assert (row["runs"], row["running"], row["queued"]) == (3, 1, 0)
def test_a_running_run_reports_how_long_it_has_been_going(
client, superuser_token_headers
):
"""A duration is only written at the end; until then, time since it began."""
with Session(db_engine) as session:
session.add(
Run(
id="in-flight",
flow="timed",
status="running",
created_at=datetime.now(UTC),
started_at=datetime.now(UTC) - timedelta(seconds=30),
)
)
session.commit()
rows = client.get(
f"{settings.API_V1_STR}/runs",
headers=superuser_token_headers,
params={"flow": "timed"},
).json()
assert rows[0]["duration_ms"] >= 30_000
def test_a_run_records_which_caller_asked_for_it(
client, superuser_token_headers, monkeypatch
):
"""The dashboard, the CLI and the SDK are told apart by what they send.
Client-supplied, so the vocabulary is closed: a column nobody can write
free text into is one a table can group by.
"""
seen: dict[str, object] = {}
class Recorder:
def submit(self, name, **kwargs):
seen.update(kwargs)
return Run(
id="cause-1",
flow=name,
cause=str(kwargs["cause"]),
created_at=datetime.now(UTC),
)
monkeypatch.setattr(client.app.state, "run_service", Recorder())
url = f"{settings.API_V1_STR}/runs/flows/demo"
answer = client.post(url, headers=superuser_token_headers, json={"cause": "cli"})
assert answer.status_code == 202
assert (seen["cause"], answer.json()["cause"]) == ("cli", "cli")
# Nothing said still means the dashboard, which is the only caller that
# does not name itself.
client.post(url, headers=superuser_token_headers, json={})
assert seen["cause"] == "api"
refused = client.post(
url, headers=superuser_token_headers, json={"cause": "somewhere else"}
)
assert refused.status_code == 422
class _Unusable:
"""Anything reaching this is something a deduplicated submit should not do."""
def __getattr__(self, name):
raise AssertionError(f"a repeated submit must not reach {name}")
def test_a_repeated_submit_returns_the_run_it_already_made():
"""The key is the answer to "did my first attempt land?".
Answered before the flow is even read: a caller retrying a submit it never
got a reply to is owed that run, whatever has been published since.
"""
service = RunService(controller=_Unusable(), queue=_Unusable())
with Session(db_engine) as session:
session.add(
Run(
id="dedup-1",
flow="study",
status="running",
idempotency_key="key-abc",
created_at=datetime.now(UTC),
)
)
session.commit()
again = service.submit("study", {"lr": 0.1}, idempotency_key="key-abc")
assert again.id == "dedup-1"
def test_a_key_nobody_used_submits_normally(
client, superuser_token_headers, monkeypatch
):
"""The route carries the key through; without one nothing changes."""
seen: dict[str, object] = {}
class Recorder:
def submit(self, name, **kwargs):
seen.update(kwargs)
return Run(id="keyed-1", flow=name, created_at=datetime.now(UTC))
monkeypatch.setattr(client.app.state, "run_service", Recorder())
answer = client.post(
f"{settings.API_V1_STR}/runs/flows/demo",
headers=superuser_token_headers,
json={"idempotency_key": "key-xyz"},
)
assert answer.status_code == 202
assert seen["idempotency_key"] == "key-xyz"
def test_overview_is_not_read_as_a_run_id(client, superuser_token_headers):
"""`/overview` is declared before `/{run_id}`, which would swallow it."""
answer = client.get(
f"{settings.API_V1_STR}/runs/overview", headers=superuser_token_headers
)
assert answer.status_code == 200
assert isinstance(answer.json(), list)
# -----------------------------------------------------------------------------
# A cached node's curve
#
# A hit replays no emissions, so the series stays in the run that recorded it
# and the run reusing it points there. Reading either one answers the same.
# -----------------------------------------------------------------------------
@pytest.fixture
def reused_run():
"""A run of `quick` whose node was restored from a run of `train`."""
with Session(db_engine) as session:
made = datetime.now(UTC)
session.add(Run(id="src-1", flow="train", status="ok", created_at=made))
session.add(Run(id="reuse-1", flow="quick", status="ok", created_at=made))
session.add(RunNode(run_id="src-1", node="train.fit", status="ok"))
session.add(
RunNode(
run_id="reuse-1",
node="quick.fit",
status="cached",
cached_from="src-1",
)
)
for step, value in enumerate([3.0, 2.0, 1.0]):
session.add(
RunMetric(
run_id="src-1",
name="train.loss",
step=step,
node="train.fit",
value=value,
)
)
session.commit()
yield
with Session(db_engine) as session:
for row in session.exec(select(RunMetric)).all():
session.delete(row)
for row in session.exec(select(RunNode)).all():
session.delete(row)
for run_id in ("src-1", "reuse-1"):
run = session.get(Run, run_id)
if run is not None:
session.delete(run)
session.commit()
@pytest.mark.usefixtures("reused_run")
def test_a_cached_node_answers_with_the_curve_it_was_restored_from(
client, superuser_token_headers
):
points = client.get(
f"{settings.API_V1_STR}/runs/reuse-1/metrics",
headers=superuser_token_headers,
).json()
# Named for the flow that asked, not the one that recorded it.
assert [point["name"] for point in points] == ["quick.loss"] * 3
assert [point["value"] for point in points] == [3.0, 2.0, 1.0]
named = client.get(
f"{settings.API_V1_STR}/runs/reuse-1/metrics",
params={"name": "quick.loss"},
headers=superuser_token_headers,
).json()
assert len(named) == 3
@pytest.mark.usefixtures("reused_run")
def test_a_curve_whose_run_is_gone_is_empty_rather_than_an_error(
client, superuser_token_headers
):
"""Deleting a flow deletes its runs; what pointed at one is left holding it."""
with Session(db_engine) as session:
session.delete(session.get(Run, "src-1"))
session.commit()
answer = client.get(
f"{settings.API_V1_STR}/runs/reuse-1/metrics",
headers=superuser_token_headers,
)
assert answer.status_code == 200
assert answer.json() == []
def _metric(run_id: str, name: str, step: int, value: float, ts: float) -> RunMetric:
return RunMetric(run_id=run_id, name=name, step=step, value=value, ts=ts)
@pytest.fixture
def plotted():
"""A run with a loss curve and an epoch counter beside it."""
run_id = new_run_id()
with Session(db_engine) as session:
session.add(
Run(id=run_id, flow="study", status="ok", created_at=datetime.now(UTC))
)
for step, (loss, epoch) in enumerate([(1.0, 10.0), (0.5, 20.0), (0.25, 30.0)]):
session.add(_metric(run_id, "study.loss", step, loss, 100.0 + step * 5))
session.add(_metric(run_id, "study.epoch", step, epoch, 100.0 + step * 5))
session.commit()
yield run_id
with Session(db_engine) as session:
for row in session.exec(
select(RunMetric).where(col(RunMetric.run_id) == run_id)
).all():
session.delete(row)
session.delete(session.get(Run, run_id))
session.commit()
def test_a_comparison_is_plotted_against_the_step_by_default(
client, superuser_token_headers, plotted
):
answer = client.get(
f"{settings.API_V1_STR}/runs/series/compare",
params={"ids": plotted, "metric": "study.loss"},
headers=superuser_token_headers,
).json()
assert answer["x"] == "step"
assert answer["lines"][0]["points"] == [[0.0, 1.0], [1.0, 0.5], [2.0, 0.25]]
def test_time_is_measured_from_this_runs_own_first_reading(
client, superuser_token_headers, plotted
):
"""Runs started hours apart still lie on top of each other."""
answer = client.get(
f"{settings.API_V1_STR}/runs/series/compare",
params={"ids": plotted, "metric": "study.loss", "x": "time"},
headers=superuser_token_headers,
).json()
assert answer["x"] == "time"
assert answer["lines"][0]["points"] == [[0.0, 1.0], [5.0, 0.5], [10.0, 0.25]]
def test_one_metric_can_be_plotted_against_another(
client, superuser_token_headers, plotted
):
answer = client.get(
f"{settings.API_V1_STR}/runs/series/compare",
params={"ids": plotted, "metric": "study.loss", "x": "study.epoch"},
headers=superuser_token_headers,
).json()
assert answer["lines"][0]["points"] == [[10.0, 1.0], [20.0, 0.5], [30.0, 0.25]]
def test_a_step_the_x_metric_never_reached_is_left_out(
client, superuser_token_headers, plotted
):
"""The join is on the step, which is the only thing two series share."""
with Session(db_engine) as session:
session.add(_metric(plotted, "study.loss", 3, 0.1, 120.0))
session.commit()
answer = client.get(
f"{settings.API_V1_STR}/runs/series/compare",
params={"ids": plotted, "metric": "study.loss", "x": "study.epoch"},
headers=superuser_token_headers,
).json()
assert [point[0] for point in answer["lines"][0]["points"]] == [10.0, 20.0, 30.0]
def test_a_run_records_the_code_it_started_with_not_the_code_it_was_queued_with():
"""A sweep queues every run at once and the tree moves while it waits.
`_restamp` is what runs at claim time, so the digest the record keeps is
the one the run actually executed.
"""
service = RunService(controller=_Unusable(), queue=_Unusable())
with Session(db_engine) as session:
session.add(
Run(
id="stamp-1",
flow="study",
status="queued",
code_digest="at-submit",
created_at=datetime.now(UTC),
)
)
session.commit()
run = session.get(Run, "stamp-1")
assert service._restamp(run, "at-claim") == "at-claim"
with Session(db_engine) as session:
assert session.get(Run, "stamp-1").code_digest == "at-claim"
def test_a_tree_that_did_not_move_is_not_written_again():
service = RunService(controller=_Unusable(), queue=_Unusable())
with Session(db_engine) as session:
session.add(
Run(
id="stamp-2",
flow="study",
status="queued",
code_digest="same",
created_at=datetime.now(UTC),
)
)
session.commit()
run = session.get(Run, "stamp-2")
assert service._restamp(run, "same") == "same"
# -----------------------------------------------------------------------------
# Export
#
# Two tables an analysis reads: the long one a curve is plotted from, and the
# wide one arms are compared in. Both carry the run id on every row.
# -----------------------------------------------------------------------------
@pytest.fixture
def exported():
"""Two runs of one flow: two curves each, and records on both sides.
The numbers a node returns are usually inside a record rather than at the
top of the result, so the fixture is shaped the way a real one is.
"""
made = datetime.now(UTC)
with Session(db_engine) as session:
for index, (lr, acc) in enumerate([(0.1, 0.5), (0.01, 0.25)]):
run_id = f"exp-{index}"
session.add(
Run(
id=run_id,
flow="export-study",
status="ok",
params={
"lr": lr,
"epochs": 10,
"config": {"model": "mlp", "depth": index + 1},
},
result={
"acc": acc,
"note": "n/a",
"final_metrics": {"train_loss": acc * 2},
"test_metrics": {"known": {"perfect": 1.0}},
},
created_at=made + timedelta(seconds=index),
)
)
for name in ("study.loss", "study.val"):
for step in range(4):
session.add(_metric(run_id, name, step, float(step), 100.0 + step))
session.commit()
yield ["exp-0", "exp-1"]
with Session(db_engine) as session:
for run_id in ("exp-0", "exp-1"):
for row in session.exec(
select(RunMetric).where(col(RunMetric.run_id) == run_id)
).all():
session.delete(row)
run = session.get(Run, run_id)
if run is not None:
session.delete(run)
session.commit()
def _lines(answer) -> list[dict]:
return [json.loads(line) for line in answer.text.splitlines() if line]
def test_an_export_strides_each_series_and_names_its_run(
client, superuser_token_headers, exported
):
"""Every second point of every curve — not every second row of all of them."""
answer = client.get(
f"{settings.API_V1_STR}/runs/export/metrics",
params={"ids": ",".join(exported), "stride": 2, "format": "jsonl"},
headers=superuser_token_headers,
)
assert answer.status_code == 200
rows = _lines(answer)
assert {row["run"] for row in rows} == set(exported)
curves: dict[tuple[str, str], list[int]] = {}
for row in rows:
curves.setdefault((row["run"], row["name"]), []).append(row["step"])
assert len(curves) == 4
assert all(steps == [0, 2] for steps in curves.values())
def test_an_exported_run_row_carries_the_inputs_that_vary(
client, superuser_token_headers, exported
):
"""The sweep axis becomes columns; what every run shares stays out of them."""
def export(**extra):
answer = client.get(
f"{settings.API_V1_STR}/runs/export/runs",
params={"ids": ",".join(exported), **extra},
headers=superuser_token_headers,
)
assert answer.status_code == 200
return answer
rows = _lines(export(format="jsonl"))
assert [row["id"] for row in rows] == ["exp-1", "exp-0"]
assert {row["param.lr"] for row in rows} == {0.1, 0.01}
# `epochs` is the same on both runs, so it is not what they differ by, and
# neither is the model inside the config — but the depth beside it is.
assert "param.epochs" not in rows[0]
assert "param.config.model" not in rows[0]
assert {row["param.config.depth"] for row in rows} == {1, 2}
assert "param.epochs" in _lines(export(format="jsonl", params="epochs"))[0]
# A number inside a record is a column of its own, however deep; a string
# is not one of the run's numbers wherever it sits.
assert rows[0]["metric.acc"] == 0.25
assert rows[0]["metric.final_metrics.train_loss"] == 0.5
assert rows[0]["metric.test_metrics.known.perfect"] == 1.0
assert "metric.note" not in rows[0]
named = _lines(export(format="jsonl", metrics="test_metrics.known.perfect"))
assert list(named[0])[-1] == "metric.test_metrics.known.perfect"
assert "metric.acc" not in named[0]
# csv is the default, and the columns are in the order the header names.
header = export().text.splitlines()[0]
assert header.startswith("id,flow,status,")
assert header.endswith(
"param.config.depth,param.lr,"
"metric.acc,metric.final_metrics.train_loss,"
"metric.test_metrics.known.perfect"
)
@pytest.fixture
def paged():
"""Three runs of one flow, a minute apart."""
made = datetime.now(UTC).replace(microsecond=0)
with Session(db_engine) as session:
for index in range(3):
session.add(
Run(
id=f"page-{index}",
flow="paged-study",
status="ok",
created_at=made + timedelta(minutes=index),
)
)
session.commit()
yield made
with Session(db_engine) as session:
for index in range(3):
run = session.get(Run, f"page-{index}")
if run is not None:
session.delete(run)
session.commit()
def test_runs_page_by_when_they_were_created(client, superuser_token_headers, paged):
"""`before` is the cursor: the last row's own timestamp reads the next page."""
def listed(**params):
answer = client.get(
f"{settings.API_V1_STR}/runs",
params={"flow": "paged-study", **params},
headers=superuser_token_headers,
)
assert answer.status_code == 200
return [row["id"] for row in answer.json()]
assert listed() == ["page-2", "page-1", "page-0"]
assert listed(before=(paged + timedelta(minutes=2)).isoformat()) == [
"page-1",
"page-0",
]
assert listed(since=(paged + timedelta(minutes=1)).isoformat()) == [
"page-2",
"page-1",
]
Reference in New Issue View Git Blame Copy Permalink