Add kubernetes objects for subjectAccessReview tests

Signed-off-by: averevki <[email protected]>

testsuite/kuadrant/policy/authorization/__init__.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from typing import Literal, Optional
 from testsuite.utils import asdict, JSONValues
-
+from testsuite.kuadrant.policy import CelExpression
 
 # pylint: disable=invalid-name
 
@@ -109,6 +109,17 @@ class PlainResponse:
     plain: ABCValue
 
 
+@dataclass(kw_only=True)
+class ResourceAttributes:
+    """Dataclass for specifying Resource Attributes in the KubernetesSubjectAccessReview authorization"""
+
+    namespace: Optional[Value | ValueFrom | CelExpression] = None
+    group: Optional[Value | ValueFrom | CelExpression] = None
+    resource: Optional[Value | ValueFrom | CelExpression] = None
+    name: Optional[Value | ValueFrom | CelExpression] = None
+    verb: Optional[Value | ValueFrom | CelExpression] = None
+
+
 @dataclass
 class WristbandSigningKeyRef:
     """Name of Kubernetes secret and corresponding signing algorithm."""

testsuite/kuadrant/policy/authorization/sections.py

@@ -13,6 +13,7 @@
     WristbandResponse,
     DenyResponse,
     Cache,
+    ResourceAttributes,
 )
 from testsuite.utils import asdict
 from testsuite.kubernetes import modify, Selector
@@ -287,7 +288,9 @@ def add_external_opa_policy(self, name, endpoint, ttl=0, **common_features):
         self.add_item(name, {"opa": {"externalPolicy": {"url": endpoint, "ttl": ttl}}}, **common_features)
 
     @modify
-    def add_kubernetes(self, name: str, user: ABCValue, resource_attributes: dict = None, **common_features):
+    def add_kubernetes(
+        self, name: str, user: ABCValue, resource_attributes: ResourceAttributes = None, **common_features
+    ):
         """Adds Kubernetes authorization
 
         :param name: name of kubernetes authorization
@@ -298,7 +301,10 @@ def add_kubernetes(self, name: str, user: ABCValue, resource_attributes: dict =
         self.add_item(
             name,
             {
-                "kubernetesSubjectAccessReview": {"user": asdict(user), "resourceAttributes": resource_attributes},
+                "kubernetesSubjectAccessReview": {
+                    "user": asdict(user),
+                    "resourceAttributes": asdict(resource_attributes) if resource_attributes else None,
+                },
             },
             **common_features,
         )

testsuite/kubernetes/cluster_role.py

@@ -1,9 +1,23 @@
 """ClusterRole and ClusterRoleBinding objects for Kubernetes"""
 
-from typing import Any
+from typing import Optional
+from dataclasses import dataclass
+
+from testsuite.utils import asdict
 from testsuite.kubernetes import KubernetesObject
 
 
+@dataclass(kw_only=True)
+class Rule:  # pylint: disable=invalid-name
+    """Dataclass for ClusterRole rule"""
+
+    verbs: list[str]
+    apiGroups: Optional[list[str]] = None
+    nonResourceURLs: Optional[list[str]] = None
+    resourceNames: Optional[list[str]] = None
+    resources: Optional[list[str]] = None
+
+
 class ClusterRole(KubernetesObject):
     """Kubernetes ClusterRole"""
 
@@ -12,7 +26,7 @@ def create_instance(
         cls,
         cluster,
         name,
-        rules: list[dict[str, Any]] = None,
+        rules: list[Rule] = None,
         labels: dict[str, str] = None,
     ):
         """Creates a new ClusterRole instance"""
@@ -23,7 +37,7 @@ def create_instance(
                 "name": name,
                 "labels": labels,
             },
-            "rules": rules,
+            "rules": [asdict(rule) for rule in rules] if rules else None,
         }
         return cls(model, context=cluster.context)
 

testsuite/tests/singlecluster/authorino/operator/tls/test_webhook.py

@@ -10,7 +10,7 @@
 
 from testsuite.certificates import CertInfo
 from testsuite.kubernetes.ingress import Ingress
-from testsuite.kuadrant.policy.authorization import ValueFrom, Pattern
+from testsuite.kuadrant.policy.authorization import ResourceAttributes, Value, ValueFrom, Pattern
 from testsuite.kuadrant.policy.authorization.auth_config import AuthConfig
 from testsuite.utils import cert_builder
 
@@ -86,30 +86,36 @@ def authorization(authorization, cluster, authorino_domain) -> AuthConfig:
         Pattern("auth.authorization.features.allow", "eq", "true"),
         Pattern("auth.authorization.features.verb", "eq", "CREATE"),
     ]
-    kube_attrs = {
-        "namespace": {"value": cluster.project},
-        "group": {"value": "networking.k8s.io"},
-        "resource": {"value": "Ingress"},
-        "verb": {"value": "create"},
-    }
     # add response for admission webhook for creating Ingress
     authorization.authorization.add_kubernetes(
-        "ingress-authn-k8s-binding-create", user_value, kube_attrs, when=when, priority=1
+        "ingress-authn-k8s-binding-create",
+        user_value,
+        ResourceAttributes(
+            namespace=Value(cluster.project),
+            group=Value("networking.k8s.io"),
+            resource=Value("Ingress"),
+            verb=Value("create"),
+        ),
+        when=when,
+        priority=1,
     )
 
     when = [
         Pattern("auth.authorization.features.allow", "eq", "true"),
         Pattern("auth.authorization.features.verb", "eq", "DELETE"),
     ]
-    kube_attrs = {
-        "namespace": {"value": cluster.project},
-        "group": {"value": "networking.k8s.io"},
-        "resource": {"value": "Ingress"},
-        "verb": {"value": "delete"},
-    }
     # add response for admission webhook for deleting Ingress
     authorization.authorization.add_kubernetes(
-        "ingress-authn-k8s-binding-delete", user_value, kube_attrs, when=when, priority=1
+        "ingress-authn-k8s-binding-delete",
+        user_value,
+        ResourceAttributes(
+            namespace=Value(cluster.project),
+            group=Value("networking.k8s.io"),
+            resource=Value("Ingress"),
+            verb=Value("delete"),
+        ),
+        when=when,
+        priority=1,
     )
     return authorization