diff --git a/tests/test_cmd.py b/tests/test_cmd.py index 9841df79..594a8127 100644 --- a/tests/test_cmd.py +++ b/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" diff --git a/tests/test_mountingaccessor.py b/tests/test_mountingaccessor.py index ecf78ab6..601b8f62 100644 --- a/tests/test_mountingaccessor.py +++ b/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) diff --git a/xcp/cmd.py b/xcp/cmd.py index 097f429f..394f0b0e 100644 --- a/xcp/cmd.py +++ b/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) diff --git a/xcp/compat.py b/xcp/compat.py index fb625b32..27bbdc69 100644 --- a/xcp/compat.py +++ b/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