From 28cfa372483e9eaadebceb19069a33bdb3d1df12 Mon Sep 17 00:00:00 2001
From: epenet <6771947+epenet@users.noreply.github.com>
Date: Fri, 29 Nov 2024 05:08:43 +0100
Subject: [PATCH] Add unique_config_entry rule to quality_scale hassfest
 validation (#131878)

* Add unique_config_entry rule to quality_scale hassfest validation

* Improve message
---
 script/hassfest/quality_scale.py              |  3 +-
 .../unique_config_entry.py                    | 49 +++++++++++++++++++
 2 files changed, 51 insertions(+), 1 deletion(-)
 create mode 100644 script/hassfest/quality_scale_validation/unique_config_entry.py

diff --git a/script/hassfest/quality_scale.py b/script/hassfest/quality_scale.py
index 543bf616952..bb6f0cae7f0 100644
--- a/script/hassfest/quality_scale.py
+++ b/script/hassfest/quality_scale.py
@@ -20,6 +20,7 @@ from .quality_scale_validation import (
     reauthentication_flow,
     reconfiguration_flow,
     strict_typing,
+    unique_config_entry,
 )
 
 QUALITY_SCALE_TIERS = {value.name.lower(): value for value in ScaledQualityScaleTiers}
@@ -53,7 +54,7 @@ ALL_RULES = [
     Rule("runtime-data", ScaledQualityScaleTiers.BRONZE),
     Rule("test-before-configure", ScaledQualityScaleTiers.BRONZE),
     Rule("test-before-setup", ScaledQualityScaleTiers.BRONZE),
-    Rule("unique-config-entry", ScaledQualityScaleTiers.BRONZE),
+    Rule("unique-config-entry", ScaledQualityScaleTiers.BRONZE, unique_config_entry),
     # SILVER
     Rule("action-exceptions", ScaledQualityScaleTiers.SILVER),
     Rule(
diff --git a/script/hassfest/quality_scale_validation/unique_config_entry.py b/script/hassfest/quality_scale_validation/unique_config_entry.py
new file mode 100644
index 00000000000..eaa879bb05e
--- /dev/null
+++ b/script/hassfest/quality_scale_validation/unique_config_entry.py
@@ -0,0 +1,49 @@
+"""Enforce that the integration prevents duplicates from being configured.
+
+https://developers.home-assistant.io/docs/core/integration-quality-scale/rules/unique-config-entry/
+"""
+
+import ast
+
+from script.hassfest.model import Integration
+
+
+def _has_method_call(module: ast.Module, name: str) -> bool:
+    """Test if the module calls a specific method."""
+    return any(
+        type(item.func) is ast.Attribute and item.func.attr == name
+        for item in ast.walk(module)
+        if isinstance(item, ast.Call)
+    )
+
+
+def _has_abort_entries_match(module: ast.Module) -> bool:
+    """Test if the module calls `_async_abort_entries_match`."""
+    return _has_method_call(module, "_async_abort_entries_match")
+
+
+def _has_abort_unique_id_configured(module: ast.Module) -> bool:
+    """Test if the module calls defines (and checks for) a unique_id."""
+    return _has_method_call(module, "async_set_unique_id") and _has_method_call(
+        module, "_abort_if_unique_id_configured"
+    )
+
+
+def validate(integration: Integration) -> list[str] | None:
+    """Validate that the integration prevents duplicate devices."""
+
+    if integration.manifest.get("single_config_entry"):
+        return None
+
+    config_flow_file = integration.path / "config_flow.py"
+    config_flow = ast.parse(config_flow_file.read_text())
+
+    if not (
+        _has_abort_entries_match(config_flow)
+        or _has_abort_unique_id_configured(config_flow)
+    ):
+        return [
+            "Integration doesn't prevent the same device or service from being "
+            f"set up twice in {config_flow_file}"
+        ]
+    return None