xcp/mount.py: Fix or silence pyright warnings

tests/test_cmd.py

@@ -65,10 +65,13 @@ def test_fileContents_pciids_bytes(self):
     def test_fileContents_pciids_binstr(self):
         contents_bytes = self.c.fileContents("tests/data/pci.ids", mode="rb")
         contents_string = self.c.fileContents("tests/data/pci.ids", mode="r")
+        contents_default = self.c.fileContents("tests/data/pci.ids")
         self.assertIsInstance(contents_bytes, bytes)
         self.assertIsInstance(contents_string, str)
+        self.assertIsInstance(contents_default, str)
         self.assertEqual(contents_bytes, six.ensure_binary(contents_string))
         self.assertEqual(contents_string, six.ensure_str(contents_bytes))
+        self.assertEqual(contents_default, six.ensure_str(contents_bytes))
 
     def test_runCmd(self):
         output_data = "line1\nline2\n"

tests/test_mountingaccessor.py

@@ -7,30 +7,61 @@
 from pyfakefs.fake_filesystem import FakeFileOpen, FakeFilesystem
 
 import xcp.accessor
+import xcp.mount
 
 from .test_httpaccessor import UTF8TEXT_LITERAL
 
+if sys.version_info >= (3, 6):
+    from pytest_subprocess.fake_process import FakeProcess
+else:
+    import pytest
+
+    pytest.skip(allow_module_level=True)
+
 binary_data = b"\x00\x1b\x5b\x95\xb1\xb2\xb3\xb4\xb5\xb6\xb7\xb8\xb9\xcc\xdd\xee\xff"
 
 
-def test_device_accessor(fs):
-    # type: (FakeFilesystem) -> None
-    accessor = xcp.accessor.createAccessor("dev:///dev/device", False)
-    check_mounting_accessor(accessor, fs)
+def expect(fp, mount):
+    fp.register_subprocess(mount)  # type: ignore[arg-type]
 
 
-def test_nfs_accessor(fs):
-    # type: (FakeFilesystem) -> None
+def test_device_accessor(fs, fp):
+    # type: (FakeFilesystem, FakeProcess) -> None
+    assert isinstance(fp, FakeProcess)
+
+    # Test xcp.mount.bindMount()
+    mount = [b"/bin/mount", b"--bind", b"src", b"mountpoint_dest"]
+    fp.register_subprocess(mount)  # type: ignore[arg-type]
+    assert xcp.mount.bindMount("src", "mountpoint_dest") is None
+
+    expect(fp, [b"/bin/mount", b"-t", b"iso9660", b"-o", b"ro", b"/dev/device", b"/tmp"])
+    accessor = xcp.accessor.createAccessor("dev:///dev/device", False)
+    check_mounting_accessor(accessor, fs, fp)
+
+
+def test_nfs_accessor(fs, fp):
+    # type: (FakeFilesystem, FakeProcess) -> None
+    assert isinstance(fp, FakeProcess)
+    mount = [
+        b"/bin/mount",
+        b"-t",
+        b"nfs",
+        b"-o",
+        b"tcp,timeo=100,retrans=1,retry=0",
+        b"server/path",
+        b"/tmp",
+    ]
+    expect(fp, mount)
     accessor = xcp.accessor.createAccessor("nfs://server/path", False)
-    check_mounting_accessor(accessor, fs)
+    check_mounting_accessor(accessor, fs, fp)
 
 
-def check_mounting_accessor(accessor, fs):
-    # type: (xcp.accessor.MountingAccessor, FakeFilesystem) -> None
+def check_mounting_accessor(accessor, fs, fp):
+    # type: (xcp.accessor.MountingAccessor, FakeFilesystem, FakeProcess) -> None
     """Test subclasses of MountingAccessor (with xcp.cmd.runCmd in xcp.mount mocked)"""
 
-    with patch("xcp.cmd.runCmd") as mount_runcmd:
-        mount_runcmd.return_value = (0, "", "")
+    with patch("tempfile.mkdtemp") as tempfile_mkdtemp:
+        tempfile_mkdtemp.return_value = "/tmp"
         accessor.start()
 
     assert accessor.location
@@ -48,8 +79,9 @@ def check_mounting_accessor(accessor, fs):
     if sys.version_info.major >= 3:
         fs.mount_points.pop(location)
 
-    with patch("xcp.cmd.runCmd"):
-        accessor.finish()
+    umount = [b"/bin/umount", b"-d", b"/tmp"]
+    fp.register_subprocess(umount)  # type: ignore[arg-type]
+    accessor.finish()
 
     assert not fs.exists(location)
 

xcp/cmd.py

@@ -25,10 +25,12 @@
 
 import subprocess
 import sys
+from typing import Any, cast
 
 from xcp import logger
 from xcp.compat import open_defaults_for_utf8_text
 
+
 def _encode_command_to_bytes(command):
     # When the locale not an UTF-8 locale, Python3.6 Popen can't deal with ord() >= 128
     # when the command contains strings, not bytes. Therefore, convert any strings to bytes:
@@ -44,6 +46,7 @@ def _encode_command_to_bytes(command):
     return command
 
 def runCmd(command, with_stdout=False, with_stderr=False, inputtext=None, **kwargs):
+    # type: (bytes | str | list[str], bool, bool, bytes | str | None, Any) -> Any
     # sourcery skip: assign-if-exp, hoist-repeated-if-condition, reintroduce-else
 
     if inputtext is not None:
@@ -59,12 +62,12 @@ def runCmd(command, with_stdout=False, with_stderr=False, inputtext=None, **kwar
 
     # pylint: disable-next=unexpected-keyword-arg
     cmd = subprocess.Popen(command, bufsize=(1 if sys.version_info < (3, 3) else -1),
-                           stdin=(inputtext and subprocess.PIPE or None),
+                           stdin=cast(int, inputtext and subprocess.PIPE or None),
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            shell=not isinstance(command, list),
                            **kwargs)
-    (out, err) = cmd.communicate(inputtext)
+    (out, err) = cmd.communicate(cast(str, inputtext))
     rv = cmd.returncode
 
     l = "ran %s; rc %d" % (str(command), rv)

xcp/compat.py

@@ -52,9 +52,7 @@ def open_defaults_for_utf8_text(args, kwargs):
         mode = args[0]
     if sys.version_info < (3, 0):
         return mode, other_kwargs
-    if not mode or not isinstance(mode, str):
-        raise ValueError("The mode argument is required! r for text, rb for binary")
     if sys.version_info >= (3, 0) and "b" not in mode:
         kwargs.setdefault("encoding", "utf-8")
         kwargs.setdefault("errors", "replace")
-    return mode, other_kwargs
+    return mode or "r", other_kwargs