Added test for rate limit with autz

testsuite/openshift/objects/rate_limit.py

@@ -45,15 +45,15 @@ def _user_id_variable(user_id):
         }
 
     @modify
-    def add_limit(self, max_value, seconds, conditions: list[str] = None, variable=None):
+    def add_limit(self, max_value, seconds, conditions: list[str] = None, enable_auth=None):
         """Add another limit"""
         limit = {"maxValue": max_value, "seconds": seconds}
         configuration = []
         if conditions:
             limit["conditions"] = conditions
-        if variable:
-            limit["variables"] = [variable]
-            configuration = [self._user_id_variable(variable)]
+        if enable_auth:
+            limit["variables"] = [enable_auth]
+            configuration = [self._user_id_variable(enable_auth)]
         rate_limit = {"limits": [limit], "configurations": configuration}
 
         self.model.spec.setdefault("rateLimits", []).append(rate_limit)

testsuite/tests/kuadrant/test_rate_limit_authz.py

@@ -0,0 +1,53 @@
+"""
+Tests for authenticated rate limiting
+http://kuadrant.io/docs/kuadrant-operator/user-guides/authenticated-rl-with-jwt-and-k8s-authnz.html
+"""
+
+import pytest
+
+from testsuite.httpx.auth import HttpxOidcClientAuth
+from testsuite.utils import fire_requests
+
+
+@pytest.fixture(scope="module")
+def rate_limit(rate_limit):
+    """Add limit to the policy"""
+    rate_limit.add_limit(5, 60, variable="userID")
+    return rate_limit
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """Adds JSON injection, that wraps the response as Envoy Dynamic Metadata for rate limit"""
+    authorization.responses.add(
+        {
+            "name": "auth-json",
+            "json": {
+                "properties": [{"name": "userID", "valueFrom": {"authJSON": "auth.identity.sub"}}],
+            },
+            "wrapper": "envoyDynamicMetadata",
+            "wrapperKey": "ext_auth_data",
+        },
+    )
+    return authorization
+
+
+@pytest.fixture(scope="module")
+def auth(oidc_provider):
+    """Returns RHSSO authentication object for HTTPX"""
+    return HttpxOidcClientAuth(oidc_provider.get_token, "authorization")
+
+
+@pytest.fixture(scope="module")
+def auth2(rhsso):
+    """Creates new RHSSO user and returns its authentication object for HTTPX"""
+    user = rhsso.realm.create_user("user2", "password", email="[email protected]")
+    return HttpxOidcClientAuth.from_user(rhsso.get_token, user=user)
+
+
+def test_multiple_iterations(client, auth, auth2):
+    """Tests that simple limit is applied successfully and works for multiple iterations"""
+    assert client.get("/get", auth=auth).status_code == 200
+    assert client.get("/get", auth=auth).status_code == 200
+    assert client.get("/get", auth=auth).status_code == 429
+    assert client.get("/get", auth=auth2).status_code == 200