test_mountingaccessor.py: use caplog to capture logs

Author: bernhardkaindl

tests/test_accessor.py

@@ -7,15 +7,16 @@
 
 if TYPE_CHECKING:
     import pyfakefs
+    import pytest
 
 
-def test_file_accessor(fs):
-    # type(pyfakefs.fake_filesystem.FakeFilesystem) -> None
+def test_file_accessor(fs, caplog):
+    # type(pyfakefs.fake_filesystem.FakeFilesystem, pytest.LogCaptureFixture) -> None
     """Test FileAccessor.writeFile(), .openAddress and .access using pyfakefs"""
     accessor = xcp.accessor.createAccessor("file://repo/", False)
     assert isinstance(accessor, xcp.accessor.FileAccessor)
     check_binary_read(accessor, "/repo", fs)
-    check_binary_write(accessor, "/repo", fs)
+    check_binary_write(accessor, "/repo", fs, caplog)
 
 
 class TestAccessor(unittest.TestCase):

tests/test_mountingaccessor.py

@@ -1,9 +1,13 @@
 """pytest tests testing subclasses of xcp.accessor.MountingAccessor using pyfakefs"""
+
+import logging
 import os
 import sys
 from io import BytesIO
 from typing import TYPE_CHECKING, cast
 
+import pytest
+from pytest import LogCaptureFixture as LogCap
 from mock import patch
 from pyfakefs.fake_filesystem import FakeFileOpen, FakeFilesystem
 
@@ -18,8 +22,6 @@
     if TYPE_CHECKING:
         from typing_extensions import Literal
 else:
-    import pytest
-
     pytest.skip(allow_module_level=True)
     sys.exit(0)  # Let pyright know that this is a dead end
 
@@ -30,8 +32,8 @@ def expect(fp, mount):
     fp.register_subprocess(mount)  # type: ignore[arg-type]
 
 
-def test_device_accessor(fs, fp, mount_dir):
-    # type: (FakeFilesystem, FakeProcess, bytes) -> None
+def test_device_accessor(fs, fp, mount_dir, caplog):
+    # type: (FakeFilesystem, FakeProcess, bytes, LogCap) -> None
     assert isinstance(fp, FakeProcess)
 
     # Test xcp.mount.bindMount()
@@ -43,11 +45,11 @@ def test_device_accessor(fs, fp, mount_dir):
     accessor = xcp.accessor.createAccessor("dev:///dev/device", (True, [""]))
 
     assert isinstance(accessor, xcp.accessor.MountingAccessorTypes)
-    check_mounting_accessor(accessor, fs, fp, mount_dir)
+    check_mounting_accessor(accessor, fs, fp, mount_dir, caplog)
 
 
-def test_nfs_accessor(fs, fp, mount_dir):
-    # type: (FakeFilesystem, FakeProcess, bytes) -> None
+def test_nfs_accessor(fs, fp, mount_dir, caplog):
+    # type: (FakeFilesystem, FakeProcess, bytes, LogCap) -> None
     assert isinstance(fp, FakeProcess)
     mount = [
         b"/bin/mount",
@@ -61,11 +63,11 @@ def test_nfs_accessor(fs, fp, mount_dir):
     expect(fp, mount)
     accessor = xcp.accessor.createAccessor("nfs://server/path", False)
     assert isinstance(accessor, xcp.accessor.NFSAccessor)
-    check_mounting_accessor(accessor, fs, fp, mount_dir)
+    check_mounting_accessor(accessor, fs, fp, mount_dir, caplog)
 
 
-def check_mounting_accessor(accessor, fs, fp, mount_point):
-    # type: (xcp.accessor.Mount, FakeFilesystem, FakeProcess, bytes) -> None
+def check_mounting_accessor(accessor, fs, fp, mount_point, caplog):
+    # type: (xcp.accessor.Mount, FakeFilesystem, FakeProcess, bytes, LogCap) -> None
     """Test subclasses of MountingAccessor (with xcp.cmd.runCmd in xcp.mount mocked)"""
 
     assert isinstance(accessor, xcp.accessor.MountingAccessorTypes)
@@ -83,7 +85,7 @@ def check_mounting_accessor(accessor, fs, fp, mount_point):
         fs.add_mount_point(location)
 
     assert check_binary_read(accessor, location, fs)
-    assert check_binary_write(accessor, location, fs)
+    assert check_binary_write(accessor, location, fs, caplog)
     assert open_text(accessor, location, fs, UTF8TEXT_LITERAL) == UTF8TEXT_LITERAL
 
     if sys.version_info.major >= 3:
@@ -119,15 +121,20 @@ def check_binary_read(accessor, location, fs):
     return cast(bytes, binary_file.read()) == binary_data
 
 
-def check_binary_write(accessor, location, fs):
-    # type: (Literal[False] | xcp.accessor.AnyAccessor, str, FakeFilesystem) -> bool
+def check_binary_write(accessor, location, fs, caplog):
+    # type: (xcp.accessor.AnyAccessor, str, FakeFilesystem, LogCap) -> bool
     """Test the writeFile() method of different types of local Accessor classes"""
 
     assert isinstance(accessor, xcp.accessor.LocalTypes)
     name = "binary_file_written_by_accessor"
-    accessor.writeFile(BytesIO(binary_data), name)
-
-    assert accessor.access(name)
+    with caplog.at_level(logging.FATAL):
+        accessor.writeFile(BytesIO(binary_data), name)
+    if caplog.record_tuples:
+        logger, level, message = caplog.record_tuples[0]
+        assert logger == "root"
+        assert level == logging.FATAL
+        assert message == "Copying to " + name
+        assert accessor.access(name)
 
     with FakeFileOpen(fs, delete_on_close=True)(location + "/" + name, "rb") as written:
         return cast(bytes, written.read()) == binary_data

tox.ini

@@ -129,7 +129,7 @@ extras      = coverage
 commands    =
     coverage xml  -o {envlogdir}/coverage.xml  --fail-under {env:XCP_COV_MIN:68}
     coverage html -d {envlogdir}/htmlcov
-    coverage html -d {envlogdir}/htmlcov-tests --fail-under {env:TESTS_COV_MIN:96}    \
+    coverage html -d {envlogdir}/htmlcov-tests --fail-under {env:TESTS_COV_MIN:95}    \
                       --include="tests/*"
     diff-cover --compare-branch=origin/master --include-untracked                     \
       {env:PY3_DIFFCOVER_OPTIONS} --fail-under {env:DIFF_COV_MIN:92}                  \

xcp/accessor.py

@@ -134,6 +134,7 @@ def openAddress(self, address):
 
 class MountingAccessor(FilesystemAccessor):
     def __init__(self, mount_types, mount_source, mount_options=None):
+        # type: (List[str], str, List[str] | None) -> None
         ro = isinstance(mount_options, list) and 'ro' in mount_options
         super(MountingAccessor, self).__init__(None, ro)