Add integration scaffolding script (#26777)

* Add integration scaffolding script

* Make easier to develop

* Update py.test -> pytest

script/scaffold/__init__.py

@@ -0,0 +1 @@
+"""Scaffold new integration."""

script/scaffold/__main__.py

@@ -0,0 +1,56 @@
+"""Validate manifests."""
+from pathlib import Path
+import subprocess
+import sys
+
+from . import gather_info, generate, error, model
+
+
+def main():
+    """Scaffold an integration."""
+    if not Path("requirements_all.txt").is_file():
+        print("Run from project root")
+        return 1
+
+    print("Creating a new integration for Home Assistant.")
+
+    if "--develop" in sys.argv:
+        print("Running in developer mode. Automatically filling in info.")
+        print()
+
+        info = model.Info(
+            domain="develop",
+            name="Develop Hub",
+            codeowner="@developer",
+            requirement="aiodevelop==1.2.3",
+        )
+    else:
+        try:
+            info = gather_info.gather_info()
+        except error.ExitApp as err:
+            print()
+            print(err.reason)
+            return err.exit_code
+
+    generate.generate(info)
+
+    print("Running hassfest to pick up new codeowner and config flow.")
+    subprocess.run("python -m script.hassfest", shell=True)
+    print()
+
+    print("Running tests")
+    print(f"$ pytest tests/components/{info.domain}")
+    if (
+        subprocess.run(f"pytest tests/components/{info.domain}", shell=True).returncode
+        != 0
+    ):
+        return 1
+    print()
+
+    print(f"Successfully created the {info.domain} integration!")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

script/scaffold/const.py

@@ -0,0 +1,5 @@
+"""Constants for scaffolding."""
+from pathlib import Path
+
+COMPONENT_DIR = Path("homeassistant/components")
+TESTS_DIR = Path("tests/components")

script/scaffold/error.py

@@ -0,0 +1,10 @@
+"""Errors for scaffolding."""
+
+
+class ExitApp(Exception):
+    """Exception to indicate app should exit."""
+
+    def __init__(self, reason, exit_code):
+        """Initialize the exit app exception."""
+        self.reason = reason
+        self.exit_code = exit_code

script/scaffold/gather_info.py

@@ -0,0 +1,79 @@
+"""Gather info for scaffolding."""
+from homeassistant.util import slugify
+
+from .const import COMPONENT_DIR
+from .model import Info
+from .error import ExitApp
+
+
+CHECK_EMPTY = ["Cannot be empty", lambda value: value]
+
+
+FIELDS = {
+    "domain": {
+        "prompt": "What is the domain?",
+        "validators": [
+            CHECK_EMPTY,
+            [
+                "Domains cannot contain spaces or special characters.",
+                lambda value: value == slugify(value),
+            ],
+            [
+                "There already is an integration with this domain.",
+                lambda value: not (COMPONENT_DIR / value).exists(),
+            ],
+        ],
+    },
+    "name": {
+        "prompt": "What is the name of your integration?",
+        "validators": [CHECK_EMPTY],
+    },
+    "codeowner": {
+        "prompt": "What is your GitHub handle?",
+        "validators": [
+            CHECK_EMPTY,
+            [
+                'GitHub handles need to start with an "@"',
+                lambda value: value.startswith("@"),
+            ],
+        ],
+    },
+    "requirement": {
+        "prompt": "What PyPI package and version do you depend on? Leave blank for none.",
+        "validators": [
+            ["Versions should be pinned using '=='.", lambda value: "==" in value]
+        ],
+    },
+}
+
+
+def gather_info() -> Info:
+    """Gather info from user."""
+    answers = {}
+
+    for key, info in FIELDS.items():
+        hint = None
+        while key not in answers:
+            if hint is not None:
+                print()
+                print(f"Error: {hint}")
+
+            try:
+                print()
+                value = input(info["prompt"] + "\n> ")
+            except (KeyboardInterrupt, EOFError):
+                raise ExitApp("Interrupted!", 1)
+
+            value = value.strip()
+            hint = None
+
+            for validator_hint, validator in info["validators"]:
+                if not validator(value):
+                    hint = validator_hint
+                    break
+
+            if hint is None:
+                answers[key] = value
+
+    print()
+    return Info(**answers)

script/scaffold/generate.py

@@ -0,0 +1,47 @@
+"""Generate an integration."""
+import json
+from pathlib import Path
+
+from .const import COMPONENT_DIR, TESTS_DIR
+from .model import Info
+
+TEMPLATE_DIR = Path(__file__).parent / "templates"
+TEMPLATE_INTEGRATION = TEMPLATE_DIR / "integration"
+TEMPLATE_TESTS = TEMPLATE_DIR / "tests"
+
+
+def generate(info: Info) -> None:
+    """Generate an integration."""
+    print(f"Generating the {info.domain} integration...")
+    integration_dir = COMPONENT_DIR / info.domain
+    test_dir = TESTS_DIR / info.domain
+
+    replaces = {
+        "NEW_DOMAIN": info.domain,
+        "NEW_NAME": info.name,
+        "NEW_CODEOWNER": info.codeowner,
+        # Special case because we need to keep the list empty if there is none.
+        '"MANIFEST_NEW_REQUIREMENT"': (
+            json.dumps(info.requirement) if info.requirement else ""
+        ),
+    }
+
+    for src_dir, target_dir in (
+        (TEMPLATE_INTEGRATION, integration_dir),
+        (TEMPLATE_TESTS, test_dir),
+    ):
+        # Guard making it for test purposes.
+        if not target_dir.exists():
+            target_dir.mkdir()
+
+        for source_file in src_dir.glob("**/*"):
+            content = source_file.read_text()
+
+            for to_search, to_replace in replaces.items():
+                content = content.replace(to_search, to_replace)
+
+            target_file = target_dir / source_file.relative_to(src_dir)
+            print(f"Writing {target_file}")
+            target_file.write_text(content)
+
+    print()

script/scaffold/model.py

@@ -0,0 +1,12 @@
+"""Models for scaffolding."""
+import attr
+
+
+@attr.s
+class Info:
+    """Info about new integration."""
+
+    domain: str = attr.ib()
+    name: str = attr.ib()
+    codeowner: str = attr.ib()
+    requirement: str = attr.ib()

script/scaffold/templates/integration/__init__.py

@@ -0,0 +1,19 @@
+"""The NEW_NAME integration."""
+
+from .const import DOMAIN
+
+
+async def async_setup(hass, config):
+    """Set up the NEW_NAME integration."""
+    hass.data[DOMAIN] = config.get(DOMAIN, {})
+    return True
+
+
+async def async_setup_entry(hass, entry):
+    """Set up a config entry for NEW_NAME."""
+    # TODO forward the entry for each platform that you want to set up.
+    # hass.async_create_task(
+    #     hass.config_entries.async_forward_entry_setup(entry, "media_player")
+    # )
+
+    return True

script/scaffold/templates/integration/config_flow.py

@@ -0,0 +1,57 @@
+"""Config flow for NEW_NAME integration."""
+import logging
+
+import voluptuous as vol
+
+from homeassistant import core, config_entries
+
+from .const import DOMAIN  # pylint:disable=unused-import
+from .error import CannotConnect, InvalidAuth
+
+_LOGGER = logging.getLogger(__name__)
+
+# TODO adjust the data schema to the data that you need
+DATA_SCHEMA = vol.Schema({"host": str, "username": str, "password": str})
+
+
+async def validate_input(hass: core.HomeAssistant, data):
+    """Validate the user input allows us to connect.
+
+    Data has the keys from DATA_SCHEMA with values provided by the user.
+    """
+    # TODO validate the data can be used to set up a connection.
+    # If you cannot connect:
+    # throw CannotConnect
+    # If the authentication is wrong:
+    # InvalidAuth
+
+    # Return some info we want to store in the config entry.
+    return {"title": "Name of the device"}
+
+
+class DomainConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for NEW_NAME."""
+
+    VERSION = 1
+    # TODO pick one of the available connection classes
+    CONNECTION_CLASS = config_entries.CONN_CLASS_UNKNOWN
+
+    async def async_step_user(self, user_input=None):
+        """Handle the initial step."""
+        errors = {}
+        if user_input is not None:
+            try:
+                info = await validate_input(self.hass, user_input)
+
+                return self.async_create_entry(title=info["title"], data=user_input)
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+        return self.async_show_form(
+            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+        )

script/scaffold/templates/integration/const.py

@@ -0,0 +1,3 @@
+"""Constants for the NEW_NAME integration."""
+
+DOMAIN = "NEW_DOMAIN"

script/scaffold/templates/integration/error.py

@@ -0,0 +1,10 @@
+"""Errors for the NEW_NAME integration."""
+from homeassistant.exceptions import HomeAssistantError
+
+
+class CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""

script/scaffold/templates/integration/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "NEW_DOMAIN",
+  "name": "NEW_NAME",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/components/NEW_DOMAIN",
+  "requirements": ["MANIFEST_NEW_REQUIREMENT"],
+  "ssdp": {},
+  "homekit": {},
+  "dependencies": [],
+  "codeowners": ["NEW_CODEOWNER"]
+}

script/scaffold/templates/integration/strings.json

@@ -0,0 +1,21 @@
+{
+  "config": {
+    "title": "NEW_NAME",
+    "step": {
+      "user": {
+        "title": "Connect to the device",
+        "data": {
+          "host": "Host"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "Failed to connect, please try again",
+      "invalid_auth": "Invalid authentication",
+      "unknown": "Unexpected error"
+    },
+    "abort": {
+      "already_configured": "Device is already configured"
+    }
+  }
+}

script/scaffold/templates/tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for the NEW_NAME integration."""

script/scaffold/templates/tests/test_config_flow.py

@@ -0,0 +1,93 @@
+"""Test the NEW_NAME config flow."""
+from unittest.mock import patch
+
+from homeassistant import config_entries, setup
+from homeassistant.components.NEW_DOMAIN.const import DOMAIN
+from homeassistant.components.NEW_DOMAIN.error import CannotConnect, InvalidAuth
+
+from tests.common import mock_coro
+
+
+async def test_form(hass):
+    """Test we get the form."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["errors"] == {}
+
+    with patch(
+        "homeassistant.components.NEW_DOMAIN.config_flow.validate_input",
+        return_value=mock_coro({"title": "Test Title"}),
+    ), patch(
+        "homeassistant.components.NEW_DOMAIN.async_setup", return_value=mock_coro(True)
+    ) as mock_setup, patch(
+        "homeassistant.components.NEW_DOMAIN.async_setup_entry",
+        return_value=mock_coro(True),
+    ) as mock_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "host": "1.1.1.1",
+                "username": "test-username",
+                "password": "test-password",
+            },
+        )
+
+    assert result2["type"] == "create_entry"
+    assert result2["title"] == "Test Title"
+    assert result2["data"] == {
+        "host": "1.1.1.1",
+        "username": "test-username",
+        "password": "test-password",
+    }
+    await hass.async_block_till_done()
+    assert len(mock_setup.mock_calls) == 1
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_form_invalid_auth(hass):
+    """Test we handle invalid auth."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.NEW_DOMAIN.config_flow.validate_input",
+        side_effect=InvalidAuth,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "host": "1.1.1.1",
+                "username": "test-username",
+                "password": "test-password",
+            },
+        )
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"base": "invalid_auth"}
+
+
+async def test_form_cannot_connect(hass):
+    """Test we handle cannot connect error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.NEW_DOMAIN.config_flow.validate_input",
+        side_effect=CannotConnect,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "host": "1.1.1.1",
+                "username": "test-username",
+                "password": "test-password",
+            },
+        )
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"base": "cannot_connect"}