kevinveenbirkenbach/computer-playbook
1
0
Fork 0
mirror of https://github.com/kevinveenbirkenbach/computer-playbook.git synced 2025-12-07 09:56:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refine deploy CLI, test-deploy workflow and Ansible output

Browse Source 
Changes:
- Update GitHub Actions test-deploy workflow to run three staged deploys (normal+debug, reset+debug, async) using inventory-generated vault password files.
- Switch Ansible stdout_callback to ansible.builtin.default and enable YAML-style result_format via callback_default.
- Refactor cli/deploy.py: typed run_ansible_playbook(), structured MODE_* handling, better error reporting, and preserved vault/interactive behaviour.
- Add unit tests for deploy CLI (bool parsing, MODE_* loading, dynamic args, validation, and ansible-playbook command construction) under tests/unit/cli/test_deploy.py.

Context: see ChatGPT conversation on 2025-12-02: https://chatgpt.com/share/692f1035-6bc4-800f-91a9-342db54e1a75
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-02 20:25:26 +01:00
parent 8a453be4b9
commit 46174125bc
4 changed files with 502 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
.github/workflows/test-deploy.yml vendored
View File

@@ -15,40 +15,58 @@ jobs:
- name: Checkout repository
uses: actions/checkout@v4
- name: Generate vault password automatically
run: |
python3 - << 'EOF' > .vault_pass
import secrets
import string
alphabet = string.ascii_letters + string.digits
pw = ''.join(secrets.choice(alphabet) for _ in range(64))
print(pw, end="")
EOF
chmod 600 .vault_pass
# Export password as environment variable
echo "VAULT_PASSWORD=$(cat .vault_pass)" >> "$GITHUB_ENV"
shell: bash
- name: Build Docker image
run: |
docker build --network=host --no-cache --pull -t infinito:latest .
# ----------------------------------------------------------------------
# 1) First deploy: NORMAL DEPLOY + DEBUG enabled
# ----------------------------------------------------------------------
# 1) First deploy: normal + debug
- name: First deploy (normal + debug)
run: |
docker run --network=host --rm \
-e VAULT_PASSWORD="${VAULT_PASSWORD}" \
infinito:latest \
/bin/sh -lc '
echo "$VAULT_PASSWORD" > /tmp/.vault_pass
chmod 600 /tmp/.vault_pass
export ANSIBLE_VAULT_PASSWORD_FILE=/tmp/.vault_pass
set -e
cd /opt/infinito-src
# Create inventory (also creates inventories/github-ci/.password if missing)
infinito create inventory inventories/github-ci \
--host localhost \
--ssl-disabled
INVENTORY_PATH="inventories/github-ci/servers.yml"
VAULT_FILE="inventories/github-ci/.password"
# First deploy with debug
infinito deploy "$INVENTORY_PATH" -T server -p "$VAULT_FILE" --debug
'
# 2) Second deploy: reset + debug
- name: Second deploy (--reset --debug)
run: |
docker run --network=host --rm \
infinito:latest \
/bin/sh -lc '
set -e
cd /opt/infinito-src
# Rebuild inventory; .password will be reused if present
infinito create inventory inventories/github-ci \
--host localhost \
--ssl-disabled
INVENTORY_PATH="inventories/github-ci/servers.yml"
VAULT_FILE="inventories/github-ci/.password"
infinito deploy "$INVENTORY_PATH" -T server -p "$VAULT_FILE" --skip-tests --reset --debug
'
# 3) Third deploy: async (no debug)
- name: Third deploy (async deploy – no debug)
run: |
docker run --network=host --rm \
infinito:latest \
/bin/sh -lc '
set -e
cd /opt/infinito-src
infinito create inventory inventories/github-ci \
@@ -56,44 +74,8 @@ jobs:
--ssl-disabled
INVENTORY_PATH="inventories/github-ci/servers.yml"
infinito deploy "$INVENTORY_PATH" -T server --debug
'
# ----------------------------------------------------------------------
# 2) Second deploy: RESET + DEBUG
# ----------------------------------------------------------------------
- name: Second deploy (--reset --debug)
run: |
docker run --network=host --rm \
-e VAULT_PASSWORD="${VAULT_PASSWORD}" \
infinito:latest \
/bin/sh -lc '
echo "$VAULT_PASSWORD" > /tmp/.vault_pass
chmod 600 /tmp/.vault_pass
export ANSIBLE_VAULT_PASSWORD_FILE=/tmp/.vault_pass
cd /opt/infinito-src
INVENTORY_PATH="inventories/github-ci/servers.yml"
infinito deploy "$INVENTORY_PATH" -T server --reset --debug
'
# ----------------------------------------------------------------------
# 3) Third deploy: ASYNC DEPLOY (no debug flag)
# ----------------------------------------------------------------------
- name: Third deploy (async deploy – no debug)
run: |
docker run --network=host --rm \
-e VAULT_PASSWORD="${VAULT_PASSWORD}" \
infinito:latest \
/bin/sh -lc '
echo "$VAULT_PASSWORD" > /tmp/.vault_pass
chmod 600 /tmp/.vault_pass
export ANSIBLE_VAULT_PASSWORD_FILE=/tmp/.vault_pass
cd /opt/infinito-src
INVENTORY_PATH="inventories/github-ci/servers.yml"
# Without --debug the deploy is asynchronous in several roles
infinito deploy "$INVENTORY_PATH" -T server
VAULT_FILE="inventories/github-ci/.password"
# Async-style deploy: no --debug, so some processes run in parallel
infinito deploy "$INVENTORY_PATH" -T server -p "$VAULT_FILE" --skip-tests
'

