Refactor VM console proxy test (#174)

- Updated vm_console_proxy_service_account_role_binding fixture to reference ClusterRole instead of Role
- Introduced a new helper function assert_resource_existence_and_availability to check resource availability status in tests.

Signed-off-by: Geetika Kapoor <gkapoor@redhat.com>

tests/infrastructure/vm_console_proxy/conftest.py

@@ -5,7 +5,6 @@
 import pytest
 from ocp_resources.cluster_role import ClusterRole
 from ocp_resources.resource import Resource
-from ocp_resources.role import Role
 from ocp_resources.role_binding import RoleBinding
 from ocp_resources.service_account import ServiceAccount
 from ocp_resources.ssp import SSP
@@ -24,6 +23,7 @@
     TOKEN_API_VERSION,
     TOKEN_ENDPOINT,
     VM_CONSOLE_PROXY,
+    VM_CONSOLE_PROXY_CLUSTER_ROLE,
     VM_CONSOLE_PROXY_USER,
 )
 from tests.infrastructure.vm_console_proxy.utils import (
@@ -103,22 +103,6 @@ def vm_for_console_proxy(namespace, unprivileged_client):
         yield vm
 
 
-@pytest.fixture(scope="class")
-def vm_console_proxy_role(namespace):
-    with Role(
-        name="vnc-token-access",
-        namespace=namespace.name,
-        rules=[
-            {
-                "apiGroups": [TOKEN_ENDPOINT],
-                "resources": ["virtualmachines/vnc"],
-                "verbs": ["get"],
-            }
-        ],
-    ) as vm_console_proxy_role:
-        yield vm_console_proxy_role
-
-
 @pytest.fixture(scope="class")
 def vm_console_proxy_service_account(namespace):
     with ServiceAccount(name=f"{VM_CONSOLE_PROXY_USER}1", namespace=namespace.name) as sa:
@@ -140,15 +124,24 @@ def vm_service_account_role_binding(namespace, vm_console_proxy_service_account)
 
 
 @pytest.fixture(scope="class")
-def vm_console_proxy_service_account_role_binding(namespace, vm_console_proxy_service_account, vm_console_proxy_role):
+def vm_console_proxy_cluster_role_exists():
+    assert ClusterRole(name=VM_CONSOLE_PROXY_CLUSTER_ROLE).exists, (
+        f"ClusterRole {VM_CONSOLE_PROXY_CLUSTER_ROLE} not found"
+    )
+
+
+@pytest.fixture(scope="class")
+def vm_console_proxy_service_account_role_binding(
+    namespace, vm_console_proxy_cluster_role_exists, vm_console_proxy_service_account
+):
     with RoleBinding(
         name=f"{VM_CONSOLE_PROXY_USER}-token-access",
         namespace=namespace.name,
         subjects_kind=vm_console_proxy_service_account.kind,
         subjects_name=vm_console_proxy_service_account.name,
         subjects_namespace=namespace.name,
-        role_ref_kind=Role.kind,
-        role_ref_name=vm_console_proxy_role.name,
+        role_ref_kind=ClusterRole.kind,
+        role_ref_name=VM_CONSOLE_PROXY_CLUSTER_ROLE,
     ) as vm_role_service_account_role_binding:
         yield vm_role_service_account_role_binding
 

tests/infrastructure/vm_console_proxy/constants.py

@@ -6,3 +6,4 @@
 TOKEN_ENDPOINT = "token.kubevirt.io"
 VM_CONSOLE_PROXY = "vm-console-proxy"
 VM_CONSOLE_PROXY_USER = f"{VM_CONSOLE_PROXY}-user"
+VM_CONSOLE_PROXY_CLUSTER_ROLE = f"{TOKEN_ENDPOINT}:generate"

tests/infrastructure/vm_console_proxy/test_vm_console_proxy.py

@@ -2,6 +2,8 @@
 
 import pytest
 
+from tests.infrastructure.vm_console_proxy.utils import assert_resource_existence_and_availability
+from utilities.constants import TIMEOUT_1MIN
 from utilities.vnc_utils import VNCConnection
 
 LOGGER = logging.getLogger(__name__)
@@ -12,17 +14,12 @@ class TestVmConsoleProxyEnablement:
     @pytest.mark.dependency(name="test_vm_proxy_cluster_resources_available")
     @pytest.mark.polarion("CNV-10416")
     def test_vm_proxy_cluster_resources_available(self, vm_console_proxy_cluster_resource):
-        assert vm_console_proxy_cluster_resource.exists, (
-            f"Missing : {vm_console_proxy_cluster_resource.kind}/{vm_console_proxy_cluster_resource.name}"
-        )
+        assert_resource_existence_and_availability(resource=vm_console_proxy_cluster_resource, timeout=TIMEOUT_1MIN)
 
     @pytest.mark.dependency(name="test_vm_proxy_namespaced_resources_available")
     @pytest.mark.polarion("CNV-10409")
     def test_vm_proxy_namespaced_resources_available(self, vm_console_proxy_namespace_resource):
-        assert vm_console_proxy_namespace_resource.exists, (
-            f"Missing : {vm_console_proxy_namespace_resource.kind}/{vm_console_proxy_namespace_resource.name} under "
-            f"{vm_console_proxy_namespace_resource.namespace}"
-        )
+        assert_resource_existence_and_availability(resource=vm_console_proxy_namespace_resource, timeout=TIMEOUT_1MIN)
 
     @pytest.mark.dependency(
         depends=[
@@ -35,7 +32,6 @@ def test_vm_proxy_namespaced_resources_available(self, vm_console_proxy_namespac
     def test_vm_console_proxy_token_access(
         self,
         vm_for_console_proxy,
-        vm_console_proxy_role,
         vm_console_proxy_service_account,
         vm_service_account_role_binding,
         vm_console_proxy_service_account_role_binding,

tests/infrastructure/vm_console_proxy/utils.py

@@ -1,7 +1,11 @@
+from __future__ import annotations
+
 import logging
+from typing import Any, Type
 
 import requests
 from ocp_resources.api_service import APIService
+from ocp_resources.resource import Resource
 
 from tests.infrastructure.vm_console_proxy.constants import (
     TOKEN_API_VERSION,
@@ -13,20 +17,20 @@
 
 
 def create_vnc_console_token(
-    url,
-    endpoint,
-    api_version,
-    namespace,
-    virtual_machine,
-    duration,
-    runtime_headers=None,
-):
+    url: str,
+    endpoint: str,
+    api_version: str,
+    namespace: str,
+    virtual_machine: str,
+    duration: int,
+    runtime_headers: dict[str, str] | None,
+) -> str:
     """
     Requests a VNC console token for a virtual machine
 
     Args:
         url (str): The base URL of API server.
-        endpoint (str): The API endpoint for  VM console proxy.
+        endpoint (str): The API endpoint for VM console proxy.
         api_version (str): The API version for the VM console proxy.
         namespace (str): The namespace of the virtual machine.
         virtual_machine (str): The name of the virtual machine.
@@ -55,7 +59,7 @@ def create_vnc_console_token(
         raise
 
 
-def get_vm_console_proxy_resource(resource_kind, namespace=None):
+def get_vm_console_proxy_resource(resource_kind: Type, namespace: str | None = None) -> Type:
     if namespace:
         vm_console_proxy_resource_object = resource_kind(
             name=VM_CONSOLE_PROXY,
@@ -66,3 +70,18 @@ def get_vm_console_proxy_resource(resource_kind, namespace=None):
             name=f"{TOKEN_API_VERSION}.{TOKEN_ENDPOINT}" if resource_kind == APIService else VM_CONSOLE_PROXY
         )
     return vm_console_proxy_resource_object
+
+
+def assert_resource_existence_and_availability(resource: Any, timeout: int) -> None:
+    if resource.kind in {"APIService", "Deployment"}:
+        LOGGER.info(f"Wait for availability/condition check for {resource.kind}.")
+        resource.wait_for_condition(
+            condition=Resource.Condition.AVAILABLE,
+            status=Resource.Condition.Status.TRUE,
+            timeout=timeout,
+        )
+        return
+
+    assert resource.exists, (
+        f"Missing: {resource.kind}/{resource.name} under {getattr(resource, 'namespace', 'cluster')}"
+    )