util: tweak _calculate_size() to _calculate_space()

Update the naming, docstring and tweak the tests.

Thanks to bcl and dustymabe!
This commit is contained in:
Michael Vogt 2023-11-20 11:26:00 +01:00 committed by Simon de Vlieger
parent 9121360f7b
commit 4b69d2e1c4
2 changed files with 29 additions and 14 deletions

View file

@ -272,25 +272,28 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
self._tracers[trace]()
@staticmethod
def _calculate_size(path_target: str) -> int:
"""Calculate total size of a directory tree
def _calculate_space(path_target: str) -> int:
"""Calculate total space of a directory tree
Calculate the total amount of storage required for a directory tree in
bytes. This does not account for metadata, but only for stored file
content.
Note that this may differ from the sum of the file sizes as it
takes sparse files into account.
Parameters:
-----------
path_target
File-system path to the directory to operate on.
"""
return sum(
return os.lstat(path_target).st_blocks * 512 + sum(
os.lstat(
os.path.join(path, f)
).st_blocks * 512 for path, dirs, files in os.walk(
path_target
) for f in files
) for f in files + dirs
)
def __fspath__(self) -> Any:
@ -925,7 +928,7 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
# Collect metadata about the new entry.
info: Dict[str, Any] = {}
info["creation-boot-id"] = self._bootid
info["size"] = self._calculate_size(path_data)
info["size"] = self._calculate_space(path_data)
# Update the total cache-size. If it exceeds the limits, bail out
# but do not trigger an error. It behaves as if the entry was

View file

@ -6,6 +6,7 @@
import json
import os
import subprocess
import tempfile
import pytest
@ -19,25 +20,36 @@ def tmpdir_fixture():
yield tmp
def test_calculate_size(tmpdir):
def test_calculate_space(tmpdir):
#
# Test the `_calculate_size()` helper and verify it only includes file
# Test the `_calculate_space()` helper and verify it only includes file
# content in its calculation.
#
def du(path_target):
env = os.environ.copy()
env["POSIXLY_CORRECT"] = "1"
output = subprocess.check_output(["du", "-s", path_target], env=env, encoding="utf8")
return int(output.split()[0].strip())*512
os.mkdir(os.path.join(tmpdir, "dir"))
assert fscache.FsCache._calculate_size(os.path.join(tmpdir, "dir")) == 0
test_dir = os.path.join(tmpdir, "dir")
os.mkdir(test_dir)
assert fscache.FsCache._calculate_space(test_dir) == du(test_dir)
with open(os.path.join(tmpdir, "dir", "file"), "x", encoding="utf8") as f:
pass
assert fscache.FsCache._calculate_size(os.path.join(tmpdir, "dir")) == 0
assert fscache.FsCache._calculate_space(test_dir) == du(test_dir)
with open(os.path.join(tmpdir, "dir", "file"), "w", encoding="utf8") as f:
f.write("foobar")
assert fscache.FsCache._calculate_space(test_dir) == du(test_dir)
assert fscache.FsCache._calculate_size(os.path.join(tmpdir, "dir")) == 6
os.makedirs(os.path.join(test_dir, "dir"))
assert fscache.FsCache._calculate_space(test_dir) == du(test_dir)
with open(os.path.join(test_dir, "sparse-file"), "w") as f:
f.truncate(10*1024*1024)
f.write("I'm not an empty file")
assert fscache.FsCache._calculate_space(test_dir) == du(test_dir)
def test_pathlike(tmpdir):
@ -351,7 +363,7 @@ def test_basic(tmpdir):
cache = fscache.FsCache("osbuild-test-appid", tmpdir)
with cache:
cache.info = cache.info._replace(maximum_size=1024)
cache.info = cache.info._replace(maximum_size=1024*1024)
with cache.stage() as rpath:
with open(os.path.join(tmpdir, rpath, "bar"), "x", encoding="utf8") as f: