tegwick
6d3414d8b7
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
import shutil
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from rtut import pack as P, run as R
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
@pytest.fixture
|
|
def pk(tmp_path):
|
|
d = tmp_path / "runbooks" / "hello"
|
|
d.mkdir(parents=True)
|
|
shutil.copy(ROOT / "runbooks/example-hello/pack.yaml", d / "pack.yaml")
|
|
return P.load(d / "pack.yaml")
|
|
|
|
|
|
def test_create_and_resume(pk):
|
|
r = R.create(pk, "bernd", {"host": "example.org"})
|
|
assert r.data["params"] == {"host": "example.org", "count": "1"}
|
|
assert R.list_runs(pk) == [r.data["run_id"]]
|
|
assert R.load(pk, r.data["run_id"]).data == r.data
|
|
assert (pk.path.parents[2] / ".local/runbook-runs").is_dir()
|
|
|
|
|
|
def test_bad_parameters_and_operator(pk):
|
|
with pytest.raises(R.RunError):
|
|
R.create(pk, "bernd", {"host": "a;b"})
|
|
with pytest.raises(R.RunError):
|
|
R.create(pk, "bernd", {"nope": "x"})
|
|
with pytest.raises(R.RunError):
|
|
R.create(pk, "two words")
|
|
with pytest.raises(R.RunError):
|
|
R.create(pk, "hvs.ABCDEFGHIJKLMN")
|
|
|
|
|
|
def test_full_run_and_exercise(pk):
|
|
r = R.create(pk, "bernd", {"host": "localhost"})
|
|
assert R.command_for(r, "ping") == "ping -c 1 localhost"
|
|
R.start(r, "ping")
|
|
with pytest.raises(R.RunError): # command needs confirmation
|
|
R.verify(r, "ping")
|
|
with pytest.raises(R.RunError): # cannot complete without passing check
|
|
R.complete(r, "ping")
|
|
res = R.verify(r, "ping", confirm=True)
|
|
assert res["kind"] == "exit-0" and "output_bytes" in res
|
|
assert "stdout" not in res and "output" not in res
|
|
if not res["passed"]:
|
|
pytest.skip("ping unavailable in this environment")
|
|
R.complete(r, "ping")
|
|
assert r.complete
|
|
assert pk.path.read_text() == P.load(pk.path).path.read_text() # the engine never writes a pack
|
|
|
|
|
|
def test_order_fail_retry_skip(pk):
|
|
r = R.create(pk, "bernd")
|
|
with pytest.raises(R.RunError):
|
|
R.fail(r, "ping", "x") # not running
|
|
R.start(r, "ping")
|
|
R.fail(r, "ping", "no route")
|
|
assert r.state("ping")["state"] == "failed"
|
|
R.retry(r, "ping")
|
|
assert r.state("ping")["state"] == "running" and "verification" not in r.state("ping")
|
|
with pytest.raises(R.RunError):
|
|
R.fail(r, "ping", "-----BEGIN RSA PRIVATE KEY-----")
|
|
r2 = R.create(pk, "bernd")
|
|
R.skip(r2, "ping", "not needed")
|
|
with pytest.raises(R.RunError, match="succeeds only when every step is done"):
|
|
R.finalize(r2, "succeeded") # a skipped step never counts as done
|
|
|
|
|
|
def test_manual_verification_needs_answer(pk):
|
|
r = R.create(pk, "bernd")
|
|
pk.data["steps"][0]["verify"] = {"done_when": "checked by eye", "expect": "manual"}
|
|
R.start(r, "ping")
|
|
with pytest.raises(R.RunError):
|
|
R.verify(r, "ping")
|
|
assert R.verify(r, "ping", manual_ok=True)["passed"]
|