test: added more openid tests

src/keycloak/keycloak_openid.py

@@ -346,7 +346,7 @@ class KeycloakOpenID:
         if data_raw.status_code == 404:
             return raise_error_from_response(data_raw, KeycloakDeprecationError)
 
-        return raise_error_from_response(data_raw, KeycloakGetError)
+        return raise_error_from_response(data_raw, KeycloakGetError)  # pragma: no cover
 
     def introspect(self, token, rpt=None, token_type_hint=None):
         """

test_keycloak_init.sh

@@ -10,7 +10,7 @@ function keycloak_stop() {
 
 function keycloak_start() {
     echo "Starting keycloak docker container"
-    docker run -d --name unittest_keycloak -e KEYCLOAK_ADMIN="${KEYCLOAK_ADMIN}" -e KEYCLOAK_ADMIN_PASSWORD="${KEYCLOAK_ADMIN_PASSWORD}" -p "${KEYCLOAK_PORT}:8080" "${KEYCLOAK_DOCKER_IMAGE}" start-dev
+    docker run -d --name unittest_keycloak -e KEYCLOAK_ADMIN="${KEYCLOAK_ADMIN}" -e KEYCLOAK_ADMIN_PASSWORD="${KEYCLOAK_ADMIN_PASSWORD}" -e KC_FEATURES="token-exchange" -p "${KEYCLOAK_PORT}:8080" "${KEYCLOAK_DOCKER_IMAGE}" start-dev
     SECONDS=0
     until curl --silent --output /dev/null localhost:$KEYCLOAK_PORT; do
       sleep 5;
@@ -28,7 +28,7 @@ keycloak_stop # In case it did not shut down correctly last time.
 keycloak_start
 
 eval ${CMD_ARGS}
-docker logs unittest_keycloak > keycloak_test_logs.txt
 RETURN_VALUE=$?
+docker logs unittest_keycloak > keycloak_test_logs.txt
 
 exit ${RETURN_VALUE}

tests/conftest.py

@@ -124,13 +124,65 @@ def oid_with_credentials(env: KeycloakTestEnv, realm: str, admin: KeycloakAdmin)
     admin.realm_name = realm
     # Create client
     client = str(uuid.uuid4())
+    secret = str(uuid.uuid4())
     client_id = admin.create_client(
         payload={
             "name": client,
             "clientId": client,
             "enabled": True,
-            "publicClient": True,
+            "publicClient": False,
+            "protocol": "openid-connect",
+            "secret": secret,
+            "clientAuthenticatorType": "client-secret",
+        }
+    )
+    # Create user
+    username = str(uuid.uuid4())
+    password = str(uuid.uuid4())
+    user_id = admin.create_user(
+        payload={
+            "username": username,
+            "email": f"{username}@test.test",
+            "enabled": True,
+            "credentials": [{"type": "password", "value": password}],
+        }
+    )
+
+    yield (
+        KeycloakOpenID(
+            server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
+            realm_name=realm,
+            client_id=client,
+            client_secret_key=secret,
+        ),
+        username,
+        password,
+    )
+
+    # Cleanup
+    admin.delete_client(client_id=client_id)
+    admin.delete_user(user_id=user_id)
+
+
+@pytest.fixture
+def oid_with_credentials_authz(env: KeycloakTestEnv, realm: str, admin: KeycloakAdmin):
+    """Fixture for an initialized KeycloakOpenID class and a random user credentials."""
+    # Set the realm
+    admin.realm_name = realm
+    # Create client
+    client = str(uuid.uuid4())
+    secret = str(uuid.uuid4())
+    client_id = admin.create_client(
+        payload={
+            "name": client,
+            "clientId": client,
+            "enabled": True,
+            "publicClient": False,
             "protocol": "openid-connect",
+            "secret": secret,
+            "clientAuthenticatorType": "client-secret",
+            "authorizationServicesEnabled": True,
+            "serviceAccountsEnabled": True,
         }
     )
     # Create user
@@ -150,6 +202,7 @@ def oid_with_credentials(env: KeycloakTestEnv, realm: str, admin: KeycloakAdmin)
             server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
             realm_name=realm,
             client_id=client,
+            client_secret_key=secret,
         ),
         username,
         password,

tests/test_keycloak_openid.py

@@ -1,8 +1,12 @@
 """Test module for KeycloakOpenID."""
 from unittest import mock
 
+import pytest
+
 from keycloak.authorization import Authorization
 from keycloak.connection import ConnectionManager
+from keycloak.exceptions import KeycloakDeprecationError, KeycloakRPTNotFound
+from keycloak.keycloak_admin import KeycloakAdmin
 from keycloak.keycloak_openid import KeycloakOpenID
 
 
@@ -105,7 +109,7 @@ def test_token(oid_with_credentials: tuple[KeycloakOpenID, str, str]):
         "not-before-policy": 0,
         "refresh_expires_in": 1800,
         "refresh_token": mock.ANY,
-        "scope": "profile email",
+        "scope": mock.ANY,
         "session_state": mock.ANY,
         "token_type": "Bearer",
     }
@@ -118,7 +122,7 @@ def test_token(oid_with_credentials: tuple[KeycloakOpenID, str, str]):
         "not-before-policy": 0,
         "refresh_expires_in": 1800,
         "refresh_token": mock.ANY,
-        "scope": "profile email",
+        "scope": mock.ANY,
         "session_state": mock.ANY,
         "token_type": "Bearer",
     }
@@ -131,7 +135,104 @@ def test_token(oid_with_credentials: tuple[KeycloakOpenID, str, str]):
         "not-before-policy": 0,
         "refresh_expires_in": 1800,
         "refresh_token": mock.ANY,
-        "scope": "profile email",
+        "scope": mock.ANY,
         "session_state": mock.ANY,
         "token_type": "Bearer",
     }
+
+
+def test_exchange_token(
+    oid_with_credentials: tuple[KeycloakOpenID, str, str], admin: KeycloakAdmin
+):
+    """Test the exchange token method."""
+    # Verify existing user
+    oid, username, password = oid_with_credentials
+
+    # Allow impersonation
+    admin.realm_name = oid.realm_name
+    admin.assign_client_role(
+        user_id=admin.get_user_id(username=username),
+        client_id=admin.get_client_id(client_name="realm-management"),
+        roles=[
+            admin.get_client_role(
+                client_id=admin.get_client_id(client_name="realm-management"),
+                role_name="impersonation",
+            )
+        ],
+    )
+
+    token = oid.token(username=username, password=password)
+    assert oid.userinfo(token=token["access_token"]) == {
+        "email": f"{username}@test.test",
+        "email_verified": False,
+        "preferred_username": username,
+        "sub": mock.ANY,
+    }
+
+    # Exchange token with the new user
+    new_token = oid.exchange_token(
+        token=token["access_token"],
+        client_id=oid.client_id,
+        audience=oid.client_id,
+        subject=username,
+    )
+    assert oid.userinfo(token=new_token["access_token"]) == {
+        "email": f"{username}@test.test",
+        "email_verified": False,
+        "preferred_username": username,
+        "sub": mock.ANY,
+    }
+    assert token != new_token
+
+
+def test_certs(oid: KeycloakOpenID):
+    """Test certificates."""
+    assert len(oid.certs()["keys"]) == 2
+
+
+def test_public_key(oid: KeycloakOpenID):
+    """Test public key."""
+    assert oid.public_key() is not None
+
+
+def test_entitlement(
+    oid_with_credentials_authz: tuple[KeycloakOpenID, str, str], admin: KeycloakAdmin
+):
+    """Test entitlement."""
+    oid, username, password = oid_with_credentials_authz
+    token = oid.token(username=username, password=password)
+    resource_server_id = admin.get_client_authz_resources(
+        client_id=admin.get_client_id(oid.client_id)
+    )[0]["_id"]
+
+    with pytest.raises(KeycloakDeprecationError):
+        oid.entitlement(token=token["access_token"], resource_server_id=resource_server_id)
+
+
+def test_introspect(oid_with_credentials: tuple[KeycloakOpenID, str, str]):
+    """Test introspect."""
+    oid, username, password = oid_with_credentials
+    token = oid.token(username=username, password=password)
+
+    assert oid.introspect(token=token["access_token"])["active"]
+    assert oid.introspect(
+        token=token["access_token"], rpt="some", token_type_hint="requesting_party_token"
+    ) == {"active": False}
+
+    with pytest.raises(KeycloakRPTNotFound):
+        oid.introspect(token=token["access_token"], token_type_hint="requesting_party_token")
+
+
+def test_decode_token(oid_with_credentials: tuple[KeycloakOpenID, str, str]):
+    """Test decode token."""
+    oid, username, password = oid_with_credentials
+    token = oid.token(username=username, password=password)
+
+    assert (
+        oid.decode_token(
+            token=token["access_token"],
+            key="-----BEGIN PUBLIC KEY-----\n" + oid.public_key() + "\n-----END PUBLIC KEY-----",
+            options={"verify_aud": False},
+        )["preferred_username"]
+        == username
+    )