|
| 1 | +import pytest |
| 2 | +import shlex |
| 3 | +import subprocess |
| 4 | +import os |
| 5 | +from typing import Generator, List, Dict, Any |
| 6 | +from ocp_resources.namespace import Namespace |
| 7 | +from ocp_resources.service_account import ServiceAccount |
| 8 | +from ocp_resources.role_binding import RoleBinding |
| 9 | +from ocp_resources.role import Role |
| 10 | +from kubernetes.dynamic import DynamicClient |
| 11 | +from pyhelper_utils.shell import run_command |
| 12 | +from tests.model_registry.utils import generate_random_name, generate_namespace_name |
| 13 | +from simple_logger.logger import get_logger |
| 14 | +from tests.model_registry.constants import MR_INSTANCE_NAME |
| 15 | + |
| 16 | + |
| 17 | +LOGGER = get_logger(name=__name__) |
| 18 | +DEFAULT_TOKEN_DURATION = "10m" |
| 19 | + |
| 20 | + |
| 21 | +@pytest.fixture(scope="function") |
| 22 | +def sa_namespace(request: pytest.FixtureRequest, admin_client: DynamicClient) -> Generator[Namespace, None, None]: |
| 23 | + """ |
| 24 | + Creates a temporary namespace using a context manager for automatic cleanup. |
| 25 | + Function scope ensures a fresh namespace for each test needing it. |
| 26 | + """ |
| 27 | + test_file = os.path.relpath(request.fspath.strpath, start=os.path.dirname(__file__)) |
| 28 | + ns_name = generate_namespace_name(file_path=test_file) |
| 29 | + LOGGER.info(f"Creating temporary namespace: {ns_name}") |
| 30 | + with Namespace(client=admin_client, name=ns_name) as ns: |
| 31 | + ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120) |
| 32 | + yield ns |
| 33 | + |
| 34 | + |
| 35 | +@pytest.fixture(scope="function") |
| 36 | +def service_account(admin_client: DynamicClient, sa_namespace: Namespace) -> Generator[ServiceAccount, None, None]: |
| 37 | + """ |
| 38 | + Creates a ServiceAccount within the temporary namespace using a context manager. |
| 39 | + Function scope ensures it's tied to the lifetime of sa_namespace for that test. |
| 40 | + """ |
| 41 | + sa_name = generate_random_name(prefix="mr-test-user") |
| 42 | + LOGGER.info(f"Creating ServiceAccount: {sa_name} in namespace {sa_namespace.name}") |
| 43 | + with ServiceAccount(client=admin_client, name=sa_name, namespace=sa_namespace.name, wait_for_resource=True) as sa: |
| 44 | + yield sa |
| 45 | + |
| 46 | + |
| 47 | +@pytest.fixture(scope="function") |
| 48 | +def sa_token(service_account: ServiceAccount) -> str: |
| 49 | + """ |
| 50 | + Retrieves a short-lived token for the ServiceAccount using 'oc create token'. |
| 51 | + Function scope because token is temporary and tied to the SA for that test. |
| 52 | + """ |
| 53 | + sa_name = service_account.name |
| 54 | + namespace = service_account.namespace |
| 55 | + LOGGER.info(f"Retrieving token for ServiceAccount: {sa_name} in namespace {namespace}") |
| 56 | + try: |
| 57 | + cmd = f"oc create token {sa_name} -n {namespace} --duration={DEFAULT_TOKEN_DURATION}" |
| 58 | + LOGGER.debug(f"Executing command: {cmd}") |
| 59 | + res, out, err = run_command(command=shlex.split(cmd), verify_stderr=False, check=True, timeout=30) |
| 60 | + token = out.strip() |
| 61 | + if not token: |
| 62 | + raise ValueError("Retrieved token is empty after successful command execution.") |
| 63 | + |
| 64 | + LOGGER.info(f"Successfully retrieved token for SA '{sa_name}'") |
| 65 | + return token |
| 66 | + |
| 67 | + except Exception as e: # Catch all exceptions from the try block |
| 68 | + error_type = type(e).__name__ |
| 69 | + log_message = ( |
| 70 | + f"Failed during token retrieval for SA '{sa_name}' in namespace '{namespace}'. " |
| 71 | + f"Error Type: {error_type}, Message: {str(e)}" |
| 72 | + ) |
| 73 | + if isinstance(e, subprocess.CalledProcessError): |
| 74 | + # Add specific details for CalledProcessError |
| 75 | + # run_command already logs the error if log_errors=True and returncode !=0, |
| 76 | + # but we can add context here. |
| 77 | + stderr_from_exception = e.stderr.strip() if e.stderr else "N/A" |
| 78 | + log_message += f". Exit Code: {e.returncode}. Stderr from exception: {stderr_from_exception}" |
| 79 | + elif isinstance(e, subprocess.TimeoutExpired): |
| 80 | + timeout_value = getattr(e, "timeout", "N/A") |
| 81 | + log_message += f". Command timed out after {timeout_value} seconds." |
| 82 | + elif isinstance(e, FileNotFoundError): |
| 83 | + # This occurs if 'oc' is not found. |
| 84 | + # e.filename usually holds the name of the file that was not found. |
| 85 | + command_not_found = e.filename if hasattr(e, "filename") and e.filename else shlex.split(cmd)[0] |
| 86 | + log_message += f". Command '{command_not_found}' not found. Is it installed and in PATH?" |
| 87 | + |
| 88 | + LOGGER.error(log_message, exc_info=True) # exc_info=True adds stack trace to the log |
| 89 | + raise |
| 90 | + |
| 91 | + |
| 92 | +# --- RBAC Fixtures --- |
| 93 | + |
| 94 | + |
| 95 | +@pytest.fixture(scope="function") |
| 96 | +def mr_access_role( |
| 97 | + admin_client: DynamicClient, |
| 98 | + model_registry_namespace: str, |
| 99 | + sa_namespace: Namespace, |
| 100 | +) -> Generator[Role, None, None]: |
| 101 | + """ |
| 102 | + Creates the MR Access Role using direct constructor parameters and a context manager. |
| 103 | + """ |
| 104 | + role_name = f"registry-user-{MR_INSTANCE_NAME}-{sa_namespace.name[:8]}" |
| 105 | + LOGGER.info(f"Defining Role: {role_name} in namespace {model_registry_namespace}") |
| 106 | + |
| 107 | + role_rules: List[Dict[str, Any]] = [ |
| 108 | + { |
| 109 | + "apiGroups": [""], |
| 110 | + "resources": ["services"], |
| 111 | + "resourceNames": [MR_INSTANCE_NAME], # Grant access only to the specific MR service object |
| 112 | + "verbs": ["get"], |
| 113 | + } |
| 114 | + ] |
| 115 | + role_labels = { |
| 116 | + "app.kubernetes.io/component": "model-registry-test-rbac", |
| 117 | + "test.opendatahub.io/namespace": sa_namespace.name, |
| 118 | + } |
| 119 | + |
| 120 | + LOGGER.info(f"Attempting to create Role: {role_name} with rules and labels.") |
| 121 | + with Role( |
| 122 | + client=admin_client, |
| 123 | + name=role_name, |
| 124 | + namespace=model_registry_namespace, |
| 125 | + rules=role_rules, |
| 126 | + label=role_labels, |
| 127 | + wait_for_resource=True, |
| 128 | + ) as role: |
| 129 | + LOGGER.info(f"Role {role.name} created successfully.") |
| 130 | + yield role |
| 131 | + LOGGER.info(f"Role {role.name} deletion initiated by context manager.") |
| 132 | + |
| 133 | + |
| 134 | +@pytest.fixture(scope="function") |
| 135 | +def mr_access_role_binding( |
| 136 | + admin_client: DynamicClient, |
| 137 | + model_registry_namespace: str, |
| 138 | + mr_access_role: Role, |
| 139 | + sa_namespace: Namespace, |
| 140 | +) -> Generator[RoleBinding, None, None]: |
| 141 | + """ |
| 142 | + Creates the MR Access RoleBinding using direct constructor parameters and a context manager. |
| 143 | + """ |
| 144 | + binding_name = f"{mr_access_role.name}-binding" |
| 145 | + |
| 146 | + LOGGER.info( |
| 147 | + f"Defining RoleBinding: {binding_name} linking Group 'system:serviceaccounts:{sa_namespace.name}' " |
| 148 | + f"to Role '{mr_access_role.name}' in namespace {model_registry_namespace}" |
| 149 | + ) |
| 150 | + binding_labels = { |
| 151 | + "app.kubernetes.io/component": "model-registry-test-rbac", |
| 152 | + "test.opendatahub.io/namespace": sa_namespace.name, |
| 153 | + } |
| 154 | + |
| 155 | + LOGGER.info(f"Attempting to create RoleBinding: {binding_name} with labels.") |
| 156 | + with RoleBinding( |
| 157 | + client=admin_client, |
| 158 | + name=binding_name, |
| 159 | + namespace=model_registry_namespace, |
| 160 | + # Subject parameters |
| 161 | + subjects_kind="Group", |
| 162 | + subjects_name=f"system:serviceaccounts:{sa_namespace.name}", |
| 163 | + subjects_api_group="rbac.authorization.k8s.io", # This is the default apiGroup for Group kind |
| 164 | + # Role reference parameters |
| 165 | + role_ref_kind="Role", |
| 166 | + role_ref_name=mr_access_role.name, |
| 167 | + label=binding_labels, |
| 168 | + wait_for_resource=True, |
| 169 | + ) as binding: |
| 170 | + LOGGER.info(f"RoleBinding {binding.name} created successfully.") |
| 171 | + yield binding |
| 172 | + LOGGER.info(f"RoleBinding {binding.name} deletion initiated by context manager.") |
0 commit comments