bvandeusenandClaude Opus 5.5
1d9ac2b46a
#5015 "Built-in host-agent playbooks make an MCP caller supply steward_url" - run_playbook, create_schedule and set_alert_action fill a steward-builtin playbook's steward_url when the caller leaves it out or empty, from Settings → General → Public base URL, else the scheme and host the MCP request arrived on — the same source the web actions use. - run_playbook's reply names what it filled (filled_by_steward), and get_playbook shows that value as the variable's default. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
135 lines
6.6 KiB
Python
135 lines
6.6 KiB
Python
"""The automation tools' pure parts and their gates: check mode by default,
|
|
secrets kept out of what is stored, roles, and refusals that run nothing.
|
|
|
|
A real check-mode run, applied run and audit rows are covered against
|
|
Postgres in tests/integration/test_mcp_automation.py.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
from types import SimpleNamespace as NS
|
|
|
|
import pytest
|
|
|
|
from steward.core.api_tokens import Caller
|
|
from steward.mcp.tools import automation
|
|
from steward.models.users import UserRole
|
|
|
|
VARS = [
|
|
{"name": "api_token", "default": "abc", "secret": True, "required": False, "prompt": None},
|
|
{"name": "greeting", "default": "hi", "secret": False, "required": False, "prompt": None},
|
|
{"name": "release", "default": "", "secret": False, "required": True, "prompt": "Release?"},
|
|
]
|
|
|
|
VIEWER_TOOLS = {"list_playbooks", "get_playbook", "list_inventory", "list_ansible_runs",
|
|
"get_ansible_run", "list_schedules", "wait_for_run"}
|
|
OPERATOR_TOOLS = {"run_playbook", "cancel_ansible_run", "create_schedule", "update_schedule",
|
|
"delete_schedule"}
|
|
|
|
|
|
def test_split_vars_keeps_secrets_apart_from_what_is_stored():
|
|
plain, secret = automation.split_vars(VARS, {"api_token": "s3cr3t", "greeting": "yo",
|
|
"release": "1.2", "other": 5})
|
|
assert plain == {"greeting": "yo", "release": "1.2", "other": 5}
|
|
assert secret == {"api_token": "s3cr3t"}
|
|
|
|
|
|
def test_split_vars_refuses_a_missing_prompt_and_odd_values():
|
|
with pytest.raises(ValueError, match="prompts for release.*Nothing was run"):
|
|
automation.split_vars(VARS, {"greeting": "yo"})
|
|
with pytest.raises(ValueError, match="string, number or boolean"):
|
|
automation.split_vars(VARS[:2], {"greeting": ["a"]})
|
|
|
|
|
|
def test_public_variables_mask_secret_defaults():
|
|
out = {v["name"]: v["default"] for v in automation.public_variables(VARS)}
|
|
assert out == {"api_token": "<secret>", "greeting": "hi", "release": ""}
|
|
|
|
|
|
def test_build_params():
|
|
assert automation.build_params({}, "", "", True) == {"check": True}
|
|
assert automation.build_params({"a": 1}, " web01 ", "deploy", False) == {
|
|
"extra_vars_map": {"a": 1}, "limit": "web01", "tags": "deploy"}
|
|
|
|
|
|
def test_writes_default_to_check_mode():
|
|
for fn in (automation.run_playbook, automation.create_schedule, automation.set_alert_action):
|
|
assert inspect.signature(fn).parameters["check"].default is True, fn.__name__
|
|
|
|
|
|
async def test_run_playbook_will_not_apply_a_confirm_playbook_unconfirmed(monkeypatch):
|
|
monkeypatch.setattr(automation, "_source", lambda name: {"name": name, "path": "/x"})
|
|
monkeypatch.setattr(automation, "_playbook",
|
|
lambda s, p: "# steward:confirm: true\n- hosts: all\n tasks: []\n")
|
|
|
|
def _no_run(*a, **k):
|
|
raise AssertionError("a run was started")
|
|
monkeypatch.setattr(automation, "trigger_run", _no_run)
|
|
with pytest.raises(ValueError, match="flagged destructive.*confirmed=true"):
|
|
await automation.run_playbook("s", "wipe.yml", check=False)
|
|
|
|
|
|
def test_recap_totals_sum_over_hosts():
|
|
results = {"hosts": {"a": {"ok": 3, "changed": 1}, "b": {"ok": 2, "failed": 1}}}
|
|
assert automation.recap_totals(results) == {
|
|
"hosts": 2, "ok": 5, "changed": 1, "failed": 1, "unreachable": 0, "skipped": 0}
|
|
assert automation.recap_totals(None) is None
|
|
|
|
|
|
def test_read_log_tail_and_grep(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(automation.executor, "artifact_path", lambda rid: str(tmp_path / "none.log"))
|
|
run = NS(id="r1", output="\n".join(["PLAY [all]", "TASK [x]", "fatal: boom", "x" * 900, "PLAY RECAP"]))
|
|
out = automation.read_log(run, tail=2, grep="FATAL")
|
|
assert out["log_lines"] == 5
|
|
assert out["tail"][1] == "PLAY RECAP" and out["tail"][0].endswith("[… 400 more]")
|
|
assert out["matches"] == [{"line_no": 3, "line": "fatal: boom"}] and out["matches_total"] == 1
|
|
(tmp_path / "r1.log").write_text("from the artifact\n")
|
|
monkeypatch.setattr(automation.executor, "artifact_path", lambda rid: str(tmp_path / f"{rid}.log"))
|
|
assert automation.read_log(run, tail=10, grep="")["tail"] == ["from the artifact"]
|
|
|
|
|
|
def test_automation_tools_are_registered_with_their_roles(app):
|
|
mcp = app.mcp_server
|
|
for name in VIEWER_TOOLS:
|
|
assert mcp.tool_roles[name] is UserRole.viewer, name
|
|
assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is True, name
|
|
for name in OPERATOR_TOOLS:
|
|
assert mcp.tool_roles[name] is UserRole.operator, name
|
|
assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is False, name
|
|
assert mcp.tool_roles["set_alert_action"] is UserRole.admin
|
|
for name in VIEWER_TOOLS | OPERATOR_TOOLS | {"set_alert_action"}:
|
|
assert (mcp._tool_manager.get_tool(name).description or "").strip(), name
|
|
|
|
|
|
def test_a_viewer_token_cannot_start_a_run(app):
|
|
viewer = Caller(user_id="u", username="v", role=UserRole.viewer, token_id=1)
|
|
operator = Caller(user_id="u", username="o", role=UserRole.operator, token_id=2)
|
|
assert "needs the operator role" in app.mcp_server.refusal("run_playbook", viewer)
|
|
assert app.mcp_server.refusal("run_playbook", operator) is None
|
|
assert "needs the admin role" in app.mcp_server.refusal("set_alert_action", operator)
|
|
|
|
|
|
def test_builtin_playbooks_get_steward_url_from_steward():
|
|
from steward.mcp import _context
|
|
app = NS(config={"PUBLIC_BASE_URL": ""})
|
|
a, u = _context._app_ctx.set(app), _context._base_url_ctx.set("http://steward.lan")
|
|
try:
|
|
declared = [{"name": "steward_url", "default": "", "secret": False, "required": False,
|
|
"prompt": None}]
|
|
builtin = {"name": "steward-builtin"}
|
|
got, filled = automation.steward_defaults(builtin, declared, {"agent_interval": 30})
|
|
assert got == {"agent_interval": 30, "steward_url": "http://steward.lan"}
|
|
assert filled == {"steward_url": "http://steward.lan"}
|
|
# The configured public URL wins over the address the request came in on.
|
|
app.config["PUBLIC_BASE_URL"] = "https://steward.example/"
|
|
assert automation.steward_defaults(builtin, declared, {})[1] == {
|
|
"steward_url": "https://steward.example"}
|
|
# A value the caller passed stands; another source's playbook is untouched.
|
|
assert automation.steward_defaults(builtin, declared, {"steward_url": "http://x"})[1] == {}
|
|
assert automation.steward_defaults({"name": "homelab"}, declared, {}) == ({}, {})
|
|
# A playbook that doesn't declare it gets nothing extra.
|
|
assert automation.steward_defaults(builtin, [], {}) == ({}, {})
|
|
finally:
|
|
_context._app_ctx.reset(a)
|
|
_context._base_url_ctx.reset(u)
|