Finished integration test for dependencies

This commit is contained in:
Kevin Veen-Birkenbach 2025-05-20 06:29:17 +02:00
parent a51a474cb3
commit 23ccaca3aa
No known key found for this signature in database
GPG Key ID: 44D8F11FD62F878E

View File

@ -1,7 +1,6 @@
import unittest
import os
import yaml
import warnings
def load_yaml_file(file_path):
"""Load a YAML file and return its content."""
@ -9,93 +8,60 @@ def load_yaml_file(file_path):
return yaml.safe_load(file) or {}
def get_meta_info(role_path):
"""Extract dependencies and run orders (before/after) from the meta/main.yml of a role."""
"""Extract dependencies from the meta/main.yml of a role."""
meta_file = os.path.join(role_path, 'meta', 'main.yml')
if not os.path.isfile(meta_file):
return [], [], []
return []
meta_data = load_yaml_file(meta_file)
dependencies = meta_data.get('dependencies', [])
run_order = meta_data.get('applications_run_order', {})
return dependencies, run_order.get('before', []), run_order.get('after', [])
return dependencies
def resolve_dependencies(roles_dir):
"""Resolve all role dependencies and detect circular dependencies."""
visited = set()
stack = set()
visited = set() # Tracks roles that have been processed
def visit(role_path):
def visit(role_path, stack):
role_name = os.path.basename(role_path)
# Check for circular dependencies
# Check for circular dependencies (if role is already in the current stack)
if role_name in stack:
warning_msg = f"Circular dependency detected with role: {role_name} in path: {' -> '.join(stack)}"
warnings.warn(warning_msg)
raise ValueError(f"Circular dependency detected with role: {role_name} in path: {' -> '.join(stack)}")
raise ValueError(f"Circular dependency detected: {' -> '.join(stack)} -> {role_name}")
# Check if role is already processed
if role_name in visited:
return [], []
return []
# Mark role as being processed
stack.add(role_name)
# Mark role as visited and add to stack
visited.add(role_name)
stack.append(role_name)
# Get dependencies and before/after orders
dependencies, before, after = get_meta_info(role_path)
# Resolve before dependencies
before_order, after_order = [], []
# Get dependencies and resolve them
dependencies = get_meta_info(role_path)
for dep in dependencies:
# Add more detailed debug information for `dep`
try:
print(f"Visiting dependency: {dep} for role {role_name}, role_path: {role_path}")
# Check if dep is a dictionary and extract the role name if necessary
if isinstance(dep, dict):
print(f"Dependency {dep} is a dictionary. Extracting role name...")
dep = list(dep.keys())[0] # Extract the key if it's a dictionary
dep_path = os.path.join(roles_dir, dep)
print(f"Resolved dependency path: {dep_path}")
visit(dep_path, stack) # Recurse into dependencies
dep_before, dep_after = visit(dep_path) # Recurse into dependencies
before_order.extend(dep_before)
after_order.extend(dep_after)
except Exception as e:
# Print error message with role path to help with debugging
raise ValueError(f"Error in role {role_name} while processing dependency {dep} at path {role_path}: {str(e)}")
# Add the current role's own before/after orders
before_order.extend(before)
after_order.extend(after)
# Mark role as processed
stack.remove(role_name)
return before_order, after_order
all_run_orders = {}
stack.pop() # Remove the current role from the stack
for role_name in os.listdir(roles_dir):
role_path = os.path.join(roles_dir, role_name)
if os.path.isdir(role_path):
try:
all_run_orders[role_path] = visit(role_path)
visit(role_path, []) # Start recursion from this role
except ValueError as e:
# Improved error reporting with role path information
raise ValueError(f"Error processing role '{role_name}' at path '{role_path}': {str(e)}")
return all_run_orders
class TestRoleDependencies(unittest.TestCase):
def test_no_circular_dependencies(self):
roles_dir = "roles"
roles_dir = "roles" # Path to the roles directory
try:
run_orders = resolve_dependencies(roles_dir)
resolve_dependencies(roles_dir)
except ValueError as e:
self.fail(f"Circular dependency detected: {e}")
# If no exception, the test passed
self.assertIsInstance(run_orders, dict)
self.assertTrue(True)
if __name__ == '__main__':
unittest.main()