"""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", "update_group_members"} 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": "", "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) async def test_run_playbook_holds_variables_to_the_callers_role(monkeypatch): monkeypatch.setattr(automation, "_source", lambda name: {"name": name, "path": "/x"}) monkeypatch.setattr(automation, "_playbook", lambda s, p: "- hosts: all\n tasks: []\n") def _no_run(*a, **k): raise AssertionError("a run was started") monkeypatch.setattr(automation, "trigger_run", _no_run) operator = Caller(user_id="u", username="op", role=UserRole.operator, token_id=1) monkeypatch.setattr(automation, "current_caller_or_none", lambda: operator) with pytest.raises(ValueError, match="Only an admin can set ansible_connection"): await automation.run_playbook("s", "site.yml", extra_vars={"ansible_connection": "local"}) with pytest.raises(ValueError, match="Ansible evaluates it"): await automation.run_playbook("s", "site.yml", extra_vars={"x": "{{ lookup('pipe','id') }}"}) def test_schedule_update_holds_only_changed_variables(monkeypatch): operator = Caller(user_id="u", username="op", role=UserRole.operator, token_id=1) monkeypatch.setattr(automation, "current_caller_or_none", lambda: operator) stored = {"ansible_ssh_common_args": "-J bastion"} automation.check_caller_vars({**stored, "release": "2"}, stored) with pytest.raises(ValueError, match="Only an admin"): automation.check_caller_vars({"ansible_ssh_common_args": "-o ProxyCommand=x"}, stored) def test_group_members_are_named_by_target_or_by_linked_host(): """#5487: a target by its name or id, or the Steward host linked to it; an unknown name refuses the lot, so a typo changes nothing.""" web = NS(id="t1", name="web01", host_id="h1") db = NS(id="t2", name="db01", host_id=None) hosts = {"h1": "VDNT-Docker02"} got = automation.member_targets(["vdnt-docker02", "t2", "web01", " "], [web, db], hosts) assert [t.name for t in got] == ["web01", "db01"] with pytest.raises(ValueError, match="no target or linked host named 'nas'.*Nothing was changed"): automation.member_targets(["web01", "nas"], [web, db], hosts)