test: convert objectstore test to pytest
Port the existing object store tests from `unittest` to `pytest`. Allow all tests that can run without root privileges to do so. No functional change of the test itself.
This commit is contained in:
Christian Kellner
Simon de Vlieger
parent
cb989f79b1
commit
fdd9e859dc
1 changed files with 193 additions and 183 deletions
|
|
@ -4,11 +4,11 @@
|
|||
|
||||
import contextlib
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from osbuild import objectstore
|
||||
|
||||
from .. import test
|
||||
|
|
@ -21,245 +21,255 @@ def store_path(store: objectstore.ObjectStore, ref: str, path: str) -> bool:
|
|||
return os.path.exists(os.path.join(obj, path))
|
||||
|
||||
|
||||
@unittest.skipUnless(test.TestBase.can_bind_mount(), "root-only")
|
||||
class TestObjectStore(unittest.TestCase):
|
||||
@pytest.fixture(name="tmpdir")
|
||||
def tmpdir_fixture():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
yield tmp
|
||||
|
||||
def setUp(self):
|
||||
self.store = tempfile.mkdtemp(prefix="osbuild-test-", dir="/var/tmp")
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.store)
|
||||
@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(self):
|
||||
# always use a temporary store so item counting works
|
||||
with objectstore.ObjectStore(self.store) as 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
|
||||
def test_basic(object_store):
|
||||
object_store.maximum_size = 1024*1024*1024
|
||||
|
||||
tree = object_store.new("a")
|
||||
# No objects or references should be in the store
|
||||
assert len(os.listdir(object_store.objects)) == 0
|
||||
|
||||
# new object should be in write mode
|
||||
assert tree.mode == objectstore.Object.Mode.WRITE
|
||||
tree = object_store.new("a")
|
||||
|
||||
p = Path(tree, "A")
|
||||
p.touch()
|
||||
# new object should be in write mode
|
||||
assert tree.mode == objectstore.Object.Mode.WRITE
|
||||
|
||||
tree.finalize() # put the object into READ mode
|
||||
assert tree.mode == objectstore.Object.Mode.READ
|
||||
p = Path(tree, "A")
|
||||
p.touch()
|
||||
|
||||
# commit makes a copy, if space
|
||||
object_store.commit(tree, "a")
|
||||
assert store_path(object_store, "a", "A")
|
||||
tree.finalize() # put the object into READ mode
|
||||
assert tree.mode == objectstore.Object.Mode.READ
|
||||
|
||||
# second object, based on the first one
|
||||
obj2 = object_store.new("b")
|
||||
obj2.init(tree)
|
||||
# commit makes a copy, if space
|
||||
object_store.commit(tree, "a")
|
||||
assert store_path(object_store, "a", "A")
|
||||
|
||||
p = Path(obj2, "B")
|
||||
p.touch()
|
||||
# second object, based on the first one
|
||||
obj2 = object_store.new("b")
|
||||
obj2.init(tree)
|
||||
|
||||
obj2.finalize() # put the object into READ mode
|
||||
assert obj2.mode == objectstore.Object.Mode.READ
|
||||
p = Path(obj2, "B")
|
||||
p.touch()
|
||||
|
||||
# commit always makes a copy, if space
|
||||
object_store.commit(tree, "b")
|
||||
obj2.finalize() # put the object into READ mode
|
||||
assert obj2.mode == objectstore.Object.Mode.READ
|
||||
|
||||
assert object_store.contains("b")
|
||||
assert store_path(object_store, "b", "A")
|
||||
assert store_path(object_store, "b", "B")
|
||||
# commit always makes a copy, if space
|
||||
object_store.commit(tree, "b")
|
||||
|
||||
assert len(os.listdir(object_store.objects)) == 2
|
||||
assert object_store.contains("b")
|
||||
assert store_path(object_store, "b", "A")
|
||||
assert store_path(object_store, "b", "B")
|
||||
|
||||
# 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
|
||||
assert len(os.listdir(object_store.objects)) == 2
|
||||
|
||||
def test_cleanup(self):
|
||||
# always use a temporary store so item counting works
|
||||
with objectstore.ObjectStore(self.store) as object_store:
|
||||
object_store.maximum_size = 1024*1024*1024
|
||||
# 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
|
||||
|
||||
stage = os.path.join(object_store, "stage")
|
||||
tree = object_store.new("a")
|
||||
self.assertEqual(len(os.listdir(stage)), 1)
|
||||
p = Path(tree, "A")
|
||||
p.touch()
|
||||
|
||||
# there should be no temporary Objects dirs anymore
|
||||
with objectstore.ObjectStore(self.store) as object_store:
|
||||
assert object_store.get("A") is None
|
||||
def test_cleanup(tmpdir):
|
||||
with objectstore.ObjectStore(tmpdir) as object_store:
|
||||
object_store.maximum_size = 1024*1024*1024
|
||||
|
||||
def test_metadata(self):
|
||||
stage = os.path.join(object_store, "stage")
|
||||
tree = object_store.new("a")
|
||||
assert len(os.listdir(stage)) == 1
|
||||
p = Path(tree, "A")
|
||||
p.touch()
|
||||
|
||||
# test metadata object directly first
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
md = objectstore.Object.Metadata(tmp)
|
||||
# there should be no temporary Objects dirs anymore
|
||||
with objectstore.ObjectStore(tmpdir) as object_store:
|
||||
assert object_store.get("A") is None
|
||||
|
||||
assert os.fspath(md) == tmp
|
||||
|
||||
with self.assertRaises(KeyError):
|
||||
with md.read("a"):
|
||||
pass
|
||||
def test_metadata(tmpdir):
|
||||
|
||||
# do not write anything to the file, it should not get stored
|
||||
with md.write("a"):
|
||||
# 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
|
||||
|
||||
assert len(list(os.scandir(tmp))) == 0
|
||||
# do not write anything to the file, it should not get stored
|
||||
with md.write("a"):
|
||||
pass
|
||||
|
||||
# also we should not write anything if an exception was raised
|
||||
with self.assertRaises(AssertionError):
|
||||
with md.write("a") as f:
|
||||
f.write("{}")
|
||||
raise AssertionError
|
||||
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
|
||||
|
||||
assert len(list(os.scandir(tmp))) == 1
|
||||
with md.write("a") as f:
|
||||
f.write("{}")
|
||||
|
||||
with md.read("a") as f:
|
||||
assert f.read() == "{}"
|
||||
assert len(list(os.scandir(tmp))) == 1
|
||||
|
||||
data = {
|
||||
"boolean": True,
|
||||
"number": 42,
|
||||
"string": "yes, please"
|
||||
}
|
||||
with md.read("a") as f:
|
||||
assert f.read() == "{}"
|
||||
|
||||
extra = {
|
||||
"extra": "data"
|
||||
}
|
||||
data = {
|
||||
"boolean": True,
|
||||
"number": 42,
|
||||
"string": "yes, please"
|
||||
}
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
md = objectstore.Object.Metadata(tmp)
|
||||
extra = {
|
||||
"extra": "data"
|
||||
}
|
||||
|
||||
d = md.get("a")
|
||||
assert d is None
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
md = objectstore.Object.Metadata(tmp)
|
||||
|
||||
md.set("a", None)
|
||||
with self.assertRaises(KeyError):
|
||||
with md.read("a"):
|
||||
pass
|
||||
d = md.get("a")
|
||||
assert d is None
|
||||
|
||||
md.set("a", data)
|
||||
assert md.get("a") == data
|
||||
md.set("a", None)
|
||||
with pytest.raises(KeyError):
|
||||
with md.read("a"):
|
||||
pass
|
||||
|
||||
with objectstore.ObjectStore(self.store) as store:
|
||||
store.maximum_size = 1024*1024*1024
|
||||
obj = store.new("a")
|
||||
p = Path(obj, "A")
|
||||
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")
|
||||
|
||||
obj.meta.set("md", data)
|
||||
assert obj.meta.get("md") == data
|
||||
# We cannot access the tree property after cleanup
|
||||
with pytest.raises(AssertionError):
|
||||
_ = host.tree
|
||||
|
||||
store.commit(obj, "x")
|
||||
obj.meta.set("extra", extra)
|
||||
assert obj.meta.get("extra") == extra
|
||||
|
||||
store.commit(obj, "a")
|
||||
# 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:
|
||||
|
||||
with objectstore.ObjectStore(self.store) as store:
|
||||
obj = store.get("a")
|
||||
store = objectstore.ObjectStore(tmpdir)
|
||||
stack.enter_context(store)
|
||||
|
||||
assert obj.meta.get("md") == data
|
||||
assert obj.meta.get("extra") == extra
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
tmp = stack.enter_context(tmp)
|
||||
|
||||
ext = store.get("x")
|
||||
server = objectstore.StoreServer(store)
|
||||
stack.enter_context(server)
|
||||
|
||||
assert ext.meta.get("md") == data
|
||||
assert ext.meta.get("extra") is None
|
||||
client = objectstore.StoreClient(server.socket_address)
|
||||
|
||||
def test_host_tree(self):
|
||||
with objectstore.ObjectStore(self.store) as store:
|
||||
host = store.host_tree
|
||||
have = client.source("org.osbuild.files")
|
||||
want = os.path.join(tmpdir, "sources")
|
||||
assert have.startswith(want)
|
||||
|
||||
assert host.tree
|
||||
assert os.fspath(host)
|
||||
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")
|
||||
|
||||
# check we actually cannot write to the path
|
||||
p = Path(host.tree, "osbuild-test-file")
|
||||
with self.assertRaises(OSError):
|
||||
p.touch()
|
||||
print("FOO")
|
||||
obj = store.new("42")
|
||||
p = Path(obj, "file.txt")
|
||||
p.write_text("osbuild")
|
||||
|
||||
# We cannot access the tree property after cleanup
|
||||
with self.assertRaises(AssertionError):
|
||||
_ = host.tree
|
||||
p = Path(obj, "directory")
|
||||
p.mkdir()
|
||||
obj.finalize()
|
||||
|
||||
# pylint: disable=too-many-statements
|
||||
def test_store_server(self):
|
||||
mountpoint = Path(tmpdir, "mountpoint")
|
||||
mountpoint.mkdir()
|
||||
|
||||
with contextlib.ExitStack() as stack:
|
||||
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"
|
||||
|
||||
store = objectstore.ObjectStore(self.store)
|
||||
stack.enter_context(store)
|
||||
# check we can mount subtrees via `read_tree_at`
|
||||
|
||||
tmpdir = tempfile.TemporaryDirectory()
|
||||
tmpdir = stack.enter_context(tmpdir)
|
||||
filemount = Path(tmpdir, "file")
|
||||
filemount.touch()
|
||||
|
||||
server = objectstore.StoreServer(store)
|
||||
stack.enter_context(server)
|
||||
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"
|
||||
|
||||
client = objectstore.StoreClient(server.socket_address)
|
||||
dirmount = Path(tmpdir, "dir")
|
||||
dirmount.mkdir()
|
||||
|
||||
have = client.source("org.osbuild.files")
|
||||
want = os.path.join(self.store, "sources")
|
||||
assert have.startswith(want)
|
||||
path = client.read_tree_at("42", dirmount, "/directory")
|
||||
dirpath = Path(path)
|
||||
assert dirpath.is_dir()
|
||||
|
||||
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")
|
||||
# check proper exceptions are raised for non existent
|
||||
# mount points and sub-trees
|
||||
|
||||
obj = store.new("42")
|
||||
p = Path(obj, "file.txt")
|
||||
p.write_text("osbuild")
|
||||
with pytest.raises(RuntimeError):
|
||||
nonexistent = os.path.join(tmpdir, "nonexistent")
|
||||
_ = client.read_tree_at("42", nonexistent)
|
||||
|
||||
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 self.assertRaises(RuntimeError):
|
||||
nonexistent = os.path.join(tmpdir, "nonexistent")
|
||||
_ = client.read_tree_at("42", nonexistent)
|
||||
|
||||
with self.assertRaises(RuntimeError):
|
||||
_ = client.read_tree_at("42", tmpdir, "/nonexistent")
|
||||
with pytest.raises(RuntimeError):
|
||||
_ = client.read_tree_at("42", tmpdir, "/nonexistent")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue