core/script/hassfest/config_flow.py

"""Generate config flow file."""
from __future__ import annotations

import json

from .model import Config, Integration

BASE = """
\"\"\"Automatically generated by hassfest.

To update, run python3 -m script.hassfest
\"\"\"

# fmt: off

FLOWS = {}
""".strip()

UNIQUE_ID_IGNORE = {"huawei_lte", "mqtt", "adguard"}


def validate_integration(config: Config, integration: Integration):
    """Validate config flow of an integration."""
    config_flow_file = integration.path / "config_flow.py"

    if not config_flow_file.is_file():
        if integration.manifest.get("config_flow"):
            integration.add_error(
                "config_flow",
                "Config flows need to be defined in the file config_flow.py",
            )
        return

    config_flow = config_flow_file.read_text()

    needs_unique_id = integration.domain not in UNIQUE_ID_IGNORE and (
        "async_step_discovery" in config_flow
        or "async_step_hassio" in config_flow
        or "async_step_homekit" in config_flow
        or "async_step_mqtt" in config_flow
        or "async_step_ssdp" in config_flow
        or "async_step_zeroconf" in config_flow
        or "async_step_dhcp" in config_flow
        or "async_step_usb" in config_flow
    )

    if not needs_unique_id:
        return

    has_unique_id = (
        "self.async_set_unique_id" in config_flow
        or "self._async_handle_discovery_without_unique_id" in config_flow
        or "register_discovery_flow" in config_flow
        or "AbstractOAuth2FlowHandler" in config_flow
    )

    if has_unique_id:
        return

    if config.specific_integrations:
        notice_method = integration.add_warning
    else:
        notice_method = integration.add_error

    notice_method(
        "config_flow", "Config flows that are discoverable need to set a unique ID"
    )


def generate_and_validate(integrations: dict[str, Integration], config: Config):
    """Validate and generate config flow data."""
    domains = {
        "integration": [],
        "helper": [],
    }

    for domain in sorted(integrations):
        integration = integrations[domain]

        if not integration.manifest or not integration.config_flow:
            continue

        validate_integration(config, integration)

        domains[integration.integration_type].append(domain)

    return BASE.format(json.dumps(domains, indent=4))


def validate(integrations: dict[str, Integration], config: Config):
    """Validate config flow file."""
    config_flow_path = config.root / "homeassistant/generated/config_flows.py"
    config.cache["config_flow"] = content = generate_and_validate(integrations, config)

    if config.specific_integrations:
        return

    with open(str(config_flow_path)) as fp:
        if fp.read().strip() != content:
            config.add_error(
                "config_flow",
                "File config_flows.py is not up to date. "
                "Run python3 -m script.hassfest",
                fixable=True,
            )
        return


def generate(integrations: dict[str, Integration], config: Config):
    """Generate config flow file."""
    config_flow_path = config.root / "homeassistant/generated/config_flows.py"
    with open(str(config_flow_path), "w") as fp:
        fp.write(f"{config.cache['config_flow']}\n")