d466d5d66a
Instead of using `Path.stat` use `os.stat` since the former only gained the `follow_symlinks` argument in 3.10 but we still need to support Python 3.6 for RHEL 7 and 8. Additionally, reduce the precision by converting timestamps to an integer to avoid false negatives due to floating point arithmetic.
328 lines
8.3 KiB
Python
328 lines
8.3 KiB
Python
#
|
|
# Tests for the 'osbuild.objectstore' module.
|
|
#
|
|
|
|
import contextlib
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from osbuild import objectstore
|
|
|
|
from .. import test
|
|
|
|
|
|
def store_path(store: objectstore.ObjectStore, ref: str, path: str) -> bool:
|
|
obj = store.get(ref)
|
|
if not obj:
|
|
return False
|
|
return os.path.exists(os.path.join(obj, path))
|
|
|
|
|
|
@pytest.fixture(name="tmpdir")
|
|
def tmpdir_fixture():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
yield tmp
|
|
|
|
|
|
@pytest.fixture(name="object_store")
|
|
def store_fixture():
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="osbuild-test-",
|
|
dir="/var/tmp",
|
|
) as tmp:
|
|
with objectstore.ObjectStore(tmp) as store:
|
|
yield store
|
|
|
|
|
|
def test_basic(object_store):
|
|
object_store.maximum_size = 1024*1024*1024
|
|
|
|
# No objects or references should be in the store
|
|
assert len(os.listdir(object_store.objects)) == 0
|
|
|
|
tree = object_store.new("a")
|
|
|
|
# new object should be in write mode
|
|
assert tree.mode == objectstore.Object.Mode.WRITE
|
|
|
|
p = Path(tree, "A")
|
|
p.touch()
|
|
|
|
tree.finalize() # put the object into READ mode
|
|
assert tree.mode == objectstore.Object.Mode.READ
|
|
|
|
# commit makes a copy, if space
|
|
object_store.commit(tree, "a")
|
|
assert store_path(object_store, "a", "A")
|
|
|
|
# second object, based on the first one
|
|
obj2 = object_store.new("b")
|
|
obj2.init(tree)
|
|
|
|
p = Path(obj2, "B")
|
|
p.touch()
|
|
|
|
obj2.finalize() # put the object into READ mode
|
|
assert obj2.mode == objectstore.Object.Mode.READ
|
|
|
|
# commit always makes a copy, if space
|
|
object_store.commit(tree, "b")
|
|
|
|
assert object_store.contains("b")
|
|
assert store_path(object_store, "b", "A")
|
|
assert store_path(object_store, "b", "B")
|
|
|
|
assert len(os.listdir(object_store.objects)) == 2
|
|
|
|
# object should exist and should be in read mode
|
|
tree = object_store.get("b")
|
|
assert tree is not None
|
|
assert tree.mode == objectstore.Object.Mode.READ
|
|
|
|
|
|
def test_cleanup(tmpdir):
|
|
with objectstore.ObjectStore(tmpdir) as object_store:
|
|
object_store.maximum_size = 1024*1024*1024
|
|
|
|
stage = os.path.join(object_store, "stage")
|
|
tree = object_store.new("a")
|
|
assert len(os.listdir(stage)) == 1
|
|
p = Path(tree, "A")
|
|
p.touch()
|
|
|
|
# there should be no temporary Objects dirs anymore
|
|
with objectstore.ObjectStore(tmpdir) as object_store:
|
|
assert object_store.get("A") is None
|
|
|
|
|
|
def test_metadata(tmpdir):
|
|
|
|
# test metadata object directly first
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
md = objectstore.Object.Metadata(tmp)
|
|
|
|
assert os.fspath(md) == tmp
|
|
|
|
with pytest.raises(KeyError):
|
|
with md.read("a"):
|
|
pass
|
|
|
|
# do not write anything to the file, it should not get stored
|
|
with md.write("a"):
|
|
pass
|
|
|
|
assert len(list(os.scandir(tmp))) == 0
|
|
|
|
# also we should not write anything if an exception was raised
|
|
with pytest.raises(AssertionError):
|
|
with md.write("a") as f:
|
|
f.write("{}")
|
|
raise AssertionError
|
|
|
|
with md.write("a") as f:
|
|
f.write("{}")
|
|
|
|
assert len(list(os.scandir(tmp))) == 1
|
|
|
|
with md.read("a") as f:
|
|
assert f.read() == "{}"
|
|
|
|
data = {
|
|
"boolean": True,
|
|
"number": 42,
|
|
"string": "yes, please"
|
|
}
|
|
|
|
extra = {
|
|
"extra": "data"
|
|
}
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
md = objectstore.Object.Metadata(tmp)
|
|
|
|
d = md.get("a")
|
|
assert d is None
|
|
|
|
md.set("a", None)
|
|
with pytest.raises(KeyError):
|
|
with md.read("a"):
|
|
pass
|
|
|
|
md.set("a", data)
|
|
assert md.get("a") == data
|
|
|
|
# use tmpdir fixture from here on
|
|
with objectstore.ObjectStore(tmpdir) as store:
|
|
store.maximum_size = 1024*1024*1024
|
|
obj = store.new("a")
|
|
p = Path(obj, "A")
|
|
p.touch()
|
|
|
|
obj.meta.set("md", data)
|
|
assert obj.meta.get("md") == data
|
|
|
|
store.commit(obj, "x")
|
|
obj.meta.set("extra", extra)
|
|
assert obj.meta.get("extra") == extra
|
|
|
|
store.commit(obj, "a")
|
|
|
|
with objectstore.ObjectStore(tmpdir) as store:
|
|
obj = store.get("a")
|
|
|
|
assert obj.meta.get("md") == data
|
|
assert obj.meta.get("extra") == extra
|
|
|
|
ext = store.get("x")
|
|
|
|
assert ext.meta.get("md") == data
|
|
assert ext.meta.get("extra") is None
|
|
|
|
|
|
@pytest.mark.skipif(not test.TestBase.can_bind_mount(), reason="Need root for bind mount")
|
|
def test_host_tree(tmpdir):
|
|
with objectstore.ObjectStore(tmpdir) as store:
|
|
host = store.host_tree
|
|
|
|
assert host.tree
|
|
assert os.fspath(host)
|
|
|
|
# check we actually cannot write to the path
|
|
p = Path(host.tree, "osbuild-test-file")
|
|
with pytest.raises(OSError):
|
|
p.touch()
|
|
print("FOO")
|
|
|
|
# We cannot access the tree property after cleanup
|
|
with pytest.raises(AssertionError):
|
|
_ = host.tree
|
|
|
|
|
|
def test_source_epoch(object_store):
|
|
tree = object_store.new("a")
|
|
tree.source_epoch = 946688461
|
|
|
|
t = Path(tree)
|
|
|
|
a = Path(tree, "A")
|
|
a.touch()
|
|
|
|
d = Path(tree, "d")
|
|
d.mkdir()
|
|
|
|
l = Path(tree, "l")
|
|
l.symlink_to(d, target_is_directory=True)
|
|
|
|
for i in (a, d, l, t):
|
|
si = os.stat(i, follow_symlinks=False)
|
|
before = int(si.st_mtime)
|
|
assert before != tree.source_epoch
|
|
|
|
# FINALIZING
|
|
tree.finalize()
|
|
|
|
for i in (a, d, l, t):
|
|
si = os.stat(i, follow_symlinks=False)
|
|
after = int(si.st_mtime)
|
|
|
|
assert after != before, f"mtime not changed for {i}"
|
|
assert after == tree.source_epoch
|
|
|
|
baum = object_store.new("b")
|
|
baum.init(tree)
|
|
|
|
assert baum.created == tree.created
|
|
|
|
b = Path(tree, "B")
|
|
b.touch()
|
|
|
|
si = os.stat(b, follow_symlinks=False)
|
|
before = int(si.st_mtime)
|
|
|
|
assert before != tree.source_epoch
|
|
|
|
# FINALIZING
|
|
baum.finalize()
|
|
si = os.stat(a, follow_symlinks=False)
|
|
after = int(si.st_mtime)
|
|
|
|
assert after != before
|
|
assert after == tree.source_epoch
|
|
|
|
# pylint: disable=too-many-statements
|
|
|
|
|
|
@pytest.mark.skipif(not test.TestBase.can_bind_mount(), reason="Need root for bind mount")
|
|
def test_store_server(tmpdir):
|
|
with contextlib.ExitStack() as stack:
|
|
|
|
store = objectstore.ObjectStore(tmpdir)
|
|
stack.enter_context(store)
|
|
|
|
tmp = tempfile.TemporaryDirectory()
|
|
tmp = stack.enter_context(tmp)
|
|
|
|
server = objectstore.StoreServer(store)
|
|
stack.enter_context(server)
|
|
|
|
client = objectstore.StoreClient(server.socket_address)
|
|
|
|
have = client.source("org.osbuild.files")
|
|
want = os.path.join(tmpdir, "sources")
|
|
assert have.startswith(want)
|
|
|
|
tmp = client.mkdtemp(suffix="suffix", prefix="prefix")
|
|
assert tmp.startswith(store.tmp)
|
|
name = os.path.basename(tmp)
|
|
assert name.startswith("prefix")
|
|
assert name.endswith("suffix")
|
|
|
|
obj = store.new("42")
|
|
p = Path(obj, "file.txt")
|
|
p.write_text("osbuild")
|
|
|
|
p = Path(obj, "directory")
|
|
p.mkdir()
|
|
obj.finalize()
|
|
|
|
mountpoint = Path(tmpdir, "mountpoint")
|
|
mountpoint.mkdir()
|
|
|
|
assert store.contains("42")
|
|
path = client.read_tree_at("42", mountpoint)
|
|
assert Path(path) == mountpoint
|
|
filepath = Path(mountpoint, "file.txt")
|
|
assert filepath.exists()
|
|
txt = filepath.read_text(encoding="utf8")
|
|
assert txt == "osbuild"
|
|
|
|
# check we can mount subtrees via `read_tree_at`
|
|
|
|
filemount = Path(tmpdir, "file")
|
|
filemount.touch()
|
|
|
|
path = client.read_tree_at("42", filemount, "/file.txt")
|
|
filepath = Path(path)
|
|
assert filepath.is_file()
|
|
txt = filepath.read_text(encoding="utf8")
|
|
assert txt == "osbuild"
|
|
|
|
dirmount = Path(tmpdir, "dir")
|
|
dirmount.mkdir()
|
|
|
|
path = client.read_tree_at("42", dirmount, "/directory")
|
|
dirpath = Path(path)
|
|
assert dirpath.is_dir()
|
|
|
|
# check proper exceptions are raised for non existent
|
|
# mount points and sub-trees
|
|
|
|
with pytest.raises(RuntimeError):
|
|
nonexistent = os.path.join(tmpdir, "nonexistent")
|
|
_ = client.read_tree_at("42", nonexistent)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
_ = client.read_tree_at("42", tmpdir, "/nonexistent")
|