Capture python warnings and report some of them as matches (#3324)

.github/workflows/tox.yml

@@ -59,7 +59,7 @@ jobs:
       WSLENV: FORCE_COLOR:PYTEST_REQPASS:TOXENV:GITHUB_STEP_SUMMARY
       # Number of expected test passes, safety measure for accidental skip of
       # tests. Update value if you add/remove tests.
-      PYTEST_REQPASS: 794
+      PYTEST_REQPASS: 795
     steps:
       - name: Activate WSL1
         if: "contains(matrix.shell, 'wsl')"

examples/playbooks/capture-warning.yml

@@ -0,0 +1,8 @@
+---
+- name: Fixture to generate a warning
+  hosts: localhost
+  tasks:
+    - name: Generate a warning
+      ansible.builtin.debug:
+        msg: "This is a warning"
+      when: "{{ false }}" # noqa: 102 jinja[spacing]

src/ansiblelint/errors.py

@@ -13,6 +13,20 @@
     from ansiblelint.utils import Task
 
 
+class LintWarning(Warning):
+    """Used by linter."""
+
+
+@dataclass
+class WarnSource:
+    """Container for warning information, so we can later create a MatchError from it."""
+
+    filename: Lintable
+    lineno: int
+    tag: str
+    message: str | None = None
+
+
 class StrictModeError(RuntimeError):
     """Raise when we encounter a warning in strict mode."""
 

src/ansiblelint/rules/__init__.py

@@ -119,7 +119,10 @@ def matchlines(self, file: Lintable) -> list[MatchError]:
             if line.lstrip().startswith("#"):
                 continue
 
-            rule_id_list = ansiblelint.skip_utils.get_rule_skips_from_line(line)
+            rule_id_list = ansiblelint.skip_utils.get_rule_skips_from_line(
+                line,
+                lintable=file,
+            )
             if self.id in rule_id_list:
                 continue
 

src/ansiblelint/rules/jinja.py

@@ -203,7 +203,10 @@ def matchyaml(self, file: Lintable) -> list[MatchError]:
                 lines = file.content.splitlines()
                 for match in raw_results:
                     # lineno starts with 1, not zero
-                    skip_list = get_rule_skips_from_line(lines[match.lineno - 1])
+                    skip_list = get_rule_skips_from_line(
+                        line=lines[match.lineno - 1],
+                        lintable=file,
+                    )
                     if match.rule.id not in skip_list and match.tag not in skip_list:
                         results.append(match)
         else:

src/ansiblelint/rules/var_naming.py

@@ -95,7 +95,10 @@ def matchplay(self, file: Lintable, data: dict[str, Any]) -> list[MatchError]:
             lines = file.content.splitlines()
             for match in raw_results:
                 # lineno starts with 1, not zero
-                skip_list = get_rule_skips_from_line(lines[match.lineno - 1])
+                skip_list = get_rule_skips_from_line(
+                    line=lines[match.lineno - 1],
+                    lintable=file,
+                )
                 if match.rule.id not in skip_list and match.tag not in skip_list:
                     results.append(match)
 
@@ -175,7 +178,10 @@ def matchyaml(self, file: Lintable) -> list[MatchError]:
                 lines = file.content.splitlines()
                 for match in raw_results:
                     # lineno starts with 1, not zero
-                    skip_list = get_rule_skips_from_line(lines[match.lineno - 1])
+                    skip_list = get_rule_skips_from_line(
+                        line=lines[match.lineno - 1],
+                        lintable=file,
+                    )
                     if match.rule.id not in skip_list and match.tag not in skip_list:
                         results.append(match)
         else:

src/ansiblelint/runner.py

@@ -5,15 +5,16 @@
 import multiprocessing
 import multiprocessing.pool
 import os
+import warnings
 from dataclasses import dataclass
 from fnmatch import fnmatch
 from typing import TYPE_CHECKING, Any
 
 import ansiblelint.skip_utils
 import ansiblelint.utils
-from ansiblelint._internal.rules import LoadingFailureRule
+from ansiblelint._internal.rules import LoadingFailureRule, WarningRule
 from ansiblelint.constants import States
-from ansiblelint.errors import MatchError
+from ansiblelint.errors import LintWarning, MatchError, WarnSource
 from ansiblelint.file_utils import Lintable, expand_dirs_in_lintables
 from ansiblelint.rules.syntax_check import AnsibleSyntaxCheckRule
 
@@ -112,8 +113,51 @@ def is_excluded(self, lintable: Lintable) -> bool:
             for path in self.exclude_paths
         )
 
-    def run(self) -> list[MatchError]:  # noqa: C901
+    def run(self) -> list[MatchError]:
         """Execute the linting process."""
