Merge pull request #207 from averevki/test-conditions

Add tests for Common feature - Conditions

testsuite/objects/__init__.py

@@ -142,6 +142,10 @@ def remove_all_hosts(self):
     def set_deny_with(self, code, value):
         """Set denyWith"""
 
+    @abc.abstractmethod
+    def add_rule(self, when: list[Rule]):
+        """Add rule for the skip of entire AuthConfig"""
+
 
 class PreexistingAuthorino(Authorino):
     """Authorino which is already deployed prior to the testrun"""

testsuite/openshift/objects/auth_config/__init__.py

@@ -2,7 +2,7 @@
 from functools import cached_property
 from typing import Dict, List
 
-from testsuite.objects import Authorization, Responses, Metadata, Identities, Authorizations
+from testsuite.objects import Authorization, Responses, Metadata, Identities, Authorizations, Rule
 from testsuite.openshift.client import OpenShiftClient
 from testsuite.openshift.objects import OpenShiftObject, modify
 from .sections import AuthorizationsSection, IdentitySection, MetadataSection, ResponsesSection
@@ -72,3 +72,9 @@ def set_deny_with(self, code, value):
         self.auth_section["denyWith"] = {
             "unauthenticated": {"code": code, "headers": [{"name": "Location", "valueFrom": {"authJSON": value}}]}
         }
+
+    @modify
+    def add_rule(self, when: list[Rule]):
+        """Add rule for the skip of entire AuthConfig"""
+        self.auth_section.setdefault("when", [])
+        self.auth_section["when"].extend([vars(x) for x in when])

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

@@ -0,0 +1,21 @@
+"""Test condition to skip the authorization section of AuthConfig"""
+import pytest
+
+from testsuite.objects import Rule
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """Add to the AuthConfig authorization with opa policy that will always reject POST requests"""
+    when_post = [Rule("context.request.http.method", "eq", "POST")]
+    authorization.authorization.opa_policy("opa", "allow { false }", when=when_post)
+    return authorization
+
+
+def test_skip_authorization(client, auth):
+    """Send GET and POST requests to the same endpoint, verify if authorization has been skipped on GET request"""
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 200
+
+    response = client.post("/post", auth=auth)
+    assert response.status_code == 403

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

@@ -0,0 +1,29 @@
+"""Test condition to skip the identity section of AuthConfig"""
+import pytest
+
+from testsuite.objects import Rule
+from testsuite.httpx.auth import HeaderApiKeyAuth
+
+
+@pytest.fixture(scope="module")
+def auth(create_api_key, module_label):
+    """Create API key and return his auth"""
+    api_key = create_api_key("api-key", module_label, "api_key_value")
+    return HeaderApiKeyAuth(api_key)
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, module_label):
+    """Add to the AuthConfig API key identity, which can only be used on requests to the /get path"""
+    when_get = [Rule("context.request.http.path", "eq", "/get")]
+    authorization.identity.api_key("api-key", match_label=module_label, when=when_get)
+    return authorization
+
+
+def test_skip_identity(client, auth):
+    """Send request to /get and /post, verify that API key evaluator is not used on /get requests"""
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 200
+
+    response = client.post("/post", auth=auth)
+    assert response.status_code == 401

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

@@ -0,0 +1,38 @@
+"""Test condition to skip the metadata section of AuthConfig"""
+import pytest
+
+from testsuite.objects import Rule
+
+
+@pytest.fixture(scope="module")
+def mockserver_expectation(request, mockserver, module_label):
+    """Creates Mockserver Expectation which returns non-empty response on hit"""
+    request.addfinalizer(lambda: mockserver.clear_expectation(module_label))
+    return mockserver.create_expectation(module_label, "response")
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, mockserver_expectation):
+    """
+    Add to the AuthConfig metadata evaluator with get http request to the mockserver,
+    which will be only triggered on POST requests to the endpoint
+    """
+    when_post = [Rule("context.request.http.method", "eq", "POST")]
+    authorization.metadata.http_metadata("mock", mockserver_expectation, "GET", when=when_post)
+    return authorization
+
+
+def test_skip_metadata(client, auth, mockserver, module_label):
+    """Send GET and POST requests to the same endpoint, verify that only POST request triggered metadata"""
+    # ensure that there are no expectation hits at the beginning
+    assert len(mockserver.retrieve_requests(module_label)) == 0
+
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 200
+    # verify that GET request did not trigger metadata http request to the mockserver
+    assert len(mockserver.retrieve_requests(module_label)) == 0
+
+    response = client.post("/post", auth=auth)
+    assert response.status_code == 200
+    # verify that POST request did trigger metadata http request to the mockserver
+    assert len(mockserver.retrieve_requests(module_label)) == 1

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

@@ -0,0 +1,40 @@
+"""Test condition to skip the response section of AuthConfig"""
+import pytest
+
+from testsuite.objects import Rule
+from testsuite.utils import extract_from_response
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """Add to the AuthConfig response, which will only trigger on POST requests"""
+    authorization.responses.add(
+        {
+            "name": "auth-json",
+            "json": {
+                "properties": [
+                    {"name": "auth", "value": "response"},
+                ]
+            },
+        },
+        when=[Rule("context.request.http.method", "eq", "POST")],
+    )
+    return authorization
+
+
+def test_skip_response(client, auth):
+    """
+    Send GET and POST requests to the same endpoint,
+    verify that POST request will return conditional response
+    """
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 200
+
+    # verify that response was not returned on a GET request
+    with pytest.raises(KeyError, match="Auth-Json"):
+        extract_from_response(response)
+
+    response = client.post("/post", auth=auth)
+    assert response.status_code == 200
+    # verify that response is returned on a POST request
+    assert extract_from_response(response)

testsuite/tests/kuadrant/authorino/conditions/test_top_level_condition.py

@@ -0,0 +1,30 @@
+"""Test condition to skip the entire AuthConfig"""
+import pytest
+
+from testsuite.objects import Rule
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, module_label):
+    """Add rule to the AuthConfig to skip entire authn/authz with certain request header"""
+    authorization.add_rule([Rule("context.request.http.headers.key", "neq", module_label)])
+    return authorization
+
+
+def test_skip_auth_config(client, auth, module_label):
+    """
+    Send requests with and without required header,
+    verify that header request trigger skip of entire AuthConfig
+    """
+    # header request ignores oidc identity
+    response = client.get("/get", headers={"key": module_label})
+    assert response.status_code == 200
+
+    # request without header uses AuthConfig
+    response = client.get("/get")
+    # request is rejected due to the oidc authentication absence
+    assert response.status_code == 401
+
+    # when oidc authentication is provided on a request without required header, request is accepted
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 200