diff --git a/pyproject.toml b/pyproject.toml index 5bf10170401..83a54c6d978 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,6 +57,7 @@ exclude_lines = [ "pass", # Other specific lines that do not need to be covered, comment in which file: "raise NbdDeviceNotFound", # python3/libexec/usb_scan.py + "params = xmlrpc.client.loads", # static-vdis ] # precision digits to use when reporting coverage (sub-percent-digits are not reported): precision = 0 @@ -261,7 +262,6 @@ expected_to_fail = [ "scripts/backup-sr-metadata.py", "scripts/restore-sr-metadata.py", # Other fixes needed: - "scripts/static-vdis", "scripts/plugins/extauth-hook-AD.py", ] diff --git a/python3/tests/test_static_vdis.py b/python3/tests/test_static_vdis.py new file mode 100644 index 00000000000..ee424c157a1 --- /dev/null +++ b/python3/tests/test_static_vdis.py @@ -0,0 +1,85 @@ +"""python3/tests/test_static_vdis.py: Test the static-vdis script""" + +import os +from pathlib import Path +from types import ModuleType + +import pytest + +from python3.tests.import_helper import import_file_as_module, mocked_modules + +# ---------------------------- Test fixtures --------------------------------- + + +@pytest.fixture(scope="function") # function scope: Re-run for each test function +def static_vdis() -> ModuleType: + """Test fixture to return the static-vdis module, mocked to avoid dependencies.""" + with mocked_modules("XenAPI", "inventory"): + return import_file_as_module("scripts/static-vdis") + + +# Hide pylint warnings for redefined-outer-name from using the static_vdis fixture: +# pylint: disable=redefined-outer-name +# Allow to access attributes of the static_vdis module from this test module: +# pyright: reportAttributeAccessIssue=false + +# ----------------------------- Test cases ----------------------------------- + + +def test_whole_file(static_vdis: ModuleType): + """Test read_whole_file() and write_whole_file()""" + + with open(__file__, encoding="utf-8") as data: + contents = data.read().strip() + assert static_vdis.read_whole_file(__file__) == contents + assert static_vdis.write_whole_file(__file__, contents) is None + with open(__file__, encoding="utf-8") as written_data: + assert written_data.read().strip() == contents + + +def test_fresh_name(static_vdis: ModuleType, tmp_path: Path): + """Test fresh_name() and list_vdis() - all code paths""" + + # When the freshly created tmp_path is empty, expect [] and "0": + static_vdis.main_dir = tmp_path.as_posix() + assert static_vdis.list_vdis() == [] + assert static_vdis.fresh_name() == "0" + + # When main_dir contains a directory with name "0", the next name should be "1": + os.mkdir(static_vdis.main_dir + "/0") + assert static_vdis.fresh_name() == "1" + + # When main_dir contains a directory with name "1", the next name should be "2": + os.mkdir(static_vdis.main_dir + "/1") + assert static_vdis.fresh_name() == "2" + + # When main_dir does not exist, an empty list and 0 should be returned: + static_vdis.main_dir = tmp_path.as_posix() + "/does-not-exist" + assert static_vdis.list_vdis() == [] + assert static_vdis.fresh_name() == "0" + + + +def test_sr_attach(static_vdis: ModuleType, mocker): + """Test sr_attach()""" + + # We need to mock those as they would attempt to load the volume plugin and + # check the clusterstack, which are not available in the test environment: + static_vdis.call_volume_plugin = mocker.MagicMock() + static_vdis.check_clusterstack = mocker.MagicMock() + + # Set the return value of the mocked functions to success: + static_vdis.call_volume_plugin.return_value = "success" + static_vdis.check_clusterstack.return_value = "success" + + # Call the sr_attach function + device_config = {"key1": "value1", "key2": "value2"} + result = static_vdis.sr_attach("plugin_name", device_config) + + # Assert the expected behavior + assert result == "success" + static_vdis.call_volume_plugin.assert_called_once_with( + "plugin_name", + "SR.attach", + ["--configuration", "key1", "value1", "--configuration", "key2", "value2"], + ) \ No newline at end of file diff --git a/scripts/static-vdis b/scripts/static-vdis index 77c9790b71e..9ca8b1d352a 100755 --- a/scripts/static-vdis +++ b/scripts/static-vdis @@ -3,10 +3,22 @@ # Common functions for managing statically-attached (ie onboot, without xapi) VDIs -import sys, os, subprocess, json, urllib.parse +import json +import os import os.path +import subprocess +import sys import time -import XenAPI, inventory, xmlrpc.client +import urllib.parse +import xmlrpc.client +from typing import TYPE_CHECKING + +import XenAPI + +import inventory + +if TYPE_CHECKING: + from typing import Any, Dict main_dir = "/etc/xensource/static-vdis" @@ -77,6 +89,7 @@ def check_clusterstack(ty): wait_for_corosync_quorum() def sr_attach(ty, device_config): + # type: (str, Dict[str, object]) -> str check_clusterstack(ty) args = [arg for (k,v) in device_config.items() @@ -238,7 +251,7 @@ def call_backend_attach(driver, config): return path def call_backend_detach(driver, config): - params = xmlrpc.client.loads(config)[0][0] + params = xmlrpc.client.loads(config)[0][0] # type: Any params['command'] = 'vdi_detach_from_config' config = xmlrpc.client.dumps(tuple([params]), params['command']) xml = doexec([ driver, config ]) @@ -388,4 +401,3 @@ if __name__ == "__main__": detach(sys.argv[2]) else: usage() - diff --git a/scripts/test_static_vdis.py b/scripts/test_static_vdis.py deleted file mode 100644 index b0ab6ad5939..00000000000 --- a/scripts/test_static_vdis.py +++ /dev/null @@ -1,56 +0,0 @@ -#!/usr/bin/env python3 -# -# unittest for static-vdis - -import unittest -from mock import MagicMock -import sys -import os -import subprocess -import tempfile - -# mock modules to avoid dependencies -sys.modules["XenAPI"] = MagicMock() -sys.modules["inventory"] = MagicMock() - -def import_from_file(module_name, file_path): - """Import a file as a module""" - if sys.version_info.major == 2: - return None - else: - from importlib import machinery, util - loader = machinery.SourceFileLoader(module_name, file_path) - spec = util.spec_from_loader(module_name, loader) - assert spec - assert spec.loader - module = util.module_from_spec(spec) - # Probably a good idea to add manually imported module stored in sys.modules - sys.modules[module_name] = module - spec.loader.exec_module(module) - return module - -def get_module(): - """Import the static-vdis script as a module for executing unit tests on functions""" - testdir = os.path.dirname(__file__) - return import_from_file("static_vdis", testdir + "/static-vdis") - -static_vdis = get_module() - -@unittest.skipIf(sys.version_info < (3, 0), reason="requires python3") -class TestReadWriteFile(unittest.TestCase): - def test_write_and_read_whole_file(self): - """Test read_whole_file and write_whole_file""" - test_file = tempfile.NamedTemporaryFile(delete=True) - filename = str(test_file.name) - content = r"""def read_whole_file(filename): - with open(filename, 'r', encoding='utf-8') as f: - return ''.join(f.readlines()).strip() - -def write_whole_file(filename, contents): - with open(filename, "w", encoding='utf-8') as f: - f.write(contents)""" - static_vdis.write_whole_file(filename, content) - expected_content = static_vdis.read_whole_file(filename) - self.assertEqual(expected_content, content) - - \ No newline at end of file