+        matches: list[MatchError] = []
+        with warnings.catch_warnings(record=True) as captured_warnings:
+            warnings.simplefilter("always")
+            matches = self._run()
+            for warn in captured_warnings:
+                # For the moment we are ignoring deprecation warnings as Ansible
+                # modules outside current content can generate them and user
+                # might not be able to do anything about them.
+                if warn.category is DeprecationWarning:
+                    continue
+                if warn.category is LintWarning:
+                    filename: None | Lintable = None
+                    if isinstance(warn.source, WarnSource):
+                        match = MatchError(
+                            message=warn.source.message or warn.category.__name__,
+                            rule=WarningRule(),
+                            filename=warn.source.filename.filename,
+                            tag=warn.source.tag,
+                            lineno=warn.source.lineno,
+                        )
+                    else:
+                        filename = warn.source
+                        # breakpoint()
+                        match = MatchError(
+                            message=warn.message
+                            if isinstance(warn.message, str)
+                            else "?",
+                            rule=WarningRule(),
+                            filename=str(filename),
+                        )
+                    matches.append(match)
+                    continue
+                _logger.warning(
+                    "%s:%s %s %s",
+                    warn.filename,
+                    warn.lineno or 1,
+                    warn.category.__name__,
+                    warn.message,
+                )
+        return matches
+
+    def _run(self) -> list[MatchError]:  # noqa: C901
+        """Run the linting (inner loop)."""
         files: list[Lintable] = []
         matches: list[MatchError] = []
 

src/ansiblelint/skip_utils.py

@@ -24,6 +24,7 @@
 import collections.abc
 import logging
 import re
+import warnings
 from functools import cache
 from itertools import product
 from typing import TYPE_CHECKING, Any
@@ -41,6 +42,7 @@
     RENAMED_TAGS,
     SKIPPED_RULES_KEY,
 )
+from ansiblelint.errors import LintWarning, WarnSource
 
 if TYPE_CHECKING:
     from collections.abc import Generator, Sequence
@@ -60,7 +62,11 @@
 # ansible.parsing.yaml.objects.AnsibleSequence
 
 
-def get_rule_skips_from_line(line: str) -> list[str]:
+def get_rule_skips_from_line(
+    line: str,
+    lintable: Lintable,
+    lineno: int = 1,
+) -> list[str]:
     """Return list of rule ids skipped via comment on the line of yaml."""
     _before_noqa, _noqa_marker, noqa_text = line.partition("# noqa")
 
@@ -69,10 +75,17 @@ def get_rule_skips_from_line(line: str) -> list[str]:
         if v in RENAMED_TAGS:
             tag = RENAMED_TAGS[v]
             if v not in _found_deprecated_tags:
-                _logger.warning(
-                    "Replaced outdated tag '%s' with '%s', replace it to avoid future regressions",
-                    v,
-                    tag,
+                msg = f"Replaced outdated tag '{v}' with '{tag}', replace it to avoid future errors"
+                warnings.warn(
+                    message=msg,
+                    category=LintWarning,
+                    source=WarnSource(
+                        filename=lintable,
+                        lineno=lineno,
+                        tag="warning[outdated-tag]",
+                        message=msg,
+                    ),
+                    stacklevel=0,
                 )
                 _found_deprecated_tags.add(v)
             v = tag
@@ -257,7 +270,11 @@ def traverse_yaml(obj: Any) -> None:  # noqa: C901
                     if _noqa_comment_re.match(comment_str):
                         line = v.start_mark.line + 1  # ruamel line numbers start at 0
                         lintable.line_skips[line].update(
-                            get_rule_skips_from_line(comment_str.strip()),
+                            get_rule_skips_from_line(
+                                comment_str.strip(),
+                                lintable=lintable,
+                                lineno=line,
+                            ),
                         )
         yaml_comment_obj_strings.append(str(obj.ca.items))
         if isinstance(obj, dict):
@@ -277,7 +294,7 @@ def traverse_yaml(obj: Any) -> None:  # noqa: C901
     rule_id_list = []
     for comment_obj_str in yaml_comment_obj_strings:
         for line in comment_obj_str.split(r"\n"):
-            rule_id_list.extend(get_rule_skips_from_line(line))
+            rule_id_list.extend(get_rule_skips_from_line(line, lintable=lintable))
 
     return [normalize_tag(tag) for tag in rule_id_list]
 

test/test_skiputils.py

@@ -8,6 +8,7 @@
 
 from ansiblelint.constants import SKIPPED_RULES_KEY
 from ansiblelint.file_utils import Lintable
+from ansiblelint.runner import Runner
 from ansiblelint.skip_utils import (
     append_skipped_rules,
     get_rule_skips_from_line,
@@ -17,6 +18,7 @@
 if TYPE_CHECKING:
     from ansible.parsing.yaml.objects import AnsibleBaseYAMLObject
 
+    from ansiblelint.rules import RulesCollection
     from ansiblelint.testing import RunFromText
 
 PLAYBOOK_WITH_NOQA = """\
@@ -43,7 +45,7 @@
 )
 def test_get_rule_skips_from_line(line: str, expected: str) -> None:
     """Validate get_rule_skips_from_line."""
-    v = get_rule_skips_from_line(line)
+    v = get_rule_skips_from_line(line, lintable=Lintable(""))
     assert v == [expected]
 
 
@@ -232,3 +234,19 @@ def test_append_skipped_rules(
 def test_is_nested_task(task: dict[str, Any], expected: bool) -> None:
     """Test is_nested_task() returns expected bool."""
     assert is_nested_task(task) == expected
+
+
+def test_capture_warning_outdated_tag(
+    default_rules_collection: RulesCollection,
+) -> None:
+    """Test that exclude paths do work."""
+    runner = Runner(
+        "examples/playbooks/capture-warning.yml",
+        rules=default_rules_collection,
+    )
+
+    matches = runner.run()
+    assert len(matches) == 1
+    assert matches[0].rule.id == "warning"
+    assert matches[0].tag == "warning[outdated-tag]"
+    assert matches[0].lineno == 8