diff --git a/src/borg/testsuite/helpers/misc_test.py b/src/borg/testsuite/helpers/misc_test.py new file mode 100644 index 000000000..7fa28a6ff --- /dev/null +++ b/src/borg/testsuite/helpers/misc_test.py @@ -0,0 +1,52 @@ +from io import StringIO, BytesIO + +import pytest + +from ...helpers.misc import ChunkIteratorFileWrapper, chunkit, iter_separated + + +def test_chunk_file_wrapper(): + cfw = ChunkIteratorFileWrapper(iter([b"abc", b"def"])) + assert cfw.read(2) == b"ab" + assert cfw.read(50) == b"cdef" + assert cfw.exhausted + + cfw = ChunkIteratorFileWrapper(iter([])) + assert cfw.read(2) == b"" + assert cfw.exhausted + + +def test_chunkit(): + it = chunkit("abcdefg", 3) + assert next(it) == ["a", "b", "c"] + assert next(it) == ["d", "e", "f"] + assert next(it) == ["g"] + with pytest.raises(StopIteration): + next(it) + with pytest.raises(StopIteration): + next(it) + + it = chunkit("ab", 3) + assert list(it) == [["a", "b"]] + + it = chunkit("", 3) + assert list(it) == [] + + +def test_iter_separated(): + # newline and utf-8 + sep, items = "\n", ["foo", "bar/baz", "αáčő"] + fd = StringIO(sep.join(items)) + assert list(iter_separated(fd)) == items + # null and bogus ending + sep, items = "\0", ["foo/bar", "baz", "spam"] + fd = StringIO(sep.join(items) + "\0") + assert list(iter_separated(fd, sep=sep)) == ["foo/bar", "baz", "spam"] + # multichar + sep, items = "SEP", ["foo/bar", "baz", "spam"] + fd = StringIO(sep.join(items)) + assert list(iter_separated(fd, sep=sep)) == items + # bytes + sep, items = b"\n", [b"foo", b"blop\t", b"gr\xe4ezi"] + fd = BytesIO(sep.join(items)) + assert list(iter_separated(fd)) == items diff --git a/src/borg/testsuite/helpers_test.py b/src/borg/testsuite/helpers_test.py index 9d82ae98c..e9ab224d6 100644 --- a/src/borg/testsuite/helpers_test.py +++ b/src/borg/testsuite/helpers_test.py @@ -1,14 +1,11 @@ from argparse import ArgumentTypeError from datetime import datetime, timezone -from io import StringIO, BytesIO import pytest from ..archiver.prune_cmd import prune_split from ..constants import * # NOQA -from ..helpers import ChunkIteratorFileWrapper, ChunkerParams -from ..helpers import chunkit -from ..helpers import iter_separated +from ..helpers import ChunkerParams from ..helpers import classify_ec, max_ec @@ -151,53 +148,6 @@ def test_prune_split_no_archives(): assert kept_because == {} -def test_chunk_file_wrapper(): - cfw = ChunkIteratorFileWrapper(iter([b"abc", b"def"])) - assert cfw.read(2) == b"ab" - assert cfw.read(50) == b"cdef" - assert cfw.exhausted - - cfw = ChunkIteratorFileWrapper(iter([])) - assert cfw.read(2) == b"" - assert cfw.exhausted - - -def test_chunkit(): - it = chunkit("abcdefg", 3) - assert next(it) == ["a", "b", "c"] - assert next(it) == ["d", "e", "f"] - assert next(it) == ["g"] - with pytest.raises(StopIteration): - next(it) - with pytest.raises(StopIteration): - next(it) - - it = chunkit("ab", 3) - assert list(it) == [["a", "b"]] - - it = chunkit("", 3) - assert list(it) == [] - - -def test_iter_separated(): - # newline and utf-8 - sep, items = "\n", ["foo", "bar/baz", "αáčő"] - fd = StringIO(sep.join(items)) - assert list(iter_separated(fd)) == items - # null and bogus ending - sep, items = "\0", ["foo/bar", "baz", "spam"] - fd = StringIO(sep.join(items) + "\0") - assert list(iter_separated(fd, sep=sep)) == ["foo/bar", "baz", "spam"] - # multichar - sep, items = "SEP", ["foo/bar", "baz", "spam"] - fd = StringIO(sep.join(items)) - assert list(iter_separated(fd, sep=sep)) == items - # bytes - sep, items = b"\n", [b"foo", b"blop\t", b"gr\xe4ezi"] - fd = BytesIO(sep.join(items)) - assert list(iter_separated(fd)) == items - - @pytest.mark.parametrize( "ec_range,ec_class", (