Repository: osbuild/osbuild
Pull request #1498 (2023-12-12 15:00:38)

fscache: implement "last_used()" helper
This is an attempt to start fixing https://github.com/osbuild/osbuild/issues/392, i.e. that when the fscache is full it will just stop caching instead of removing the oldest piece from the cache and continuing. This behavior is now explicitly tested as part of the PR so that we can later change it for something that removes the oldest entries.

When looking at this I noticed that AFAICT we have no mechanism right now to identify how recently an object in the cache was used. This PR adds a helper "FsCache._last_used()" and uses the per object lock files `atime` for this. It has to be at the file level because the cache is designed to be accessible from multiple programs at once and this seems to be the least invasive and robust way. But better ideas are of course welcome!

Once this foundation is done followup PRs would
a) provide a helper to sort cache entries by oldest
b) when the cache is full, loop over oldest entries and remove until there is enough space for the new object (this needs to be done carefully as (a) will be either be racy or slow (I think!) but we can discuss this in the followups :)

dustymabe: > When looking at this I noticed that AFAICT we have no mechanism right now to identify how recently an object in the cache was used. This PR adds a helper "FsCache._last_used()" and uses the per object lock files `atime` for this. It has to be at the file level because the cache is designed to be accessible from multiple programs at once and this seems to be the least invasive and robust way. But better ideas are of course welcome!

SGTM - I was thinking of just evicting the oldest entries (FIFO), but LRU works too if we have a good mechanism to detect it.
mvo5: > > When looking at this I noticed that AFAICT we have no mechanism right now to identify how recently an object in the cache was used. This PR adds a helper "FsCache._last_used()" and uses the per object lock files `atime` for this. It has to be at the file level because the cache is designed to be accessible from multiple programs at once and this seems to be the least invasive and robust way. But better ideas are of course welcome!
> 
> SGTM - I was thinking of just evicting the oldest entries (FIFO), but LRU works too if we have a good mechanism to detect it.

Thanks for your feedback! Maybe I'm overthinking things here, I agree that FIFO is way easier - let's hear what the others think, I'm happy either way (but I like the LRU slightly better, feels more precise).
achilleas-k: I think `noatime` is a rather common mount option.  Would it make sense to `touch` the root of the object (or the `object.info` file) whenever it's used and use the mtime instead?

As for using FIFO, it would be better than what we have now but in realistic cases might be strange.  I'm imagining a scenario where a build is created, used 10 times to build 10 images, and then is removed because it's the oldest object in the cache, even though it's been used by every image build since it was created.
mvo5: > I think `noatime` is a rather common mount option. Would it make sense to `touch` the root of the object (or the `object.info` file) whenever it's used and use the mtime instead?
> 
> As for using FIFO, it would be better than what we have now but in realistic cases might be strange. I'm imagining a scenario where a build is created, used 10 times to build 10 images, and then is removed because it's the oldest object in the cache, even though it's been used by every image build since it was created.

Thanks for your feedback on this!  I should have added a comment (or a test) but I'm 85% confident that an explicit utime() will override `noatime`. I will add a test just to be sure that I'm not misremembering :) If that fails I'm happy to use mtime, conceptually it feels more correct to use `atime` to me though (but again, maybe I'm overthinking things here).
achilleas-k: > I'm 85% confident that an explicit utime() will override `noatime`
 
Thanks!  I wasn't aware :)

Patch:
diff --git a/osbuild/util/fscache.py b/osbuild/util/fscache.py
index 669735b5..c8fc99aa 100644
--- a/osbuild/util/fscache.py
+++ b/osbuild/util/fscache.py
@@ -9,6 +9,7 @@ the cache under a given limit.
 # pylint: disable=too-many-lines
 
 import contextlib
+import ctypes
 import errno
 import json
 import os
@@ -1016,14 +1017,12 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
             # Use an ExitStack so we can catch exceptions raised by the
             # `__enter__()` call on the context-manager. We want to catch
             # `OSError` exceptions and convert them to cache-misses.
+            obj_lock_path = os.path.join(
+                self._dirname_objects, name, self._filename_object_lock)
             try:
