Fix comma seperated parsing with test

Author: olaals

workflow-notifier/src/workflow_notifier/config/settings.py

@@ -1,17 +1,9 @@
-from typing import Annotated, Optional
+from typing import Optional
 
 from dotenv import load_dotenv
-from pydantic import BeforeValidator, Field
+from pydantic import Field
 from pydantic_settings import BaseSettings
 
-
-def _parse_comma_separated(value: object) -> list[str]:
-    """Accept a comma-separated string **or** an existing list and return a list of stripped, non-empty strings."""
-    if isinstance(value, str):
-        return [item.strip() for item in value.split(",") if item.strip()]
-    return value  # type: ignore[return-value]
-
-
 load_dotenv()
 
 
@@ -21,14 +13,18 @@ class Settings(BaseSettings):
     NOTIFIER_CLIENT_ID: str
     SARA_APP_REG_SCOPE: str
 
-    # Optional client secret for local development.
+    # Optional client secret for local development. In cloud (AKS with Azure
+    # Workload Identity) this is not provided and the federated token file is
+    # used instead. Include "ClientSecret" in ALLOWED_AUTH_METHODS to enable
+    # the ClientSecretCredential path.
     NOTIFIER_CLIENT_SECRET: Optional[str] = Field(default=None)
 
-    # In environment variables or ConfigMaps, supply a comma-separated string
+    # Ordered, comma-separated list of credential types that may be used to
+    # acquire an Azure AD access token. Allowed values: "WorkloadIdentity",
+    # "ClientSecret". When more than one method is configured, the order
+    # determines the priority inside the resulting ChainedTokenCredential.
     # (e.g. ALLOWED_AUTH_METHODS=WorkloadIdentity,ClientSecret).
-    ALLOWED_AUTH_METHODS: Annotated[
-        list[str], BeforeValidator(_parse_comma_separated)
-    ] = Field(default_factory=lambda: ["WorkloadIdentity"])
+    ALLOWED_AUTH_METHODS: str = Field(default="WorkloadIdentity")
 
     OTEL_SERVICE_NAME: str = Field(default="workflow-notifier")
     OTEL_EXPORTER_OTLP_ENDPOINT: str = Field(default="http://localhost:4317")
@@ -44,6 +40,11 @@ def authority(self) -> str:
     def scopes(self) -> list[str]:
         return [self.SARA_APP_REG_SCOPE]
 
+    @property
+    def allowed_auth_methods(self) -> list[str]:
+        """Parse the comma-separated ALLOWED_AUTH_METHODS string into a list."""
+        return [m.strip() for m in self.ALLOWED_AUTH_METHODS.split(",") if m.strip()]
+
     @property
     def workflow_notification_url(self) -> str:
         return f"{self.SARA_SERVER_URL}/workflow-notification"

workflow-notifier/src/workflow_notifier/notifier.py

@@ -35,7 +35,7 @@ def _get_credential() -> TokenCredential:
     Build a TokenCredential.
 
     The set of credential types to try is configured via
-    ``settings.ALLOWED_AUTH_METHODS``, an ordered list whose entries may be
+    ``settings.allowed_auth_methods``, an ordered list whose entries may be
     ``"WorkloadIdentity"`` and/or ``"ClientSecret"`` (case-insensitive). When
     more than one method is configured, the order determines the order inside
     the resulting ``ChainedTokenCredential``.
@@ -57,7 +57,7 @@ def _get_credential() -> TokenCredential:
     credentials: list[TokenCredential] = []
     activated: list[str] = []
 
-    allowed_methods = settings.ALLOWED_AUTH_METHODS or ["WorkloadIdentity"]
+    allowed_methods = settings.allowed_auth_methods or ["WorkloadIdentity"]
 
     for method in allowed_methods:
         normalized = method.strip().lower()

workflow-notifier/tests/test_settings.py

@@ -0,0 +1,62 @@
+from workflow_notifier.config.settings import Settings
+
+# Required env vars that Settings always needs.
+_REQUIRED = {
+    "SARA_SERVER_URL": "http://localhost:8100/api",
+    "TENANT_ID": "00000000-0000-0000-0000-000000000000",
+    "NOTIFIER_CLIENT_ID": "dummy-client-id",
+    "SARA_APP_REG_SCOPE": "api://dummy/.default",
+}
+
+
+class TestSettingsAllowedAuthMethods:
+    """Ensure ALLOWED_AUTH_METHODS is correctly parsed from env vars."""
+
+    def test_default_value(self, monkeypatch):
+        for k, v in _REQUIRED.items():
+            monkeypatch.setenv(k, v)
+        monkeypatch.delenv("ALLOWED_AUTH_METHODS", raising=False)
+
+        s = Settings()
+        assert s.allowed_auth_methods == ["WorkloadIdentity"]
+
+    def test_single_value_from_env(self, monkeypatch):
+        for k, v in _REQUIRED.items():
+            monkeypatch.setenv(k, v)
+        monkeypatch.setenv("ALLOWED_AUTH_METHODS", "ClientSecret")
+
+        s = Settings()
+        assert s.allowed_auth_methods == ["ClientSecret"]
+
+    def test_comma_separated_from_env(self, monkeypatch):
+        for k, v in _REQUIRED.items():
+            monkeypatch.setenv(k, v)
+        monkeypatch.setenv("ALLOWED_AUTH_METHODS", "WorkloadIdentity,ClientSecret")
+
+        s = Settings()
+        assert s.allowed_auth_methods == ["WorkloadIdentity", "ClientSecret"]
+
+    def test_whitespace_is_stripped(self, monkeypatch):
+        for k, v in _REQUIRED.items():
+            monkeypatch.setenv(k, v)
+        monkeypatch.setenv("ALLOWED_AUTH_METHODS", " WorkloadIdentity , ClientSecret ")
+
+        s = Settings()
+        assert s.allowed_auth_methods == ["WorkloadIdentity", "ClientSecret"]
+
+    def test_empty_segments_ignored(self, monkeypatch):
+        for k, v in _REQUIRED.items():
+            monkeypatch.setenv(k, v)
+        monkeypatch.setenv("ALLOWED_AUTH_METHODS", ",WorkloadIdentity,,ClientSecret,")
+
+        s = Settings()
+        assert s.allowed_auth_methods == ["WorkloadIdentity", "ClientSecret"]
+
+    def test_raw_field_is_str(self, monkeypatch):
+        for k, v in _REQUIRED.items():
+            monkeypatch.setenv(k, v)
+        monkeypatch.setenv("ALLOWED_AUTH_METHODS", "WorkloadIdentity")
+
+        s = Settings()
+        assert isinstance(s.ALLOWED_AUTH_METHODS, str)
+        assert s.ALLOWED_AUTH_METHODS == "WorkloadIdentity"