115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
"""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_bluetooth" 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")
|