Responses section refactor

testsuite/objects/__init__.py

@@ -121,6 +121,76 @@ class ValueFrom(ABCValue):
     selector: str
 
 
+@dataclass
+class JsonResponse:
+    """Response item as JSON injection."""
+
+    properties: dict[str, ABCValue]
+
+    def asdict(self):
+        """Custom asdict due to nested structure."""
+        asdict_properties = {}
+        for key, value in self.properties.items():
+            asdict_properties[key] = asdict(value)
+        return {"json": {"properties": asdict_properties}}
+
+
+@dataclass
+class PlainResponse:
+    """Response item as plain text value."""
+
+    plain: ABCValue
+
+
+@dataclass
+class WristbandSigningKeyRef:
+    """Name of Kubernetes secret and corresponding signing algorithm."""
+
+    name: str
+    algorithm: str = "RS256"
+
+
+@dataclass(kw_only=True)
+class WristbandResponse:
+    """
+    Response item as Festival Wristband Token.
+
+    :param issuer: Endpoint to the Authorino service that issues the wristband
+    :param signingKeyRefs: List of  Kubernetes secrets of dataclass `WristbandSigningKeyRef`
+    :param customClaims: Custom claims added to the wristband token.
+    :param tokenDuration: Time span of the wristband token, in seconds.
+    """
+
+    issuer: str
+    signingKeyRefs: list[WristbandSigningKeyRef]
+    customClaims: Optional[list[dict[str, ABCValue]]] = None
+    tokenDuration: Optional[int] = None
+
+    def asdict(self):
+        """Custom asdict due to nested structure."""
+
+        asdict_key_refs = [asdict(i) for i in self.signingKeyRefs]
+        asdict_custom_claims = [asdict(i) for i in self.customClaims] if self.customClaims else None
+        return {
+            "wristband": {
+                "issuer": self.issuer,
+                "signingKeyRefs": asdict_key_refs,
+                "customClaims": asdict_custom_claims,
+                "tokenDuration": self.tokenDuration,
+            }
+        }
+
+
+@dataclass(kw_only=True)
+class DenyResponse:
+    """Dataclass for custom responses deny reason."""
+
+    code: Optional[int] = None
+    message: Optional[ABCValue] = None
+    headers: Optional[dict[str, ABCValue]] = None
+    body: Optional[ABCValue] = None
+
+
 @dataclass
 class Cache:
     """Dataclass for specifying Cache in Authorization"""

testsuite/openshift/objects/auth_config/sections.py

@@ -1,5 +1,5 @@
 """AuthConfig CR object"""
-from typing import Literal, Iterable, TYPE_CHECKING
+from typing import Literal, Iterable, TYPE_CHECKING, Union
 
 from testsuite.objects import (
     asdict,
@@ -9,6 +9,10 @@
     Selector,
     Credentials,
     ValueFrom,
+    JsonResponse,
+    PlainResponse,
+    WristbandResponse,
+    DenyResponse,
 )
 from testsuite.openshift.objects import modify
 
@@ -199,100 +203,46 @@ def add_uma(self, name, endpoint, credentials_secret, **common_features):
 class ResponseSection(Section):
     """Section which contains response configuration."""
 
-    @property
-    def success_headers(self):
-        """Nested dict for items wrapped as HTTP headers."""
-        return self.section.setdefault("success", {}).setdefault("headers", {})
-
-    @property
-    def success_dynamic_metadata(self):
-        """Nested dict for items wrapped as Envoy Dynamic Metadata."""
-        return self.section.setdefault("success", {}).setdefault("dynamicMetadata", {})
+    SUCCESS_RESPONSE = Union[JsonResponse, PlainResponse, WristbandResponse]
 
-    def _add(
-        self,
-        name: str,
-        value: dict,
-        wrapper: Literal["headers", "dynamicMetadata"] = "headers",
-        **common_features,
-    ):
+    def add_simple(self, auth_json: str, name="simple", key="data", **common_features):
+        """
+        Add simple response to AuthConfig, used for configuring response for debugging purposes,
+        which can be easily read back using extract_response
         """
-        Add response to AuthConfig.
+        self.add_success_header(name, JsonResponse({key: ValueFrom(auth_json)}), **common_features)
 
-        :param wrapper: This variable configures if the response should be wrapped as HTTP headers or
-         as Envoy Dynamic Metadata. Default is "headers"
+    def add_success_header(self, name: str, value: SUCCESS_RESPONSE, **common_features):
+        """
+        Add item to responses.success.headers section.
+        This section is for items wrapped as HTTP headers.
         """
-        add_common_features(value, **common_features)
-        if wrapper == "headers":
-            self.success_headers.update({name: value})
-        if wrapper == "dynamicMetadata":
-            self.success_dynamic_metadata.update({name: value})
 