-                es.enter_context(
+                lock_fd = es.enter_context(
                     self._atomic_open(
-                        os.path.join(
-                            self._dirname_objects,
-                            name,
-                            self._filename_object_lock,
-                        ),
+                        obj_lock_path,
                         write=False,
                         wait=False,
                     )
@@ -1033,12 +1032,33 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
                     raise self.MissError() from None
                 raise e
 
+            libc = linux.Libc.default()
+            libc.futimens(lock_fd, ctypes.byref(linux.c_timespec_times2(
+                atime=linux.c_timespec(tv_sec=0, tv_nsec=libc.UTIME_NOW),
+                mtime=linux.c_timespec(tv_sec=0, tv_nsec=libc.UTIME_OMIT),
+            )))
+
             yield os.path.join(
                 self._dirname_objects,
                 name,
                 self._dirname_data,
             )
 
+    def _last_used(self, name: str) -> float:
+        """Return the last time the given object was last used.
+
+        Note that the resolution is only as good as what the filesystem "atime"
+        gives us.
+        """
+        obj_lock_path = os.path.join(
+            self._dirname_objects, name, self._filename_object_lock)
+        try:
+            return os.stat(self._path(obj_lock_path)).st_atime
+        except OSError as e:
+            if e.errno in [errno.EAGAIN, errno.ENOENT, errno.ENOTDIR]:
+                raise self.MissError() from None
+            raise e
+
     @property
     def info(self) -> FsCacheInfo:
         """Query Cache Information
diff --git a/osbuild/util/linux.py b/osbuild/util/linux.py
index b0650bd3..d09af8fc 100644
--- a/osbuild/util/linux.py
+++ b/osbuild/util/linux.py
@@ -422,6 +422,14 @@ def fcntl_flock(fd: int, lock_type: int, wait: bool = False):
     fcntl.fcntl(fd, lock_cmd, arg_flock64)
 
 
+class c_timespec(ctypes.Structure):
+    _fields_ = [('tv_sec', ctypes.c_long), ('tv_nsec', ctypes.c_long)]
+
+
+class c_timespec_times2(ctypes.Structure):
+    _fields_ = [('atime', c_timespec), ('mtime', c_timespec)]
+
+
 class Libc:
     """Safe Access to libc
 
@@ -434,6 +442,10 @@ class Libc:
     RENAME_NOREPLACE = ctypes.c_uint(1)
     RENAME_WHITEOUT = ctypes.c_uint(4)
 
+    # see /usr/include/x86_64-linux-gnu/bits/stat.h
+    UTIME_NOW = ctypes.c_long(((1 << 30) - 1))
+    UTIME_OMIT = ctypes.c_long(((1 << 30) - 2))
+
     _lock = threading.Lock()
     _inst = None
 
@@ -462,6 +474,22 @@ class Libc:
         setattr(proto, "errcheck", self._errcheck_errno)
         setattr(proto, "__name__", "renameat2")
         self.renameat2 = proto
+        # prototype: futimens
+        proto = ctypes.CFUNCTYPE(
+            ctypes.c_int,
+            ctypes.c_int,
+            ctypes.POINTER(c_timespec_times2),
+            use_errno=True,
+        )(
+            ("futimens", self._lib),
+            (
+                (1, "fd"),
+                (1, "timespec"),
+            ),
+        )
+        setattr(proto, "errcheck", self._errcheck_errno)
+        setattr(proto, "__name__", "futimens")
+        self.futimens = proto
 
     @staticmethod
     def make() -> "Libc":


Test patch:
diff --git a/osbuild/testutil/__init__.py b/osbuild/testutil/__init__.py
index b429f343..e4ab1e8c 100644
--- a/osbuild/testutil/__init__.py
+++ b/osbuild/testutil/__init__.py
@@ -1,6 +1,7 @@
 """
 Test related utilities
 """
+import os
 import shutil
 
 
@@ -13,3 +14,20 @@ def assert_dict_has(v, keys, expected_value):
         assert key in v
         v = v[key]
     assert v == expected_value
+
+
+def make_fake_input_tree(tmpdir, fake_content: dict) -> str:
+    """Create a directory tree of files with content.
+
+    Call it with:
+        {"filename": "content", "otherfile": "content"}
+
+    filename paths will have their parents created as needed, under tmpdir.
+    """
+    basedir = os.path.join(tmpdir, "tree")
+    for path, content in fake_content.items():
+        dirp, name = os.path.split(os.path.join(basedir, path.lstrip("/")))
+        os.makedirs(dirp, exist_ok=True)
+        with open(os.path.join(dirp, name), "w", encoding="utf-8") as fp:
+            fp.write(content)
+    return basedir
diff --git a/stages/test/test_erofs.py b/stages/test/test_erofs.py
index da534344..bcda4313 100644
--- a/stages/test/test_erofs.py
+++ b/stages/test/test_erofs.py
@@ -7,20 +7,9 @@ from unittest import mock
 import pytest
 
 import osbuild.meta
-from osbuild.testutil import has_executable
+from osbuild.testutil import has_executable, make_fake_input_tree
 from osbuild.testutil.imports import import_module_from_path
 
-
-def make_fake_input_tree(tmpdir, fake_content: dict) -> str:
-    basedir = os.path.join(tmpdir, "tree")
-    for path, content in fake_content.items():
-        dirp, name = os.path.split(os.path.join(basedir, path.lstrip("/")))
-        os.makedirs(dirp, exist_ok=True)
-        with open(os.path.join(dirp, name), "w", encoding="utf-8") as fp:
-            fp.write(content)
-    return basedir
-
-
 TEST_INPUT = [
     ({}, []),
     ({"compression": {"method": "lz4"}}, ["-z", "lz4"]),
diff --git a/stages/test/test_xz.py b/stages/test/test_xz.py
new file mode 100644
index 00000000..c69a9c15
--- /dev/null
+++ b/stages/test/test_xz.py
@@ -0,0 +1,97 @@
+#!/usr/bin/python3
+
+import os.path
+import subprocess
+from unittest import mock
+
+import pytest
+
+import osbuild.meta
+from osbuild.testutil import has_executable, make_fake_input_tree
+from osbuild.testutil.imports import import_module_from_path
+
+
+@pytest.mark.parametrize("test_data,expected_err", [
+    # bad
+    ({}, "'filename' is a required property"),
+    # good
+    ({"filename": "image.xz"}, ""),
+])
+def test_schema_validation_xz(test_data, expected_err):
+    name = "org.osbuild.xz"
+    root = os.path.join(os.path.dirname(__file__), "../..")
+    mod_info = osbuild.meta.ModuleInfo.load(root, "Stage", name)
+    schema = osbuild.meta.Schema(mod_info.get_schema(version="2"), name)
+
+    test_input = {
+        "type": "org.osbuild.xz",
+        "options": {
+        }
+    }
+    test_input["options"].update(test_data)
+    res = schema.validate(test_input)
+
+    if expected_err == "":
+        assert res.valid is True, f"err: {[e.as_dict() for e in res.errors]}"
+    else:
+        assert res.valid is False
+        assert len(res.errors) == 1, [e.as_dict() for e in res.errors]
+        err_msgs = [e.as_dict()["message"] for e in res.errors]
+        assert expected_err in err_msgs[0]
+
+
+@pytest.fixture(name="fake_input_tree")
+def fake_input(tmp_path):
+    fake_input_tree = make_fake_input_tree(tmp_path, {
+        "/sha256:ff3e910ceb79fd66e67b7ff4a9073b1d584c748a5f3d1e6b1873afa1cf35ec59": "A file to be compressed",
+    })
+    inputs = {
+        "file": {
+            "path": fake_input_tree,
+            "data": {
+                "files": {
+                    "sha256:ff3e910ceb79fd66e67b7ff4a9073b1d584c748a5f3d1e6b1873afa1cf35ec59": {}
+                }
+            }
+        }
+    }
+    return (fake_input_tree, inputs)
+
+
+@pytest.mark.skipif(not has_executable("xz"), reason="no xz executable")
+def test_xz_integration(tmp_path, fake_input_tree):  # pylint: disable=unused-argument
+    inputs = fake_input_tree[1]
+    stage_path = os.path.join(os.path.dirname(__file__), "../org.osbuild.xz")
+    stage = import_module_from_path("xz_stage", stage_path)
+    options = {
+        "filename": "image.txt.xz",
+    }
+    stage.main(inputs, tmp_path, options)
+
+    img_path = os.path.join(tmp_path, "image.txt.xz")
+    assert os.path.exists(img_path)
+    output = subprocess.check_output([
+        "xz", "--decompress", "--stdout", img_path], encoding="utf-8")
+    assert "A file to be compressed" in output
+
+
+@mock.patch("subprocess.run")
+def test_xz_cmdline(mock_run, tmp_path, fake_input_tree):
+    fake_input_path = fake_input_tree[0]
+    inputs = fake_input_tree[1]
+    stage_path = os.path.join(os.path.dirname(__file__), "../org.osbuild.xz")
+    stage = import_module_from_path("xz_stage", stage_path)
+    filename = "image.txt.xz"
+    options = {
+        "filename": filename,
+    }
+    stage.main(inputs, tmp_path, options)
+
+    expected = [
+        "xz",
+        "--keep",
+        "--stdout",
+        "-0",
+        os.path.join(fake_input_path, "sha256:ff3e910ceb79fd66e67b7ff4a9073b1d584c748a5f3d1e6b1873afa1cf35ec59")
+    ]
+    mock_run.assert_called_with(expected, stdout=mock.ANY, check=True, env={'XZ_OPT': '--threads 0'})
diff --git a/stages/test/test_zstd.py b/stages/test/test_zstd.py
new file mode 100644
index 00000000..665601e9
--- /dev/null
+++ b/stages/test/test_zstd.py
@@ -0,0 +1,98 @@
+#!/usr/bin/python3
+
+import os.path
+import subprocess
+from unittest import mock
+
+import pytest
+
+import osbuild.meta
+from osbuild.testutil import has_executable, make_fake_input_tree
+from osbuild.testutil.imports import import_module_from_path
+
+
+@pytest.mark.parametrize("test_data,expected_err", [
+    # bad
+    ({}, "'filename' is a required property"),
+    # good
+    ({"filename": "image.zst"}, ""),
+])
+def test_schema_validation_zstd(test_data, expected_err):
+    name = "org.osbuild.zstd"
+    root = os.path.join(os.path.dirname(__file__), "../..")
+    mod_info = osbuild.meta.ModuleInfo.load(root, "Stage", name)
+    schema = osbuild.meta.Schema(mod_info.get_schema(version="2"), name)
+
+    test_input = {
+        "type": "org.osbuild.zstd",
+        "options": {
+        }
+    }
+    test_input["options"].update(test_data)
+    res = schema.validate(test_input)
+
+    if expected_err == "":
+        assert res.valid is True, f"err: {[e.as_dict() for e in res.errors]}"
+    else:
+        assert res.valid is False
+        assert len(res.errors) == 1, [e.as_dict() for e in res.errors]
+        err_msgs = [e.as_dict()["message"] for e in res.errors]
+        assert expected_err in err_msgs[0]
+
+
+@pytest.fixture(name="fake_input_tree")
+def fake_input(tmp_path):
+    fake_input_tree = make_fake_input_tree(tmp_path, {
+        "/sha256:ff3e910ceb79fd66e67b7ff4a9073b1d584c748a5f3d1e6b1873afa1cf35ec59": "A file to be compressed",
+    })
+    inputs = {
+        "file": {
+            "path": fake_input_tree,
+            "data": {
+                "files": {
+                    "sha256:ff3e910ceb79fd66e67b7ff4a9073b1d584c748a5f3d1e6b1873afa1cf35ec59": {}
+                }
+            }
+        }
+    }
+    return (fake_input_tree, inputs)
+
+
+@pytest.mark.skipif(not has_executable("zstd"), reason="no zstd executable")
+def test_zstd_integration(tmp_path, fake_input_tree):  # pylint: disable=unused-argument
+    inputs = fake_input_tree[1]
+    stage_path = os.path.join(os.path.dirname(__file__), "../org.osbuild.zstd")
+    stage = import_module_from_path("zstd_stage", stage_path)
+    options = {
+        "filename": "image.txt.zst",
+    }
+    stage.main(inputs, tmp_path, options)
+
+    img_path = os.path.join(tmp_path, "image.txt.zst")
+    assert os.path.exists(img_path)
+    output = subprocess.check_output([
+        "zstd", "--decompress", "--stdout", img_path], encoding="utf-8")
+    assert "A file to be compressed" in output
+
+
+@mock.patch("subprocess.run")
+def test_zstd_cmdline(mock_run, tmp_path, fake_input_tree):
+    fake_input_path = fake_input_tree[0]
+    inputs = fake_input_tree[1]
+    stage_path = os.path.join(os.path.dirname(__file__), "../org.osbuild.zstd")
+    stage = import_module_from_path("zstd_stage", stage_path)
+    filename = "image.txt.zst"
+    options = {
+        "filename": filename,
+    }
+    stage.main(inputs, tmp_path, options)
+
+    expected = [
+        "zstd",
+        "--keep",
+        "--stdout",
+        "-T0",
+        "-1",
+        os.path.join(fake_input_path, "sha256:ff3e910ceb79fd66e67b7ff4a9073b1d584c748a5f3d1e6b1873afa1cf35ec59")
+    ]
+    mock_run.assert_called_with(expected, stdout=mock.ANY, check=True)
diff --git a/test/mod/test_testutil_fake_tree.py b/test/mod/test_testutil_fake_tree.py
new file mode 100644
index 00000000..e929949c
--- /dev/null
+++ b/test/mod/test_testutil_fake_tree.py
@@ -0,0 +1,20 @@
+#
+# Tests for the 'osbuild.util.testutil' module.
+#
+import os.path
+
+import pytest
+
+from osbuild.testutil import make_fake_input_tree
+
+
+def test_make_fake_tree(tmp_path):  # pylint: disable=unused-argument
+    fake_input_tree = make_fake_input_tree(tmp_path, {
+        "/fake-file-one": "Some content",
+        "/second/fake-file-two": "Second content",
+    })
+    assert os.path.isdir(fake_input_tree)
+    assert os.path.exists(os.path.join(fake_input_tree, "fake-file-one"))
+    assert open(os.path.join(fake_input_tree, "fake-file-one")).read() == "Some content"
+    assert os.path.exists(os.path.join(fake_input_tree, "second/fake-file-two"))
+    assert open(os.path.join(fake_input_tree, "second/fake-file-two")).read() == "Second content"
diff --git a/test/mod/test_util_fscache.py b/test/mod/test_util_fscache.py
index 1741b52c..d7428493 100644
--- a/test/mod/test_util_fscache.py
+++ b/test/mod/test_util_fscache.py
@@ -4,10 +4,14 @@
 
 # pylint: disable=protected-access
 
+import contextlib
 import json
 import os
+import pathlib
 import subprocess
+import sys
 import tempfile
+import time
 
 import pytest
 
@@ -20,6 +24,22 @@ def tmpdir_fixture():
         yield tmp
 
 
+def sleep_for_fs():
+    """Sleep a tiny amount of time for atime/mtime updates to show up in fs"""
+    time.sleep(0.05)
+
+
+def has_precise_fs_timestamps():
+    with tempfile.TemporaryDirectory(dir="/var/tmp") as tmpdir:
+        stamp_path = pathlib.Path(tmpdir) / "stamp"
+        stamp_path.write_bytes(b"m1")
+        mtime1 = stamp_path.stat().st_mtime
+        sleep_for_fs()
+        stamp_path.write_bytes(b"m2")
+        mtime2 = stamp_path.stat().st_mtime
+        return mtime2 > mtime1
+
+
 def test_calculate_space(tmpdir):
     #
     # Test the `_calculate_space()` helper and verify it only includes file
@@ -397,3 +417,83 @@ def test_size_discard(tmpdir):
         with pytest.raises(fscache.FsCache.MissError):
             with cache.load("foo") as rpath:
                 pass
+
+def test_cache_last_used_noent(tmpdir):
+    cache = fscache.FsCache("osbuild-test-appid", tmpdir)
+    with pytest.raises(fscache.FsCache.MissError):
+        cache._last_used("non-existant-entry")
+
+
+@pytest.mark.skipif(not has_precise_fs_timestamps(), reason="need precise fs timestamps")
+def test_cache_load_updates_last_used(tmpdir):
+    cache = fscache.FsCache("osbuild-test-appid", tmpdir)
+    with cache:
+        cache.info = cache.info._replace(maximum_size=1024*1024)
+        with cache.store("foo") as rpath:
+            pass
+        with cache.load("foo") as rpath:
+            pass
+        load_time1 = cache._last_used("foo")
+        # would be nice to have a helper for this in cache
+        obj_lock_path = os.path.join(
+            cache._dirname_objects, "foo", cache._filename_object_lock)
+        mtime1 = os.stat(cache._path(obj_lock_path)).st_mtime
+        assert load_time1 > 0
+        sleep_for_fs()
+        with cache.load("foo") as rpath:
+            pass
+        # load time is updated
+        load_time2 = cache._last_used("foo")
+        assert load_time2 > load_time1
+        # mtime is unchanged
+        mtime2 = os.stat(cache._path(obj_lock_path)).st_mtime
+        assert mtime1 == mtime2
+
+
+@pytest.mark.skipif(os.getuid() != 0, reason="needs root")
+def test_cache_load_updates_last_used_on_noatime(tmp_path):
+    mnt_path = tmp_path / "mnt"
+    mnt_path.mkdir()
+    with contextlib.ExitStack() as cm:
+        subprocess.check_call(
+            ["mount", "-t", "tmpfs", "-o", "noatime", "none", os.fspath(mnt_path)],
+            stdout=sys.stdout,
+            stderr=sys.stderr,
+        )
+        cm.callback(subprocess.check_call, ["umount", os.fspath(mnt_path)], stdout=sys.stdout, stderr=sys.stderr)
+        test_cache_load_updates_last_used(mnt_path)
+
+
+def test_cache_full_behavior(tmp_path):
+    cache = fscache.FsCache("osbuild-cache-evict", tmp_path)
+    with cache:
+        # use big sizes to mask the effect of dirs using 4k of space too
+        cache.info = cache.info._replace(maximum_size=192 * 1024)
+        # add one object to the store, we are below the limit
+        with cache.store("o1") as rpath:
+            rpath_f1 = os.path.join(tmp_path, rpath, "f1")
+            with open(rpath_f1, "wb") as fp:
+                fp.write(b'a'*64*1024)
+        assert cache._calculate_space(tmp_path) > 64 * 1024
+        assert cache._calculate_space(tmp_path) < 128 * 1024
+        with cache.load("o1") as o:
+            assert o != ""
+        # and one more
+        with cache.store("o2") as rpath:
+            rpath_f2 = os.path.join(tmp_path, rpath, "f2")
+            with open(rpath_f2, "wb") as fp:
+                fp.write(b'b'*64*1024)
+        assert cache._calculate_space(tmp_path) > 128 * 1024
+        assert cache._calculate_space(tmp_path) < 192 * 1024
+        with cache.load("o2") as o:
+            assert o != ""
+        # adding a third one will (silently) fail because the cache is full
+        with cache.store("o3") as rpath:
+            rpath_f3 = os.path.join(tmp_path, rpath, "f3")
+            with open(rpath_f3, "wb") as fp:
+                fp.write(b'b'*128*1024)
+        assert cache._calculate_space(tmp_path) > 128 * 1024
+        assert cache._calculate_space(tmp_path) < 192 * 1024
+        with pytest.raises(fscache.FsCache.MissError):
+            with cache.load("o3") as o:
+                pass
diff --git a/test/mod/test_util_linux.py b/test/mod/test_util_linux.py
index 050d8513..b098bf66 100644
--- a/test/mod/test_util_linux.py
+++ b/test/mod/test_util_linux.py
@@ -2,6 +2,7 @@
 # Tests for the `osbuild.util.linux` module.
 #
 
+import ctypes
 import os
 import subprocess
 import tempfile
@@ -240,3 +241,24 @@ def test_proc_boot_id():
 
     bootid3 = linux.proc_boot_id("foobar")
     assert bootid.int != bootid3.int
+
+
+def test_libc_futimens_errcheck():
+    libc = linux.Libc.default()
+    with pytest.raises(OSError):
+        libc.futimens(-1, None)
+
+
+def test_libc_futimes_works(tmpdir):
+    libc = linux.Libc.default()
+    stamp_file = os.path.join(tmpdir, "foo")
+    with open(stamp_file, "w") as fp:
+        fp.write("meep")
+    mtime1 = os.stat(stamp_file).st_mtime
+    with open(stamp_file, "w") as fp:
+        libc.futimens(fp.fileno(), ctypes.byref(linux.c_timespec_times2(
+            atime=linux.c_timespec(tv_sec=3, tv_nsec=300*1000*1000),
+            mtime=linux.c_timespec(tv_sec=0, tv_nsec=libc.UTIME_OMIT),
+        )))
+    assert os.stat(stamp_file).st_atime == 3.3
+    assert os.stat(stamp_file).st_mtime == mtime1