test(lint): recognise compose-only task gates + skip comment / YAML metadata lines

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.
This commit is contained in:
kevinveenbirkenbach committed 2026-06-14 19:08:45 +02:00
1 parent 673ccc5f40
commit fb977e975a
5 files changed
+480 -4

No files matched your search

@@ -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:
@@ -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"|(?<![\w.-])compose\s+(?:exec|run|restart|logs|up|down|ps|config)\b"
)
# Exclude Ansible handler triggers (`notify: compose up`) and comment lines.
_HANDLER_TRIGGER = re.compile(r"^\s*notify\s*:")
_HANDLER_LIST_ITEM = re.compile(
r"^\s*-\s+compose\s+(?:up|down|build|run|restart|logs|exec|ps|config)\s*$"
)
_COMMENT_LINE = re.compile(r"^\s*#")
_YAML_NAME_FIELD = re.compile(r"^\s*[-]?\s*(?:name|msg|title|label|description)\s*:")
_QUOTED_LIST_ITEM = re.compile(r'^\s*-\s*[\'"]')
def _is_scan_target(rel_path: str) -> 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:
@@ -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()))
@@ -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()
+68
View File
@@ -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 <verb>` / `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<indent>\s*)-\s+(?:name|block|include_tasks|import_tasks|hosts)\s*:"
)
_WHEN_RE = re.compile(r"^\s*when\s*:\s*(?P<expr>.+?)\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