-    def add_simple(self, auth_json: str, name="simple", key="data", **common_features):
+        success_headers = self.section.setdefault("success", {}).setdefault("headers", {})
+        asdict_value = asdict(value)
+        add_common_features(asdict_value, **common_features)
+        success_headers.update({name: asdict_value})
+
+    def add_success_dynamic(self, name: str, value: SUCCESS_RESPONSE, **common_features):
         """
-        Add simple response to AuthConfig, used for configuring response for debugging purposes,
-        which can be easily read back using extract_response
+        Add item to responses.success.dynamicMetadata section.
+        This section is for items wrapped as Envoy Dynamic Metadata.
         """
-        self.add_json(name, {key: ValueFrom(auth_json)}, **common_features)
 
-    @modify
-    def add_json(self, name: str, properties: dict[str, ABCValue], **common_features):
-        """Adds json response to AuthConfig"""
-        asdict_properties = {}
-        for key, value in properties.items():
-            asdict_properties[key] = asdict(value)
-        self._add(name, {"json": {"properties": asdict_properties}}, **common_features)
+        success_dynamic_metadata = self.section.setdefault("success", {}).setdefault("dynamicMetadata", {})
+        asdict_value = asdict(value)
+        add_common_features(asdict_value, **common_features)
+        success_dynamic_metadata.update({name: asdict_value})
 
-    @modify
-    def add_plain(self, name: str, value: ABCValue, **common_features):
-        """Adds plain response to AuthConfig"""
-        self._add(name, {"plain": asdict(value)}, **common_features)
+    def set_unauthenticated(self, deny_response: DenyResponse):
+        """Set custom deny response for unauthenticated error."""
 
-    @modify
-    def add_wristband(self, name: str, issuer: str, secret_name: str, algorithm: str = "RS256", **common_features):
-        """Adds wristband response to AuthConfig"""
-        self._add(
-            name,
-            {
-                "wristband": {
-                    "issuer": issuer,
-                    "signingKeyRefs": [
-                        {
-                            "name": secret_name,
-                            "algorithm": algorithm,
-                        }
-                    ],
-                },
-            },
-            **common_features,
-        )
+        self.add_item("unauthenticated", asdict(deny_response))
 
-    @modify
-    def set_deny_with(
-        self,
-        category: Literal["unauthenticated", "unauthorized"],
-        code: int = None,
-        message: ABCValue = None,
-        headers: dict[str, ABCValue] = None,
-        body: ABCValue = None,
-    ):
-        """Set default deny code, message, headers, and body for 'unauthenticated' and 'unauthorized' error."""
-        asdict_message = asdict(message) if message else None
-        asdict_body = asdict(body) if body else None
-        asdict_headers = None
-        if headers:
-            asdict_headers = {}
-            for key, value in headers.items():
-                asdict_headers[key] = asdict(value)
-        self.add_item(
-            category,
-            {
-                "code": code,
-                "message": asdict_message,
-                "headers": asdict_headers,
-                "body": asdict_body,
-            },
-        )
+    def set_unauthorized(self, deny_response: DenyResponse):
+        """Set custom deny response for unauthorized error."""
+
+        self.add_item("unauthorized", asdict(deny_response))
 
 
 class AuthorizationSection(Section):

testsuite/tests/kuadrant/authorino/conditions/section_conditions/test_response_condition.py

@@ -1,15 +1,15 @@
 """Test condition to skip the response section of AuthConfig"""
 import pytest
 
-from testsuite.objects import Rule, Value
+from testsuite.objects import Rule, Value, JsonResponse
 from testsuite.utils import extract_response
 
 
 @pytest.fixture(scope="module")
 def authorization(authorization):
     """Add to the AuthConfig response, which will only trigger on POST requests"""
