test(integration): fail if reset.yml exists but is never included

Updated test_mode_reset.py to also validate roles that contain a reset
task file (*_reset.yml or reset.yml) even when no mode_reset keyword is
found. The test now:

- Detects roles with reset files but no include, and fails accordingly.
- Ignores commented include_tasks and when lines.
- Ensures exactly one non-commented include of the reset file exists.
- Requires that the include is guarded in the same task block by a
  when containing mode_reset | bool (with optional extra conditions).

This prevents silent omissions of reset task integration.

https://chatgpt.com/share/6899b745-7150-800f-98f3-ca714486f5ba
This commit is contained in:
Kevin Veen-Birkenbach 2025-08-11 11:27:15 +02:00
parent 1fcf072257
commit 21b6362bc1
No known key found for this signature in database
GPG Key ID: 44D8F11FD62F878E

View File

@ -1,4 +1,6 @@
#!/usr/bin/env python3
import os import os
import re
import unittest import unittest
# Base directory for roles (adjust if needed) # Base directory for roles (adjust if needed)
@ -6,9 +8,13 @@ BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../roles'
class TestModeResetIntegration(unittest.TestCase): class TestModeResetIntegration(unittest.TestCase):
""" """
Integration test to verify that when 'mode_reset' is used in any task file, Verify that a role either mentioning 'mode_reset' under tasks/ OR containing a reset file:
the role provides a *_reset.yml (or reset.yml) and includes it correctly in main.yml, - provides a *_reset.yml (or reset.yml) in tasks/,
and that the include_tasks for that file with the mode_reset condition appears only once. - includes it exactly once across tasks/*.yml,
- and the include is guarded in the SAME task block by a non-commented `when`
that contains `mode_reset | bool` (inline, list, or array).
Additional conditions (e.g., `and something`) are allowed.
Commented-out conditions (e.g., `#when: ...` or `# include_tasks: ...`) do NOT count.
""" """
def test_mode_reset_tasks(self): def test_mode_reset_tasks(self):
@ -20,77 +26,143 @@ class TestModeResetIntegration(unittest.TestCase):
if not os.path.isdir(tasks_dir): if not os.path.isdir(tasks_dir):
self.skipTest(f"Role '{role_name}' has no tasks directory.") self.skipTest(f"Role '{role_name}' has no tasks directory.")
# Look for 'mode_reset' in task files # Gather all task files
mode_reset_found = False task_files = []
for root, _, files in os.walk(tasks_dir): for root, _, files in os.walk(tasks_dir):
for fname in files: for fname in files:
if not fname.lower().endswith(('.yml', '.yaml')): if fname.lower().endswith(('.yml', '.yaml')):
continue task_files.append(os.path.join(root, fname))
file_path = os.path.join(root, fname)
with open(file_path, 'r', encoding='utf-8') as f: # Detect any 'mode_reset' usage
mode_reset_found = False
for fp in task_files:
try:
with open(fp, 'r', encoding='utf-8') as f:
if 'mode_reset' in f.read(): if 'mode_reset' in f.read():
mode_reset_found = True mode_reset_found = True
break break
if mode_reset_found: except (UnicodeDecodeError, OSError):
break continue
if not mode_reset_found: # Detect reset files in tasks/ root
self.skipTest(f"Role '{role_name}': no mode_reset usage detected.") try:
task_root_listing = os.listdir(tasks_dir)
# Check *_reset.yml exists except OSError:
task_root_listing = []
reset_files = [ reset_files = [
fname for fname in os.listdir(tasks_dir) fname for fname in task_root_listing
if fname.endswith('_reset.yml') or fname == 'reset.yml' if fname.endswith('_reset.yml') or fname == 'reset.yml'
] ]
# Decide if this role must be validated:
# - if it mentions mode_reset anywhere under tasks/, OR
# - if it has a reset file in tasks/ root
should_check = mode_reset_found or bool(reset_files)
if not should_check:
self.skipTest(f"Role '{role_name}': no mode_reset usage and no reset file found.")
# If we check, a reset file MUST exist
self.assertTrue( self.assertTrue(
reset_files, reset_files,
f"Role '{role_name}': 'mode_reset' used but no *_reset.yml or reset.yml found in tasks/." f"Role '{role_name}': expected a *_reset.yml or reset.yml in tasks/."
) )
# Check main.yml exists # Patterns to find non-commented reset include occurrences
main_yml = os.path.join(tasks_dir, 'main.yml') def include_patterns(rf: str):
self.assertTrue( # Accept:
os.path.isfile(main_yml), # - include_tasks: reset.yml (quoted or unquoted)
f"Role '{role_name}': tasks/main.yml is missing." # - ansible.builtin.include_tasks: reset.yml
# - include_tasks:\n file: reset.yml
# All must be non-commented (no leading '#')
q = r'(?:' + re.escape(rf) + r'|"' + re.escape(rf) + r'"|\'' + re.escape(rf) + r'\')'
return [
re.compile(
rf'(?m)^(?<!#)\s*-?\s*(?:ansible\.builtin\.)?include_tasks:\s*{q}\s*$'
),
re.compile(
rf'(?ms)^(?<!#)\s*-?\s*(?:ansible\.builtin\.)?include_tasks:\s*\n[^-\S\r\n]*file:\s*{q}\s*$'
),
]
include_occurrences = [] # (file_path, reset_file, (span_start, span_end))
# Search every tasks/*.yml for exactly one include of any reset file
for fp in task_files:
try:
with open(fp, 'r', encoding='utf-8') as f:
content = f.read()
for rf in reset_files:
for patt in include_patterns(rf):
for m in patt.finditer(content):
include_occurrences.append((fp, rf, m.span()))
except (UnicodeDecodeError, OSError):
continue
self.assertGreater(
len(include_occurrences), 0,
f"Role '{role_name}': must include one of {reset_files} in some non-commented tasks/*.yml."
)
self.assertEqual(
len(include_occurrences), 1,
f"Role '{role_name}': reset include must appear exactly once across tasks/*.yml, "
f"found {len(include_occurrences)}."
) )
with open(main_yml, 'r', encoding='utf-8') as f: # Verify a proper 'when' containing 'mode_reset | bool' exists in the SAME task block
include_fp, included_rf, span = include_occurrences[0]
with open(include_fp, 'r', encoding='utf-8') as f:
content = f.read() content = f.read()
lines = content.splitlines()
# Match the actual reset file name used in include_tasks # Compute the line index where the include occurs
found_include = None include_line_idx = content.count('\n', 0, span[0])
for reset_file in reset_files:
if f'include_tasks: {reset_file}' in content:
found_include = reset_file
break
self.assertIsNotNone( def is_task_start(line: str) -> bool:
found_include, # new task begins with "- " at current indentation
f"Role '{role_name}': tasks/main.yml must include one of {reset_files} with 'include_tasks'." return re.match(r'^\s*-\s', line) is not None
# Expand upwards to task start
start_idx = include_line_idx
while start_idx > 0 and not is_task_start(lines[start_idx]):
start_idx -= 1
# Expand downwards to next task start or EOF
end_idx = include_line_idx
while end_idx + 1 < len(lines) and not is_task_start(lines[end_idx + 1]):
end_idx += 1
task_block = "\n".join(lines[start_idx:end_idx + 1])
# Build regexes that:
# - DO NOT match commented lines (require ^\s*when: not preceded by '#')
# - Allow additional conditions inline (and/or/parentheses/etc.)
# - Support list form and yaml array form
when_inline = re.search(
r'(?m)^(?<!#)\s*when:\s*(?!\[)(?:(?!\n).)*mode_reset\s*\|\s*bool',
task_block
)
when_list = re.search(
r'(?ms)^(?<!#)\s*when:\s*\n' # non-commented when:
r'(?:(?:\s*#.*\n)|(?:\s*-\s*.*\n))*' # comments or other list items
r'\s*-\s*[^#\n]*mode_reset\s*\|\s*bool[^#\n]*$', # list item with mode_reset | bool (not commented)
task_block
)
when_array = re.search(
r'(?m)^(?<!#)\s*when:\s*\[[^\]\n]*mode_reset\s*\|\s*bool[^\]\n]*\]',
task_block
) )
# Check the inclusion has the correct when condition when_ok = bool(when_inline or when_list or when_array)
include_line = f'include_tasks: {found_include}'
when_line = 'when: mode_reset | bool'
self.assertIn( self.assertTrue(
include_line, when_ok,
content, (
f"Role '{role_name}': tasks/main.yml missing '{include_line}'." f"Role '{role_name}': file '{include_fp}' must guard the reset include "
) f"with a non-commented 'when' containing 'mode_reset | bool'. "
self.assertIn( f"Commented-out conditions do not count."
when_line, )
content,
f"Role '{role_name}': tasks/main.yml missing '{when_line}'."
)
self.assertEqual(
content.count(include_line), 1,
f"Role '{role_name}': '{include_line}' must appear exactly once."
)
self.assertEqual(
content.count(when_line), 1,
f"Role '{role_name}': '{when_line}' must appear exactly once."
) )
if __name__ == '__main__': if __name__ == '__main__':
unittest.main(verbosity=2) unittest.main(verbosity=2)