diff --git a/pyproject.toml b/pyproject.toml index 83a54c6d978..5dd6d1ee8e5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,8 @@ exclude_lines = [ # 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 + "assert.*# must not be None", # static-vdis + "except Exception:", # static-vdis ] # precision digits to use when reporting coverage (sub-percent-digits are not reported): precision = 0 diff --git a/python3/tests/test_static_vdis.py b/python3/tests/test_static_vdis.py index ee424c157a1..1b7efc0bcf0 100644 --- a/python3/tests/test_static_vdis.py +++ b/python3/tests/test_static_vdis.py @@ -1,6 +1,7 @@ """python3/tests/test_static_vdis.py: Test the static-vdis script""" import os +import sys from pathlib import Path from types import ModuleType @@ -26,17 +27,66 @@ def static_vdis() -> ModuleType: # ----------------------------- Test cases ----------------------------------- -def test_whole_file(static_vdis: ModuleType): +def test_whole_file(static_vdis: ModuleType, tmp_path): """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 static_vdis.write_whole_file(tmp_path / "temp_file", contents) is None + with open(tmp_path / "temp_file", encoding="utf-8") as written_data: assert written_data.read().strip() == contents +def test_attach(static_vdis: ModuleType, tmpdir, mocker, capsys): + """Test five common and SMAPIv1 code paths in the attach() function""" + + # Path 1: When the VDI is not found, expect attach() to raise an exception: + static_vdis.list_vdis = lambda: [{"vdi-uuid": "existing-uuid"}] + with pytest.raises(Exception) as exc_info: + static_vdis.attach("nonexisting-uuid") + assert exc_info.value.args[0] == "Disk configuration not found" + + # Path 2: When the VDI is already attached, expect main():attach to return None\n: + static_vdis.list_vdis = lambda: [{"vdi-uuid": "attached", "path": "/attached"}] + sys.argv = ["static-vdis", "attach", "attached"] + static_vdis.main() + with capsys.disabled(): + assert capsys.readouterr().out == "None\n" + + # Path 3: When the VDI is not attached, attach() to return "the-id/disk": + vdis: list[dict[str, str]] = [{"vdi-uuid": "attach-uuid", "id": "the-id"}] + static_vdis.list_vdis = lambda: vdis + static_vdis.call_backend_attach = lambda driver, config: "/mock-attached-path" + static_vdis.read_whole_file = lambda path: '{"json":true}' + disk = tmpdir.mkdir(vdis[0]["id"]).join("disk") + static_vdis.main_dir = str(tmpdir) + assert static_vdis.attach("attach-uuid") == disk + assert os.readlink(disk) == "/mock-attached-path" + os.unlink(disk) + + # Path 4: Create the disk file expect it to be deleted and replaced by a symlink: + disk.write("mock-disk-contents-to-delete") + assert static_vdis.attach("attach-uuid") == disk + assert os.readlink(disk) == "/mock-attached-path" + + # Path 5: When the backend call returns None, expect attach() to raise TypeError + static_vdis.call_backend_attach = lambda driver, config: None + with pytest.raises(TypeError) as exc_info: + static_vdis.attach("attach-uuid") + + # Path 6: When the backend returns an empty str, attach() raises FileNotFoundError: + static_vdis.call_backend_attach = lambda driver, config: "" + with pytest.raises(FileNotFoundError) as exc_info: + static_vdis.attach("attach-uuid") + + # Path 7: If the smapiv3_config exists, but not the volume plugin, attach() fails: + with pytest.raises(FileNotFoundError) as exc_info: + mocker.patch("os.path.exists", return_value=True) + static_vdis.MULTIPATH_FLAG = __file__ + static_vdis.attach("attach-uuid") + + def test_fresh_name(static_vdis: ModuleType, tmp_path: Path): """Test fresh_name() and list_vdis() - all code paths""" @@ -59,7 +109,6 @@ def test_fresh_name(static_vdis: ModuleType, tmp_path: Path): assert static_vdis.fresh_name() == "0" - def test_sr_attach(static_vdis: ModuleType, mocker): """Test sr_attach()""" @@ -82,4 +131,4 @@ def test_sr_attach(static_vdis: ModuleType, mocker): "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 ec24848e934..ff3a01da596 100755 --- a/scripts/static-vdis +++ b/scripts/static-vdis @@ -11,7 +11,7 @@ import sys import time import urllib.parse import xmlrpc.client -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast import XenAPI @@ -166,6 +166,7 @@ def add(session, vdi_uuid, reason): sm = None all_sm = session.xenapi.SM.get_all_records() + sm_ref = "" # pragma: no cover for sm_ref in all_sm: if all_sm[sm_ref]['type'] == ty: sm = all_sm[sm_ref] @@ -183,6 +184,7 @@ def add(session, vdi_uuid, reason): if "VDI_ATTACH_OFFLINE" in sm["features"]: data["volume-plugin"] = ty data[smapiv3_config] = json.dumps(device_config) + assert device_config # must not be None sr = sr_attach(ty, device_config) location = session.xenapi.VDI.get_location(vdi) stat = call_volume_plugin(ty, "Volume.stat", [ sr, location ]) @@ -238,7 +240,7 @@ def call_backend_attach(driver, config): xml = doexec(args) if xml[0] != 0: raise Exception("SM_BACKEND_FAILURE(%d, %s, %s)" % xml) - xml_rpc = xmlrpc.client.loads(xml[1]) + xml_rpc = xmlrpc.client.loads(xml[1]) # type: Any # pragma: no cover if 'params_nbd' in xml_rpc[0][0]: # Prefer NBD if available @@ -259,8 +261,8 @@ def call_backend_detach(driver, config): raise Exception("SM_BACKEND_FAILURE(%d, %s, %s)" % xml) xml_rpc = xmlrpc.client.loads(xml[1]) try: - res = xml_rpc[0][0]['params'] - except: + res = cast(dict, xml_rpc[0][0])['params'] # pragma: no cover + except Exception: res = xml_rpc[0][0] return res @@ -301,7 +303,7 @@ def attach(vdi_uuid): os.unlink(d + "/disk") except: pass - path = None + path = None # Raise TypeError if path is not set at the end if not (os.path.exists(d + "/" + smapiv3_config)): # SMAPIv1 config = read_whole_file(d + "/config") @@ -333,10 +335,13 @@ def attach(vdi_uuid): (path, exportname) = parse_nbd_uri(uri) path = connect_nbd(path=path, exportname=exportname) + if path is None: + raise TypeError("static-vdis: attach(): path was not set") os.symlink(path, d + "/disk") return d + "/disk" if not found: raise Exception("Disk configuration not found") + return None def detach(vdi_uuid): found = False @@ -375,8 +380,9 @@ def usage(): print(" %s attach -- attach the VDI immediately" % sys.argv[0]) print(" %s detach -- detach the VDI immediately" % sys.argv[0]) sys.exit(1) - -if __name__ == "__main__": + + +def main(): if len(sys.argv) < 2: usage() @@ -395,9 +401,13 @@ if __name__ == "__main__": elif sys.argv[1] == "del" and len(sys.argv) == 3: delete(sys.argv[2]) elif sys.argv[1] == "attach" and len(sys.argv) == 3: - path = attach(sys.argv[2]) - print(path) + disk_path = attach(sys.argv[2]) + print(disk_path) elif sys.argv[1] == "detach" and len(sys.argv) == 3: detach(sys.argv[2]) else: usage() + + +if __name__ == "__main__": # pragma: no cover + main()