tests/test_cli_graph.py
Rama main · réplica de lectura del repositorio original.
"""Tests for `job graph` cross-project DAG view."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import pytest
from typer.testing import CliRunner
class _FakeResp:
def __init__(self, data):
self._data = data
def raise_for_status(self):
pass
def json(self):
return self._data
def _now_iso(offset_seconds: int = 0) -> str:
return (datetime.now(UTC) + timedelta(seconds=offset_seconds)).isoformat()
def _job(
job_id: int,
*,
state: str = "completed",
project: str = "project-c",
cmd: list[str] | None = None,
depends_on: list[int] | None = None,
age_seconds: int = 60,
) -> dict:
return {
"id": job_id,
"state": state,
"project": project,
"cmd": cmd or ["python", "-m", "project-c"],
"depends_on": depends_on or [],
"submitted_at": _now_iso(-age_seconds),
}
# ---------------------------------------------------------------------------
# _parse_since
# ---------------------------------------------------------------------------
def test_parse_since_hours():
from job_cli.cli import _parse_since
assert _parse_since("24h") == timedelta(hours=24)
def test_parse_since_days():
from job_cli.cli import _parse_since
assert _parse_since("3d") == timedelta(days=3)
def test_parse_since_weeks():
from job_cli.cli import _parse_since
assert _parse_since("2w") == timedelta(weeks=2)
def test_parse_since_rejects_garbage():
import typer
from job_cli.cli import _parse_since
with pytest.raises(typer.BadParameter):
_parse_since("forever")
def test_parse_since_rejects_zero():
import typer
from job_cli.cli import _parse_since
with pytest.raises(typer.BadParameter):
_parse_since("0h")
# ---------------------------------------------------------------------------
# _filter_recent
# ---------------------------------------------------------------------------
def test_filter_recent_drops_old():
from job_cli.cli import _filter_recent
jobs = [
_job(1, age_seconds=60), # recent
_job(2, age_seconds=3600 * 48), # 2d old
]
out = _filter_recent(jobs, timedelta(hours=24))
assert [j["id"] for j in out] == [1]
def test_filter_recent_keeps_all_when_window_wide():
from job_cli.cli import _filter_recent
jobs = [_job(1, age_seconds=3600 * 48), _job(2, age_seconds=60)]
out = _filter_recent(jobs, timedelta(weeks=4))
assert {j["id"] for j in out} == {1, 2}
# ---------------------------------------------------------------------------
# _orphan_ids
# ---------------------------------------------------------------------------
def test_orphan_ids_flags_child_of_failed_parent():
from job_cli.cli import _orphan_ids
jobs = [
_job(1, state="failed"),
_job(2, state="cancelled", depends_on=[1]),
]
assert _orphan_ids(jobs) == {2}
def test_orphan_ids_does_not_flag_completed_chain():
from job_cli.cli import _orphan_ids
jobs = [
_job(1, state="completed"),
_job(2, state="completed", depends_on=[1]),
]
assert _orphan_ids(jobs) == set()
def test_orphan_ids_flags_child_of_preempted_or_cancelled():
from job_cli.cli import _orphan_ids
jobs = [
_job(1, state="preempted"),
_job(2, state="cancelled", depends_on=[1]),
_job(3, state="cancelled"),
_job(4, state="cancelled", depends_on=[3]),
]
assert _orphan_ids(jobs) == {2, 4}
def test_orphan_ids_unknown_parent_not_orphan():
"""If the parent isn't in the filtered window, don't speculate."""
from job_cli.cli import _orphan_ids
jobs = [_job(2, state="cancelled", depends_on=[999])]
assert _orphan_ids(jobs) == set()
# ---------------------------------------------------------------------------
# _render_ascii
# ---------------------------------------------------------------------------
def test_render_ascii_shows_root_and_indented_child():
from job_cli.cli import _render_ascii
jobs = [
_job(10, state="completed", project="project-c", cmd=["fit_sweep"]),
_job(
11,
state="completed",
project="project-c",
cmd=["evaluate"],
depends_on=[10],
),
]
out = _render_ascii(jobs, orphan_ids=set())
lines = out.splitlines()
# Root appears before child; child is indented
root_idx = next(i for i, ln in enumerate(lines) if "10" in ln and "fit_sweep" in ln)
child_idx = next(i for i, ln in enumerate(lines) if "11" in ln and "evaluate" in ln)
assert root_idx < child_idx
assert lines[child_idx].startswith(" ") # indented
def test_render_ascii_marks_orphan():
from job_cli.cli import _render_ascii
jobs = [
_job(20, state="failed", project="project-c", cmd=["fit"]),
_job(
21,
state="cancelled",
project="project-c",
cmd=["downstream"],
depends_on=[20],
),
]
out = _render_ascii(jobs, orphan_ids={21})
assert "orphaned" in out.lower() or "parent_failed" in out.lower()
def test_render_ascii_handles_multi_parent_fanin():
"""A job depending on two parents is referenced under the second parent."""
from job_cli.cli import _render_ascii
jobs = [
_job(30, project="a", cmd=["a"]),
_job(31, project="b", cmd=["b"]),
_job(32, project="c", cmd=["c"], depends_on=[30, 31]),
]
out = _render_ascii(jobs, orphan_ids=set())
# "32" appears at least twice (once full, once as a fan-in ref) OR
# the output contains an explicit fan-in marker; we accept either.
occurrences = sum(1 for ln in out.splitlines() if "32" in ln)
assert occurrences >= 2 or "fan-in" in out.lower() or "…" in out
# ---------------------------------------------------------------------------
# _render_dot
# ---------------------------------------------------------------------------
def test_render_dot_emits_digraph_header():
from job_cli.cli import _render_dot
jobs = [_job(1)]
out = _render_dot(jobs, orphan_ids=set())
assert out.lstrip().startswith("digraph")
assert "}" in out
def test_render_dot_emits_edge_for_dependency():
from job_cli.cli import _render_dot
jobs = [_job(1), _job(2, depends_on=[1])]
out = _render_dot(jobs, orphan_ids=set())
# Edge: 1 -> 2 (parent points to child)
assert "1 -> 2" in out or "1->2" in out
def test_render_dot_colors_by_state():
from job_cli.cli import _render_dot
jobs = [_job(1, state="completed"), _job(2, state="failed")]
out = _render_dot(jobs, orphan_ids=set())
# Both nodes get a color attribute somewhere in their attribute block
assert "color=" in out
# Distinct colors for distinct states
assert "green" in out.lower()
assert "red" in out.lower()
# ---------------------------------------------------------------------------
# end-to-end CLI
# ---------------------------------------------------------------------------
def _patch_client(monkeypatch, jobs):
import job_cli.cli as cli_mod
class FakeClient:
def __enter__(self):
return self
def __exit__(self, *a):
pass
def get(self, path, params=None):
if path == "/jobs":
# Mirror the real broker's server-side filters (see jobd.app /jobs).
filtered = jobs
if params and params.get("project"):
filtered = [j for j in filtered if j["project"] == params["project"]]
if params and params.get("state_filter"):
filtered = [j for j in filtered if j["state"] == params["state_filter"]]
return _FakeResp(filtered)
if path == "/workers":
return _FakeResp([])
return _FakeResp([])
monkeypatch.setattr(cli_mod, "_client", lambda: FakeClient())
return cli_mod
def test_graph_command_renders_ascii_by_default(monkeypatch):
cli_mod = _patch_client(
monkeypatch,
[
_job(100, project="project-c", cmd=["fit_sweep"]),
_job(101, project="project-c", cmd=["evaluate"], depends_on=[100]),
],
)
r = CliRunner().invoke(cli_mod.app, ["graph"])
assert r.exit_code == 0, r.stdout
assert "100" in r.stdout
assert "101" in r.stdout
def test_graph_command_filters_by_project(monkeypatch):
cli_mod = _patch_client(
monkeypatch,
[
_job(200, project="project-c", cmd=["a"]),
_job(201, project="project-a", cmd=["b"]),
],
)
r = CliRunner().invoke(cli_mod.app, ["graph", "--project", "project-c"])
assert r.exit_code == 0, r.stdout
assert "200" in r.stdout
assert "201" not in r.stdout
def test_graph_command_format_dot(monkeypatch):
cli_mod = _patch_client(monkeypatch, [_job(300)])
r = CliRunner().invoke(cli_mod.app, ["graph", "--format", "dot"])
assert r.exit_code == 0, r.stdout
assert "digraph" in r.stdout
def test_graph_command_since_drops_old_jobs(monkeypatch):
cli_mod = _patch_client(
monkeypatch,
[
_job(400, age_seconds=60),
_job(401, age_seconds=3600 * 48),
],
)
r = CliRunner().invoke(cli_mod.app, ["graph", "--since", "24h"])
assert r.exit_code == 0, r.stdout
assert "400" in r.stdout
assert "401" not in r.stdout
def test_graph_command_rejects_bad_since(monkeypatch):
cli_mod = _patch_client(monkeypatch, [_job(500)])
r = CliRunner().invoke(cli_mod.app, ["graph", "--since", "forever"])
assert r.exit_code != 0