util/fscache: provide store_tree() helper

Add a helper that copies an entire directory tree including all metadata
into the cache. Use it in the ObjectStore to commit entries.

Unlike FsCache.store() this does not require entering the context from
the call-site. Instead, all data is directly passed to the cache and the
operation is under full control of the cache.

The ObjectStore is adjusted to make use of this. This requires exposing
the root-path (rather than the tree-path) to be accessible for
individual objects, hence a `path`-@property is added alongside the
`tree`-@property. Note that `__fspath__` still refers to the tree-path,
since this is the only path really required for outside access other
than from the object-manager itself.

Signed-off-by: David Rheinsberg <david.rheinsberg@gmail.com>
This commit is contained in:
David Rheinsberg 2022-12-15 12:38:16 +01:00
parent 50f8f6ac47
commit 8a9efa89fc
3 changed files with 109 additions and 27 deletions

View file

@ -290,6 +290,44 @@ def test_load(tmpdir):
pass
def test_store_tree(tmpdir):
#
# API tests for the `store_tree()` method.
#
cache = fscache.FsCache("osbuild-test-appid", tmpdir)
with pytest.raises(AssertionError):
cache.store_tree("foobar", "invalid/dir")
with cache:
cache.info = cache.info._replace(maximum_size=1024*1024*1024)
with pytest.raises(ValueError):
cache.store_tree("", "invalid/dir")
with pytest.raises(RuntimeError):
cache.store_tree("key", "invalid/dir")
with tempfile.TemporaryDirectory(dir="/var/tmp") as tmp:
with open(os.path.join(tmp, "outside"), "x", encoding="utf8") as f:
f.write("foo")
os.mkdir(os.path.join(tmp, "tree"))
with open(os.path.join(tmp, "tree", "inside"), "x", encoding="utf8") as f:
f.write("bar")
with open(os.path.join(tmp, "tree", "more-inside"), "x", encoding="utf8") as f:
f.write("foobar")
cache.store_tree("key", os.path.join(tmp, "tree"))
with cache.load("key") as rpath:
assert len(list(os.scandir(os.path.join(cache, rpath)))) == 1
assert len(list(os.scandir(os.path.join(cache, rpath, "tree")))) == 2
with open(os.path.join(cache, rpath, "tree", "inside"), "r", encoding="utf8") as f:
assert f.read() == "bar"
with open(os.path.join(cache, rpath, "tree", "more-inside"), "r", encoding="utf8") as f:
assert f.read() == "foobar"
def test_basic(tmpdir):
#
# A basic cache store+load test.