Merge pull request #1815 from alan-turing-institute/unique_list_validator

Add UniqueList annotated type

Author: JimMadge

data_safe_haven/config/config.py

@@ -26,6 +26,7 @@
     b64decode,
     b64encode,
 )
+from data_safe_haven.functions.validators import validate_unique_list
 from data_safe_haven.utility import (
     DatabaseSystem,
     LoggingSingleton,
@@ -39,6 +40,7 @@
     Guid,
     IpAddress,
     TimeZone,
+    UniqueList,
 )
 
 from .context_settings import Context
@@ -150,7 +152,9 @@ def update(
 
 
 class ConfigSectionSRE(BaseModel, validate_assignment=True):
-    databases: list[DatabaseSystem] = Field(..., default_factory=list[DatabaseSystem])
+    databases: UniqueList[DatabaseSystem] = Field(
+        ..., default_factory=list[DatabaseSystem]
+    )
     data_provider_ip_addresses: list[IpAddress] = Field(
         ..., default_factory=list[IpAddress]
     )
@@ -164,17 +168,6 @@ class ConfigSectionSRE(BaseModel, validate_assignment=True):
     )
     software_packages: SoftwarePackageCategory = SoftwarePackageCategory.NONE
 
-    @field_validator("databases")
-    @classmethod
-    def all_databases_must_be_unique(
-        cls, v: list[DatabaseSystem | str]
-    ) -> list[DatabaseSystem]:
-        v_ = [DatabaseSystem(d) for d in v]
-        if len(v_) != len(set(v_)):
-            msg = "all databases must be unique"
-            raise ValueError(msg)
-        return v_
-
     def update(
         self,
         *,
@@ -258,9 +251,7 @@ def all_sre_indices_must_be_unique(
         cls, v: dict[str, ConfigSectionSRE]
     ) -> dict[str, ConfigSectionSRE]:
         indices = [s.index for s in v.values()]
-        if len(indices) != len(set(indices)):
-            msg = "all SRE indices must be unique"
-            raise ValueError(msg)
+        validate_unique_list(indices)
         return v
 
     @property

data_safe_haven/functions/validators.py

@@ -1,5 +1,7 @@
 import ipaddress
 import re
+from collections.abc import Hashable
+from typing import TypeVar
 
 import fqdn
 import pytz
@@ -57,3 +59,13 @@ def validate_timezone(timezone: str) -> str:
         msg = "Expected valid timezone, for example 'Europe/London'."
         raise ValueError(msg)
     return timezone
+
+
+TH = TypeVar("TH", bound=Hashable)
+
+
+def validate_unique_list(items: list[TH]) -> list[TH]:
+    if len(items) != len(set(items)):
+        msg = "All items must be unique."
+        raise ValueError(msg)
+    return items

data_safe_haven/utility/annotated_types.py

@@ -1,4 +1,5 @@
-from typing import Annotated
+from collections.abc import Hashable
+from typing import Annotated, TypeAlias, TypeVar
 
 from pydantic import Field
 from pydantic.functional_validators import AfterValidator
@@ -11,6 +12,7 @@
     validate_fqdn,
     validate_ip_address,
     validate_timezone,
+    validate_unique_list,
 )
 
 AzureShortName = Annotated[str, Field(min_length=1, max_length=24)]
@@ -22,3 +24,9 @@
 Guid = Annotated[str, AfterValidator(validate_aad_guid)]
 IpAddress = Annotated[str, AfterValidator(validate_ip_address)]
 TimeZone = Annotated[str, AfterValidator(validate_timezone)]
+TH = TypeVar("TH", bound=Hashable)
+# type UniqueList[TH] = Annotated[list[TH], AfterValidator(validate_unique_list)]
+# mypy doesn't support PEP695 type statements
+UniqueList: TypeAlias = Annotated[  # noqa:UP040
+    list[TH], AfterValidator(validate_unique_list)
+]

pyproject.toml

@@ -91,15 +91,15 @@ all = [
 
 [tool.hatch.envs.test]
 dependencies = [
-  "pytest~=7.4"
+  "pytest~=8.1"
 ]
 pre-install-commands = ["pip install -r requirements.txt"]
 
 [tool.hatch.envs.test.scripts]
 test = "pytest {args:-vvv tests}"
 
 [tool.black]
-target-version = ["py311", "py312"]
+target-version = ["py312"]
 
 [tool.ruff.lint]
 select = [

tests/config/test_config.py

@@ -137,7 +137,7 @@ def test_constructor_defaults(self, remote_desktop_config):
         assert sre_config.software_packages == SoftwarePackageCategory.NONE
 
     def test_all_databases_must_be_unique(self):
-        with pytest.raises(ValueError, match="all databases must be unique"):
+        with pytest.raises(ValueError, match="All items must be unique."):
             ConfigSectionSRE(
                 index=1,
                 databases=[DatabaseSystem.POSTGRESQL, DatabaseSystem.POSTGRESQL],
@@ -239,7 +239,7 @@ def test_constructor(self, context, azure_config, pulumi_config, shm_config):
     def test_all_sre_indices_must_be_unique(
         self, context, azure_config, pulumi_config, shm_config
     ):
-        with pytest.raises(ValueError, match="all SRE indices must be unique"):
+        with pytest.raises(ValueError, match="All items must be unique."):
             sre_config_1 = ConfigSectionSRE(index=1)
             sre_config_2 = ConfigSectionSRE(index=1)
             Config(

tests/functions/test_validators.py

@@ -1,6 +1,11 @@
 import pytest
 
-from data_safe_haven.functions.validators import validate_aad_guid, validate_fqdn
+from data_safe_haven.functions.validators import (
+    validate_aad_guid,
+    validate_fqdn,
+    validate_unique_list,
+)
+from data_safe_haven.utility.enums import DatabaseSystem
 
 
 class TestValidateAadGuid:
@@ -53,3 +58,28 @@ def test_validate_fqdn_fail(self, fqdn):
             ValueError, match="Expected valid fully qualified domain name"
         ):
             validate_fqdn(fqdn)
+
+
+class TestValidateUniqueList:
+    @pytest.mark.parametrize(
+        "items",
+        [
+            [1, 2, 3],
+            ["a", 5, len],
+        ],
+    )
+    def test_validate_unique_list(self, items):
+        validate_unique_list(items)
+
+    @pytest.mark.parametrize(
+        "items",
+        [
+            [DatabaseSystem.POSTGRESQL, DatabaseSystem.POSTGRESQL],
+            [DatabaseSystem.POSTGRESQL, 2, DatabaseSystem.POSTGRESQL],
+            [1, 1],
+            ["abc", "abc"],
+        ],
+    )
+    def test_validate_unique_list_fail(self, items):
+        with pytest.raises(ValueError, match="All items must be unique."):
+            validate_unique_list(items)