"""Generate config flow file.""" import json from typing import Dict from homeassistant.requirements import DISCOVERY_INTEGRATIONS from .model import Config, Integration BASE = """ \"\"\"Automatically generated by hassfest. To update, run python3 -m script.hassfest \"\"\" # fmt: off FLOWS = {} """.strip() UNIQUE_ID_IGNORE = {"esphome", "fritzbox", "heos", "huawei_lte"} 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(): integration.add_error( "config_flow", "Config flows need to be defined in the file config_flow.py" ) return needs_unique_id = integration.domain not in UNIQUE_ID_IGNORE and any( key in integration.manifest for keys in DISCOVERY_INTEGRATIONS.values() for key in keys ) if not needs_unique_id: return config_flow = config_flow_file.read_text() has_unique_id = ( "self.async_set_unique_id" in config_flow or "config_entry_flow.register_discovery_flow" in config_flow or "config_entry_oauth2_flow.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 = [] for domain in sorted(integrations): integration = integrations[domain] if not integration.manifest: continue config_flow = integration.manifest.get("config_flow") if not config_flow: continue validate_integration(config, integration) domains.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")