Merge pull request #5565 from xenserver-next/test-and-fix-usb_reset-m…

…ount

CA-390883: Move usb_reset.py to python3, merge import_file.py

python3/Makefile

@@ -13,6 +13,7 @@ install:
 
 	$(IDATA) packages/observer.py $(DESTDIR)$(SITE3_DIR)/
 
+	$(IPROG) libexec/usb_reset.py $(DESTDIR)$(LIBEXECDIR)
 	$(IPROG) libexec/usb_scan.py $(DESTDIR)$(LIBEXECDIR)
 	$(IPROG) libexec/nbd_client_manager.py $(DESTDIR)$(LIBEXECDIR)
 

python3/tests/conftest.py

@@ -0,0 +1,10 @@
+"""scripts/unit_test/conftest.py: Common pytest module for shared pytest fixtures"""
+import pytest
+
+from .rootless_container import enter_private_mount_namespace
+
+
+@pytest.fixture(scope="session")
+def private_mount_namespace():
+    """Enter a private mount namespace that allows us to test mount and unmount"""
+    return enter_private_mount_namespace()

python3/tests/import_helper.py

@@ -0,0 +1,70 @@
+"""helpers for unit-testing functions in scripts without permanent global mocks"""
+import os
+import sys
+from contextlib import contextmanager
+from types import ModuleType
+
+from typing import Generator
+from mock import Mock
+
+
+@contextmanager
+def mocked_modules(*module_names):  # type:(str) -> Generator[None, None, None]
+    """Context manager that temporarily mocks the specified modules.
+
+    :param module_names: Variable number of names of the modules to be mocked.
+    :yields: None
+
+    During the context, the specified modules are added to the sys.modules
+    dictionary as instances of the ModuleType class.
+    This effectively mocks the modules, allowing them to be imported and used
+    within the context. After the context, the mocked modules are removed
+    from the sys.modules dictionary.
+
+    Example usage:
+    ```python
+    with mocked_modules("module1", "module2"):
+        # Code that uses the mocked modules
+    ```
+    """
+    for module_name in module_names:
+        sys.modules[module_name] = Mock()
+    yield
+    for module_name in module_names:
+        sys.modules.pop(module_name)
+
+
+def import_file_as_module(relative_script_path):  # type:(str) -> ModuleType
+    """Import a Python script without the .py extension as a python module.
+
+    :param relative_script_path (str): The relative path of the script to import.
+    :returns module: The imported module.
+    :raises: AssertionError: If the spec or loader is not available.
+
+    Note:
+    - This function uses different methods depending on the Python version.
+    - For Python 2, it uses the imp module.
+    - For Python 3, it uses the importlib module.
+
+    Example:
+    - import_script_as_module('scripts/mail-alarm')  # Returns the imported module.
+    """
+    script_path = os.path.dirname(__file__) + "/../../" + relative_script_path
+    module_name = os.path.basename(script_path.replace(".py", ""))
+
+    # For Python 3.11+: Import Python script without the .py extension:
+    # https://gist.github.com/bernhardkaindl/1aaa04ea925fdc36c40d031491957fd3:
+    # pylint: disable-next=import-outside-toplevel
+    from importlib import (  # pylint: disable=no-name-in-module
+        machinery,
+        util,
+    )
+
+    loader = machinery.SourceFileLoader(module_name, script_path)
+    spec = util.spec_from_loader(module_name, loader)
+    assert spec
+    assert spec.loader
+    module = util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)
+    return module

python3/tests/rootless_container.py

