More type (#1496)

.pre-commit-config.yaml

@@ -76,7 +76,7 @@ repos:
     - flake8-docstrings>=1.5.0
     - flake8-pytest-style>=1.2.2
 - repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v0.800
+  rev: v0.812
   hooks:
   - id: mypy
     # empty args needed in order to match mypy cli behavior
@@ -85,8 +85,15 @@ repos:
     - Sphinx>=3.1.2
     - enrich
     - flaky
-    - yamllint
+    - pytest
+    - types-PyYAML
     - wcmatch
+    - yamllint
+    exclude: >
+      (?x)^(
+        test/.*|
+        plugins/.*
+      )$
 - repo: https://github.com/pre-commit/mirrors-pylint
   rev: v2.6.0
   hooks:

conftest.py

@@ -2,12 +2,13 @@
 import os
 import re
 import sys
+from typing import List
 
 os.environ["NO_COLOR"] = "1"
 pytest_plugins = ["ansiblelint.testing.fixtures"]
 
 
-def pytest_cmdline_preparse(args):
+def pytest_cmdline_preparse(args: List[str]) -> None:
     """Pytest hook."""
     # disable xdist when called with -k args (filtering)
     # https://stackoverflow.com/questions/66407583/how-to-disable-pytest-xdist-only-when-pytest-is-called-with-filters

docs/rules_table_generator_ext.py

@@ -36,7 +36,7 @@ def _nodes_from_rst(
         ),
         node=node,
     )
-    return node.children
+    return node.children  # type: ignore
 
 
 class AnsibleLintDefaultRulesDirective(SphinxDirective):

mypy.ini

@@ -3,11 +3,23 @@ python_version = 3.6
 color_output = True
 error_summary = True
 disallow_untyped_calls = True
-; warn_redundant_casts=True
+; disallow_untyped_defs = True
+; disallow_any_generics = True
+; disallow_any_unimported = True
+; warn_redundant_casts = True
+; warn_return_any = True
+; warn_unused_configs = True
 
 [mypy-ansiblelint.*]
 ignore_missing_imports = True
 
+[mypy-ansiblelint.testing.*]
+disallow_untyped_defs = False
+
+[mypy-test.*]
+disallow_untyped_defs = False
+
+
 # 3rd party ignores
 [mypy-ansible]
 ignore_missing_imports = True

src/ansiblelint/__main__.py

@@ -26,8 +26,9 @@
 import pathlib
 import subprocess
 import sys
+from argparse import Namespace
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, Iterator, List, Optional
 
 from enrich.console import should_do_markup
 
@@ -108,7 +109,9 @@ def initialize_options(arguments: Optional[List[str]] = None) -> None:
     options.configured = True
 
 
-def report_outcome(result: "LintResult", options, mark_as_success=False) -> int:
+def report_outcome(
+    result: "LintResult", options: Namespace, mark_as_success: bool = False
+) -> int:
     """Display information about how to skip found rules.
 
     Returns exit code, 2 if errors were found, 0 when only warnings were found.
@@ -249,7 +252,7 @@ def main(argv: List[str] = None) -> int:
 
 
 @contextmanager
-def _previous_revision():
+def _previous_revision() -> Iterator[None]:
     """Create or update a temporary workdir containing the previous revision."""
     worktree_dir = ".cache/old-rev"
     revision = subprocess.run(

src/ansiblelint/app.py

@@ -5,6 +5,7 @@
 
 from ansiblelint import formatters
 from ansiblelint.color import console
+from ansiblelint.errors import MatchError
 
 if TYPE_CHECKING:
     from argparse import Namespace
@@ -26,7 +27,7 @@ def __init__(self, options: "Namespace"):
         formatter_factory = choose_formatter_factory(options)
         self.formatter = formatter_factory(options.cwd, options.display_relative_path)
 
-    def render_matches(self, matches: List) -> None:
+    def render_matches(self, matches: List[MatchError]) -> None:
         """Display given matches."""
         if isinstance(self.formatter, formatters.CodeclimateJSONFormatter):
             # If formatter CodeclimateJSONFormatter is chosen,

src/ansiblelint/color.py

@@ -25,7 +25,7 @@
 console_stderr = Console(**console_options_stderr)
 
 
-def reconfigure(new_options: Dict[str, Any]):
+def reconfigure(new_options: Dict[str, Any]) -> None:
     """Reconfigure console options."""
     global console_options  # pylint: disable=global-statement
     global console_stderr  # pylint: disable=global-statement
@@ -39,6 +39,6 @@ def reconfigure(new_options: Dict[str, Any]):
     console_stderr.__dict__ = tmp_console.__dict__
 
 
-def render_yaml(text: str):
+def render_yaml(text: str) -> Syntax:
     """Colorize YAMl for nice display."""
     return Syntax(text, 'yaml', theme="ansi_dark")

src/ansiblelint/file_utils.py

@@ -67,7 +67,7 @@ def expand_paths_vars(paths: List[str]) -> List[str]:
     return paths
 
 
-def kind_from_path(path: Path, base=False) -> str:
+def kind_from_path(path: Path, base: bool = False) -> str:
     """Determine the file kind based on its name.
 
     When called with base=True, it will return the base file type instead

