From 8c64f91a6dd975a05dcad06bd568e8875f1969d1 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Thu, 4 Dec 2025 16:59:13 +0100 Subject: [PATCH] Refactor deploy CLI into cli/deploy/dedicated.py, add package inits, and update unit tests (see ChatGPT conversation: https://chatgpt.com/share/6931af9e-dad0-800f-9a8f-6d01c373de87) --- cli/deploy/__init__.py | 0 cli/{deploy.py => deploy/dedicated.py} | 264 +++++++++--------- tests/unit/cli/deploy/__init__.py | 0 .../test_dedicated.py} | 176 +++++++----- 4 files changed, 233 insertions(+), 207 deletions(-) create mode 100644 cli/deploy/__init__.py rename cli/{deploy.py => deploy/dedicated.py} (55%) create mode 100644 tests/unit/cli/deploy/__init__.py rename tests/unit/cli/{test_deploy.py => deploy/test_dedicated.py} (68%) diff --git a/cli/deploy/__init__.py b/cli/deploy/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/cli/deploy.py b/cli/deploy/dedicated.py similarity index 55% rename from cli/deploy.py rename to cli/deploy/dedicated.py index 729e91d7..e3a2119c 100644 --- a/cli/deploy.py +++ b/cli/deploy/dedicated.py @@ -1,12 +1,13 @@ -#!/usr/bin/env python3 + #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -Infinito.Nexus deploy CLI +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. +phases. It supports partial deployments, dynamic MODE flag generation, +inventory validation, and structured execution flow. """ import argparse @@ -18,6 +19,20 @@ import re from typing import Optional, Dict, Any, List +# -------------------------------------------------------------------------------------- +# Path resolution +# -------------------------------------------------------------------------------------- + +# Current file: .../cli/deploy/deploy.py +SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__)) # → cli/deploy +CLI_ROOT = os.path.dirname(SCRIPT_DIR) # → cli +REPO_ROOT = os.path.dirname(CLI_ROOT) # → project root + + +# -------------------------------------------------------------------------------------- +# Main execution logic +# -------------------------------------------------------------------------------------- + def run_ansible_playbook( inventory: str, modes: Dict[str, Any], @@ -30,96 +45,97 @@ def run_ansible_playbook( logs: bool = False, diff: bool = False, ) -> None: - """Run ansible-playbook with the given parameters and modes.""" + """Run ansible-playbook with the given parameters and execution modes.""" start_time = datetime.datetime.now() print(f"\n▶️ Script started at: {start_time.isoformat()}\n") - # 1) Cleanup phase (MODE_CLEANUP) + # --------------------------------------------------------- + # 1) Cleanup Phase + # --------------------------------------------------------- if modes.get("MODE_CLEANUP", False): - cleanup_command = ["make", "clean-keep-logs"] if logs else ["make", "clean"] - print(f"\n🧹 Cleaning up project ({' '.join(cleanup_command)})...\n") - subprocess.run(cleanup_command, check=True) + cleanup_cmd = ["make", "clean-keep-logs"] if logs else ["make", "clean"] + print(f"\n🧹 Running cleanup ({' '.join(cleanup_cmd)})...\n") + subprocess.run(cleanup_cmd, check=True) else: - print("\n🧹 Cleanup skipped (MODE_CLEANUP=false or not set)\n") + print("\n🧹 Cleanup skipped (MODE_CLEANUP not set or False)\n") - # 2) Build phase + # --------------------------------------------------------- + # 2) Build Phase + # --------------------------------------------------------- if not skip_build: - print("\n🛠️ Building project (make messy-build)...\n") + print("\n🛠️ Running project build (make messy-build)...\n") subprocess.run(["make", "messy-build"], check=True) else: print("\n🛠️ Build skipped (--skip-build)\n") - script_dir = os.path.dirname(os.path.realpath(__file__)) - repo_root = os.path.dirname(script_dir) - playbook = os.path.join(repo_root, "playbook.yml") + # The Ansible playbook is located in the repo root + playbook_path = os.path.join(REPO_ROOT, "playbook.yml") - # 3) Inventory validation phase (MODE_ASSERT) + # --------------------------------------------------------- + # 3) Inventory Validation Phase + # --------------------------------------------------------- if modes.get("MODE_ASSERT", None) is False: print("\n🔍 Inventory assertion explicitly disabled (MODE_ASSERT=false)\n") - elif "MODE_ASSERT" not in modes or modes["MODE_ASSERT"] is True: + else: print("\n🔍 Validating inventory before deployment...\n") + validator_path = os.path.join(CLI_ROOT, "validate", "inventory.py") try: subprocess.run( - [ - sys.executable, - os.path.join(script_dir, "validate", "inventory.py"), - os.path.dirname(inventory), - ], + [sys.executable, validator_path, os.path.dirname(inventory)], check=True, ) except subprocess.CalledProcessError: print( - "\n[ERROR] Inventory validation failed. Aborting deploy.\n", + "\n[ERROR] Inventory validation failed. Aborting deployment.\n", file=sys.stderr, ) sys.exit(1) - # 4) Test phase + # --------------------------------------------------------- + # 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") + # --------------------------------------------------------- # 5) Build ansible-playbook command - cmd: List[str] = ["ansible-playbook", "-i", inventory, playbook] + # --------------------------------------------------------- + cmd: List[str] = ["ansible-playbook", "-i", inventory, playbook_path] - # --limit / -l + # Limit hosts if limit: cmd.extend(["-l", limit]) - # extra var: allowed_applications + # Allowed applications (partial deployment) if allowed_applications: joined = ",".join(allowed_applications) cmd.extend(["-e", f"allowed_applications={joined}"]) - # inject MODE_* variables as extra vars + # MODE_* flags 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 + # Vault password file if password_file: - # If a file is explicitly provided, pass it through cmd.extend(["--vault-password-file", password_file]) - # 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 + # Enable diff mode if diff: cmd.append("--diff") - # MODE_DEBUG=true → always at least -vvv + # MODE_DEBUG → enforce high verbosity if modes.get("MODE_DEBUG", False): verbose = max(verbose, 3) - # verbosity flags + # Verbosity flags if verbose: cmd.append("-" + "v" * verbose) print("\n🚀 Launching Ansible Playbook...\n") - # Capture output so the real Ansible error is visible before exit result = subprocess.run(cmd) if result.returncode != 0: @@ -130,14 +146,17 @@ def run_ansible_playbook( sys.exit(result.returncode) end_time = datetime.datetime.now() + print(f"\n✅ Script ended at: {end_time.isoformat()}\n") + print(f"⏱️ Total execution time: {end_time - start_time}\n") - duration = end_time - start_time - print(f"⏱️ Total execution time: {duration}\n") +# -------------------------------------------------------------------------------------- +# Application ID validation +# -------------------------------------------------------------------------------------- def validate_application_ids(inventory: str, app_ids: List[str]) -> None: - """Use ValidDeployId helper to ensure all requested IDs are valid.""" + """Validate requested application IDs using ValidDeployId.""" if not app_ids: return @@ -145,25 +164,30 @@ def validate_application_ids(inventory: str, app_ids: List[str]) -> None: validator = ValidDeployId() invalid = validator.validate(inventory, app_ids) + if invalid: print("\n[ERROR] Some application_ids are invalid for this inventory:\n") for app_id, status in invalid.items(): - reasons: List[str] = [] + reasons = [] 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)) + print(f" - {app_id}: {', '.join(reasons)}") sys.exit(1) +# -------------------------------------------------------------------------------------- +# MODE_* parsing logic +# -------------------------------------------------------------------------------------- + MODE_LINE_RE = re.compile( r"""^\s*(?P[A-Z0-9_]+)\s*:\s*(?P.+?)\s*(?:#\s*(?P.*))?\s*$""" ) def _parse_bool_literal(text: str) -> Optional[bool]: - """Parse a simple true/false/yes/no/on/off into bool or None.""" + """Convert simple true/false/yes/no/on/off into boolean.""" t = text.strip().lower() if t in ("true", "yes", "on"): return True @@ -173,12 +197,9 @@ def _parse_bool_literal(text: str) -> Optional[bool]: def load_modes_from_yaml(modes_yaml_path: str) -> List[Dict[str, Any]]: - """ - Load MODE_* metadata from a simple key: value file. - - Each non-comment, non-empty line is parsed via MODE_LINE_RE. - """ + """Load MODE_* definitions from YAML-like key/value file.""" modes: List[Dict[str, Any]] = [] + if not os.path.exists(modes_yaml_path): raise FileNotFoundError(f"Modes file not found: {modes_yaml_path}") @@ -187,9 +208,11 @@ def load_modes_from_yaml(modes_yaml_path: str) -> List[Dict[str, Any]]: line = line.rstrip() if not line or line.lstrip().startswith("#"): continue + m = MODE_LINE_RE.match(line) if not m: continue + key = m.group("key") val = m.group("value").strip() cmt = (m.group("cmt") or "").strip() @@ -198,6 +221,7 @@ def load_modes_from_yaml(modes_yaml_path: str) -> List[Dict[str, Any]]: continue default_bool = _parse_bool_literal(val) + modes.append( { "name": key, @@ -205,20 +229,23 @@ def load_modes_from_yaml(modes_yaml_path: str) -> List[Dict[str, Any]]: "help": cmt or f"Toggle {key}", } ) + return modes +# -------------------------------------------------------------------------------------- +# Dynamic argparse mode injection +# -------------------------------------------------------------------------------------- + def add_dynamic_mode_args( parser: argparse.ArgumentParser, modes_meta: List[Dict[str, Any]] ) -> Dict[str, Dict[str, Any]]: """ - 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) + Add command-line arguments dynamically based on MODE_* metadata. """ + spec: Dict[str, Dict[str, Any]] = {} + for m in modes_meta: name = m["name"] default = m["default"] @@ -226,25 +253,24 @@ def add_dynamic_mode_args( short = name.replace("MODE_", "").lower() if default is True: + # MODE_FOO: true → --skip-foo disables it opt = f"--skip-{short}" dest = f"skip_{short}" - help_txt = desc or f"Skip/disable {short} (default: enabled)" - parser.add_argument(opt, action="store_true", help=help_txt, dest=dest) + parser.add_argument(opt, action="store_true", dest=dest, help=desc) spec[name] = {"dest": dest, "default": True, "kind": "bool_true"} + elif default is False: + # MODE_BAR: false → --bar enables it opt = f"--{short}" dest = short - help_txt = desc or f"Enable {short} (default: disabled)" - parser.add_argument(opt, action="store_true", help=help_txt, dest=dest) + parser.add_argument(opt, action="store_true", dest=dest, help=desc) spec[name] = {"dest": dest, "default": False, "kind": "bool_false"} + else: + # Explicit: MODE_XYZ: null → --xyz true|false opt = f"--{short}" dest = short - 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) + parser.add_argument(opt, choices=["true", "false"], dest=dest, help=desc) spec[name] = {"dest": dest, "default": None, "kind": "explicit"} return spec @@ -253,116 +279,76 @@ def add_dynamic_mode_args( def build_modes_from_args( spec: Dict[str, Dict[str, Any]], args_namespace: argparse.Namespace ) -> Dict[str, Any]: - """ - Build a MODE_* dict from parsed CLI args and the dynamic spec. - """ + """Resolve CLI arguments into a MODE_* dictionary.""" modes: Dict[str, Any] = {} + for mode_name, info in spec.items(): dest = info["dest"] kind = info["kind"] - val = getattr(args_namespace, dest, None) + value = getattr(args_namespace, dest, None) if kind == "bool_true": - # default True, flag means "skip" → False - modes[mode_name] = False if val else True + modes[mode_name] = False if value else True + elif kind == "bool_false": - # default False, flag enables → True - modes[mode_name] = True if val else False + modes[mode_name] = True if value else False + else: # explicit - if val is not None: - modes[mode_name] = True if val == "true" else False + if value is not None: + modes[mode_name] = (value == "true") + return modes +# -------------------------------------------------------------------------------------- +# Main entrypoint +# -------------------------------------------------------------------------------------- + def main() -> None: parser = argparse.ArgumentParser( - description="Deploy the Infinito.Nexus stack via ansible-playbook." + description="Deploy the Infinito.Nexus stack using ansible-playbook." ) + # Standard arguments + parser.add_argument("inventory", help="Path to the inventory file.") + parser.add_argument("-l", "--limit", help="Limit execution to certain hosts or groups.") parser.add_argument( - "inventory", - help="Path to the inventory file (INI or YAML) containing hosts and variables.", + "-T", "--host-type", choices=["server", "desktop"], default="server", + help="Specify target type: server or desktop." ) parser.add_argument( - "-l", - "--limit", - help="Restrict execution to a specific host or host group from the inventory.", + "-p", "--password-file", + help="Vault password file for encrypted variables." + ) + parser.add_argument("-B", "--skip-build", action="store_true", help="Skip build phase.") + parser.add_argument("-t", "--skip-tests", action="store_true", help="Skip test phase.") + parser.add_argument( + "-i", "--id", nargs="+", default=[], dest="id", + help="List of application_ids for partial deployment." ) parser.add_argument( - "-T", - "--host-type", - choices=["server", "desktop"], - default="server", - 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, ansible-vault will prompt interactively." - ), - ) - parser.add_argument( - "-B", - "--skip-build", - action="store_true", - help="Skip running 'make messy-build' before deployment.", - ) - parser.add_argument( - "-t", - "--skip-tests", - action="store_true", - help="Skip running 'make messy-test' before deployment.", - ) - parser.add_argument( - "-i", - "--id", - 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." - ), - ) - 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)." - ), - ) - parser.add_argument( - "--logs", - action="store_true", - 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.", + "-v", "--verbose", action="count", default=0, + help="Increase verbosity (e.g. -vvv)." ) + parser.add_argument("--logs", action="store_true", help="Keep logs during cleanup.") + parser.add_argument("--diff", action="store_true", help="Enable Ansible diff mode.") - 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") + # Dynamic MODE_* parsing + modes_yaml_path = os.path.join(REPO_ROOT, "group_vars", "all", "01_modes.yml") modes_meta = load_modes_from_yaml(modes_yaml_path) modes_spec = add_dynamic_mode_args(parser, modes_meta) args = parser.parse_args() + + # Validate application IDs validate_application_ids(args.inventory, args.id) + # Build final mode map modes = build_modes_from_args(modes_spec, args) modes["MODE_LOGS"] = args.logs modes["host_type"] = args.host_type + # Run playbook run_ansible_playbook( inventory=args.inventory, modes=modes, diff --git a/tests/unit/cli/deploy/__init__.py b/tests/unit/cli/deploy/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/cli/test_deploy.py b/tests/unit/cli/deploy/test_dedicated.py similarity index 68% rename from tests/unit/cli/test_deploy.py rename to tests/unit/cli/deploy/test_dedicated.py index 436c8caf..87d4802a 100644 --- a/tests/unit/cli/test_deploy.py +++ b/tests/unit/cli/deploy/test_dedicated.py @@ -1,15 +1,12 @@ -#!/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 +from cli.deploy import dedicated as deploy + class TestParseBoolLiteral(unittest.TestCase): def test_true_values(self): @@ -32,13 +29,13 @@ class TestParseBoolLiteral(unittest.TestCase): 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 +MODE_CLEANUP: true # cleanup before deploy +MODE_DEBUG: false # debug output +MODE_ASSERT: null # inventory assertion +OTHER_KEY: true # should be ignored (no MODE_ prefix) """ + with tempfile.NamedTemporaryFile("w+", delete=False, encoding="utf-8") as f: path = f.name f.write(content) @@ -49,7 +46,6 @@ INVALID_KEY: true # ignored because no MODE_ prefix 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} @@ -63,10 +59,29 @@ INVALID_KEY: true # ignored because no MODE_ prefix self.assertIsNone(by_name["MODE_ASSERT"]["default"]) self.assertEqual(by_name["MODE_CLEANUP"]["help"], "cleanup before deploy") + def test_load_modes_ignores_non_mode_keys(self): + content = """\ +FOO: true +BAR: false +MODE_FOO: true +""" + + 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) + + self.assertEqual(len(modes), 1) + self.assertEqual(modes[0]["name"], "MODE_FOO") + 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"}, @@ -74,59 +89,62 @@ class TestDynamicModes(unittest.TestCase): ] 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) + # No flags passed → defaults apply 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 + self.assertNotIn("MODE_ASSERT", modes) # explicit, not set - def test_add_dynamic_mode_args_and_build_modes_flags(self): + def test_add_dynamic_mode_args_and_build_modes_flags_true(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"] - ) + # MODE_CLEANUP: true → --skip-cleanup sets it to False + # MODE_DEBUG: false → --debug sets it to True + # MODE_ASSERT: None → --assert true sets it to 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"]) + def test_add_dynamic_mode_args_and_build_modes_flags_false_explicit(self): + from argparse import ArgumentParser + + parser = ArgumentParser() + spec = deploy.add_dynamic_mode_args(parser, self.modes_meta) + + # explicit false for MODE_ASSERT + args = parser.parse_args(["--assert", "false"]) + modes = deploy.build_modes_from_args(spec, args) + + self.assertTrue(modes["MODE_CLEANUP"]) # still default True + self.assertFalse(modes["MODE_DEBUG"]) # still default False + self.assertIn("MODE_ASSERT", modes) + self.assertFalse(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. - """ + # Should not raise 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}, @@ -153,20 +171,18 @@ class TestRunAnsiblePlaybook(unittest.TestCase): 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.) + Side effect for subprocess.run that: + - records all commands to calls_store + - returns a CompletedProcess with ansible_rc for ansible-playbook + - returns success (rc=0) for all other commands """ + 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, @@ -175,28 +191,19 @@ class TestRunAnsiblePlaybook(unittest.TestCase): 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 + "MODE_DEBUG": True, } inventory_path = "inventories/github-ci/servers.yml" @@ -210,16 +217,14 @@ class TestRunAnsiblePlaybook(unittest.TestCase): limit=limit, allowed_applications=allowed_apps, password_file=password_file, - verbose=1, # explicitly set, then raised by MODE_DEBUG + verbose=1, 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. + # Cleanup, build, tests self.assertTrue( any(call == ["make", "clean"] for call in calls), "Expected 'make clean' when MODE_CLEANUP is true", @@ -241,12 +246,12 @@ class TestRunAnsiblePlaybook(unittest.TestCase): "Expected inventory validation call (... inventory.py ...)", ) - # Last command should be ansible-playbook + # The last call 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 + # Inventory and playbook ordering: -i self.assertIn("-i", last_cmd) self.assertIn(inventory_path, last_cmd) @@ -254,22 +259,22 @@ class TestRunAnsiblePlaybook(unittest.TestCase): 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")), + playbook_arg.endswith("playbook.yml"), f"playbook argument should end with 'playbook.yml', got: {playbook_arg}", ) - # Check --limit / -l + # Limit handling self.assertIn("-l", last_cmd) self.assertIn(limit, last_cmd) - # allowed_applications extra var + # 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 + # MODE_* variables self.assertIn("MODE_CLEANUP=true", last_cmd_str) self.assertIn("MODE_ASSERT=true", last_cmd_str) self.assertIn("MODE_DEBUG=true", last_cmd_str) @@ -278,27 +283,22 @@ class TestRunAnsiblePlaybook(unittest.TestCase): self.assertIn("--vault-password-file", last_cmd) self.assertIn(password_file, last_cmd) - # --diff should be present + # Diff flag self.assertIn("--diff", last_cmd) - # Verbosity: MODE_DEBUG enforces at least -vvv regardless of initial -v + # Verbosity should be at least -vvv when MODE_DEBUG=True 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 + # Ensure we did not accidentally set text/capture_output in the final run last_call = mock_run.call_args_list[-1] - # In streaming mode we no longer pass text/capture_output to subprocess.run self.assertNotIn("text", last_call.kwargs) self.assertNotIn("capture_output", last_call.kwargs) @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. - """ + def test_run_ansible_playbook_failure_exits_with_code_and_skips_phases(self, mock_run): calls: List[List[str]] = [] mock_run.side_effect = self._fake_run_side_effect(calls, ansible_rc=4) @@ -327,6 +327,46 @@ class TestRunAnsiblePlaybook(unittest.TestCase): any(call and call[0] == "ansible-playbook" for call in calls), "ansible-playbook should have been invoked once in the error path", ) + # No cleanup, no build, no tests, no inventory validation + self.assertFalse(any(call == ["make", "clean"] for call in calls)) + self.assertFalse(any(call == ["make", "messy-build"] for call in calls)) + self.assertFalse(any(call == ["make", "messy-test"] for call in calls)) + self.assertFalse( + any( + isinstance(call, list) + and any("inventory.py" in part for part in call) + for call in calls + ) + ) + + @mock.patch("subprocess.run") + def test_run_ansible_playbook_cleanup_with_logs_uses_clean_keep_logs(self, mock_run): + 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": False, + "MODE_DEBUG": False, + } + + 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=True, + diff=False, + ) + + self.assertTrue( + any(call == ["make", "clean-keep-logs"] for call in calls), + "Expected 'make clean-keep-logs' when MODE_CLEANUP=true and logs=True", + ) if __name__ == "__main__":