@@ -0,0 +1,83 @@
+"""rootless_container.py: Create a rootless container on any Linux and GitHub CI"""
+import ctypes
+import os
+
+# Unshare the user namespace, so that the calling process is moved into a new
+# user namespace which is not shared with any previously existing process.
+# Needed so that the current user id can be mapped to 0 for getting a new
+# mount namespace.
+CLONE_NEWUSER = 0x10000000
+# Unshare the mount namespace, so that the calling process has a private copy
+# of its root directory namespace which is not shared with any other process:
+CLONE_NEWNS = 0x00020000
+# Flags for mount(2):
+MS_BIND = 4096
+MS_REC = 16384
+MS_PRIVATE = 1 << 18
+
+
+def unshare(flags):  # type:(int) -> None
+    """Wrapper for the library call to unshare Linux kernel namespaces"""
+    lib = ctypes.CDLL(None, use_errno=True)
+    lib.unshare.argtypes = [ctypes.c_int]
+    rc = lib.unshare(flags)
+    if rc != 0:  # pragma: no cover
+        errno = ctypes.get_errno()
+        raise OSError(errno, os.strerror(errno), flags)
+
+
+def mount(source="none", target="", fs="", flags=0, options=""):
+    # type:(str, str, str, int, str) -> None
+    """Wrapper for the library call mount(). Supports Python2.7 and Python3.x"""
+    lib = ctypes.CDLL(None, use_errno=True)
+    lib.mount.argtypes = (
+        ctypes.c_char_p,
+        ctypes.c_char_p,
+        ctypes.c_char_p,
+        ctypes.c_ulong,
+        ctypes.c_char_p,
+    )
+    result = lib.mount(
+        source.encode(), target.encode(), fs.encode(), flags, options.encode()
+    )
+    if result < 0:  # pragma: no cover
+        errno = ctypes.get_errno()
+        raise OSError(
+            errno,
+            "mount " + target + " (" + options + "): " + os.strerror(errno),
+        )
+
+
+def umount(target):  # type:(str) -> None
+    """Wrapper for the Linux umount system call, supports Python2.7 and Python3.x"""
+    lib = ctypes.CDLL(None, use_errno=True)
+    result = lib.umount(ctypes.c_char_p(target.encode()))
+    if result < 0:  # pragma: no cover
+        errno = ctypes.get_errno()
+        raise OSError(errno, "umount " + target + ": " + os.strerror(errno))
+
+
+def enter_private_mount_namespace():
+    """Enter a private mount and user namespace with the user and simulate uid 0
+
+    Some code like mount() requires to be run as root. The container simulates
+    root-like privileges and a new mount namespace that allows mount() in it.
+
+    Implements the equivalent of `/usr/bin/unshare --map-root-user --mount`
+    """
+
+    # Read the actual user and group ids before entering the new user namespace:
+    real_uid = os.getuid()
+    real_gid = os.getgid()
+    unshare(CLONE_NEWUSER | CLONE_NEWNS)
+    # Setup user map to map the user id to behave like uid 0:
+    with open("/proc/self/uid_map", "wb") as proc_self_user_map:
+        proc_self_user_map.write(b"0 %d 1" % real_uid)
+    with open("/proc/self/setgroups", "wb") as proc_self_set_groups:
+        proc_self_set_groups.write(b"deny")
+    # Setup group map for the user's gid to behave like gid 0:
+    with open("/proc/self/gid_map", "wb") as proc_self_group_map:
+        proc_self_group_map.write(b"0 %d 1" % real_gid)
+    # Private root mount in the mount namespace top support mounting a private tmpfs:
+    mount(target="/", flags=MS_REC | MS_PRIVATE)
+    return True

python3/tests/test_observer.py

@@ -10,7 +10,7 @@
 with patch("os.listdir") as mock_listdir:
     # Prevent it finding an observer.conf
     mock_listdir.return_value = []
-    from packages import observer
+    from python3.packages import observer
 
 # mock modules to avoid dependencies
 sys.modules["opentelemetry"] = MagicMock()
@@ -29,7 +29,7 @@
     OTEL_RESOURCE_ATTRIBUTES='service.name=sm'
     """
 TEST_OBSERVER_CONF = "test-observer.conf"
-OBSERVER_OPEN = "packages.observer.open"
+OBSERVER_OPEN = "python3.packages.observer.open"
 
 
 # pylint: disable=missing-function-docstring,protected-access

python3/tests/test_usb_reset_mount.py

@@ -0,0 +1,14 @@
+"""scripts/unit_test/test_usb_reset_mount.py: Test usb_reset.mount and .umount"""
+from __future__ import print_function
+
+from python3.tests.import_helper import import_file_as_module, mocked_modules
+
+
+def test_usb_reset_mount_umount(private_mount_namespace):
+    """Test usb_reset.mount and .umount"""
+    assert private_mount_namespace
+    with mocked_modules("xcp", "xcp.logger"):
+        usb_reset = import_file_as_module("python3/libexec/usb_reset.py")
+        usb_reset.log.error = print
+        usb_reset.mount(source="tmpfs", target="/tmp", fs="tmpfs")
+        usb_reset.umount("/tmp")

python3/unittest/test_hfx_filename.py

@@ -7,12 +7,12 @@
 import sys
 import unittest
 from mock import MagicMock, patch, call
-from import_file import get_module
+from python3.tests.import_helper import import_file_as_module
 
 # mock modules to avoid dependencies
 sys.modules["XenAPI"] = MagicMock()
 
-hfx_filename = get_module("hfx_filename", "../bin/hfx_filename")
+hfx_filename = import_file_as_module("python3/bin/hfx_filename")
 
 
 @patch("socket.socket")
@@ -82,7 +82,7 @@ def test_rpc_international_character(self, mock_socket):
 
     def test_db_get_uuid(self, mock_socket):
         """