src/ansiblelint/rules/AnsibleSyntaxCheckRule.py

@@ -3,7 +3,7 @@
 import re
 import subprocess
 import sys
-from typing import List
+from typing import Any, List
 
 from ansiblelint._internal.rules import BaseRule, RuntimeErrorRule
 from ansiblelint.config import options
@@ -141,7 +141,7 @@ def test_empty_playbook() -> None:
         assert result[0].message == "Empty playbook, nothing to do"
         assert len(result) == 1
 
-    def test_extra_vars_passed_to_command(config_options) -> None:
+    def test_extra_vars_passed_to_command(config_options: Any) -> None:
         """Validate `extra-vars` are passed to syntax check command."""
         config_options.extra_vars = {
             'foo': 'bar',

src/ansiblelint/rules/NoSameOwnerRule.py

@@ -109,6 +109,7 @@ def handle_unarchive(task: Any) -> bool:
 
     import pytest
 
+    from ansiblelint.rules import RulesCollection  # pylint: disable=ungrouped-imports
     from ansiblelint.runner import Runner  # pylint: disable=ungrouped-imports
 
     @pytest.mark.parametrize(
@@ -122,7 +123,9 @@ def handle_unarchive(task: Any) -> bool:
             ),
         ),
     )
-    def test_no_same_owner_rule(default_rules_collection, test_file, failures) -> None:
+    def test_no_same_owner_rule(
+        default_rules_collection: RulesCollection, test_file: str, failures: int
+    ) -> None:
         """Test rule matches."""
         results = Runner(test_file, rules=default_rules_collection).run()
         assert len(results) == failures

src/ansiblelint/rules/__init__.py

@@ -5,9 +5,10 @@
 import logging
 import os
 import re
+from argparse import Namespace
 from collections import defaultdict
 from importlib.abc import Loader
