WIP disks: add filtering using --disks=host:device

FIXME: settle on the CLI syntax

Author: ydirson

conftest.py

@@ -8,12 +8,12 @@
 import tempfile
 
 from packaging import version
-from typing import Dict, Generator
+from typing import Dict, Generator, Iterable
 
 import lib.config as global_config
 
 from lib import pxe
-from lib.common import DiskDevName
+from lib.common import DiskDevName, HostAddress
 from lib.common import callable_marker, shortened_nodeid, prefix_object_name
 from lib.common import wait_for, vm_image, is_uuid
 from lib.common import setup_formatted_and_mounted_disk, teardown_formatted_and_mounted_disk
@@ -75,6 +75,14 @@ def pytest_addoption(parser):
         default=20,
         help="Max lines to output in a ssh log (0 if no limit)"
     )
+    parser.addoption(
+        "--disk",
+        action="append",
+        default=[],
+        # FIXME: do we want explicit --disk=* to authorize all?
+        help="HOST:DISK to authorize for use by tests. "
+             "No mention of a given host authorizes use of all its disks."
+    )
     parser.addoption(
         "--sr-disk",
         action="append",
@@ -219,6 +227,13 @@ def cleanup_hosts():
 
     cleanup_hosts()
 
+@pytest.fixture(scope='session')
+def pools_hosts_by_name_or_ip(hosts: list[Host]) -> Generator[dict[HostAddress, Host], None, None]:
+    yield {host.hostname_or_ip: host
+           for pool_master in hosts
+           for host in pool_master.pool.hosts
+           }
+
 @pytest.fixture(scope='session')
 def registered_xo_cli():
     # The fixture is not responsible for establishing the connection.
@@ -331,9 +346,29 @@ def local_sr_on_hostB1(hostB1):
     yield sr
 
 @pytest.fixture(scope='session')
-def disks(hosts: list[Host]) -> Generator[dict[Host, list[DiskDevName]], None, None]:
-    ret = {host: host.disks()
-           for pool_master in hosts
+def disks(pytestconfig, pools_hosts_by_name_or_ip: dict[HostAddress, Host]
+          ) -> Generator[dict[Host, list[DiskDevName]], None, None]:
+    def _parse_disk_option(option_text: str) -> tuple[HostAddress, DiskDevName]:
+        parsed = tuple(option_text.split(sep=":", maxsplit=1))
+        assert len(parsed) == 2, f"--disk option {option_text!r} is not <host>:<disk>"
+        return parsed
+    def _cli_disks_per_host(options_list: Iterable[tuple[HostAddress, DiskDevName]]
+                            ) -> dict[Host, list[DiskDevName]]:
+        ret: dict[Host, list[DiskDevName]] = {host: []
+                                              for pool_master in pools_hosts_by_name_or_ip.values()
+                                              for host in pool_master.pool.hosts
+                                              }
+        for (hostname, disk) in options_list:
+            host = pools_hosts_by_name_or_ip[hostname]
+            assert host in ret
+            ret[host].append(disk)
+        return ret
+
+    cli_disks = _cli_disks_per_host(_parse_disk_option(option_text)
+                                    for option_text in pytestconfig.getoption("disk"))
+    ret = {host: [disk for disk in host.disks()
+                  if not cli_disks[host] or disk in cli_disks[host]]
+           for pool_master in pools_hosts_by_name_or_ip.values()
            for host in pool_master.pool.hosts
            }
     logging.debug("disks collected: %s", {host.hostname_or_ip: value for host, value in ret.items()})