refactor: move validate_metrics_value to utilities/monitoring

conftest.py

@@ -33,6 +33,7 @@
 from utilities.bitwarden import get_cnv_tests_secret_by_name
 from utilities.constants import (
     AMD_64,
+    MULTIARCH,
     QUARANTINED,
     SETUP_ERROR,
     TIMEOUT_5MIN,
@@ -558,6 +559,15 @@ def filter_sno_only_tests(items: list[Item], config: Config) -> list[Item]:
     return items
 
 
+def filter_multiarch_tests(items: list[Item], config: Config) -> list[Item]:
+    if py_config.get("cluster_type") == MULTIARCH:
+        return items
+    discard_tests, items_to_return = remove_tests_from_list(items=items, filter_str="multiarch")
+    if discard_tests:
+        config.hook.pytest_deselected(items=discard_tests)
+    return items_to_return
+
+
 def remove_tests_from_list(items: list[Item], filter_str: str) -> tuple[list[Item], list[Item]]:
     discard_tests: list[Item] = []
     items_to_return: list[Item] = []
@@ -635,6 +645,7 @@ def pytest_collection_modifyitems(session, config, items):
         config.hook.pytest_deselected(items=discard)
     items[:] = filter_deprecated_api_tests(items=items, config=config)
     items[:] = filter_sno_only_tests(items=items, config=config)
+    items[:] = filter_multiarch_tests(items=items, config=config)
     items[:] = mark_nmstate_dependent_tests(items=items)
 
 

tests/conftest.py

@@ -73,6 +73,7 @@
 from libs.net.ip import filter_link_local_addresses, random_ipv4_address, random_ipv6_address
 from libs.net.vmspec import lookup_iface_status
 from tests.utils import download_and_extract_tar
+from utilities.architecture import get_cluster_architecture
 from utilities.artifactory import get_artifactory_header, get_http_image_url, get_test_artifact_server_url
 from utilities.bitwarden import get_cnv_tests_secret_by_name
 from utilities.cluster import cache_admin_client, get_oc_whoami_username
@@ -446,6 +447,7 @@ def schedulable_nodes(nodes):
     """
     schedulable_label = "kubevirt.io/schedulable"
     cpu_arch = py_config.get("cpu_arch")
+    cpu_archs = [cpu_arch] if isinstance(cpu_arch, str) else cpu_arch
     schedulable = [
         node
         for node in nodes
@@ -454,7 +456,7 @@ def schedulable_nodes(nodes):
         and not node.instance.spec.unschedulable
         and not kubernetes_taint_exists(node)
         and node.kubelet_ready
-        and (not cpu_arch or node.labels.get(KUBERNETES_ARCH_LABEL) == cpu_arch)
+        and (not cpu_archs or node.labels.get(KUBERNETES_ARCH_LABEL) in cpu_archs)
     ]
 
     LOGGER.info(f"Schedulable nodes: {[node.name for node in schedulable]}, node architecture: {cpu_arch or 'all'}")
@@ -1026,7 +1028,7 @@ def mac_pool(admin_client, hco_namespace):
 
 @pytest.fixture(scope="session")
 def nodes_cpu_architecture():
-    return py_config["cpu_arch"]
+    return py_config.get("cpu_arch")
 
 
 @pytest.fixture(scope="session")
@@ -1478,7 +1480,7 @@ def cluster_info(
         f"\tOCS version: {ocs_current_version}\n"
         f"\tCNI type: {get_cluster_cni_type(admin_client=admin_client)}\n"
         f"\tWorkers type: {workers_type}\n"
-        f"\tCluster CPU Architecture: {nodes_cpu_architecture}\n"
+        f"\tCluster CPU Architecture: {nodes_cpu_architecture or ', '.join(sorted(get_cluster_architecture()))}\n"
         f"\tIPv4 cluster: {ipv4_supported_cluster()}\n"
         f"\tIPv6 cluster: {ipv6_supported_cluster()}\n"
         f"\tVirtctl version: \n\t{virtctl_client_version}\n\t{virtctl_server_version}\n"

tests/observability/metrics/conftest.py

@@ -36,7 +36,6 @@
     network_packets_received,
     vnic_info_from_vm_or_vmi,
 )
-from tests.observability.utils import validate_metrics_value
 from tests.utils import create_vms, start_stress_on_vm
 from utilities import console
 from utilities.constants import (
@@ -72,7 +71,7 @@
     get_pod_by_name_prefix,
     unique_name,
 )
-from utilities.monitoring import get_metrics_value
+from utilities.monitoring import get_metrics_value, validate_metrics_value
 from utilities.network import assert_ping_successful, get_ip_from_vm_or_virt_handler_pod, ping
 from utilities.ssp import verify_ssp_pod_is_running
 from utilities.storage import (

tests/observability/metrics/test_aaq_metrics.py

@@ -4,7 +4,7 @@
     timestamp_to_seconds,
     validate_values_from_kube_application_aware_resourcequota_metric,
 )
-from tests.observability.utils import validate_metrics_value
+from utilities.monitoring import validate_metrics_value
 
 pytestmark = [
     pytest.mark.usefixtures(

tests/observability/metrics/test_general_metrics.py

@@ -7,7 +7,7 @@
 from libs.net.cluster import is_ipv6_single_stack_cluster
 from tests.observability.metrics.constants import KUBEVIRT_VMI_NODE_CPU_AFFINITY
 from tests.observability.metrics.utils import validate_vmi_node_cpu_affinity_with_prometheus
-from tests.observability.utils import validate_metrics_value
+from utilities.monitoring import validate_metrics_value
 from utilities.virt import VirtualMachineForTests, fedora_vm_body, running_vm
 
 KUBEVIRT_VM_TAG = f"{Resource.ApiGroup.KUBEVIRT_IO}/vm"

tests/observability/metrics/test_metrics.py

@@ -9,10 +9,10 @@
     assert_vm_metric_virt_handler_pod,
     compare_kubevirt_vmi_info_metric_with_vm_info,
 )
-from tests.observability.utils import validate_metrics_value
 from utilities.constants import (
     KUBEVIRT_HCO_HYPERCONVERGED_CR_EXISTS,
 )
+from utilities.monitoring import validate_metrics_value
 
 pytestmark = [pytest.mark.post_upgrade, pytest.mark.sno]
 

tests/observability/metrics/test_migration_metrics.py

@@ -13,7 +13,7 @@
     timestamp_to_seconds,
     wait_for_non_empty_metrics_value,
 )
-from tests.observability.utils import validate_metrics_value
+from utilities.monitoring import validate_metrics_value
 
 LOGGER = logging.getLogger(__name__)
 

tests/observability/metrics/test_ssp_metrics.py

@@ -7,12 +7,12 @@
     validate_metric_value_with_round_down,
     validate_metric_value_within_range,
 )
-from tests.observability.utils import validate_metrics_value
 from utilities.constants import (
     SSP_OPERATOR,
     VIRT_TEMPLATE_VALIDATOR,
 )
 from utilities.hco import ResourceEditorValidateHCOReconcile
+from utilities.monitoring import validate_metrics_value
 from utilities.virt import VirtualMachineForTests
 
 KUBEVIRT_SSP_OPERATOR_UP = "kubevirt_ssp_operator_up"

tests/observability/metrics/test_vms_metrics.py

@@ -27,7 +27,6 @@
     validate_metric_value_greater_than_initial_value,
     validate_vnic_info,
 )
-from tests.observability.utils import validate_metrics_value
 from utilities.constants import (
     CAPACITY,
     MIGRATION_POLICY_VM_LABEL,
@@ -38,7 +37,7 @@
     USED,
 )
 from utilities.infra import get_node_selector_dict
-from utilities.monitoring import get_metrics_value
+from utilities.monitoring import get_metrics_value, validate_metrics_value
 from utilities.virt import VirtualMachineForTests, fedora_vm_body, running_vm
 
 LOGGER = logging.getLogger(__name__)

tests/observability/upgrade/test_upgrade_observability.py

@@ -1,9 +1,9 @@
 import pytest
 
 from tests.observability.constants import KUBEVIRT_VMI_NUMBER_OF_OUTDATED
-from tests.observability.utils import validate_metrics_value
 from tests.upgrade_params import IUO_UPGRADE_TEST_DEPENDENCY_NODE_ID
 from utilities.constants import DEPENDENCY_SCOPE_SESSION
+from utilities.monitoring import validate_metrics_value
 
 
 @pytest.mark.cnv_upgrade

tests/observability/utils.py

@@ -1,46 +1,13 @@
-import datetime
 import logging
 
 from ocp_utilities.monitoring import Prometheus
-from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 
 from tests.observability.constants import SSP_COMMON_TEMPLATES_MODIFICATION_REVERTED
-from utilities.constants import (
-    TIMEOUT_4MIN,
-    TIMEOUT_15SEC,
-)
-from utilities.monitoring import get_metrics_value
 
 LOGGER = logging.getLogger(__name__)
 ALLOW_ALERTS_ON_HEALTHY_CLUSTER_LIST = [SSP_COMMON_TEMPLATES_MODIFICATION_REVERTED]
 
 
-def validate_metrics_value(
-    prometheus: Prometheus, metric_name: str, expected_value: str, timeout: int = TIMEOUT_4MIN
-) -> None:
-    samples = TimeoutSampler(
-        wait_timeout=timeout,
-        sleep=TIMEOUT_15SEC,
-        func=get_metrics_value,
-        prometheus=prometheus,
-        metrics_name=metric_name,
-    )
-    sample = None
-    comparison_values_log = {}
-    try:
-        for sample in samples:
-            if sample:
-                comparison_values_log[datetime.datetime.now()] = (
-                    f"metric: {metric_name} value is: {sample}, the expected value is {expected_value}"
-                )
-                if sample == expected_value:
-                    LOGGER.info("Metrics value matches the expected value!")
-                    return
-    except TimeoutExpiredError:
-        LOGGER.error(f"Metrics value: {sample}, expected: {expected_value}, comparison log: {comparison_values_log}")
-        raise
-
-
 def verify_no_listed_alerts_on_cluster(prometheus: Prometheus, alerts_list: list[str]) -> None:
     """
     It gets a list of alerts and verifies that none of them are firing on a cluster.

utilities/monitoring.py

@@ -1,7 +1,12 @@
+import datetime
 import logging
+from typing import TYPE_CHECKING
 
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 
+if TYPE_CHECKING:
+    from ocp_utilities.monitoring import Prometheus
+
 from utilities.constants import (
     FIRING_STATE,
     KUBEVIRT_HYPERCONVERGED_OPERATOR_HEALTH_STATUS,
@@ -175,6 +180,43 @@ def get_metrics_value(prometheus, metrics_name):
     return 0
 
 
+def validate_metrics_value(
+    prometheus: "Prometheus", metric_name: str, expected_value: str | int, timeout: int = TIMEOUT_5MIN
+) -> None:
+    """Wait until the metric matches the expected value.
+
+    Args:
+        prometheus: Prometheus instance.
+        metric_name: Name of the metric to query.
+        expected_value: Expected value to match. Use str for emitted values (e.g. "0"),
+            int 0 for absent/not-emitted metrics (get_metrics_value returns int 0 when absent).
+        timeout: Maximum wait time in seconds.
+
+    Raises:
+        TimeoutExpiredError: If the metric does not match within timeout.
+    """
+    samples = TimeoutSampler(
+        wait_timeout=timeout,
+        sleep=TIMEOUT_5SEC,
+        func=get_metrics_value,
+        prometheus=prometheus,
+        metrics_name=metric_name,
+    )
+    sample = None
+    comparison_values_log = {}
+    try:
+        for sample in samples:
+            comparison_values_log[datetime.datetime.now()] = (
+                f"metric: {metric_name} value is: {sample}, the expected value is {expected_value}"
+            )
+            if sample == expected_value:
+                LOGGER.info(f"Metrics value matches the expected value: {expected_value}")
+                return
+    except TimeoutExpiredError:
+        LOGGER.error(f"Metrics value: {sample}, expected: {expected_value}, comparison log: {comparison_values_log}")
+        raise
+
+
 def wait_for_gauge_metrics_value(prometheus, query, expected_value, timeout=TIMEOUT_5MIN):
     samples = TimeoutSampler(
         wait_timeout=timeout,

utilities/unittests/test_monitoring.py

@@ -13,6 +13,7 @@
     get_metrics_value,
     validate_alert_cnv_labels,
     validate_alerts,
+    validate_metrics_value,
     wait_for_alert,
     wait_for_firing_alert_clean_up,
     wait_for_gauge_metrics_value,
@@ -351,6 +352,52 @@ def test_get_metrics_value_no_data(self):
         assert result == 0
 
 
+class TestValidateMetricsValue:
+    """Test cases for validate_metrics_value function"""
+
+    @patch("utilities.monitoring.TimeoutSampler")
+    def test_matches_emitted_string_value(self, mock_sampler_cls):
+        """Test matching when metric is emitted with a string value."""
+        mock_sampler_cls.return_value = iter(["0"])
+        validate_metrics_value(prometheus=MagicMock(), metric_name="test_metric", expected_value="0")
+
+    @patch("utilities.monitoring.TimeoutSampler")
+    def test_matches_absent_metric_with_int_zero(self, mock_sampler_cls):
+        """Test matching absent metric (int 0) with expected_value=0."""
+        mock_sampler_cls.return_value = iter([0])
+        validate_metrics_value(prometheus=MagicMock(), metric_name="test_metric", expected_value=0)
+
+    @patch("utilities.monitoring.TimeoutSampler")
+    def test_absent_metric_does_not_match_string_zero(self, mock_sampler_cls):
+        """Test that absent metric (int 0) does NOT match expected_value='0'."""
+
+        def raise_after_samples(*args, **kwargs):
+            def _iter():
+                yield 0
+                raise TimeoutExpiredError("Timeout")
+
+            return _iter()
+
+        mock_sampler_cls.side_effect = raise_after_samples
+        with pytest.raises(TimeoutExpiredError):
+            validate_metrics_value(prometheus=MagicMock(), metric_name="test_metric", expected_value="0")
+
+    @patch("utilities.monitoring.TimeoutSampler")
+    def test_timeout_when_value_does_not_match(self, mock_sampler_cls):
+        """Test timeout when metric value never matches expected."""
+
+        def raise_after_samples(*args, **kwargs):
+            def _iter():
+                yield "5"
+                raise TimeoutExpiredError("Timeout")
+
+            return _iter()
+
+        mock_sampler_cls.side_effect = raise_after_samples
+        with pytest.raises(TimeoutExpiredError):
+            validate_metrics_value(prometheus=MagicMock(), metric_name="test_metric", expected_value="0")
+
+
 class TestWaitForGaugeMetricsValue:
     """Test cases for wait_for_gauge_metrics_value function"""