-from typing import Iterator, List, Optional, Union
+from typing import Iterator, List, Optional, Set, Union
 
 import ansiblelint.utils
 from ansiblelint._internal.rules import (
@@ -130,14 +131,6 @@ def matchtasks(self, file: Lintable) -> List[MatchError]:
             matches.append(m)
         return matches
 
-    @staticmethod
-    def _matchplay_linenumber(play, optional_linenumber):
-        try:
-            (linenumber,) = optional_linenumber
-        except ValueError:
-            linenumber = play[ansiblelint.utils.LINE_NUMBER_KEY]
-        return linenumber
-
     def matchyaml(self, file: Lintable) -> List[MatchError]:
         matches: List[MatchError] = []
         if not self.matchplay or file.base_kind != 'text/yaml':
@@ -191,7 +184,9 @@ def load_plugins(directory: str) -> List[AnsibleLintRule]:
 
 
 class RulesCollection:
-    def __init__(self, rulesdirs: Optional[List[str]] = None, options=options) -> None:
+    def __init__(
+        self, rulesdirs: Optional[List[str]] = None, options: Namespace = options
+    ) -> None:
         """Initialize a RulesCollection instance."""
         self.options = options
         if rulesdirs is None:
@@ -226,7 +221,7 @@ def extend(self, more: List[AnsibleLintRule]) -> None:
         self.rules.extend(more)
 
     def run(
-        self, file: Lintable, tags=set(), skip_list: List[str] = []
+        self, file: Lintable, tags: Set[str] = set(), skip_list: List[str] = []
     ) -> List[MatchError]:
         matches: List[MatchError] = list()
 

src/ansiblelint/skip_utils.py

@@ -22,7 +22,7 @@
 import logging
 from functools import lru_cache
 from itertools import product
-from typing import TYPE_CHECKING, Any, Generator, List, Sequence
+from typing import TYPE_CHECKING, Any, Generator, List, Optional, Sequence
 
 import ruamel.yaml
 
@@ -33,7 +33,6 @@
 if TYPE_CHECKING:
     from ansible.parsing.yaml.objects import AnsibleBaseYAMLObject
 
-
 _logger = logging.getLogger(__name__)
 
 
@@ -52,7 +51,7 @@ def get_rule_skips_from_line(line: str) -> List[str]:
 
 def append_skipped_rules(
     pyyaml_data: "AnsibleBaseYAMLObject", lintable: Lintable
-) -> Sequence[Any]:
+) -> "AnsibleBaseYAMLObject":
     """Append 'skipped_rules' to individual tasks or single metadata block.
 
     For a file, uses 2nd parser (ruamel.yaml) to pull comments out of
@@ -71,6 +70,10 @@ def append_skipped_rules(
         # Notify user of skip error, do not stop, do not change exit code
         _logger.error('Error trying to append skipped rules', exc_info=True)
         return pyyaml_data
+
+    if not yaml_skip:
+        return pyyaml_data
+
     return yaml_skip
 
 
@@ -88,8 +91,8 @@ def load_data(file_text: str) -> Any:
 
 
 def _append_skipped_rules(
-    pyyaml_data: Sequence[Any], lintable: Lintable
-) -> Sequence[Any]:
+    pyyaml_data: "AnsibleBaseYAMLObject", lintable: Lintable
+) -> Optional["AnsibleBaseYAMLObject"]:
     # parse file text using 2nd parser library
     ruamel_data = load_data(lintable.content)
 
@@ -114,7 +117,7 @@ def _append_skipped_rules(
         return pyyaml_data
     else:
         # For unsupported file types, we return empty skip lists
-        return []
+        return None
 
     # get tasks from blocks of tasks
     pyyaml_tasks = _get_tasks_from_blocks(pyyaml_task_blocks)

src/ansiblelint/utils.py

@@ -35,7 +35,7 @@
 from ansible.parsing.dataloader import DataLoader
 from ansible.parsing.mod_args import ModuleArgsParser
 from ansible.parsing.splitter import split_args
-from ansible.parsing.yaml.constructor import AnsibleConstructor
+from ansible.parsing.yaml.constructor import AnsibleConstructor, AnsibleMapping
 from ansible.parsing.yaml.loader import AnsibleLoader
 from ansible.parsing.yaml.objects import AnsibleBaseYAMLObject, AnsibleSequence
 from ansible.plugins.loader import add_all_plugin_dirs
@@ -89,7 +89,7 @@ def path_dwim(basedir: str, given: str) -> str:
     return str(dl.path_dwim(given))
 
 
-def ansible_template(basedir: str, varname: Any, templatevars, **kwargs) -> Any:
+def ansible_template(basedir: str, varname: Any, templatevars: Any, **kwargs) -> Any:
     dl = DataLoader()
     dl.set_basedir(basedir)
     templar = Templar(dl, variables=templatevars)
@@ -186,7 +186,7 @@ def _playbook_items(pb_data: dict) -> ItemsView:
     return [item for play in pb_data if play for item in play.items()]
 
 
-def _set_collections_basedir(basedir: str):
+def _set_collections_basedir(basedir: str) -> None:
     # Sets the playbook directory as playbook_paths for the collection loader
     try:
         # Ansible 2.10+
@@ -480,16 +480,7 @@ def _look_for_role_files(basedir: str, role: str, main='main') -> List[Lintable]
     return results
 
 
-def rolename(filepath):
-    idx = filepath.find('roles/')
-    if idx < 0:
-        return ''
-    role = filepath[idx + 6 :]
-    role = role[: role.find('/')]
-    return role
-
-
-def _kv_to_dict(v):
+def _kv_to_dict(v: str) -> Dict[str, Any]:
     (command, args, kwargs) = tokenize(v)
     return dict(__ansible_module__=command, __ansible_arguments__=args, **kwargs)
 
@@ -555,12 +546,12 @@ def normalize_task_v2(task: Dict[str, Any]) -> Dict[str, Any]:
     return result
 
 
-def normalize_task_v1(task):  # noqa: C901
+def normalize_task_v1(task) -> Dict[str, Any]:  # noqa: C901
     result = dict()
     for (k, v) in task.items():
         if k in VALID_KEYS or k.startswith('with_'):
             if k in ('local_action', 'action'):
-                if not isinstance(v, dict):
+                if isinstance(v, str):
                     v = _kv_to_dict(v)
                 v['__ansible_arguments__'] = v.get('__ansible_arguments__', list())
                 result['action'] = v
@@ -712,14 +703,18 @@ def parse_yaml_linenumbers(lintable: Lintable) -> AnsibleBaseYAMLObject:
     The line numbers are stored in each node's LINE_NUMBER_KEY key.
     """
 
-    def compose_node(parent, index):
+    def compose_node(parent: yaml.nodes.Node, index: int) -> yaml.nodes.Node:
         # the line number where the previous token has ended (plus empty lines)
         line = loader.line
-        node = Composer.compose_node(loader, parent, index)
+        node = Composer.compose_node(loader, parent, index)  # type: ignore
         node.__line__ = line + 1
+        if not isinstance(node, yaml.nodes.Node):
+            raise RuntimeError("Unexpected yaml data.")
         return node
 
-    def construct_mapping(node, deep=False):
+    def construct_mapping(
+        node: AnsibleBaseYAMLObject, deep: bool = False
+    ) -> AnsibleMapping:
         mapping = AnsibleConstructor.construct_mapping(loader, node, deep=deep)
         if hasattr(node, '__line__'):
             mapping[LINE_NUMBER_KEY] = node.__line__