diff --git a/test/mod/test_objectstore.py b/test/mod/test_objectstore.py index c68e7ce0..0cbedd2f 100644 --- a/test/mod/test_objectstore.py +++ b/test/mod/test_objectstore.py @@ -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")