Add pylint checker to ensure flow menus have backing step handlers (#174796)

pylint/plugins/README.md

@@ -138,6 +138,7 @@ Every check has a code following the
 | `W7431` | [`home-assistant-options-flow-field-not-translated`](#w7431-home-assistant-options-flow-field-not-translated) | Options flow form field missing translation in `strings.json` |
 | `W7432` | [`home-assistant-subentry-flow-field-not-translated`](#w7432-home-assistant-subentry-flow-field-not-translated) | Subentry flow form field missing translation in `strings.json` |
 | `W7433` | [`home-assistant-missing-test-before-configure`](#w7433-home-assistant-missing-test-before-configure) | Config flow should test the connection before creating an entry |
+| `W7434` | [`home-assistant-config-flow-menu-missing-step`](#w7434-home-assistant-config-flow-menu-missing-step) | `async_show_menu` option has no matching `async_step_*` method |
 | `W7435` | [`home-assistant-json-fixture`](#w7435-home-assistant-json-fixture) | Use a JSON fixture helper instead of parsing a loaded fixture |
 
 
@@ -366,6 +367,25 @@ in config flows; they come automatically from the device or are set by
 the integration.
 
 
+## `home_assistant_config_flow_menu_options` checker
+
+Validates that every option passed to
+`self.async_show_menu(menu_options=...)` corresponds to an
+`async_step_<option>` method on the flow class. Each option becomes a
+`next_step_id` the flow manager dispatches to that method; a missing method
+raises `UnknownStep` at runtime when the user selects the option.
+
+Only statically resolvable forms are checked: a literal list/tuple/set of
+strings, or a literal dict keyed by the step ids. Dynamic forms
+(comprehensions, unresolved variables) and flows with an unresolvable base
+class are skipped to avoid false positives.
+
+### `W7434`: `home-assistant-config-flow-menu-missing-step`
+
+A `menu_options` entry does not match any `async_step_*` method defined on
+the flow class or its ancestors.
+
+
 ## `home_assistant_unused_test_fixture_args` checker
 
 **Disabled by default** while existing violations are being cleaned up.

pylint/plugins/pylint_home_assistant/checkers/config_flow/menu_options.py

@@ -0,0 +1,182 @@
+"""Checker for menu_options in config flow async_show_menu calls.
+
+Every option passed to ``self.async_show_menu(menu_options=...)`` becomes a
+``next_step_id`` the user can select, which the flow manager dispatches to an
+``async_step_<option>`` method. If no such method exists the flow raises
+``UnknownStep`` at runtime when the user picks that option.
+
+This checker validates the statically resolvable forms of ``menu_options``
+(a literal list/tuple/set of strings, or a literal dict whose keys are the
+step ids) against the ``async_step_*`` methods defined on the enclosing flow
+class and its ancestors. Dynamic forms (comprehensions, unresolved variables)
+are skipped to avoid false positives.
+"""
+
+import astroid
+from astroid import nodes
+from pylint.checkers import BaseChecker
+from pylint.lint import PyLinter
+
+from pylint_home_assistant.const import Module
+from pylint_home_assistant.helpers.ast_utils import extended_ancestors
+from pylint_home_assistant.helpers.module_info import parse_module
+
+_STEP_PREFIX = "async_step_"
+
+
+def _safe_infer(node: nodes.NodeNG) -> nodes.NodeNG | None:
+    """Infer a single unambiguous value for *node*, else None.
+
+    A name assigned in several branches infers to more than one value; we
+    reject that (rather than trusting the first) so the checker only acts on
+    an unambiguous inference. ``pylint``'s ``safe_infer`` only rejects results
+    of *differing type*, so it would accept two different string lists here --
+    hence the explicit single-result check.
+    """
+    try:
+        inferred = list(node.infer())
+    except astroid.InferenceError:
+        return None
+    if len(inferred) != 1 or inferred[0] is astroid.Uninferable:
+        return None
+    return inferred[0]
+
+
+def _const_str_elements(elements: list[nodes.NodeNG]) -> set[str] | None:
+    """Return the string values of *elements*, or None if any is not a string."""
+    step_ids: set[str] = set()
+    for element in elements:
+        if not isinstance(element, nodes.Const) or not isinstance(element.value, str):
+            return None
+        step_ids.add(element.value)
+    return step_ids
+
+
+def _resolve_step_ids(node: nodes.NodeNG) -> set[str] | None:
+    """Resolve the step ids referenced by a ``menu_options`` value.
+
+    Returns the set of step ids for the statically resolvable forms (a literal
+    sequence of strings or a literal dict keyed by strings), or None when the
+    value cannot be resolved (and should therefore be skipped).
+    """
+    if isinstance(node, (nodes.List, nodes.Tuple, nodes.Set)):
+        return _const_str_elements(node.elts)
+    if isinstance(node, nodes.Dict):
+        return _const_str_elements([key for key, _ in node.items])
+
+    # Variables/constants: only trust an inferred literal collection.
+    inferred = _safe_infer(node)
+    if inferred is None or inferred is node:
+        return None
+    if isinstance(inferred, (nodes.List, nodes.Tuple, nodes.Set)):
+        return _const_str_elements(inferred.elts)
+    if isinstance(inferred, nodes.Dict):
+        return _const_str_elements([key for key, _ in inferred.items])
+    return None
+
+
+def _enclosing_class(node: nodes.NodeNG) -> nodes.ClassDef | None:
+    """Walk up the tree to find the enclosing class."""
+    current = node.parent
+    while current is not None:
+        if isinstance(current, nodes.ClassDef):
+            return current
+        current = current.parent
+    return None
+
+
+def _bases_resolved(klass: nodes.ClassDef) -> bool:
+    """Return True if every base in *klass*'s full ancestry resolves to a class.
+
+    When a base cannot be resolved its methods are invisible, so a step could
+    be defined there without us seeing it -- in that case we must not flag.
+    The whole chain is walked, not just the direct bases: a resolvable base
+    with an unresolvable ancestor hides methods just the same.
+    """
+    seen: set[str] = set()
+    stack: list[nodes.ClassDef] = [klass]
+    while stack:
+        current = stack.pop()
+        if current.qname() in seen:
+            continue
+        seen.add(current.qname())
+        for base in current.bases:
+            target = base.value if isinstance(base, nodes.Subscript) else base
+            inferred = _safe_infer(target)
+            if not isinstance(inferred, nodes.ClassDef):
+                return False
+            stack.append(inferred)
+    return True
+
+
+def _step_handler_names(klass: nodes.ClassDef) -> set[str]:
+    """Collect ``async_step_*`` handler names on *klass* and its ancestors.
+
+    The class namespace is used rather than only method definitions, so that
+    aliases (e.g. ``async_step_user = async_step_location``) -- a valid and
+    used pattern -- are recognised as handlers too.
+    """
+    names: set[str] = set()
+    for current in (klass, *extended_ancestors(klass)):
+        for name in current.locals:
+            if name.startswith(_STEP_PREFIX):
+                names.add(name)
+    return names
+
+
+class HassConfigFlowMenuOptionsChecker(BaseChecker):
+    """Checker for menu_options referencing missing async_step_* methods."""
+
+    name = "home_assistant_config_flow_menu_options"
+    priority = -1
+    msgs = {
+        "W7434": (
+            "Config flow menu option '%s' has no matching `async_step_%s` "
+            "method on the flow",
+            "home-assistant-config-flow-menu-missing-step",
+            "Used when async_show_menu is called with a menu option that does "
+            "not correspond to an async_step_<option> method on the flow "
+            "class. Selecting that option would raise UnknownStep at runtime.",
+        ),
+    }
+    options = ()
+
+    def visit_call(self, node: nodes.Call) -> None:
+        """Check async_show_menu calls for unresolved menu options."""
+        if (
+            not isinstance(node.func, nodes.Attribute)
+            or node.func.attrname != "async_show_menu"
+        ):
+            return
+
+        parsed = parse_module(node.root().name)
+        if parsed is None or parsed.module != Module.CONFIG_FLOW:
+            return
+
+        menu_options = next(
+            (kw.value for kw in node.keywords if kw.arg == "menu_options"), None
+        )
+        if menu_options is None:
+            return
+
+        step_ids = _resolve_step_ids(menu_options)
+        if not step_ids:
+            return
+
+        klass = _enclosing_class(node)
+        if klass is None or not _bases_resolved(klass):
+            return
+
+        step_handlers = _step_handler_names(klass)
+        for step_id in sorted(step_ids):
+            if f"{_STEP_PREFIX}{step_id}" not in step_handlers:
+                self.add_message(
+                    "home-assistant-config-flow-menu-missing-step",
+                    node=menu_options,
+                    args=(step_id, step_id),
+                )
+
+
+def register(linter: PyLinter) -> None:
+    """Register the checker."""
+    linter.register_checker(HassConfigFlowMenuOptionsChecker(linter))

tests/pylint/config_flow/test_menu_options.py

@@ -0,0 +1,297 @@
+"""Tests for the pylint config_flow menu_options checker."""
+
+from __future__ import annotations
+
+import astroid
+from astroid import nodes
+from pylint.testutils import MessageTest, UnittestLinter
+from pylint_home_assistant.checkers.config_flow.menu_options import (
+    HassConfigFlowMenuOptionsChecker,
+)
+import pytest
+
+from tests.pylint import assert_adds_messages, assert_no_messages, walk_checker
+
+CONFIG_FLOW_MODULE = "homeassistant.components.test.config_flow"
+
+
+@pytest.fixture(name="checker")
+def checker_fixture(linter: UnittestLinter) -> HassConfigFlowMenuOptionsChecker:
+    """Fixture to provide a config_flow menu_options checker."""
+    checker = HassConfigFlowMenuOptionsChecker(linter)
+    checker.module = "homeassistant.components.pylint_test"
+    return checker
+
+
+def _find_menu_options_node(root_node: nodes.Module) -> nodes.NodeNG:
+    """Find the ``menu_options`` value node of the ``async_show_menu`` call."""
+    for call in root_node.nodes_of_class(nodes.Call):
+        if (
+            isinstance(call.func, nodes.Attribute)
+            and call.func.attrname == "async_show_menu"
+        ):
+            for keyword in call.keywords:
+                if keyword.arg == "menu_options":
+                    return keyword.value
+    raise AssertionError("no async_show_menu(menu_options=...) call found")
+
+
+def _expect_missing_step(node: nodes.NodeNG, step_id: str) -> MessageTest:
+    """Build the expected MessageTest for a missing ``async_step`` handler."""
+    return MessageTest(
+        msg_id="home-assistant-config-flow-menu-missing-step",
+        node=node,
+        line=node.lineno,
+        col_offset=node.col_offset,
+        end_line=node.end_lineno,
+        end_col_offset=node.end_col_offset,
+        args=(step_id, step_id),
+    )
+
+
+@pytest.mark.parametrize(
+    ("code", "module_name"),
+    [
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(
+                step_id="user",
+                menu_options=["local", "cloud"],
+            )
+
+        async def async_step_local(self, user_input=None):
+            pass
+
+        async def async_step_cloud(self, user_input=None):
+            pass
+    """,
+            CONFIG_FLOW_MODULE,
+            id="list_all_present",
+        ),
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(
+                menu_options={"local": "Local", "cloud": "Cloud"},
+            )
+
+        async def async_step_local(self, user_input=None):
+            pass
+
+        async def async_step_cloud(self, user_input=None):
+            pass
+    """,
+            CONFIG_FLOW_MODULE,
+            id="dict_all_present",
+        ),
+        pytest.param(
+            """
+    _MENU_OPTIONS = ["local", "cloud"]
+
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(menu_options=_MENU_OPTIONS)
+
+        async def async_step_local(self, user_input=None):
+            pass
+
+        async def async_step_cloud(self, user_input=None):
+            pass
+    """,
+            CONFIG_FLOW_MODULE,
+            id="inferred_constant_all_present",
+        ),
+        pytest.param(
+            """
+    class BaseFlow:
+        async def async_step_pick_implementation(self, user_input=None):
+            pass
+
+    class TestFlow(BaseFlow):
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(
+                menu_options=["pick_implementation", "manual"],
+            )
+
+        async def async_step_manual(self, user_input=None):
+            pass
+    """,
+            CONFIG_FLOW_MODULE,
+            id="inherited_step_present",
+        ),
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(
+                menu_options=["location", "reconfigure"],
+            )
+
+        async def async_step_location(self, user_input=None):
+            pass
+
+        async_step_reconfigure = async_step_location
+    """,
+            CONFIG_FLOW_MODULE,
+            id="aliased_step_present",
+        ),
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_init(self, user_input=None):
+            return self.async_show_menu(
+                menu_options=[option.value for option in SomeEnum],
+            )
+    """,
+            CONFIG_FLOW_MODULE,
+            id="comprehension_skipped",
+        ),
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            if self.show_cloud:
+                options = ["local", "cloud"]
+            else:
+                options = ["local"]
+            return self.async_show_menu(menu_options=options)
+
+        async def async_step_local(self, user_input=None):
+            pass
+    """,
+            CONFIG_FLOW_MODULE,
+            id="ambiguous_inference_skipped",
+        ),
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_init(self, user_input=None):
+            options = self._build_options()
+            return self.async_show_menu(menu_options=options)
+    """,
+            CONFIG_FLOW_MODULE,
+            id="unresolved_variable_skipped",
+        ),
+        pytest.param(
+            """
+    class TestFlow(SomeUnresolvedBase):
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(menu_options=["from_base"])
+    """,
+            CONFIG_FLOW_MODULE,
+            id="unresolved_base_skipped",
+        ),
+        pytest.param(
+            """
+    class BaseFlow(SomeUnresolvedBase):
+        async def async_step_from_base(self, user_input=None):
+            pass
+
+    class TestFlow(BaseFlow):
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(
+                menu_options=["from_base", "cloud"],
+            )
+    """,
+            CONFIG_FLOW_MODULE,
+            id="unresolved_ancestor_skipped",
+        ),
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(menu_options=["only"])
+
+        async def async_step_only(self, user_input=None):
+            pass
+    """,
+            "homeassistant.components.test.options_flow",
+            id="non_config_flow_module_skipped",
+        ),
+    ],
+)
+def test_menu_options_good(
+    linter: UnittestLinter,
+    checker: HassConfigFlowMenuOptionsChecker,
+    code: str,
+    module_name: str,
+) -> None:
+    """Good test cases that should not raise a message."""
+    root_node = astroid.parse(code, module_name)
+
+    with assert_no_messages(linter):
+        walk_checker(linter, checker, root_node)
+
+
+@pytest.mark.parametrize(
+    ("code", "expected_steps"),
+    [
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(menu_options=["local", "cloud"])
+
+        async def async_step_local(self, user_input=None):
+            pass
+    """,
+            ["cloud"],
+            id="list_missing_one",
+        ),
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(menu_options=["local", "cloud"])
+    """,
+            ["cloud", "local"],
+            id="list_missing_all",
+        ),
+        pytest.param(
+            """
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(
+                menu_options={"local": "Local", "cloud": "Cloud"},
+            )
+
+        async def async_step_local(self, user_input=None):
+            pass
+    """,
+            ["cloud"],
+            id="dict_missing_one",
+        ),
+        pytest.param(
+            """
+    _MENU_OPTIONS = ["local", "cloud"]
+
+    class TestFlow:
+        async def async_step_user(self, user_input=None):
+            return self.async_show_menu(menu_options=_MENU_OPTIONS)
+
+        async def async_step_local(self, user_input=None):
+            pass
+    """,
+            ["cloud"],
+            id="inferred_constant_missing_one",
+        ),
+    ],
+)
+def test_menu_options_bad(
+    linter: UnittestLinter,
+    checker: HassConfigFlowMenuOptionsChecker,
+    code: str,
+    expected_steps: list[str],
+) -> None:
+    """Bad test cases that should raise a message per missing step."""
+    root_node = astroid.parse(code, CONFIG_FLOW_MODULE)
+    menu_options_node = _find_menu_options_node(root_node)
+
+    with assert_adds_messages(
+        linter,
+        *(_expect_missing_step(menu_options_node, step) for step in expected_steps),
+    ):
+        walk_checker(linter, checker, root_node)