5
ansible.cfg
View File

@@ -11,7 +11,7 @@ deprecation_warnings = True
interpreter_python = auto_silent
# --- Output & Profiling ---
stdout_callback = yaml
stdout_callback = ansible.builtin.default
callbacks_enabled = profile_tasks,timer
# --- Plugin paths ---
@@ -27,3 +27,6 @@ transfer_method = smart
[persistent_connection]
connect_timeout = 30
command_timeout = 60
[callback_default]
result_format = yaml

181
cli/deploy.py
View File

@@ -1,4 +1,13 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Infinito.Nexus deploy CLI
This script is the main entrypoint for running the Ansible playbook with
dynamic MODE_* flags, automatic inventory validation, and optional build/test
steps.
"""
import argparse
import subprocess
@@ -10,40 +19,43 @@ from typing import Optional, Dict, Any, List
def run_ansible_playbook(
inventory,
modes,
limit=None,
allowed_applications=None,
password_file=None,
verbose=0,
skip_build=False,
skip_tests=False,
logs=False,
diff=False,
):
inventory: str,
modes: Dict[str, Any],
limit: Optional[str] = None,
allowed_applications: Optional[List[str]] = None,
password_file: Optional[str] = None,
verbose: int = 0,
skip_build: bool = False,
skip_tests: bool = False,
logs: bool = False,
diff: bool = False,
) -> None:
"""Run ansible-playbook with the given parameters and modes."""
start_time = datetime.datetime.now()
print(f"\n▶️ Script started at: {start_time.isoformat()}\n")
# Cleanup is now handled via MODE_CLEANUP
# 1) Cleanup phase (MODE_CLEANUP)
if modes.get("MODE_CLEANUP", False):
cleanup_command = ["make", "clean-keep-logs"] if logs else ["make", "clean"]
print("\n🧹 Cleaning up project (" + " ".join(cleanup_command) + ")...\n")
print(f"\n🧹 Cleaning up project ({' '.join(cleanup_command)})...\n")
subprocess.run(cleanup_command, check=True)
else:
print("\n⚠️ Skipping cleanup as requested.\n")
print("\n🧹 Cleanup skipped (MODE_CLEANUP=false or not set)\n")
# 2) Build phase
if not skip_build:
print("\n🛠️ Building project (make messy-build)...\n")
subprocess.run(["make", "messy-build"], check=True)
else:
print("\n⚠️ Skipping build as requested.\n")
print("\n🛠️ Build skipped (--skip-build)\n")
script_dir = os.path.dirname(os.path.realpath(__file__))
playbook = os.path.join(os.path.dirname(script_dir), "playbook.yml")
repo_root = os.path.dirname(script_dir)
playbook = os.path.join(repo_root, "playbook.yml")
# Inventory validation is controlled via MODE_ASSERT
# 3) Inventory validation phase (MODE_ASSERT)
if modes.get("MODE_ASSERT", None) is False:
print("\n⚠️ Skipping inventory validation as requested.\n")
print("\n🔍 Inventory assertion explicitly disabled (MODE_ASSERT=false)\n")
elif "MODE_ASSERT" not in modes or modes["MODE_ASSERT"] is True:
print("\n🔍 Validating inventory before deployment...\n")
try:
@@ -56,40 +68,71 @@ def run_ansible_playbook(
check=True,
)
except subprocess.CalledProcessError:
print("\n❌ Inventory validation failed. Deployment aborted.\n", file=sys.stderr)
print(
"\n[ERROR] Inventory validation failed. Aborting deploy.\n",
file=sys.stderr,
)
sys.exit(1)
# 4) Test phase
if not skip_tests:
print("\n🧪 Running tests (make messy-test)...\n")
subprocess.run(["make", "messy-test"], check=True)
else:
print("\n🧪 Tests skipped (--skip-tests)\n")
# Build ansible-playbook command
cmd = ["ansible-playbook", "-i", inventory, playbook]
# 5) Build ansible-playbook command
cmd: List[str] = ["ansible-playbook", "-i", inventory, playbook]
# --limit / -l
if limit:
cmd.extend(["--limit", limit])
cmd.extend(["-l", limit])
# extra var: allowed_applications
if allowed_applications:
joined = ",".join(allowed_applications)
cmd.extend(["-e", f"allowed_applications={joined}"])
# inject MODE_* variables as extra vars
for key, value in modes.items():
val = str(value).lower() if isinstance(value, bool) else str(value)
cmd.extend(["-e", f"{key}={val}"])
# vault password handling
if password_file:
# If a file is explicitly provided, pass it through
cmd.extend(["--vault-password-file", password_file])
else:
cmd.extend(["--ask-vault-pass"])
# else:
# No explicit vault option → ansible will prompt if it needs a password.
# This keeps the old behaviour and the CLI help text correct.
# diff mode
if diff:
cmd.append("--diff")
cmd.append("--diff")
# MODE_DEBUG=true → always at least -vvv
if modes.get("MODE_DEBUG", False):
verbose = max(verbose, 3)
# verbosity flags
if verbose:
cmd.append("-" + "v" * verbose)
print("\n🚀 Launching Ansible Playbook...\n")
subprocess.run(cmd, check=True)
# Capture output so the real Ansible error is visible before exit
result = subprocess.run(cmd, text=True, capture_output=True)
if result.stdout:
print(result.stdout, end="")
if result.stderr:
print(result.stderr, file=sys.stderr, end="")
if result.returncode != 0:
print(
f"\n[ERROR] ansible-playbook exited with status {result.returncode}\n",
file=sys.stderr,
)
sys.exit(result.returncode)
end_time = datetime.datetime.now()
print(f"\n✅ Script ended at: {end_time.isoformat()}\n")
@@ -98,22 +141,23 @@ def run_ansible_playbook(
print(f"⏱️ Total execution time: {duration}\n")
def validate_application_ids(inventory, app_ids):
"""
Abort the script if any application IDs are invalid, with detailed reasons.
"""
def validate_application_ids(inventory: str, app_ids: List[str]) -> None:
"""Use ValidDeployId helper to ensure all requested IDs are valid."""
if not app_ids:
return
from module_utils.valid_deploy_id import ValidDeployId
validator = ValidDeployId()
invalid = validator.validate(inventory, app_ids)
if invalid:
print("\n❌ Detected invalid application_id(s):\n")
print("\n[ERROR] Some application_ids are invalid for this inventory:\n")
for app_id, status in invalid.items():
reasons = []
if not status["in_roles"]:
reasons.append("not defined in roles (infinito)")
if not status["in_inventory"]:
reasons.append("not found in inventory file")
reasons: List[str] = []
if not status.get("allowed", True):
reasons.append("not allowed by configuration")
if not status.get("in_inventory", True):
reasons.append("not present in inventory")
print(f" - {app_id}: " + ", ".join(reasons))
sys.exit(1)
@@ -124,6 +168,7 @@ MODE_LINE_RE = re.compile(
def _parse_bool_literal(text: str) -> Optional[bool]:
"""Parse a simple true/false/yes/no/on/off into bool or None."""
t = text.strip().lower()
if t in ("true", "yes", "on"):
return True
@@ -134,12 +179,11 @@ def _parse_bool_literal(text: str) -> Optional[bool]:
def load_modes_from_yaml(modes_yaml_path: str) -> List[Dict[str, Any]]:
"""
Parse group_vars/all/01_modes.yml line-by-line to recover:
- name (e.g., MODE_TEST)
- default (True/False/None if templated/unknown)
- help (from trailing # comment, if present)
Load MODE_* metadata from a simple key: value file.
Each non-comment, non-empty line is parsed via MODE_LINE_RE.
"""
modes = []
modes: List[Dict[str, Any]] = []
if not os.path.exists(modes_yaml_path):
raise FileNotFoundError(f"Modes file not found: {modes_yaml_path}")
@@ -173,8 +217,11 @@ def add_dynamic_mode_args(
parser: argparse.ArgumentParser, modes_meta: List[Dict[str, Any]]
) -> Dict[str, Dict[str, Any]]:
"""
Add argparse options based on modes metadata.
Returns a dict mapping mode name -> { 'dest': <argparse_dest>, 'default': <bool/None>, 'kind': 'bool_true'|'bool_false'|'explicit' }.
Add dynamic CLI flags based on MODE_* metadata.
- MODE_FOO: true -> --skip-foo (default enabled, flag disables it)
- MODE_BAR: false -> --bar (default disabled, flag enables it)
- MODE_BAZ: null -> --baz {true,false} (explicit)
"""
spec: Dict[str, Dict[str, Any]] = {}
for m in modes_meta:
@@ -198,7 +245,10 @@ def add_dynamic_mode_args(
else:
opt = f"--{short}"
dest = short
help_txt = desc or f"Set {short} explicitly (true/false). If omitted, keep inventory default."
help_txt = (
desc
or f"Set {short} explicitly (true/false). If omitted, keep inventory default."
)
parser.add_argument(opt, choices=["true", "false"], help=help_txt, dest=dest)
spec[name] = {"dest": dest, "default": None, "kind": "explicit"}
@@ -209,7 +259,7 @@ def build_modes_from_args(
spec: Dict[str, Dict[str, Any]], args_namespace: argparse.Namespace
) -> Dict[str, Any]:
"""
Using the argparse results and the spec, compute the `modes` dict to pass to Ansible.
Build a MODE_* dict from parsed CLI args and the dynamic spec.
"""
modes: Dict[str, Any] = {}
for mode_name, info in spec.items():
@@ -218,18 +268,20 @@ def build_modes_from_args(
val = getattr(args_namespace, dest, None)
if kind == "bool_true":
# default True, flag means "skip" → False
modes[mode_name] = False if val else True
elif kind == "bool_false":
# default False, flag enables → True
modes[mode_name] = True if val else False
else:
else: # explicit
if val is not None:
modes[mode_name] = True if val == "true" else False
return modes
def main():
def main() -> None:
parser = argparse.ArgumentParser(
description="Run the central Ansible deployment script to manage infrastructure, updates, and tests."
description="Deploy the Infinito.Nexus stack via ansible-playbook."
)
parser.add_argument(
@@ -246,24 +298,30 @@ def main():
"--host-type",
choices=["server", "desktop"],
default="server",
help="Specify whether the target is a server or a personal computer. Affects role selection and variables.",
help=(
"Specify whether the target is a server or a personal computer. "
"Affects role selection and variables."
),
)
parser.add_argument(
"-p",
"--password-file",
help="Path to the file containing the Vault password. If not provided, prompts for the password interactively.",
help=(
"Path to the file containing the Vault password. "
"If not provided, ansible-vault will prompt interactively."
),
)
parser.add_argument(
"-B",
"--skip-build",
action="store_true",
help="Skip running 'make build' before deployment.",
help="Skip running 'make messy-build' before deployment.",
)
parser.add_argument(
"-t",
"--skip-tests",
action="store_true",
help="Skip running 'make messy-tests' before deployment.",
help="Skip running 'make messy-test' before deployment.",
)
parser.add_argument(
"-i",
@@ -271,28 +329,32 @@ def main():
nargs="+",
default=[],
dest="id",
help="List of application_id's for partial deploy. If not set, all application IDs defined in the inventory will be executed.",
help=(
"List of application_id's for partial deploy. "
"If not set, all application IDs defined in the inventory will be executed."
),
)
parser.add_argument(
"-v",
"--verbose",
action="count",
default=0,
help="Increase verbosity level. Multiple -v flags increase detail (e.g., -vvv for maximum log output).",
help=(
"Increase verbosity level. Multiple -v flags increase detail "
"(e.g., -vvv for maximum log output)."
),
)
parser.add_argument(
"--logs",
action="store_true",
help="Keep the CLI logs during cleanup command",
help="Keep the CLI logs during cleanup command.",
)
parser.add_argument(
"--diff",
action="store_true",
help="Pass --diff to ansible-playbook to show configuration changes.",
)
# ---- Dynamically add mode flags from group_vars/all/01_modes.yml ----
script_dir = os.path.dirname(os.path.realpath(__file__))
repo_root = os.path.dirname(script_dir)
modes_yaml_path = os.path.join(repo_root, "group_vars", "all", "01_modes.yml")
@@ -302,10 +364,7 @@ def main():
args = parser.parse_args()
validate_application_ids(args.inventory, args.id)
# Build modes from dynamic args
modes = build_modes_from_args(modes_spec, args)
# Additional non-dynamic flags
modes["MODE_LOGS"] = args.logs
modes["host_type"] = args.host_type

332
tests/unit/cli/test_deploy.py Normal file
View File

@@ -0,0 +1,332 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import tempfile
import unittest
from typing import Any, Dict, List
from unittest import mock
import cli.deploy as deploy
import subprocess
class TestParseBoolLiteral(unittest.TestCase):
def test_true_values(self):
self.assertTrue(deploy._parse_bool_literal("true"))
self.assertTrue(deploy._parse_bool_literal("True"))
self.assertTrue(deploy._parse_bool_literal(" yes "))
self.assertTrue(deploy._parse_bool_literal("ON"))
def test_false_values(self):
self.assertFalse(deploy._parse_bool_literal("false"))
self.assertFalse(deploy._parse_bool_literal("False"))
self.assertFalse(deploy._parse_bool_literal(" no "))
self.assertFalse(deploy._parse_bool_literal("off"))
def test_unknown_value(self):
self.assertIsNone(deploy._parse_bool_literal("maybe"))
self.assertIsNone(deploy._parse_bool_literal(""))
self.assertIsNone(deploy._parse_bool_literal(" "))
class TestLoadModesFromYaml(unittest.TestCase):
def test_load_modes_basic(self):
# Create a temporary "01_modes.yml"-like file
content = """\
MODE_CLEANUP: true # cleanup before deploy
MODE_DEBUG: false # enable debug
MODE_ASSERT: null # explicitly set via CLI
INVALID_KEY: true # ignored because no MODE_ prefix
"""
with tempfile.NamedTemporaryFile("w+", delete=False, encoding="utf-8") as f:
path = f.name
f.write(content)
f.flush()
try:
modes = deploy.load_modes_from_yaml(path)
finally:
os.unlink(path)
# We expect 3 MODE_* entries, INVALID_KEY is ignored
self.assertEqual(len(modes), 3)
by_name = {m["name"]: m for m in modes}
self.assertIn("MODE_CLEANUP", by_name)
self.assertIn("MODE_DEBUG", by_name)
self.assertIn("MODE_ASSERT", by_name)
self.assertEqual(by_name["MODE_CLEANUP"]["default"], True)
self.assertEqual(by_name["MODE_DEBUG"]["default"], False)
self.assertIsNone(by_name["MODE_ASSERT"]["default"])
self.assertEqual(by_name["MODE_CLEANUP"]["help"], "cleanup before deploy")
class TestDynamicModes(unittest.TestCase):
def setUp(self):
# Simple meta as if parsed from 01_modes.yml
self.modes_meta = [
{"name": "MODE_CLEANUP", "default": True, "help": "Cleanup before run"},
{"name": "MODE_DEBUG", "default": False, "help": "Debug output"},
{"name": "MODE_ASSERT", "default": None, "help": "Inventory assertion"},
]
def test_add_dynamic_mode_args_and_build_modes_defaults(self):
parser = unittest.mock.MagicMock()
# Use a real ArgumentParser for build_modes_from_args
from argparse import ArgumentParser
real_parser = ArgumentParser()
spec = deploy.add_dynamic_mode_args(real_parser, self.modes_meta)
# We expect three entries
self.assertIn("MODE_CLEANUP", spec)
self.assertIn("MODE_DEBUG", spec)
self.assertIn("MODE_ASSERT", spec)
# No flags given: use defaults (True/False/None)
args = real_parser.parse_args([])
modes = deploy.build_modes_from_args(spec, args)
self.assertTrue(modes["MODE_CLEANUP"]) # default True
self.assertFalse(modes["MODE_DEBUG"]) # default False
self.assertNotIn("MODE_ASSERT", modes) # default None → not present
def test_add_dynamic_mode_args_and_build_modes_flags(self):
from argparse import ArgumentParser
parser = ArgumentParser()
spec = deploy.add_dynamic_mode_args(parser, self.modes_meta)
# CLI: --skip-cleanup → MODE_CLEANUP=False
# --debug → MODE_DEBUG=True
# --assert true → MODE_ASSERT=True
args = parser.parse_args(
["--skip-cleanup", "--debug", "--assert", "true"]
)
modes = deploy.build_modes_from_args(spec, args)
self.assertFalse(modes["MODE_CLEANUP"])
self.assertTrue(modes["MODE_DEBUG"])
self.assertTrue(modes["MODE_ASSERT"])
class TestValidateApplicationIds(unittest.TestCase):
def test_no_ids_does_nothing(self):
"""
When no application_ids are passed, the function should just return
without trying to validate anything.
"""
deploy.validate_application_ids("inventories/github-ci/servers.yml", [])
@mock.patch("module_utils.valid_deploy_id.ValidDeployId")
def test_invalid_ids_raise_system_exit(self, mock_vdi_cls):
"""
When ValidDeployId reports invalid IDs, validate_application_ids should
print an error and exit with code 1.
"""
instance = mock_vdi_cls.return_value
instance.validate.return_value = {
"web-app-foo": {"allowed": False, "in_inventory": True},
"web-app-bar": {"allowed": True, "in_inventory": False},
}
with self.assertRaises(SystemExit) as ctx:
deploy.validate_application_ids(
"inventories/github-ci/servers.yml",
["web-app-foo", "web-app-bar"],
)
self.assertEqual(ctx.exception.code, 1)
instance.validate.assert_called_once_with(
"inventories/github-ci/servers.yml",
["web-app-foo", "web-app-bar"],
)
class TestRunAnsiblePlaybook(unittest.TestCase):
def _fake_run_side_effect(
self,
calls_store: List[List[str]],
ansible_rc: int = 0,
):
"""
side_effect for subprocess.run that:
- Records every command into calls_store
- Returns 'ansible_rc' for ansible-playbook
- Returns rc=0 for all other commands (make, validation, etc.)
"""
def _side_effect(cmd, *args, **kwargs):
# Normalize into a list for easier assertions
if isinstance(cmd, list):
calls_store.append(cmd)
else:
calls_store.append([cmd])
# Special handling for ansible-playbook
if isinstance(cmd, list) and cmd and cmd[0] == "ansible-playbook":
return subprocess.CompletedProcess(
cmd,
ansible_rc,
stdout="ANSIBLE_STDOUT\n",
stderr="ANSIBLE_STDERR\n",
)
# Everything else (make, python inventory) is treated as success
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
return _side_effect
@mock.patch("subprocess.run")
def test_run_ansible_playbook_builds_correct_command(self, mock_run):
"""
Happy-path test:
- MODE_CLEANUP=True → 'make clean' is executed
- build/test phases are executed
- ansible-playbook is called with expected arguments
- commands are fully mocked (no real execution)
"""
calls: List[List[str]] = []
mock_run.side_effect = self._fake_run_side_effect(calls, ansible_rc=0)
modes: Dict[str, Any] = {
"MODE_CLEANUP": True,
"MODE_ASSERT": True,
"MODE_DEBUG": True, # should enforce at least -vvv
}
inventory_path = "inventories/github-ci/servers.yml"
limit = "localhost"
allowed_apps = ["web-app-foo", "web-app-bar"]
password_file = "inventories/github-ci/.password"
deploy.run_ansible_playbook(
inventory=inventory_path,
modes=modes,
limit=limit,
allowed_applications=allowed_apps,
password_file=password_file,
verbose=1, # explicitly set, then raised by MODE_DEBUG
skip_build=False,
skip_tests=False,
logs=False,
diff=True,
)
# We expect at least: make clean, make messy-build, make messy-test,
# plus the ansible-playbook invocation. Inventory validation is also
# done via python ... inventory.py but we do not assert the full path.
self.assertTrue(
any(call == ["make", "clean"] for call in calls),
"Expected 'make clean' when MODE_CLEANUP is true",
)
self.assertTrue(
any(call == ["make", "messy-build"] for call in calls),
"Expected 'make messy-build' when skip_build=False",
)
self.assertTrue(
any(call == ["make", "messy-test"] for call in calls),
"Expected 'make messy-test' when skip_tests=False",
)
self.assertTrue(
any(
isinstance(call, list)
and any("inventory.py" in part for part in call)
for call in calls
),
"Expected inventory validation call (... inventory.py ...)",
)
# Last command should be ansible-playbook
self.assertGreaterEqual(len(calls), 1)
last_cmd = calls[-1]
self.assertEqual(last_cmd[0], "ansible-playbook")
# Check inventory and playbook args
self.assertIn("-i", last_cmd)
self.assertIn(inventory_path, last_cmd)
idx_inv = last_cmd.index(inventory_path)
self.assertGreater(len(last_cmd), idx_inv + 1)
playbook_arg = last_cmd[idx_inv + 1]
self.assertTrue(
playbook_arg.endswith(os.path.join("", "playbook.yml")),
f"playbook argument should end with 'playbook.yml', got: {playbook_arg}",
)
# Check --limit / -l
self.assertIn("-l", last_cmd)
self.assertIn(limit, last_cmd)
# allowed_applications extra var
last_cmd_str = " ".join(last_cmd)
self.assertIn(
"allowed_applications=web-app-foo,web-app-bar",
last_cmd_str,
)
# Modes passed as -e
self.assertIn("MODE_CLEANUP=true", last_cmd_str)
self.assertIn("MODE_ASSERT=true", last_cmd_str)
self.assertIn("MODE_DEBUG=true", last_cmd_str)
# Vault password file
self.assertIn("--vault-password-file", last_cmd)
self.assertIn(password_file, last_cmd)
# --diff should be present
self.assertIn("--diff", last_cmd)
# Verbosity: MODE_DEBUG enforces at least -vvv regardless of initial -v
self.assertTrue(
any(arg.startswith("-vvv") for arg in last_cmd),
"Expected at least -vvv because MODE_DEBUG=True",
)
# Verify that for ansible-playbook, we asked for text+capture_output
last_call = mock_run.call_args_list[-1]
self.assertTrue(last_call.kwargs.get("text", False))
self.assertTrue(last_call.kwargs.get("capture_output", False))
@mock.patch("subprocess.run")
def test_run_ansible_playbook_failure_exits_with_code(self, mock_run):
"""
If ansible-playbook returns non-zero, run_ansible_playbook should exit
with the same code. All external commands are mocked.
"""
calls: List[List[str]] = []
mock_run.side_effect = self._fake_run_side_effect(calls, ansible_rc=4)
modes: Dict[str, Any] = {
"MODE_CLEANUP": False,
"MODE_ASSERT": False,
"MODE_DEBUG": False,
}
with self.assertRaises(SystemExit) as ctx:
deploy.run_ansible_playbook(
inventory="inventories/github-ci/servers.yml",
modes=modes,
limit=None,
allowed_applications=None,
password_file=None,
verbose=0,
skip_build=True,
skip_tests=True,
logs=False,
diff=False,
)
self.assertEqual(ctx.exception.code, 4)
self.assertTrue(
any(call and call[0] == "ansible-playbook" for call in calls),
"ansible-playbook should have been invoked once in the error path",
)
if __name__ == "__main__":
unittest.main()