CP-49928 static-vdis: Fix pyright: Add type hints, assert, var = ""

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

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"],
-    )
+    )

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 <uuid>        -- attach the VDI immediately" % sys.argv[0])    
     print(" %s detach <uuid>        -- 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()