-        Tests db_get_uuid 
+        Tests db_get_uuid
         """
         mock_connected_socket = MagicMock()
         mock_socket.return_value = mock_connected_socket
@@ -100,7 +100,7 @@ def test_db_get_uuid(self, mock_socket):
 
     def test_read_field(self, mock_socket):
         """
-        Tests read_field 
+        Tests read_field
         """
         mock_connected_socket = MagicMock()
         mock_socket.return_value = mock_connected_socket

python3/unittest/test_nbd_client_manager.py

@@ -3,13 +3,12 @@
 This module provides unittest for nbd_client_manager.py
 """
 
-import sys
 import unittest
 import subprocess
 from mock import MagicMock, patch, mock_open, call
-from import_file import get_module
+from python3.tests.import_helper import import_file_as_module
 
-nbd_client_manager = get_module("nbd_client_manager", "../libexec/nbd_client_manager.py")
+nbd_client_manager = import_file_as_module("python3/libexec/nbd_client_manager.py")
 
 @patch('subprocess.Popen')
 class TestCallFunction(unittest.TestCase):

python3/unittest/test_perfmon.py

@@ -7,12 +7,12 @@
 import math
 import unittest
 from mock import MagicMock, patch, mock_open
-from import_file import get_module
+from python3.tests.import_helper import import_file_as_module
 
 # mock modules to avoid dependencies
 sys.modules["XenAPI"] = MagicMock()
 
-perfmon = get_module("perfmon", "../bin/perfmon")
+perfmon = import_file_as_module("python3/bin/perfmon")
 
 
 @patch("subprocess.getoutput")
