Allow hassfest to validate specific integrations (#34277)

azure-pipelines-ci.yml

@@ -96,7 +96,7 @@ stages:
           pip install -e .
     - script: |
         . venv/bin/activate
-        python -m script.hassfest validate
+        python -m script.hassfest --action validate
       displayName: 'Validate manifests'
     - script: |
         . venv/bin/activate

script/hassfest/__main__.py

@@ -1,4 +1,5 @@
 """Validate manifests."""
+import argparse
 import pathlib
 import sys
 from time import monotonic
@@ -16,10 +17,9 @@ from . import (
 )
 from .model import Config, Integration
 
-PLUGINS = [
+INTEGRATION_PLUGINS = [
     codeowners,
     config_flow,
-    coverage,
     dependencies,
     manifest,
     services,
@@ -27,16 +27,52 @@ PLUGINS = [
     translations,
     zeroconf,
 ]
+HASS_PLUGINS = [
+    coverage,
+]
+
+
+def valid_integration_path(integration_path):
+    """Test if it's a valid integration."""
+    path = pathlib.Path(integration_path)
+    if not path.is_dir():
+        raise argparse.ArgumentTypeError(f"{integration_path} is not a directory.")
+
+    return path
 
 
 def get_config() -> Config:
     """Return config."""
-    if not pathlib.Path("requirements_all.txt").is_file():
-        raise RuntimeError("Run from project root")
+    parser = argparse.ArgumentParser(description="Hassfest")
+    parser.add_argument(
+        "--action", type=str, choices=["validate", "generate"], default=None
+    )
+    parser.add_argument(
+        "--integration-path",
+        action="append",
+        type=valid_integration_path,
+        help="Validate a single integration",
+    )
+    parsed = parser.parse_args()
+
+    if parsed.action is None:
+        parsed.action = "validate" if parsed.integration_path else "generate"
+
+    if parsed.action == "generate" and parsed.integration_path:
+        raise RuntimeError(
+            "Generate is not allowed when limiting to specific integrations"
+        )
+
+    if (
+        not parsed.integration_path
+        and not pathlib.Path("requirements_all.txt").is_file()
+    ):
+        raise RuntimeError("Run from Home Assistant root")
 
     return Config(
         root=pathlib.Path(".").absolute(),
-        action="validate" if sys.argv[-1] == "validate" else "generate",
+        specific_integrations=parsed.integration_path,
+        action=parsed.action,
     )
 
 
@@ -48,9 +84,21 @@ def main():
         print(err)
         return 1
 
-    integrations = Integration.load_dir(pathlib.Path("homeassistant/components"))
+    plugins = INTEGRATION_PLUGINS
 
-    for plugin in PLUGINS:
+    if config.specific_integrations:
+        integrations = {}
+
+        for int_path in config.specific_integrations:
+            integration = Integration(int_path)
+            integration.load_manifest()
+            integrations[integration.domain] = integration
+
+    else:
+        integrations = Integration.load_dir(pathlib.Path("homeassistant/components"))
+        plugins += HASS_PLUGINS
+
+    for plugin in plugins:
         try:
             start = monotonic()
             print(f"Validating {plugin.__name__.split('.')[-1]}...", end="", flush=True)
@@ -77,14 +125,15 @@ def main():
         general_errors = config.errors
         invalid_itg = [itg for itg in integrations.values() if itg.errors]
 
+    print()
     print("Integrations:", len(integrations))
     print("Invalid integrations:", len(invalid_itg))
 
     if not invalid_itg and not general_errors:
-        for plugin in PLUGINS:
-            if hasattr(plugin, "generate"):
-                plugin.generate(integrations, config)
-
+        if config.action == "generate":
+            for plugin in plugins:
+                if hasattr(plugin, "generate"):
+                    plugin.generate(integrations, config)
         return 0
 
     print()
@@ -99,7 +148,8 @@ def main():
         print()
 
     for integration in sorted(invalid_itg, key=lambda itg: itg.domain):
-        print(f"Integration {integration.domain}:")
+        extra = f" - {integration.path}" if config.specific_integrations else ""
+        print(f"Integration {integration.domain}{extra}:")
         for error in integration.errors:
             print("*", error)
         print()

script/hassfest/codeowners.py

@@ -60,6 +60,9 @@ def validate(integrations: Dict[str, Integration], config: Config):
     codeowners_path = config.root / "CODEOWNERS"
     config.cache["codeowners"] = content = generate_and_validate(integrations)
 
+    if config.specific_integrations:
+        return
+
     with open(str(codeowners_path)) as fp:
         if fp.read().strip() != content:
             config.add_error(

script/hassfest/config_flow.py

@@ -68,6 +68,9 @@ def validate(integrations: Dict[str, Integration], config: Config):
     config_flow_path = config.root / "homeassistant/generated/config_flows.py"
     config.cache["config_flow"] = content = generate_and_validate(integrations)
 
+    if config.specific_integrations:
+        return
+
     with open(str(config_flow_path)) as fp:
         if fp.read().strip() != content:
             config.add_error(

script/hassfest/dependencies.py

@@ -249,6 +249,9 @@ def validate(integrations: Dict[str, Integration], config):
 
         validate_dependencies(integrations, integration)
 
+        if config.specific_integrations:
+            continue
+
         # check that all referenced dependencies exist
         for dep in integration.manifest.get("dependencies", []):
             if dep not in integrations:

script/hassfest/manifest.py

@@ -10,7 +10,7 @@ from .model import Integration
 DOCUMENTATION_URL_SCHEMA = "https"
 DOCUMENTATION_URL_HOST = "www.home-assistant.io"
 DOCUMENTATION_URL_PATH_PREFIX = "/integrations/"
-DOCUMENTATION_URL_EXCEPTIONS = ["https://www.home-assistant.io/hassio"]
+DOCUMENTATION_URL_EXCEPTIONS = {"https://www.home-assistant.io/hassio"}
 
 SUPPORTED_QUALITY_SCALES = ["gold", "internal", "platinum", "silver"]
 
@@ -23,9 +23,9 @@ def documentation_url(value: str) -> str:
     parsed_url = urlparse(value)
     if not parsed_url.scheme == DOCUMENTATION_URL_SCHEMA:
         raise vol.Invalid("Documentation url is not prefixed with https")
-    if not parsed_url.netloc == DOCUMENTATION_URL_HOST:
-        raise vol.Invalid("Documentation url not hosted at www.home-assistant.io")
-    if not parsed_url.path.startswith(DOCUMENTATION_URL_PATH_PREFIX):
+    if parsed_url.netloc == DOCUMENTATION_URL_HOST and not parsed_url.path.startswith(
+        DOCUMENTATION_URL_PATH_PREFIX
+    ):
         raise vol.Invalid(
             "Documentation url does not begin with www.home-assistant.io/integrations"
         )

script/hassfest/model.py

@@ -2,7 +2,7 @@
 import importlib
 import json
 import pathlib
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 import attr
 
@@ -24,10 +24,11 @@ class Error:
 class Config:
     """Config for the run."""
 
-    root = attr.ib(type=pathlib.Path)
-    action = attr.ib(type=str)
-    errors = attr.ib(type=List[Error], factory=list)
-    cache = attr.ib(type=Dict[str, Any], factory=dict)
+    specific_integrations: Optional[pathlib.Path] = attr.ib()
+    root: pathlib.Path = attr.ib()
+    action: str = attr.ib()
+    errors: List[Error] = attr.ib(factory=list)
+    cache: Dict[str, Any] = attr.ib(factory=dict)
 
     def add_error(self, *args, **kwargs):
         """Add an error."""

script/hassfest/ssdp.py

@@ -65,6 +65,9 @@ def validate(integrations: Dict[str, Integration], config: Config):
     ssdp_path = config.root / "homeassistant/generated/ssdp.py"
     config.cache["ssdp"] = content = generate_and_validate(integrations)
 
+    if config.specific_integrations:
+        return
+
     with open(str(ssdp_path)) as fp:
         if fp.read().strip() != content:
             config.add_error(

script/hassfest/zeroconf.py

@@ -120,6 +120,9 @@ def validate(integrations: Dict[str, Integration], config: Config):
     zeroconf_path = config.root / "homeassistant/generated/zeroconf.py"
     config.cache["zeroconf"] = content = generate_and_validate(integrations)
 
+    if config.specific_integrations:
+        return
+
     with open(str(zeroconf_path)) as fp:
         current = fp.read().strip()
         if current != content: