[DOP-21482] - move keycloak test configs to fixtures

.env.local

@@ -18,15 +18,6 @@ export SYNCMASTER__CRYPTO_KEY=UBgPTioFrtH2unlC4XFDiGf5sYfzbdSf_VgiUSaQc94=
 # Postgres
 export SYNCMASTER__DATABASE__URL=postgresql+asyncpg://syncmaster:changeme@localhost:5432/syncmaster
 
-# Keycloack Auth
-export SYNCMASTER__AUTH__SERVER_URL=http://keycloak:8080/
-export SYNCMASTER__AUTH__REALM_NAME=manually_created
-export SYNCMASTER__AUTH__CLIENT_ID=manually_created
-export SYNCMASTER__AUTH__CLIENT_SECRET=generated_by_keycloak
-export SYNCMASTER__AUTH__REDIRECT_URI=http://localhost:8000/auth/callback
-export SYNCMASTER__AUTH__SCOPE=email
-export SYNCMASTER__AUTH__PROVIDER=syncmaster.backend.providers.auth.keycloak_provider.KeycloakAuthProvider
-
 # Dummy Auth
 export SYNCMASTER__AUTH__PROVIDER=syncmaster.backend.providers.auth.dummy_provider.DummyAuthProvider
 export SYNCMASTER__AUTH__ACCESS_TOKEN__SECRET_KEY=secret
@@ -69,3 +60,13 @@ export TEST_HDFS_IPC_PORT=9820
 export SPARK_CONF_DIR=./tests/spark/hive/conf/
 export HADOOP_CONF_DIR=./tests/spark/hadoop/
 export HIVE_CONF_DIR=./tests/spark/hive/conf/
+
+
+# Keycloack Auth
+export TEST_AUTH__SERVER_URL=http://localhost:8080
+export TEST_AUTH__REALM_NAME=manually_created
+export TEST_AUTH__CLIENT_ID=manually_created
+export TEST_AUTH__CLIENT_SECRET=generated_by_keycloak
+export TEST_AUTH__REDIRECT_URI=http://localhost:8000/auth/callback
+export TEST_AUTH__SCOPE=email
+export TEST_AUTH__PROVIDER=syncmaster.backend.providers.auth.keycloak_provider.KeycloakAuthProvider

tests/conftest.py