@@ -292,13 +292,13 @@ def test_process_rrd_updates(self, mock_get_percent_mem_usage,
         obj_report = perfmon.ObjectReport("vm", uuid)
         obj_report.vars = {
             'cpu0': [0.0063071, 0.0048038, 0.0045862, 0.0048865, 0.0048923],
-            'cpu1': [0.0067629, 0.0055811, 0.0058988, 0.0058809, 0.0053645], 
-            'cpu2': [0.0088599, 0.0078701, 0.0058573, 0.0063993, 0.0056833], 
-            'cpu3': [0.0085826, 0.0056874, 0.005697, 0.0061155, 0.0048769], 
-            'cpu4': [0.0051265, 0.0045452, 0.0046137, 0.0066399, 0.0050993], 
-            'cpu5': [0.0062369, 0.0053982, 0.0056624, 0.00606, 0.0062017], 
-            'cpu6': [0.006235, 0.0041764, 0.0048101, 0.0053798, 0.0050934], 
-            'cpu7': [0.0050709, 0.005482, 0.0058926, 0.0052934, 0.0049544], 
+            'cpu1': [0.0067629, 0.0055811, 0.0058988, 0.0058809, 0.0053645],
+            'cpu2': [0.0088599, 0.0078701, 0.0058573, 0.0063993, 0.0056833],
+            'cpu3': [0.0085826, 0.0056874, 0.005697, 0.0061155, 0.0048769],
+            'cpu4': [0.0051265, 0.0045452, 0.0046137, 0.0066399, 0.0050993],
+            'cpu5': [0.0062369, 0.0053982, 0.0056624, 0.00606, 0.0062017],
+            'cpu6': [0.006235, 0.0041764, 0.0048101, 0.0053798, 0.0050934],
+            'cpu7': [0.0050709, 0.005482, 0.0058926, 0.0052934, 0.0049544],
             'memory': [2785000000.0, 2785000000.0, 2785000000.0,
                        2785000000.0, 2785000000.0]
         }
@@ -332,13 +332,13 @@ def test_process_rrd_updates(self, mock_xapisession):
         obj_report = perfmon.ObjectReport("vm", uuid)
         obj_report.vars = {
             'cpu0': [0.0063071, 0.0048038, 0.0045862, 0.0048865, 0.0048923],
-            'cpu1': [0.0067629, 0.0055811, 0.0058988, 0.0058809, 0.0053645], 
-            'cpu2': [0.0088599, 0.0078701, 0.0058573, 0.0063993, 0.0056833], 
-            'cpu3': [0.0085826, 0.0056874, 0.005697, 0.0061155, 0.0048769], 
-            'cpu4': [0.0051265, 0.0045452, 0.0046137, 0.0066399, 0.0050993], 
-            'cpu5': [0.0062369, 0.0053982, 0.0056624, 0.00606, 0.0062017], 
-            'cpu6': [0.006235, 0.0041764, 0.0048101, 0.0053798, 0.0050934], 
-            'cpu7': [0.0050709, 0.005482, 0.0058926, 0.0052934, 0.0049544], 
+            'cpu1': [0.0067629, 0.0055811, 0.0058988, 0.0058809, 0.0053645],
+            'cpu2': [0.0088599, 0.0078701, 0.0058573, 0.0063993, 0.0056833],
+            'cpu3': [0.0085826, 0.0056874, 0.005697, 0.0061155, 0.0048769],
+            'cpu4': [0.0051265, 0.0045452, 0.0046137, 0.0066399, 0.0050993],
+            'cpu5': [0.0062369, 0.0053982, 0.0056624, 0.00606, 0.0062017],
+            'cpu6': [0.006235, 0.0041764, 0.0048101, 0.0053798, 0.0050934],
+            'cpu7': [0.0050709, 0.005482, 0.0058926, 0.0052934, 0.0049544],
             'memory': [2785000000.0, 2785000000.0, 2785000000.0,
                        2785000000.0, 2785000000.0]
         }
@@ -415,7 +415,7 @@ def test_process_rrd_updates(self, mock_xapisession):
         obj_report = perfmon.ObjectReport("vm", uuid)
         obj_report.vars = {
             'size': [100, 200, 300, 400, 500],
-            'physical_utilisation': [2000, 3000, 4000, 5000, 6000], 
+            'physical_utilisation': [2000, 3000, 4000, 5000, 6000],
         }
         rrd_updates.report.obj_reports[uuid] = obj_report
         rrd_updates.report.rows = 5

python3/unittest/test_usb_scan.py

@@ -8,9 +8,11 @@
 import tempfile
 import unittest
 from collections.abc import Mapping
+from typing import cast
 
 import mock
-from import_file import get_module
+
+from python3.tests.import_helper import import_file_as_module
 
 sys.modules["xcp"] = mock.Mock()
 sys.modules["xcp.logger"] = mock.Mock()
@@ -107,9 +109,10 @@ def verify_usb_common(
         self, moc_devices,
         moc_interfaces,
         moc_results,
-        path="./scripts/usb-policy.conf"
+        # Use relative path to allow tests to be started in subdirectories
+        path = os.path.dirname(__file__) + "/../../scripts/usb-policy.conf"
     ):
-        usb_scan = get_module("usb_scan", "../libexec/usb_scan.py")
+        usb_scan = import_file_as_module("python3/libexec/usb_scan.py")
 
         mock_setup(usb_scan, moc_devices, moc_interfaces, path)
 
@@ -134,7 +137,7 @@ def verify_usb_exit(
             # cm.exception.code is int type whose format
             # looks like "duplicated tag'vid' found,
             # malformed line ALLOW:vid=056a vid=0314 class=03"
-            self.assertIn(msg, cm.exception.code)  # pytype: disable=wrong-arg-types
+            self.assertIn(msg, cast(str, cm.exception.code))  # code is a str
 
     def test_usb_dongle(self):
         devices = [

scripts/Makefile

@@ -108,7 +108,6 @@ install:
 	$(IPROG) pam.d-xapi $(DESTDIR)/etc/pam.d/xapi
 	$(IPROG) upload-wrapper logs-download $(DESTDIR)$(LIBEXECDIR)
 	$(IDATA) usb-policy.conf $(DESTDIR)$(ETCXENDIR)
-	$(IPROG) usb_reset.py $(DESTDIR)$(LIBEXECDIR)
 	mkdir -p $(DESTDIR)$(OPTDIR)/packages/iso #omg XXX
 	$(IPROG) xapi-rolling-upgrade-miami $(DESTDIR)$(LIBEXECDIR)/xapi-rolling-upgrade
 	$(IPROG) set-hostname $(DESTDIR)$(LIBEXECDIR)