feat(cli): add decnet swarm check wrapper for POST /swarm/check

The swarmctl API already exposes POST /swarm/check — an active mTLS
probe that refreshes SwarmHost.status + last_heartbeat for every
enrolled worker. The CLI was missing a wrapper, so operators had to
curl the endpoint directly (which is how the VM validation run did it,
and how the wiki Deployment-Modes / SWARM-Mode pages ended up doc'ing
a command that didn't exist yet).

Matches the existing list/enroll/decommission pattern: typer subcommand
under swarm_app, --url override, Rich table output plus --json for
scripting. Three tests: populated table, empty-swarm path, and --json
emission.
This commit is contained in:
2026-04-18 20:28:34 -04:00
parent 3da5a2c4ee
commit 411a797120
2 changed files with 81 additions and 0 deletions

View File

@@ -455,6 +455,45 @@ def swarm_list(
console.print(table)
@swarm_app.command("check")
def swarm_check(
url: Optional[str] = typer.Option(None, "--url", help="Override swarm controller URL"),
json_out: bool = typer.Option(False, "--json", help="Emit JSON instead of a table"),
) -> None:
"""Actively probe every enrolled worker and refresh status + last_heartbeat."""
resp = _http_request("POST", _swarmctl_base_url(url) + "/swarm/check", timeout=60.0)
payload = resp.json()
results = payload.get("results", [])
if json_out:
console.print_json(data=payload)
return
if not results:
console.print("[dim]No workers enrolled.[/]")
return
table = Table(title="DECNET swarm check")
for col in ("name", "address", "reachable", "detail"):
table.add_column(col)
for r in results:
reachable = r.get("reachable")
mark = "[green]yes[/]" if reachable else "[red]no[/]"
detail = r.get("detail")
detail_str = ""
if isinstance(detail, dict):
detail_str = detail.get("status") or ", ".join(f"{k}={v}" for k, v in detail.items())
elif detail is not None:
detail_str = str(detail)
table.add_row(
r.get("name") or "",
r.get("address") or "",
mark,
detail_str,
)
console.print(table)
@swarm_app.command("decommission")
def swarm_decommission(
name: Optional[str] = typer.Option(None, "--name", help="Worker hostname"),

View File

@@ -110,6 +110,48 @@ def test_swarm_enroll_writes_bundle(http_stub, tmp_path: pathlib.Path) -> None:
assert body["sans"] == ["decky01.lan", "10.0.0.1"]
# ------------------------------------------------------------- swarm check
def test_swarm_check_prints_table(http_stub) -> None:
http_stub.script[("POST", "/swarm/check")] = _FakeResp({
"results": [
{"host_uuid": "u-a", "name": "decky01", "address": "10.0.0.1",
"reachable": True, "detail": {"status": "ok"}},
{"host_uuid": "u-b", "name": "decky02", "address": "10.0.0.2",
"reachable": False, "detail": "connection refused"},
]
})
result = runner.invoke(app, ["swarm", "check"])
assert result.exit_code == 0, result.output
assert "decky01" in result.output
assert "decky02" in result.output
# Both reachable=true and reachable=false render.
assert "yes" in result.output.lower()
assert "no" in result.output.lower()
def test_swarm_check_empty(http_stub) -> None:
http_stub.script[("POST", "/swarm/check")] = _FakeResp({"results": []})
result = runner.invoke(app, ["swarm", "check"])
assert result.exit_code == 0
assert "No workers" in result.output
def test_swarm_check_json_output(http_stub) -> None:
http_stub.script[("POST", "/swarm/check")] = _FakeResp({
"results": [
{"host_uuid": "u-a", "name": "decky01", "address": "10.0.0.1",
"reachable": True, "detail": {"status": "ok"}},
]
})
result = runner.invoke(app, ["swarm", "check", "--json"])
assert result.exit_code == 0
# JSON mode emits structured output, not the rich table.
assert '"reachable"' in result.output
assert '"decky01"' in result.output
# ------------------------------------------------------------- swarm decommission