adding tests

Author: c0mpiler

tests/__init__.py

@@ -0,0 +1,5 @@
+# -*- coding: utf-8 -*-
+"""
+Author: Harsha Krishnareddy
+Email: c0mpiler@ins8s.dev
+"""

tests/conftest.py

@@ -0,0 +1,330 @@
+# -*- coding: utf-8 -*-
+"""
+Author: Harsha Krishnareddy
+Email: c0mpiler@ins8s.dev
+"""
+import importlib
+import json
+import os
+import threading
+import time
+from http.server import HTTPServer
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+from jose import jwk
+from jose import jwt as jose_jwt
+
+# --- IMPORTANT: Import modules using the NEW package name 'novactl' --- #
+from novactl import auth, config, main # Import main as well
+
+# --- Constants for Mocking ---
+TEST_SERVICE_NAME = "novactl-test"
+# Use derived username logic from config for consistency
+TEST_KEYRING_USERNAME = config._derive_keyring_username(TEST_SERVICE_NAME)
+TEST_REALM = "ibm-eda"
+TEST_CLIENT_ID = "eda-ui"
+TEST_SERVER_URL = "https://auth.eda.res.ibm.com:8443/auth"
+MOCK_KID = "test-key-id"
+
+# --- Test Key Pair Generation --- #
+private_key_pem, public_key_pem = None, None
+mock_public_jwk = None
+try:
+    from cryptography.hazmat.primitives import serialization
+    from cryptography.hazmat.primitives.asymmetric import rsa
+
+    _private_key_obj = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    _public_key_obj = _private_key_obj.public_key()
+    private_key_pem = _private_key_obj.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    public_key_pem = _public_key_obj.public_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    mock_public_jwk = jwk.construct(_public_key_obj, algorithm="RS256").to_dict()
+    mock_public_jwk["kid"] = MOCK_KID
+    mock_public_jwk["use"] = "sig"
+except ImportError:
+    print("WARNING: cryptography library not found, JWT tests will be limited.")
+    mock_public_jwk = {
+        "kty": "RSA", "alg": "RS256", "kid": MOCK_KID, "use": "sig",
+        "n": "dummy", "e": "AQAB",
+    }
+
+# --- Base Test Configuration Data --- #
+TEST_CONFIG_DATA = {
+    "keycloak_server_url": TEST_SERVER_URL,
+    "keycloak_realm": TEST_REALM,
+    "keycloak_client_id": TEST_CLIENT_ID,
+    "keyring_service_name": TEST_SERVICE_NAME,
+    "additional_scopes": "openid email profile offline_access",
+    "token_leeway_seconds": 10,
+    "ssl_verify": True,
+    "log_level": "DEBUG",
+    "console_log_level": "DEBUG",
+}
+
+# --- Fixtures --- #
+
+@pytest.fixture(scope="function", autouse=True)
+def mock_config_and_reload_modules(monkeypatch):
+    """Mocks config loading, derives values, and reloads affected modules."""
+    # Start with default config and override with test data
+    current_test_config = config.DEFAULT_CONFIG.copy()
+    current_test_config.update(TEST_CONFIG_DATA)
+
+    # Manually derive values that load_config would normally handle
+    current_test_config["keyring_username"] = config._derive_keyring_username(
+        current_test_config["keyring_service_name"]
+    )
+    current_test_config["ssl_verify"] = config._process_ssl_verify(
+        current_test_config["ssl_verify"]
+    )
+    server_url = current_test_config["keycloak_server_url"].rstrip("/")
+    realm = current_test_config["keycloak_realm"]
+    if server_url and realm:
+        base_oidc_path = f"{server_url}/realms/{realm}/protocol/openid-connect"
+        current_test_config["oidc_token_endpoint"] = f"{base_oidc_path}/token"
+        current_test_config["oidc_logout_endpoint"] = f"{base_oidc_path}/logout"
+        current_test_config["oidc_jwks_uri"] = f"{base_oidc_path}/certs"
+        current_test_config["oidc_issuer"] = f"{server_url}/realms/{realm}"
+        current_test_config["oidc_device_auth_endpoint"] = f"{base_oidc_path}/auth/device"
+
+    # --- Mock Config Module --- #
+    mock_load = MagicMock(return_value=current_test_config)
+    monkeypatch.setattr(config, "load_config", mock_load)
+    monkeypatch.setattr(config, "_config", current_test_config)
+    # Ensure get_config_value uses the mocked config
+    def mock_get_config_value(key, default=None):
+        return current_test_config.get(key, default)
+    monkeypatch.setattr(config, "get_config_value", mock_get_config_value)
+    # Mock logging setup within config as it might run on import
+    monkeypatch.setattr(config, "_configure_logging", MagicMock())
+
+    # --- Reload Auth Module --- #
+    # Reload to make it pick up the mocked config functions
+    importlib.reload(auth)
+
+    # --- Mock Dependencies within Auth Module --- #
+    # Mock JWKS fetching within the reloaded auth module
+    mock_jwks_fetch = MagicMock(return_value={MOCK_KID: mock_public_jwk})
+    monkeypatch.setattr(auth, "_get_public_keys", mock_jwks_fetch)
+    auth._public_keys = None # Reset cache
+    auth._jwks_uri = current_test_config.get("oidc_jwks_uri")
+
+    # Mock time.sleep used in polling
+    mock_sleep = MagicMock()
+    monkeypatch.setattr(auth.time, "sleep", mock_sleep)
+
+    # Mock Rich Progress used in polling
+    mock_progress_instance = MagicMock()
+    mock_progress_instance.add_task.return_value = 1 # Dummy task ID
+    mock_progress_instance.update = MagicMock()
+    mock_progress_class = MagicMock()
+    mock_progress_class.return_value.__enter__.return_value = mock_progress_instance
+    mock_progress_class.return_value.__exit__.return_value = False # No exception on exit
+    monkeypatch.setattr(auth, "Progress", mock_progress_class)
+
+    # --- Reload Main Module --- #
+    # Reload main to pick up reloaded auth/config
+    importlib.reload(main)
+
+    # Yield the prepared config for tests to use
+    yield current_test_config
+
+    # --- Cleanup --- #
+    # Reset caches and module state after test
+    config._config = None
+    config._config_schema = None
+    config._logging_configured = False
+    auth._public_keys = None
+    auth._jwks_uri = None
+    # Consider reloading original modules if necessary, though pytest usually isolates tests
+
+
+@pytest.fixture(scope="function", autouse=True)
+def mock_keyring(monkeypatch):
+    """Mocks the keyring module used by auth.py."""
+    keyring_store: Dict[tuple, str] = {}
+
+    def set_password(service, username, password):
+        keyring_store[(service, username)] = password
+
+    def get_password(service, username):
+        return keyring_store.get((service, username))
+
+    def delete_password(service, username):
+        # Simulate PasswordDeleteError if not found
+        try:
+            from keyring.errors import PasswordDeleteError
+            if (service, username) not in keyring_store:
+                 raise PasswordDeleteError("Not found")
+            keyring_store.pop((service, username), None)
+        except ImportError:
+             keyring_store.pop((service, username), None) # Fallback if error class not available
+
+    # Patch the keyring functions within the reloaded auth module
+    monkeypatch.setattr(auth.keyring, "set_password", set_password)
+    monkeypatch.setattr(auth.keyring, "get_password", get_password)
+    monkeypatch.setattr(auth.keyring, "delete_password", delete_password)
+
+    # Dynamically patch PasswordDeleteError if keyring library is available
+    try:
+        from keyring.errors import PasswordDeleteError
+        monkeypatch.setattr(auth.keyring.errors, "PasswordDeleteError", PasswordDeleteError)
+    except (ImportError, AttributeError):
+        # If keyring or its errors module isn't available, skip patching the error class
+        pass
+
+    yield keyring_store # Provide the store to tests for inspection/manipulation
+    keyring_store.clear() # Clear store after test
+
+
+@pytest.fixture(scope="function")
+def mock_requests(requests_mock, mock_config_and_reload_modules):
+    """Sets up common Keycloak endpoint mocks for device flow using requests_mock."""
+    # Get config from the other fixture
+    cfg = mock_config_and_reload_modules
+
+    # --- Endpoint URLs from Config --- #
+    device_auth_url = cfg.get("oidc_device_auth_endpoint")
+    token_url = cfg.get("oidc_token_endpoint")
+    jwks_uri = cfg.get("oidc_jwks_uri")
+
+    if not all([device_auth_url, token_url, jwks_uri]):
+         pytest.fail("Missing OIDC endpoint configuration in mock_config")
+
+    # --- Mock JWKS Endpoint --- #
+    requests_mock.get(jwks_uri, json={"keys": [mock_public_jwk] if mock_public_jwk else []})
+
+    # --- Token Generation Helper --- #
+    def generate_mock_token(
+        payload_override=None, expiry_offset_secs=3600, token_type="Bearer"
+    ):
+        """Generates a mock JWT token signed with the test private key."""
+        if private_key_pem is None:
+             # Fallback for limited testing without cryptography
+             timestamp = int(time.time())
+             return f"mock_{token_type}_token_{timestamp}"
+
+        now = int(time.time())
+        iat = int(payload_override.get("iat", now)) if payload_override else now
+        payload = {
+            "iss": cfg["oidc_issuer"],
+            "sub": "mock-user-id",
+            "aud": cfg["keycloak_client_id"],
+            "exp": iat + expiry_offset_secs,
+            "iat": iat,
+            "jti": f"jwt-id-{iat}-{token_type}",
+            "typ": token_type,
+            "azp": cfg["keycloak_client_id"],
+            "preferred_username": "mockuser",
+            "email": "mockuser@test.local", "email_verified": True,
+            "name": "Mock User",
+            "realm_access": {"roles": ["default-role", "realm-test-role"]},
+            "resource_access": {cfg["keycloak_client_id"]: {"roles": ["client-test-role"]}},
+        }
+        # Add at_hash for ID tokens if access token can be generated consistently
+        if token_type.lower() == "id":
+            try:
+                 # Generate a *consistent* access token based on the same iat
+                 consistent_access_token = generate_mock_token(payload_override={'iat': iat}, token_type="Bearer")
+                 from jose.utils import calculate_at_hash
+                 # Use the correct algorithm specified in the header (RS256)
+                 payload['at_hash'] = calculate_at_hash(consistent_access_token, jwk.ALGORITHMS.RS256)
+            except Exception as e:
+                 print(f"Warning: Could not generate at_hash for mock ID token: {e}")
+
+        if payload_override: payload.update(payload_override)
+        # Recalculate expiry if iat was overridden
+        if payload_override and 'iat' in payload_override: payload['exp'] = payload_override['iat'] + expiry_offset_secs
+
+        headers = {"alg": "RS256", "typ": "JWT", "kid": MOCK_KID}
+        return jose_jwt.encode(payload, private_key_pem, algorithm="RS256", headers=headers)
+
+    # --- Token Response Generation Helper --- #
+    def generate_token_response(
+        iat_override=None, access_token_expiry=300, refresh_token_expiry=1800, include_refresh=True
+    ):
+        """Generates a dictionary mimicking a successful token endpoint response."""
+        now = int(time.time())
+        iat = iat_override if iat_override is not None else now
+        access_payload = {"iat": iat}
+        id_payload = {"iat": iat} # Nonce not relevant for device flow
+
+        response = {
+            "access_token": generate_mock_token(payload_override=access_payload, expiry_offset_secs=access_token_expiry, token_type="Bearer"),
+            "expires_in": access_token_expiry,
+            "refresh_expires_in": refresh_token_expiry,
+            "token_type": "Bearer",
+            "id_token": generate_mock_token(payload_override=id_payload, expiry_offset_secs=access_token_expiry, token_type="ID"),
+            "session_state": f"mock-session-state-{iat}",
+            "scope": cfg.get("additional_scopes", "openid email profile"),
+        }
+        if include_refresh and "offline_access" in response["scope"]:
+            response["refresh_token"] = f"mock_refresh_token_{iat}"
+        return response
+
+    # --- Mock Device Authorization Endpoint --- #
+    mock_device_response = {
+        "device_code": "mock-device-code-123",
+        "user_code": "ABCD-EFGH",
+        "verification_uri": f"{cfg['oidc_issuer']}/device", # Example URI
+        "verification_uri_complete": f"{cfg['oidc_issuer']}/device?user_code=ABCD-EFGH",
+        "expires_in": 180, # 3 minutes
+        "interval": 5, # Default polling interval
+    }
+    requests_mock.post(device_auth_url, json=mock_device_response)
+
+    # --- Mock Token Endpoint (Default: Pending) --- #
+    # *** CORRECTED: Use standard requests_mock.post ***
+    requests_mock.post(token_url, json={"error": "authorization_pending"}, status_code=400)
+
+    # --- Test Helpers --- #
+    def set_token_endpoint_success(response_data=None):
+        """Updates the token endpoint mock to return a successful response."""
+        if response_data is None:
+            response_data = generate_token_response()
+        # Use requests_mock directly, it handles overwriting previous mocks for the same URL/method
+        requests_mock.post(token_url, json=response_data, status_code=200)
+
+    def set_token_endpoint_error(error_code="invalid_grant", status=400):
+        """Updates the token endpoint mock to return an error response."""
+        requests_mock.post(token_url, json={"error": error_code}, status_code=status)
+
+    # Provide mocks and helpers to the test function
+    test_helpers = {
+        "generate_token": generate_mock_token,
+        "generate_token_response": generate_token_response,
+        "device_auth_url": device_auth_url,
+        "token_url": token_url,
+        "jwks_uri": jwks_uri,
+        "mock_device_response": mock_device_response,
+        "set_token_endpoint_success": set_token_endpoint_success,
+        "set_token_endpoint_error": set_token_endpoint_error,
+    }
+    yield requests_mock, test_helpers # Yield requests_mock and the helpers dict
+
+
+@pytest.fixture(autouse=True)
+def reset_auth_caches(mock_keyring): # Depends on mock_keyring fixture
+    """Resets auth state and clears keyring before each test."""
+    # Clear auth module caches
+    auth._public_keys = None
+    auth._jwks_uri = None
+    # Clear the mocked keyring store via the fixture
+    mock_keyring.clear()
+    # Reset config cache (handled by mock_config_and_reload_modules fixture)
+
+@pytest.fixture
+def runner():
+    """Provides a Typer CliRunner instance."""
+    from typer.testing import CliRunner
+    # Ensure the main module's app instance is used
+    return CliRunner(mix_stderr=False) # Use the app from the reloaded main module