From fb977e975acbb1523a45835338b8ee252c3a9ee2 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Fri, 12 Jun 2026 18:41:16 +0200 Subject: [PATCH] test(lint): recognise compose-only task gates + skip comment / YAML metadata lines MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The swarm-compat lints (`compose-chdir-in-task`, `compose-verb-in-task`, `docker-compose-raw-call`) currently flag every offending line and only accept per-line / file-head `# nocheck:` markers. That is noisy for two patterns that are objectively safe: 1. Tasks gated by `when: DEPLOYMENT_MODE \!= 'swarm'` (or `== 'compose'`) never reach a swarm host - the compose verbs inside are legitimate. 2. YAML comment lines, handler-trigger list items, and human-readable metadata fields (`name:`, `msg:`, `title:`, `label:`, `description:`) contain the word `compose` as documentation, not as a shell call. The lints now skip both classes automatically. - utils/annotations/task_gate.py: new helper `is_task_compose_only_gated(lines, idx)`. Walks the surrounding task block via a line-based heuristic and matches `when:` against the compose-only DEPLOYMENT_MODE expressions. Documented as deliberately shallow (no YAML parse). - tests/lint/ansible/shell/test_no_compose_chdir_in_shell_tasks.py and tests/lint/ansible/shell/test_no_compose_verb_in_shell_tasks.py: add a file-level `is_suppressed_in_head` skip, the `is_task_compose_only_gated` skip, and -- for the verb lint -- three extra regex exclusions for notify-list items, YAML metadata fields, and quoted list items. - tests/lint/ansible/shell/test_no_docker_compose_raw_call.py: skip YAML comment lines (bluesky and jenkins each had a `# … docker compose build …` doc-comment that was being flagged as a real call). - tests/unit/utils/annotations/test_task_gate.py: 21 cases covering block-bounds + gate detection, including a `TestKnownLimitations` class that pins the heuristic's blind spots (folded-scalar `when:`, list-form `when:`, `in`/`not in`, unquoted literal, outer-`when:` inheritance into `block:` children) so a future YAML-aware refactor knows exactly which assertions to flip. --- .../test_no_compose_chdir_in_shell_tasks.py | 7 +- .../test_no_compose_verb_in_shell_tasks.py | 21 +- .../shell/test_no_docker_compose_raw_call.py | 2 + .../unit/utils/annotations/test_task_gate.py | 386 ++++++++++++++++++ utils/annotations/task_gate.py | 68 +++ 5 files changed, 480 insertions(+), 4 deletions(-) create mode 100644 tests/unit/utils/annotations/test_task_gate.py create mode 100644 utils/annotations/task_gate.py diff --git a/tests/lint/ansible/shell/test_no_compose_chdir_in_shell_tasks.py b/tests/lint/ansible/shell/test_no_compose_chdir_in_shell_tasks.py index 2a9c0e3a58..d4403f7671 100644 --- a/tests/lint/ansible/shell/test_no_compose_chdir_in_shell_tasks.py +++ b/tests/lint/ansible/shell/test_no_compose_chdir_in_shell_tasks.py @@ -14,7 +14,8 @@ import re import unittest from pathlib import Path -from utils.annotations.suppress import is_suppressed_at +from utils.annotations.suppress import is_suppressed_at, is_suppressed_in_head +from utils.annotations.task_gate import is_task_compose_only_gated from utils.cache.files import iter_project_files_with_content from . import PROJECT_ROOT @@ -48,11 +49,15 @@ class TestNoComposeChdirInShellTasks(unittest.TestCase): if not _is_scan_target(rel): continue lines = content.splitlines() + if is_suppressed_in_head(lines, _RULE): + continue for idx, line in enumerate(lines): if not _COMPOSE_CHDIR.search(line): continue if is_suppressed_at(lines, idx + 1, _RULE, mode="same-or-above"): continue + if is_task_compose_only_gated(lines, idx): + continue findings.append((rel, idx + 1, line.strip())) if findings: diff --git a/tests/lint/ansible/shell/test_no_compose_verb_in_shell_tasks.py b/tests/lint/ansible/shell/test_no_compose_verb_in_shell_tasks.py index 508581b48f..d0193a9a73 100644 --- a/tests/lint/ansible/shell/test_no_compose_verb_in_shell_tasks.py +++ b/tests/lint/ansible/shell/test_no_compose_verb_in_shell_tasks.py @@ -14,7 +14,8 @@ import re import unittest from pathlib import Path -from utils.annotations.suppress import is_suppressed_at +from utils.annotations.suppress import is_suppressed_at, is_suppressed_in_head +from utils.annotations.task_gate import is_task_compose_only_gated from utils.cache.files import iter_project_files_with_content from . import PROJECT_ROOT @@ -25,9 +26,13 @@ _COMPOSE_VERB = re.compile( r"\{\{\s*BIN_COMPOSE\s*\}\}\s+(?:exec|run|restart|logs|up|down|ps|config)\b" r"|(? bool: @@ -51,13 +56,23 @@ class TestNoComposeVerbInShellTasks(unittest.TestCase): if not _is_scan_target(rel): continue lines = content.splitlines() + if is_suppressed_in_head(lines, _RULE): + continue for idx, line in enumerate(lines): - if _COMMENT_LINE.match(line) or _HANDLER_TRIGGER.match(line): + if ( + _COMMENT_LINE.match(line) + or _HANDLER_TRIGGER.match(line) + or _HANDLER_LIST_ITEM.match(line) + or _YAML_NAME_FIELD.match(line) + or _QUOTED_LIST_ITEM.match(line) + ): continue if not _COMPOSE_VERB.search(line): continue if is_suppressed_at(lines, idx + 1, _RULE, mode="same-or-above"): continue + if is_task_compose_only_gated(lines, idx): + continue findings.append((rel, idx + 1, line.strip())) if findings: diff --git a/tests/lint/ansible/shell/test_no_docker_compose_raw_call.py b/tests/lint/ansible/shell/test_no_docker_compose_raw_call.py index 27aeab83c1..5ebac3b385 100644 --- a/tests/lint/ansible/shell/test_no_docker_compose_raw_call.py +++ b/tests/lint/ansible/shell/test_no_docker_compose_raw_call.py @@ -48,6 +48,8 @@ class TestNoDockerComposeRawCall(unittest.TestCase): for idx, line in enumerate(lines): if not _DOCKER_COMPOSE.search(line): continue + if line.lstrip().startswith("#"): + continue if is_suppressed_at(lines, idx + 1, _RULE, mode="same-or-above"): continue findings.append((rel, idx + 1, line.strip())) diff --git a/tests/unit/utils/annotations/test_task_gate.py b/tests/unit/utils/annotations/test_task_gate.py new file mode 100644 index 0000000000..edc750869d --- /dev/null +++ b/tests/unit/utils/annotations/test_task_gate.py @@ -0,0 +1,386 @@ +"""Unit tests for the line-based Ansible task-gate detector. + +Covers ``_task_block_bounds`` and ``is_task_compose_only_gated`` in +``utils.annotations.task_gate``. The module is consumed by the +swarm-compatibility lints (``compose-chdir-in-task``, +``compose-verb-in-task``); regressing it would either re-raise false +positives on already-gated tasks or silently let new violations +through. +""" + +from __future__ import annotations + +import textwrap +import unittest + +from utils.annotations.task_gate import ( + _task_block_bounds, + is_task_compose_only_gated, +) + + +def _lines(text: str) -> list[str]: + """Strip the leading newline and uniform indent off triple-quoted + YAML fixtures, then split into lines (matching the helper's input + format).""" + return textwrap.dedent(text).lstrip("\n").splitlines() + + +def _line_index(lines: list[str], needle: str) -> int: + """Return the index of the first line containing *needle*.""" + for i, line in enumerate(lines): + if needle in line: + return i + raise AssertionError(f"needle {needle!r} not found in fixture") + + +class TestTaskBlockBounds(unittest.TestCase): + def test_single_task_block_spans_whole_file(self): + lines = _lines( + """ + - name: Only task + ansible.builtin.shell: | + compose stop x + args: + chdir: /opt/compose/x + when: DEPLOYMENT_MODE != 'swarm' + """ + ) + chdir_idx = _line_index(lines, "chdir:") + self.assertEqual(_task_block_bounds(lines, chdir_idx), (0, len(lines))) + + def test_two_sibling_tasks_isolate_each_other(self): + lines = _lines( + """ + - name: First task + ansible.builtin.shell: | + compose stop x + args: + chdir: /opt/compose/x + + - name: Second task + ansible.builtin.shell: | + compose stop y + args: + chdir: /opt/compose/y + """ + ) + first_chdir = _line_index(lines, "/opt/compose/x") + second_chdir = _line_index(lines, "/opt/compose/y") + second_name = _line_index(lines, "Second task") + + start_a, end_a = _task_block_bounds(lines, first_chdir) + start_b, end_b = _task_block_bounds(lines, second_chdir) + + self.assertEqual(start_a, 0) + self.assertEqual(end_a, second_name) + self.assertEqual(start_b, second_name) + self.assertEqual(end_b, len(lines)) + + def test_idx_before_any_task_falls_back_to_full_span(self): + lines = _lines( + """ + --- + # role-wide preamble + - name: First task + ansible.builtin.shell: echo hi + """ + ) + self.assertEqual(_task_block_bounds(lines, 0), (0, len(lines))) + + def test_block_start_marker_is_recognised(self): + lines = _lines( + """ + - block: + - name: Inner + ansible.builtin.shell: | + compose stop x + args: + chdir: /opt/compose/x + when: DEPLOYMENT_MODE != 'swarm' + """ + ) + chdir_idx = _line_index(lines, "chdir:") + start, end = _task_block_bounds(lines, chdir_idx) + # The nearest matching start is the inner `- name:` (more + # indented), not the outer `- block:`. Documents that + # semantics so future readers know the helper is shallow. + inner_name = _line_index(lines, "- name: Inner") + self.assertEqual(start, inner_name) + self.assertEqual(end, len(lines)) + + def test_include_tasks_and_import_tasks_are_start_markers(self): + lines = _lines( + """ + - include_tasks: setup.yml + when: DEPLOYMENT_MODE != 'swarm' + + - import_tasks: run.yml + when: DEPLOYMENT_MODE == 'compose' + """ + ) + include_idx = _line_index(lines, "include_tasks:") + import_idx = _line_index(lines, "import_tasks:") + self.assertEqual(_task_block_bounds(lines, include_idx), (0, import_idx)) + self.assertEqual( + _task_block_bounds(lines, import_idx), (import_idx, len(lines)) + ) + + def test_deeper_indented_marker_does_not_close_block(self): + lines = _lines( + """ + - name: Outer + block: + - name: Inner + ansible.builtin.shell: echo hi + when: DEPLOYMENT_MODE != 'swarm' + """ + ) + outer_idx = _line_index(lines, "- name: Outer") + # `Inner` is more indented than `Outer`, so it does NOT end the + # outer block - the whole file is the outer's block. + self.assertEqual(_task_block_bounds(lines, outer_idx), (0, len(lines))) + + +class TestIsTaskComposeOnlyGated(unittest.TestCase): + def test_when_after_offending_line_is_seen(self): + # The lint scans top-down and hits the chdir line before the + # `when:` clause; the helper must look at the whole task body, + # not just lines above the violation. + lines = _lines( + """ + - name: Stop the stack (compose-only) + ansible.builtin.shell: | + compose stop x + args: + chdir: /opt/compose/x + executable: /bin/bash + when: DEPLOYMENT_MODE != 'swarm' + """ + ) + chdir_idx = _line_index(lines, "chdir:") + self.assertTrue(is_task_compose_only_gated(lines, chdir_idx)) + + def test_when_before_offending_line_is_seen(self): + lines = _lines( + """ + - name: Stop the stack + when: DEPLOYMENT_MODE == 'compose' + ansible.builtin.shell: | + compose stop x + args: + chdir: /opt/compose/x + """ + ) + chdir_idx = _line_index(lines, "chdir:") + self.assertTrue(is_task_compose_only_gated(lines, chdir_idx)) + + def test_double_quoted_swarm_literal_is_accepted(self): + lines = _lines( + """ + - name: Task + shell: compose stop x + when: DEPLOYMENT_MODE != "swarm" + """ + ) + idx = _line_index(lines, "compose stop x") + self.assertTrue(is_task_compose_only_gated(lines, idx)) + + def test_task_with_no_when_is_not_gated(self): + lines = _lines( + """ + - name: Unconditional task + shell: compose stop x + args: + chdir: /opt/compose/x + """ + ) + chdir_idx = _line_index(lines, "chdir:") + self.assertFalse(is_task_compose_only_gated(lines, chdir_idx)) + + def test_unrelated_when_is_not_compose_gate(self): + lines = _lines( + """ + - name: Task + shell: compose stop x + args: + chdir: /opt/compose/x + when: enable_x | bool + """ + ) + chdir_idx = _line_index(lines, "chdir:") + self.assertFalse(is_task_compose_only_gated(lines, chdir_idx)) + + def test_swarm_only_when_is_not_compose_gate(self): + # `DEPLOYMENT_MODE == 'swarm'` means the task runs ONLY on + # swarm - exactly where compose verbs would break. Must NOT + # be treated as compose-only. + lines = _lines( + """ + - name: Task + shell: compose stop x + args: + chdir: /opt/compose/x + when: DEPLOYMENT_MODE == 'swarm' + """ + ) + chdir_idx = _line_index(lines, "chdir:") + self.assertFalse(is_task_compose_only_gated(lines, chdir_idx)) + + def test_when_in_sibling_task_does_not_leak(self): + lines = _lines( + """ + - name: Gated task + shell: compose stop x + when: DEPLOYMENT_MODE != 'swarm' + + - name: Ungated task + shell: compose stop y + args: + chdir: /opt/compose/y + """ + ) + ungated_chdir = _line_index(lines, "/opt/compose/y") + self.assertFalse(is_task_compose_only_gated(lines, ungated_chdir)) + + def test_compound_when_expression_with_compose_gate_is_accepted(self): + # Authors often combine the mode gate with another predicate; + # the helper looks for substring presence, not equality. + lines = _lines( + """ + - name: Task + shell: compose stop x + args: + chdir: /opt/compose/x + when: DEPLOYMENT_MODE != 'swarm' and (enable_x | bool) + """ + ) + chdir_idx = _line_index(lines, "chdir:") + self.assertTrue(is_task_compose_only_gated(lines, chdir_idx)) + + def test_idx_before_any_task_marker(self): + # idx 0 with no preceding `- name:` - falls back to scanning + # the whole file. The fixture has no compose gate at all, so + # the result must be False (not silently True). + lines = _lines( + """ + --- + # Preamble lines only + # No task with a compose-only when clause anywhere + """ + ) + self.assertFalse(is_task_compose_only_gated(lines, 0)) + + def test_idx_before_any_task_with_global_gate_falls_back(self): + # When idx falls outside any task block, the helper scans the + # full file - documenting that behaviour explicitly. + lines = _lines( + """ + --- + - name: Some task + when: DEPLOYMENT_MODE != 'swarm' + shell: compose stop x + """ + ) + self.assertTrue(is_task_compose_only_gated(lines, 0)) + + +class TestKnownLimitations(unittest.TestCase): + """The helper is a deliberately shallow, line-based heuristic - it + does NOT load YAML. These tests pin the current limitations so a + future YAML-aware refactor knows exactly which assertions to flip, + and so a maintainer reading the lint output for one of these forms + can recognise the false-positive as a known pattern rather than + chasing a phantom bug. + + Real-world impact: a role hitting one of these forms gets a lint + violation it cannot silence by gating - the only escape today is a + file-level or per-line ``# nocheck:`` marker. + """ + + def test_folded_scalar_when_is_not_detected(self): + # `when: >` puts the expression on a continuation line; the + # line-based regex sees `>` as the expression and never reads + # the next line. + lines = _lines( + """ + - name: Task + shell: compose stop x + when: > + DEPLOYMENT_MODE != 'swarm' + """ + ) + idx = _line_index(lines, "compose stop x") + self.assertFalse(is_task_compose_only_gated(lines, idx)) + + def test_list_form_when_is_not_detected(self): + # Idiomatic AND-list form. The `when:` line carries no inline + # expression, and the helper does not aggregate child list + # items. + lines = _lines( + """ + - name: Task + shell: compose stop x + when: + - DEPLOYMENT_MODE != 'swarm' + - enable_x | bool + """ + ) + idx = _line_index(lines, "compose stop x") + self.assertFalse(is_task_compose_only_gated(lines, idx)) + + def test_in_operator_gate_is_not_detected(self): + # `DEPLOYMENT_MODE in ['compose']` / `not in ['swarm']` are + # both real Ansible idioms; the helper is wired to `==` / `!=` + # only. + lines = _lines( + """ + - name: Task in-form + shell: compose stop x + when: DEPLOYMENT_MODE in ['compose'] + + - name: Task not-in-form + shell: compose stop y + when: DEPLOYMENT_MODE not in ['swarm'] + """ + ) + x_idx = _line_index(lines, "compose stop x") + y_idx = _line_index(lines, "compose stop y") + self.assertFalse(is_task_compose_only_gated(lines, x_idx)) + self.assertFalse(is_task_compose_only_gated(lines, y_idx)) + + def test_unquoted_literal_is_not_detected(self): + # `when: DEPLOYMENT_MODE != swarm` (no quotes around the + # literal) is rare but legal Ansible. The regex insists on + # quotes for safety against partial-word matches. + lines = _lines( + """ + - name: Task + shell: compose stop x + when: DEPLOYMENT_MODE != swarm + """ + ) + idx = _line_index(lines, "compose stop x") + self.assertFalse(is_task_compose_only_gated(lines, idx)) + + def test_outer_when_does_not_propagate_to_inner_block_child(self): + # In real Ansible the outer task's `when:` does propagate to + # children of its `block:`, but the helper's block-bounds + # snap to the inner `- name:` and miss the outer gate. + # Roles relying on this propagation must instead carry a + # `# nocheck:` marker on the inner task or file. + lines = _lines( + """ + - name: Outer (compose-only) + when: DEPLOYMENT_MODE != 'swarm' + block: + - name: Inner + shell: compose stop x + """ + ) + idx = _line_index(lines, "compose stop x") + self.assertFalse(is_task_compose_only_gated(lines, idx)) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/annotations/task_gate.py b/utils/annotations/task_gate.py new file mode 100644 index 0000000000..2b5f4279d3 --- /dev/null +++ b/utils/annotations/task_gate.py @@ -0,0 +1,68 @@ +"""Detect Ansible task-level `when:` gates that exempt compose-only +shell calls from swarm-compat lints. + +A task whose `when:` evaluates compose-only at deploy time (e.g. +`when: DEPLOYMENT_MODE != 'swarm'` or `when: DEPLOYMENT_MODE == 'compose'`) +never reaches a swarm host, so `compose ` / `chdir: directories.instance` +inside that task body is legitimate. This module gives the lints a +heuristic to recognise that gate without parsing YAML. + +The detection is line-based: the task block is the span between the +nearest `- name:` (at indent N) above the offending line and the next +top-level structural marker (`- name:` at indent <= N, or end of +file). Within that span we look for any `when:` clause whose textual +value matches a compose-only DEPLOYMENT_MODE expression. +""" + +from __future__ import annotations + +import re +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Sequence + +_TASK_NAME_RE = re.compile( + r"^(?P\s*)-\s+(?:name|block|include_tasks|import_tasks|hosts)\s*:" +) +_WHEN_RE = re.compile(r"^\s*when\s*:\s*(?P.+?)\s*$") +_COMPOSE_ONLY_EXPR = re.compile( + r"DEPLOYMENT_MODE\s*!=\s*['\"]swarm['\"]" + r"|DEPLOYMENT_MODE\s*==\s*['\"]compose['\"]" +) + + +def _task_block_bounds(lines: Sequence[str], idx: int) -> tuple[int, int]: + """Return (start, end_exclusive) of the task block containing *idx*. + + *start* is the index of the nearest `- name:` (or `- block:`/`- include_tasks:`) + above *idx*. *end* is the index of the next sibling `- name:` at the same + indent, or `len(lines)` if none. + """ + start = -1 + indent = "" + for i in range(idx, -1, -1): + m = _TASK_NAME_RE.match(lines[i]) + if m: + start = i + indent = m.group("indent") + break + if start < 0: + return (0, len(lines)) + + for j in range(start + 1, len(lines)): + m = _TASK_NAME_RE.match(lines[j]) + if m and len(m.group("indent")) <= len(indent): + return (start, j) + return (start, len(lines)) + + +def is_task_compose_only_gated(lines: Sequence[str], idx: int) -> bool: + """True iff the task body containing *idx* carries a `when:` clause + whose textual expression evaluates compose-only at deploy time.""" + start, end = _task_block_bounds(lines, idx) + for k in range(start, end): + m = _WHEN_RE.match(lines[k]) + if m and _COMPOSE_ONLY_EXPR.search(m.group("expr")): + return True + return False