quicktest: split tests into SR specific and common ones

Currently we are calling quicktest for all SRs. The problem is that it
takes long time to run and some tests are not related to the type of the
SR. This patch creates one new test that runs the common suites one,
and we only run SR specific tests per SR.

A new option has been added to Quicktest (-list-tags/-with-tags) that
allows to select tests. _quicktest_scoping() detects which the target
host supports and picks accordingly.

Signed-off-by: Guillaume Thouvenin <guillaume.thouvenin@vates.tech>

Author: gthvn1

lib/sr.py

@@ -1,6 +1,12 @@
 from __future__ import annotations
 
+import pytest
+
+import enum
+import functools
 import logging
+import re
+import shlex
 import time
 
 import lib.commands as commands
@@ -26,6 +32,36 @@
     from lib.host import Host
     from lib.pool import Pool
 
+QUICKTEST_BIN = "/opt/xensource/debug/quicktest"
+
+QUICKTEST_SR_SUITES = (
+    "cbt,copy,SR tests,Quicktest_vdi,Quicktest_async_calls,"
+    "Quicktest_vm_import_export,Quicktest_vm_lifecycle,Quicktest_vm_snapshot,"
+    "Quicktest_vdi_ops_data_integrity,Quicktest_max_vdi_size,Quicktest_static_vdis"
+)
+
+QUICKTEST_COMMON_SUITES = (
+    "Quicktest_example,Quicktest_message,xenstore,event,import_raw_vdi,"
+    "Quicktest_date,Quicktest_crypt_r,http,unixext,Timer"
+)
+
+class QuicktestScoping(enum.Enum):
+    WITH_TAG_PARAM = enum.auto()
+    RUN_ONLY_PARAM = enum.auto()
+    NO_PARAM = enum.auto()
+
+@functools.lru_cache(maxsize=None)
+def _quicktest_scoping(hostname_or_ip: str) -> QuicktestScoping:
+    tags_output = commands.ssh_with_result(hostname_or_ip, f"{QUICKTEST_BIN} -list-tags")
+    if tags_output.returncode == 0 and re.search(r"^sr:", tags_output.stdout, re.MULTILINE):
+        return QuicktestScoping.WITH_TAG_PARAM
+
+    help_output = commands.ssh(hostname_or_ip, f"{QUICKTEST_BIN} -help", check=False)
+    if "-run-only" in help_output:
+        return QuicktestScoping.RUN_ONLY_PARAM
+
+    return QuicktestScoping.NO_PARAM
+
 class SR:
     xe_prefix = 'sr'
 
@@ -230,14 +266,29 @@ def create_vdi(
         vdi_uuid = self.pool.master.xe('vdi-create', args)
         return VDI(vdi_uuid, sr=self)
 
-    def run_quicktest(self) -> None:
-        logging.info(f"Run quicktest on SR {self.uuid}")
+    def run_quicktest(self, sr_specific: bool = True) -> None:
+        scoping = _quicktest_scoping(self.pool.master.hostname_or_ip)
+        cmd = f"{QUICKTEST_BIN} -sr {self.uuid}"
+
+        if scoping is QuicktestScoping.WITH_TAG_PARAM:
+            cmd += " -with-tag sr" if sr_specific else " -without-tag sr"
+        elif scoping is QuicktestScoping.RUN_ONLY_PARAM:
+            suites = QUICKTEST_SR_SUITES if sr_specific else QUICKTEST_COMMON_SUITES
+            cmd += f" -run-only {shlex.quote(suites)}"
+        elif not sr_specific:
+            # QuicktestScoping.NO_PARAM: no way to select just the common suites, and every
+            # per-SR pass on this host is already unfiltered, so this run adds nothing.
+            pytest.skip("quicktest has no scoping support on this host; "
+                        "common suites are already covered by the per-SR runs.")
+
+        logging.info(f"Run quicktest on SR {self.uuid}: {cmd}")
+
         # Always display the output of quicktest, failed or not.
         # This will duplicate the output in some cases, but it ensures we always have it for failure analysis,
         # even when quicktest leaves SRs in a state which makes teardown fail (in this case, pytest often doesn't
         # manage to display the details of the failed command, for a reason unknown - no usable reproducer found)
         try:
-            output = self.pool.master.ssh(f'/opt/xensource/debug/quicktest -sr {self.uuid}')
+            output = self.pool.master.ssh(cmd)
             logging.info(f"Quicktest output: {output}")
         except commands.SSHCommandFailed as e:
             logging.error(f"Quicktest output: {e.stdout}")

tests/storage/test_quicktest_common.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+import pytest
+
+from lib.sr import SR
+
+@pytest.mark.quicktest
+def test_common_quicktest(local_sr_on_hostA1: SR) -> None:
+    local_sr_on_hostA1.run_quicktest(sr_specific=False)