Add diagnostics rule to quality_scale hassfest validation (#131859)

script/hassfest/quality_scale.py

@@ -16,6 +16,7 @@ from .quality_scale_validation import (
     RuleValidationProtocol,
     config_entry_unloading,
     config_flow,
+    diagnostics,
     reauthentication_flow,
     reconfiguration_flow,
 )
@@ -69,7 +70,7 @@ ALL_RULES = [
     Rule("test-coverage", ScaledQualityScaleTiers.SILVER),
     # GOLD: [
     Rule("devices", ScaledQualityScaleTiers.GOLD),
-    Rule("diagnostics", ScaledQualityScaleTiers.GOLD),
+    Rule("diagnostics", ScaledQualityScaleTiers.GOLD, diagnostics),
     Rule("discovery", ScaledQualityScaleTiers.GOLD),
     Rule("discovery-update-info", ScaledQualityScaleTiers.GOLD),
     Rule("docs-data-update", ScaledQualityScaleTiers.GOLD),

script/hassfest/quality_scale_validation/config_entry_unloading.py

@@ -8,10 +8,11 @@ import ast
 from script.hassfest.model import Integration
 
 
-def _has_async_function(module: ast.Module, name: str) -> bool:
-    """Test if the module defines a function."""
+def _has_unload_entry_function(module: ast.Module) -> bool:
+    """Test if the module defines `async_unload_entry` function."""
     return any(
-        type(item) is ast.AsyncFunctionDef and item.name == name for item in module.body
+        type(item) is ast.AsyncFunctionDef and item.name == "async_unload_entry"
+        for item in module.body
     )
 
 
@@ -21,7 +22,7 @@ def validate(integration: Integration) -> list[str] | None:
     init_file = integration.path / "__init__.py"
     init = ast.parse(init_file.read_text())
 
-    if not _has_async_function(init, "async_unload_entry"):
+    if not _has_unload_entry_function(init):
         return [
             "Integration does not support config entry unloading "
             "(is missing `async_unload_entry` in __init__.py)"

script/hassfest/quality_scale_validation/diagnostics.py

@@ -0,0 +1,42 @@
+"""Enforce that the integration implements diagnostics.
+
+https://developers.home-assistant.io/docs/core/integration-quality-scale/rules/diagnostics/
+"""
+
+import ast
+
+from script.hassfest.model import Integration
+
+DIAGNOSTICS_FUNCTIONS = {
+    "async_get_config_entry_diagnostics",
+    "async_get_device_diagnostics",
+}
+
+
+def _has_diagnostics_function(module: ast.Module) -> bool:
+    """Test if the module defines at least one of diagnostic functions."""
+    return any(
+        type(item) is ast.AsyncFunctionDef and item.name in DIAGNOSTICS_FUNCTIONS
+        for item in ast.walk(module)
+    )
+
+
+def validate(integration: Integration) -> list[str] | None:
+    """Validate that the integration implements diagnostics."""
+
+    diagnostics_file = integration.path / "diagnostics.py"
+    if not diagnostics_file.exists():
+        return [
+            "Integration does implement diagnostics platform "
+            "(is missing diagnostics.py)",
+        ]
+
+    diagnostics = ast.parse(diagnostics_file.read_text())
+
+    if not _has_diagnostics_function(diagnostics):
+        return [
+            f"Integration is missing one of {DIAGNOSTICS_FUNCTIONS} "
+            f"in {diagnostics_file}"
+        ]
+
+    return None

script/hassfest/quality_scale_validation/reauthentication_flow.py

@@ -8,10 +8,10 @@ import ast
 from script.hassfest.model import Integration
 
 
-def _has_async_function(module: ast.Module, name: str) -> bool:
-    """Test if the module defines a function."""
+def _has_step_reauth_function(module: ast.Module) -> bool:
+    """Test if the module defines `async_step_reauth` function."""
     return any(
-        type(item) is ast.AsyncFunctionDef and item.name == name
+        type(item) is ast.AsyncFunctionDef and item.name == "async_step_reauth"
         for item in ast.walk(module)
     )
 
@@ -22,7 +22,7 @@ def validate(integration: Integration) -> list[str] | None:
     config_flow_file = integration.path / "config_flow.py"
     config_flow = ast.parse(config_flow_file.read_text())
 
-    if not _has_async_function(config_flow, "async_step_reauth"):
+    if not _has_step_reauth_function(config_flow):
         return [
             "Integration does not support a reauthentication flow "
             f"(is missing `async_step_reauth` in {config_flow_file})"

script/hassfest/quality_scale_validation/reconfiguration_flow.py

@@ -8,10 +8,10 @@ import ast
 from script.hassfest.model import Integration
 
 
-def _has_async_function(module: ast.Module, name: str) -> bool:
+def _has_step_reconfigure_function(module: ast.Module) -> bool:
     """Test if the module defines a function."""
     return any(
-        type(item) is ast.AsyncFunctionDef and item.name == name
+        type(item) is ast.AsyncFunctionDef and item.name == "async_step_reconfigure"
         for item in ast.walk(module)
     )
 
@@ -22,7 +22,7 @@ def validate(integration: Integration) -> list[str] | None:
     config_flow_file = integration.path / "config_flow.py"
     config_flow = ast.parse(config_flow_file.read_text())
 
-    if not _has_async_function(config_flow, "async_step_reconfigure"):
+    if not _has_step_reconfigure_function(config_flow):
         return [
             "Integration does not support a reconfiguration flow "
             f"(is missing `async_step_reconfigure` in {config_flow_file})"