@@ -32,6 +32,7 @@
 
 pytest_plugins = [
     "tests.test_unit.test_transfers.transfer_fixtures",
+    "tests.test_unit.test_auth.auth_fixtures",
     "tests.test_unit.test_runs.run_fixtures",
     "tests.test_unit.test_connections.connection_fixtures",
     "tests.test_unit.test_scheduler.scheduler_fixtures",

tests/settings.py

@@ -40,6 +40,14 @@ class TestSettings(BaseSettings):
     TEST_S3_PROTOCOL: str = "http"
     TEST_S3_ADDITIONAL_PARAMS: dict = {}
 
+    TEST_AUTH__SERVER_URL: str
+    TEST_AUTH__REALM_NAME: str
+    TEST_AUTH__CLIENT_ID: str
+    TEST_AUTH__CLIENT_SECRET: str
+    TEST_AUTH__REDIRECT_URI: str
+    TEST_AUTH__SCOPE: str
+    TEST_AUTH__PROVIDER: str
+
     @model_validator(mode="before")
     def check_sid_and_service_name(cls, values):
         sid = values.get("TEST_ORACLE_SID")

tests/test_unit/test_auth/auth_fixtures/__init__.py

@@ -0,0 +1,6 @@
+from tests.test_unit.test_auth.auth_fixtures.keycloak_fixture import (
+    create_session_cookie,
+    mock_keycloak_realm,
+    mock_keycloak_well_known,
+    rsa_keys,
+)

tests/test_unit/test_auth/auth_fixtures/keycloak_fixture.py

@@ -0,0 +1,122 @@
+import json
+import time
+from base64 import b64encode
+
+import pytest
+import responses
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
+from itsdangerous import TimestampSigner
+from jose import jwt
+
+
+@pytest.fixture(scope="session")
+def rsa_keys():
+    # create private & public keys to emulate Keycloak signing
+    private_key = rsa.generate_private_key(
+        public_exponent=65537,
+        key_size=2048,
+    )
+    private_pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    public_key = private_key.public_key()
+
+    return {
+        "private_key": private_key,
+        "private_pem": private_pem,
+        "public_key": public_key,
+    }
+
+
+def get_public_key_pem(public_key):
+    public_pem = public_key.public_bytes(
+        encoding=Encoding.PEM,
+        format=PublicFormat.SubjectPublicKeyInfo,
+    )
+    public_pem_str = public_pem.decode("utf-8")
+    public_pem_str = public_pem_str.replace("-----BEGIN PUBLIC KEY-----\n", "")
+    public_pem_str = public_pem_str.replace("-----END PUBLIC KEY-----\n", "")
+    public_pem_str = public_pem_str.replace("\n", "")
+    return public_pem_str
+
+
+@pytest.fixture
+def create_session_cookie(rsa_keys, settings):
+    def _create_session_cookie(user, expire_in_msec=1000) -> str:
+        private_pem = rsa_keys["private_pem"]
+        session_secret_key = settings.server.session.secret_key
+
+        payload = {
+            "sub": str(user.id),
+            "preferred_username": user.username,
+            "email": user.email,
+            "given_name": user.first_name,
+            "middle_name": user.middle_name,
+            "family_name": user.last_name,
+            "exp": int(time.time()) + (expire_in_msec / 1000),
+        }
+
+        access_token = jwt.encode(payload, private_pem, algorithm="RS256")
+        refresh_token = "mock_refresh_token"
+
+        session_data = {
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+        }
+
+        signer = TimestampSigner(session_secret_key)
+        json_bytes = json.dumps(session_data).encode("utf-8")
+        base64_bytes = b64encode(json_bytes)
+        signed_data = signer.sign(base64_bytes)
+        session_cookie = signed_data.decode("utf-8")
+
+        return session_cookie
+
+    return _create_session_cookie
+
+
+@pytest.fixture
+def mock_keycloak_well_known(test_settings):
+    server_url = test_settings.TEST_AUTH__SERVER_URL
+    realm_name = test_settings.TEST_AUTH__REALM_NAME
+    well_known_url = f"{server_url}/realms/{realm_name}/.well-known/openid-configuration"
+
+    responses.add(
+        responses.GET,
+        well_known_url,
+        json={
+            "authorization_endpoint": f"{server_url}/realms/{realm_name}/protocol/openid-connect/auth",
+            "token_endpoint": f"{server_url}/realms/{realm_name}/protocol/openid-connect/token",
+            "userinfo_endpoint": f"{server_url}/realms/{realm_name}/protocol/openid-connect/userinfo",
+            "end_session_endpoint": f"{server_url}/realms/{realm_name}/protocol/openid-connect/logout",
+            "jwks_uri": f"{server_url}/realms/{realm_name}/protocol/openid-connect/certs",
+            "issuer": f"{server_url}/realms/{realm_name}",
+        },
+        status=200,
+        content_type="application/json",
+    )
+
+
+@pytest.fixture
+def mock_keycloak_realm(test_settings, rsa_keys):
+    server_url = test_settings.TEST_AUTH__SERVER_URL
+    realm_name = test_settings.TEST_AUTH__REALM_NAME
+    realm_url = f"{server_url}/realms/{realm_name}"
+    public_pem_str = get_public_key_pem(rsa_keys["public_key"])
+
+    responses.add(
+        responses.GET,
+        realm_url,
+        json={
+            "realm": realm_name,
+            "public_key": public_pem_str,
+            "token-service": f"{server_url}/realms/{realm_name}/protocol/openid-connect/token",
+            "account-service": f"{server_url}/realms/{realm_name}/account",
+        },
+        status=200,
+        content_type="application/json",
+    )