mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2025-11-27 23:26:47 +00:00
Refine role dependency graph/tree builders and tests
- Refactor cli/build/graph.py to use cached metadata and dependency indices for faster graph generation and cleaner separation of concerns - Refactor cli/build/tree.py to delegate per-role processing to process_role() and support parallel execution via ProcessPoolExecutor - Add unit tests for graph helper functions and build_mappings() under tests/unit/cli/build/test_graph.py - Add unit tests for find_roles() and process_role() behaviour under tests/unit/cli/build/test_tree.py - Remove the old include_role dependency integration test which relied on the previous tree.json dependencies bucket For details see ChatGPT conversation: https://chatgpt.com/share/6926b805-28a0-800f-a075-e5250aab5c4a
This commit is contained in:
233
tests/unit/cli/build/test_graph.py
Normal file
233
tests/unit/cli/build/test_graph.py
Normal file
@@ -0,0 +1,233 @@
|
||||
import os
|
||||
import json
|
||||
import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
from io import StringIO
|
||||
from contextlib import redirect_stdout
|
||||
|
||||
from cli.build.graph import (
|
||||
load_meta,
|
||||
load_tasks,
|
||||
build_mappings,
|
||||
output_graph,
|
||||
ALL_KEYS,
|
||||
)
|
||||
|
||||
|
||||
class TestGraphHelpers(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
self.addCleanup(lambda: shutil.rmtree(self.tmpdir, ignore_errors=True))
|
||||
|
||||
def _write_file(self, rel_path: str, content: str) -> str:
|
||||
path = os.path.join(self.tmpdir, rel_path)
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
return path
|
||||
|
||||
def test_load_meta_parses_run_after_and_dependencies(self):
|
||||
meta_path = self._write_file(
|
||||
"roles/role_a/meta/main.yml",
|
||||
"""
|
||||
galaxy_info:
|
||||
author: Test Author
|
||||
run_after:
|
||||
- role_b
|
||||
- role_c
|
||||
dependencies:
|
||||
- role_d
|
||||
- role_e
|
||||
""",
|
||||
)
|
||||
|
||||
meta = load_meta(meta_path)
|
||||
|
||||
self.assertIn("galaxy_info", meta)
|
||||
self.assertEqual(meta["galaxy_info"]["author"], "Test Author")
|
||||
self.assertEqual(meta["run_after"], ["role_b", "role_c"])
|
||||
self.assertEqual(meta["dependencies"], ["role_d", "role_e"])
|
||||
|
||||
def test_load_tasks_filters_out_jinja_and_reads_names(self):
|
||||
tasks_path = self._write_file(
|
||||
"roles/role_a/tasks/main.yml",
|
||||
"""
|
||||
- name: include plain file
|
||||
include_tasks: "subtasks.yml"
|
||||
|
||||
- name: include with dict
|
||||
include_tasks:
|
||||
name: "other.yml"
|
||||
|
||||
- name: include jinja, should be ignored
|
||||
include_tasks: "{{ dynamic_file }}"
|
||||
|
||||
- name: import plain file
|
||||
import_tasks: "legacy.yml"
|
||||
|
||||
- name: import with dict
|
||||
import_tasks:
|
||||
name: "more.yml"
|
||||
|
||||
- name: import jinja, should be ignored
|
||||
import_tasks: "{{ legacy_file }}"
|
||||
""",
|
||||
)
|
||||
|
||||
include_files = load_tasks(tasks_path, "include_tasks")
|
||||
import_files = load_tasks(tasks_path, "import_tasks")
|
||||
|
||||
self.assertEqual(sorted(include_files), ["other.yml", "subtasks.yml"])
|
||||
self.assertEqual(sorted(import_files), ["legacy.yml", "more.yml"])
|
||||
|
||||
|
||||
class TestBuildMappings(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.roles_dir = tempfile.mkdtemp()
|
||||
self.addCleanup(lambda: shutil.rmtree(self.roles_dir, ignore_errors=True))
|
||||
|
||||
def _write_file(self, rel_path: str, content: str) -> str:
|
||||
path = os.path.join(self.roles_dir, rel_path)
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
return path
|
||||
|
||||
def _create_minimal_role(self, name: str, with_meta: bool = False) -> None:
|
||||
os.makedirs(os.path.join(self.roles_dir, name), exist_ok=True)
|
||||
if with_meta:
|
||||
self._write_file(
|
||||
f"{name}/meta/main.yml",
|
||||
"""
|
||||
galaxy_info:
|
||||
author: Minimal
|
||||
""",
|
||||
)
|
||||
|
||||
def test_build_mappings_collects_all_dependency_types(self):
|
||||
# Create roles directory structure
|
||||
self._create_minimal_role("role_b")
|
||||
self._create_minimal_role("role_c")
|
||||
self._create_minimal_role("role_d")
|
||||
self._create_minimal_role("role_e")
|
||||
|
||||
# Role A with meta (run_after + dependencies)
|
||||
self._write_file(
|
||||
"role_a/meta/main.yml",
|
||||
"""
|
||||
galaxy_info:
|
||||
author: Role A Author
|
||||
run_after:
|
||||
- role_b
|
||||
dependencies:
|
||||
- role_c
|
||||
""",
|
||||
)
|
||||
|
||||
# Role A tasks with include_role, import_role, include_tasks, import_tasks
|
||||
self._write_file(
|
||||
"role_a/tasks/main.yml",
|
||||
"""
|
||||
- name: use docker style role
|
||||
include_role:
|
||||
name: role_d
|
||||
|
||||
- name: use import role
|
||||
import_role:
|
||||
name: role_e
|
||||
|
||||
- name: include static tasks file
|
||||
include_tasks: "subtasks.yml"
|
||||
|
||||
- name: import static tasks file
|
||||
import_tasks:
|
||||
name: "legacy.yml"
|
||||
""",
|
||||
)
|
||||
|
||||
# Dummy tasks/meta for other roles not required, but create dirs so they
|
||||
# are recognized as roles.
|
||||
self._create_minimal_role("role_a") # dirs already exist but harmless
|
||||
|
||||
graphs = build_mappings("role_a", self.roles_dir, max_depth=2)
|
||||
|
||||
# Ensure we got all expected graph keys
|
||||
for key in ALL_KEYS:
|
||||
self.assertIn(key, graphs, msg=f"Missing graph key {key!r} in result")
|
||||
|
||||
# Helper to find links in a graph
|
||||
def links_of(key: str):
|
||||
return graphs[key]["links"]
|
||||
|
||||
# run_after_to: role_a -> role_b
|
||||
run_after_links = links_of("run_after_to")
|
||||
self.assertIn(
|
||||
{"source": "role_a", "target": "role_b", "type": "run_after"},
|
||||
run_after_links,
|
||||
)
|
||||
|
||||
# dependencies_to: role_a -> role_c
|
||||
dep_links = links_of("dependencies_to")
|
||||
self.assertIn(
|
||||
{"source": "role_a", "target": "role_c", "type": "dependencies"},
|
||||
dep_links,
|
||||
)
|
||||
|
||||
# include_role_to: role_a -> role_d
|
||||
inc_role_links = links_of("include_role_to")
|
||||
self.assertIn(
|
||||
{"source": "role_a", "target": "role_d", "type": "include_role"},
|
||||
inc_role_links,
|
||||
)
|
||||
|
||||
# import_role_to: role_a -> role_e
|
||||
imp_role_links = links_of("import_role_to")
|
||||
self.assertIn(
|
||||
{"source": "role_a", "target": "role_e", "type": "import_role"},
|
||||
imp_role_links,
|
||||
)
|
||||
|
||||
# include_tasks_to: role_a -> "subtasks.yml"
|
||||
inc_tasks_links = links_of("include_tasks_to")
|
||||
self.assertIn(
|
||||
{"source": "role_a", "target": "subtasks.yml", "type": "include_tasks"},
|
||||
inc_tasks_links,
|
||||
)
|
||||
|
||||
# import_tasks_to: role_a -> "legacy.yml"
|
||||
imp_tasks_links = links_of("import_tasks_to")
|
||||
self.assertIn(
|
||||
{"source": "role_a", "target": "legacy.yml", "type": "import_tasks"},
|
||||
imp_tasks_links,
|
||||
)
|
||||
|
||||
def test_output_graph_console_prints_header_and_yaml(self):
|
||||
graph_data = {"nodes": [{"id": "role_a"}], "links": []}
|
||||
buf = StringIO()
|
||||
with redirect_stdout(buf):
|
||||
output_graph(graph_data, "console", "role_a", "include_role_to")
|
||||
|
||||
out = buf.getvalue()
|
||||
self.assertIn("--- role_a_include_role_to ---", out)
|
||||
self.assertIn("nodes:", out)
|
||||
self.assertIn("role_a", out)
|
||||
|
||||
def test_output_graph_writes_json_file(self):
|
||||
graph_data = {"nodes": [{"id": "role_a"}], "links": []}
|
||||
# Use current working directory; file is small and cleaned manually.
|
||||
fname = "role_a_include_role_to.json"
|
||||
try:
|
||||
output_graph(graph_data, "json", "role_a", "include_role_to")
|
||||
self.assertTrue(os.path.exists(fname))
|
||||
|
||||
with open(fname, "r", encoding="utf-8") as f:
|
||||
loaded = json.load(f)
|
||||
self.assertEqual(graph_data, loaded)
|
||||
finally:
|
||||
if os.path.exists(fname):
|
||||
os.remove(fname)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
109
tests/unit/cli/build/test_tree.py
Normal file
109
tests/unit/cli/build/test_tree.py
Normal file
@@ -0,0 +1,109 @@
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
from io import StringIO
|
||||
from contextlib import redirect_stdout
|
||||
from unittest.mock import patch
|
||||
|
||||
from cli.build import tree as tree_module
|
||||
|
||||
|
||||
class TestFindRoles(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.roles_dir = tempfile.mkdtemp()
|
||||
self.addCleanup(lambda: shutil.rmtree(self.roles_dir, ignore_errors=True))
|
||||
|
||||
def test_find_roles_returns_only_directories(self):
|
||||
# Create some role directories and a non-directory entry
|
||||
os.makedirs(os.path.join(self.roles_dir, "role_a"))
|
||||
os.makedirs(os.path.join(self.roles_dir, "role_b"))
|
||||
with open(os.path.join(self.roles_dir, "not_a_role.txt"), "w", encoding="utf-8") as f:
|
||||
f.write("dummy")
|
||||
|
||||
roles = dict(tree_module.find_roles(self.roles_dir))
|
||||
self.assertEqual(set(roles.keys()), {"role_a", "role_b"})
|
||||
self.assertTrue(all(os.path.isdir(path) for path in roles.values()))
|
||||
|
||||
|
||||
class TestProcessRole(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
# We use a temporary "roles" directory and a separate shadow folder.
|
||||
self.roles_dir = tempfile.mkdtemp()
|
||||
self.shadow_dir = tempfile.mkdtemp()
|
||||
self.addCleanup(lambda: shutil.rmtree(self.roles_dir, ignore_errors=True))
|
||||
self.addCleanup(lambda: shutil.rmtree(self.shadow_dir, ignore_errors=True))
|
||||
|
||||
# Create a minimal role directory
|
||||
os.makedirs(os.path.join(self.roles_dir, "myrole"), exist_ok=True)
|
||||
|
||||
def test_process_role_writes_tree_json_and_does_not_mutate_graphs(self):
|
||||
graphs = {
|
||||
"include_role_to": {"nodes": [{"id": "myrole"}], "links": []},
|
||||
"custom_key": {"value": 42}, # sentinel to ensure we do not modify the dict
|
||||
}
|
||||
|
||||
with patch.object(tree_module, "build_mappings", return_value=graphs) as mocked_build:
|
||||
tree_module.process_role(
|
||||
role_name="myrole",
|
||||
roles_dir=self.roles_dir,
|
||||
depth=0,
|
||||
shadow_folder=self.shadow_dir,
|
||||
output="json",
|
||||
preview=False,
|
||||
verbose=False,
|
||||
no_include_role=False,
|
||||
no_import_role=False,
|
||||
no_dependencies=False,
|
||||
no_run_after=False,
|
||||
)
|
||||
|
||||
mocked_build.assert_called_once()
|
||||
|
||||
tree_file = os.path.join(self.shadow_dir, "myrole", "meta", "tree.json")
|
||||
self.assertTrue(os.path.exists(tree_file), "tree.json was not written")
|
||||
|
||||
with open(tree_file, "r", encoding="utf-8") as f:
|
||||
written_graphs = json.load(f)
|
||||
|
||||
# The written file must be exactly what build_mappings returned
|
||||
self.assertEqual(graphs, written_graphs)
|
||||
# Especially: no extra top-level "dependencies" block is added
|
||||
self.assertNotIn("dependencies", written_graphs)
|
||||
|
||||
def test_process_role_preview_calls_output_graph_and_does_not_write_file(self):
|
||||
graphs = {
|
||||
"graph_a": {"nodes": [{"id": "myrole"}], "links": []},
|
||||
"graph_b": {"nodes": [], "links": []},
|
||||
}
|
||||
|
||||
with patch.object(tree_module, "build_mappings", return_value=graphs), patch.object(
|
||||
tree_module, "output_graph"
|
||||
) as mocked_output:
|
||||
buf = StringIO()
|
||||
with redirect_stdout(buf):
|
||||
tree_module.process_role(
|
||||
role_name="myrole",
|
||||
roles_dir=self.roles_dir,
|
||||
depth=0,
|
||||
shadow_folder=self.shadow_dir,
|
||||
output="json",
|
||||
preview=True,
|
||||
verbose=True,
|
||||
no_include_role=False,
|
||||
no_import_role=False,
|
||||
no_dependencies=False,
|
||||
no_run_after=False,
|
||||
)
|
||||
|
||||
# output_graph must be called once per graph entry
|
||||
self.assertEqual(mocked_output.call_count, len(graphs))
|
||||
|
||||
# In preview mode, no tree.json should be written
|
||||
tree_file = os.path.join(self.shadow_dir, "myrole", "meta", "tree.json")
|
||||
self.assertFalse(os.path.exists(tree_file))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,143 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import tempfile
|
||||
import shutil
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
# Absoluter Pfad zum tree.py Script (wie im vorhandenen Test)
|
||||
SCRIPT_PATH = os.path.abspath(
|
||||
os.path.join(os.path.dirname(__file__), "../../../../cli/build/tree.py")
|
||||
)
|
||||
|
||||
class TestTreeIncludeRoleDependencies(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# Temp roles root
|
||||
self.roles_dir = tempfile.mkdtemp()
|
||||
|
||||
# Producer-Role (die wir scannen) + Zielrollen für Matches
|
||||
self.producer = "producer"
|
||||
self.producer_path = os.path.join(self.roles_dir, self.producer)
|
||||
os.makedirs(os.path.join(self.producer_path, "tasks"))
|
||||
os.makedirs(os.path.join(self.producer_path, "meta"))
|
||||
|
||||
# Rollen, die durch Pattern/Loops gematcht werden sollen
|
||||
self.roles_to_create = [
|
||||
"sys-ctl-hlth-webserver",
|
||||
"sys-ctl-hlth-csp",
|
||||
"svc-db-postgres",
|
||||
"svc-db-mysql",
|
||||
"axb", # für a{{ database_type }}b → a*b
|
||||
"ayyb", # für a{{ database_type }}b → a*b
|
||||
"literal-role", # für reinen Literalnamen
|
||||
]
|
||||
for r in self.roles_to_create:
|
||||
os.makedirs(os.path.join(self.roles_dir, r, "meta"), exist_ok=True)
|
||||
|
||||
# tasks/main.yml mit allen geforderten Varianten
|
||||
tasks_yaml = """
|
||||
- name: Include health dependencies
|
||||
include_role:
|
||||
name: "{{ item }}"
|
||||
loop:
|
||||
- sys-ctl-hlth-webserver
|
||||
- sys-ctl-hlth-csp
|
||||
|
||||
- name: Pattern with literal + var suffix
|
||||
include_role:
|
||||
name: "svc-db-{{ database_type }}"
|
||||
|
||||
- name: Pattern with literal prefix/suffix around var
|
||||
include_role:
|
||||
name: "a{{ database_type }}b"
|
||||
|
||||
- name: Pure variable only (should be ignored)
|
||||
include_role:
|
||||
name: "{{ database_type }}"
|
||||
|
||||
- name: Pure literal include
|
||||
include_role:
|
||||
name: "literal-role"
|
||||
"""
|
||||
with open(os.path.join(self.producer_path, "tasks", "main.yml"), "w", encoding="utf-8") as f:
|
||||
f.write(tasks_yaml)
|
||||
|
||||
# shadow folder
|
||||
self.shadow_dir = tempfile.mkdtemp()
|
||||
|
||||
# Patch argv
|
||||
self.orig_argv = sys.argv[:]
|
||||
sys.argv = [
|
||||
SCRIPT_PATH,
|
||||
"-d", self.roles_dir,
|
||||
"-s", self.shadow_dir,
|
||||
"-o", "json",
|
||||
]
|
||||
|
||||
# Ensure project root on sys.path
|
||||
project_root = os.path.abspath(
|
||||
os.path.join(os.path.dirname(__file__), "../../../../")
|
||||
)
|
||||
if project_root not in sys.path:
|
||||
sys.path.insert(0, project_root)
|
||||
|
||||
def tearDown(self):
|
||||
sys.argv = self.orig_argv
|
||||
shutil.rmtree(self.roles_dir)
|
||||
shutil.rmtree(self.shadow_dir)
|
||||
|
||||
@patch("cli.build.tree.output_graph")
|
||||
@patch("cli.build.tree.build_mappings")
|
||||
def test_include_role_dependencies_detected(self, mock_build_mappings, mock_output_graph):
|
||||
# Basis-Graph leer, damit nur unsere Dependencies sichtbar sind
|
||||
mock_build_mappings.return_value = {}
|
||||
|
||||
# Import und Ausführen
|
||||
import importlib
|
||||
tree_mod = importlib.import_module("cli.build.tree")
|
||||
tree_mod.main()
|
||||
|
||||
# Erwarteter Pfad im Shadow-Folder
|
||||
expected_tree_path = os.path.join(
|
||||
self.shadow_dir, self.producer, "meta", "tree.json"
|
||||
)
|
||||
self.assertTrue(
|
||||
os.path.isfile(expected_tree_path),
|
||||
f"tree.json not found at {expected_tree_path}"
|
||||
)
|
||||
|
||||
# JSON laden und Abhängigkeiten prüfen
|
||||
with open(expected_tree_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
# Erwartete include_role-Dependenzen:
|
||||
expected = sorted([
|
||||
"sys-ctl-hlth-webserver", # aus loop
|
||||
"sys-ctl-hlth-csp", # aus loop
|
||||
"svc-db-postgres", # aus svc-db-{{ database_type }}
|
||||
"svc-db-mysql", # aus svc-db-{{ database_type }}
|
||||
"axb", # aus a{{ database_type }}b
|
||||
"ayyb", # aus a{{ database_type }}b
|
||||
"literal-role", # reiner Literalname
|
||||
])
|
||||
|
||||
deps = (
|
||||
data
|
||||
.get("dependencies", {})
|
||||
.get("include_role", [])
|
||||
)
|
||||
self.assertEqual(deps, expected, "include_role dependencies mismatch")
|
||||
|
||||
# Sicherstellen, dass der pure Variable-Name "{{ database_type }}" NICHT aufgenommen wurde
|
||||
self.assertNotIn("{{ database_type }}", deps, "pure variable include should be ignored")
|
||||
|
||||
# Sicherstellen, dass im Original-meta der Producer-Role nichts geschrieben wurde
|
||||
original_tree_path = os.path.join(self.producer_path, "meta", "tree.json")
|
||||
self.assertFalse(
|
||||
os.path.exists(original_tree_path),
|
||||
"tree.json should NOT be written to the real meta/ folder"
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user