-    authorization.responses.add_json(
-        "simple", {"data": Value("response")}, when=[Rule("context.request.http.method", "eq", "POST")]
+    authorization.responses.add_success_header(
+        "simple", JsonResponse({"data": Value("response")}), when=[Rule("context.request.http.method", "eq", "POST")]
     )
     return authorization
 

testsuite/tests/kuadrant/authorino/identity/rhsso/test_rhsso_context.py

@@ -4,18 +4,20 @@
 
 import pytest
 
-from testsuite.objects import ValueFrom
+from testsuite.objects import ValueFrom, JsonResponse
 
 
 @pytest.fixture(scope="module")
 def authorization(authorization):
     """Setup AuthConfig for test"""
-    authorization.responses.add_json(
+    authorization.responses.add_success_header(
         "auth-json",
-        {
-            "auth": ValueFrom("auth.identity"),
-            "context": ValueFrom("context.request.http.headers.authorization"),
-        },
+        JsonResponse(
+            {
+                "auth": ValueFrom("auth.identity"),
+                "context": ValueFrom("context.request.http.headers.authorization"),
+            }
+        ),
     )
     return authorization
 

testsuite/tests/kuadrant/authorino/metrics/test_deep_metrics.py

@@ -1,7 +1,7 @@
 """Tests for the functionality of the deep-evaluator metric samples"""
 import pytest
 
-from testsuite.objects import Value
+from testsuite.objects import Value, JsonResponse
 
 
 @pytest.fixture(scope="module")
@@ -25,7 +25,7 @@ def authorization(authorization, mockserver_expectation):
     authorization.identity.add_anonymous("anonymous", metrics=True)
     authorization.authorization.add_opa_policy("opa", "allow { true }", metrics=True)
     authorization.metadata.add_http("http", mockserver_expectation, "GET", metrics=True)
-    authorization.responses.add_json("json", {"auth": Value("response")}, metrics=True)
+    authorization.responses.add_success_header("json", JsonResponse({"auth": Value("response")}), metrics=True)
 
     return authorization
 

testsuite/tests/kuadrant/authorino/operator/clusterwide/test_wildcard_collision.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from testsuite.objects import Value
+from testsuite.objects import Value, JsonResponse
 from testsuite.openshift.objects.auth_config import AuthConfig
 
 
@@ -15,7 +15,7 @@ def authorization(authorino, blame, openshift, module_label, proxy, wildcard_dom
     auth = AuthConfig.create_instance(
         openshift, blame("ac"), None, hostnames=[wildcard_domain], labels={"testRun": module_label}
     )
-    auth.responses.add_json("header", {"anything": Value("one")})
+    auth.responses.add_success_header("header", JsonResponse({"anything": Value("one")}))
     return auth
 
 
@@ -26,7 +26,7 @@ def authorization2(authorino, blame, openshift2, module_label, proxy, wildcard_d
     auth = AuthConfig.create_instance(
         openshift2, blame("ac"), None, hostnames=[wildcard_domain], labels={"testRun": module_label}
     )
-    auth.responses.add_json("header", {"anything": Value("two")})
+    auth.responses.add_success_header("header", JsonResponse({"anything": Value("two")}))
     return auth
 
 

testsuite/tests/kuadrant/authorino/operator/http/conftest.py

@@ -1,7 +1,7 @@
 """Conftest for all tests requiring custom deployment of Authorino"""
 import pytest
 
-from testsuite.objects import Value
+from testsuite.objects import Value, JsonResponse
 from testsuite.httpx import HttpxBackoffClient
 from testsuite.openshift.objects.auth_config import AuthConfig
 from testsuite.openshift.objects.route import OpenshiftRoute
@@ -13,7 +13,7 @@ def authorization(authorization, wildcard_domain, openshift, module_label) -> Au
     """In case of Authorino, AuthConfig used for authorization"""
     authorization.remove_all_hosts()
     authorization.add_host(wildcard_domain)
-    authorization.responses.add_json("x-ext-auth-other-json", {"propX": Value("valueX")})
+    authorization.responses.add_success_header("x-ext-auth-other-json", JsonResponse({"propX": Value("valueX")}))
     return authorization
 
 

testsuite/tests/kuadrant/authorino/operator/sharding/conftest.py

@@ -1,7 +1,7 @@
 """Conftest for authorino sharding tests"""
 import pytest
 
-from testsuite.objects import Value
+from testsuite.objects import Value, JsonResponse
 from testsuite.openshift.envoy import Envoy
 from testsuite.openshift.objects.auth_config import AuthConfig
 
@@ -34,7 +34,7 @@ def _authorization(hostname=None, sharding_label=None):
             hostnames=[hostname],
             labels={"testRun": module_label, "sharding": sharding_label},
         )
-        auth.responses.add_json("header", {"anything": Value(sharding_label)})
+        auth.responses.add_success_header("header", JsonResponse({"anything": Value(sharding_label)}))
         request.addfinalizer(auth.delete)
         auth.commit()
         return auth

testsuite/tests/kuadrant/authorino/response/test_auth_json.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from testsuite.objects import ValueFrom
+from testsuite.objects import ValueFrom, JsonResponse
 
 
 @pytest.fixture(scope="module")
@@ -31,7 +31,7 @@ def authorization(authorization, path_and_value):
     path, _ = path_and_value
 
     authorization.responses.clear_all()  # delete previous responses due to the parametrization
-    authorization.responses.add_json("header", {"anything": ValueFrom(path)})
+    authorization.responses.add_success_header("header", JsonResponse({"anything": ValueFrom(path)}))
     return authorization
 
 

testsuite/tests/kuadrant/authorino/response/test_base64.py

@@ -6,14 +6,14 @@
 
 import pytest
 
-from testsuite.objects import ValueFrom
+from testsuite.objects import ValueFrom, JsonResponse
 
 
 @pytest.fixture(scope="module")
 def authorization(authorization):
     """Add response to Authorization"""
-    authorization.responses.add_json(
-        "header", {"anything": ValueFrom("context.request.http.headers.test|@base64:decode")}
+    authorization.responses.add_success_header(
+        "header", JsonResponse({"anything": ValueFrom("context.request.http.headers.test|@base64:decode")})
     )
     return authorization