From a918deefa90ccf1ab58530cbced20473ad37c04a Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 15 May 2026 06:22:25 +0530 Subject: [PATCH 1/3] testsuite: add LegacyArchives unit tests, refs #9556 --- src/borg/testsuite/legacy_archives_test.py | 471 +++++++++++++++++++++ 1 file changed, 471 insertions(+) create mode 100644 src/borg/testsuite/legacy_archives_test.py diff --git a/src/borg/testsuite/legacy_archives_test.py b/src/borg/testsuite/legacy_archives_test.py new file mode 100644 index 0000000000..c17be15740 --- /dev/null +++ b/src/borg/testsuite/legacy_archives_test.py @@ -0,0 +1,471 @@ +"""Tests for borg.legacy.archives (LegacyArchives).""" + +from argparse import Namespace +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from ..crypto.key import PlaintextKey +from ..helpers.errors import CommandError, Error +from ..legacy.archives import LegacyArchives +from ..legacy.repository import LegacyRepository +from ..manifest import ArchiveInfo, ArchivesInterface, Manifest + + +# ── helpers ────────────────────────────────────────────────────────────────────── + + +def _id(n): + return bytes([n]) * 32 + + +TS = "2020-06-01T12:00:00.000000" +TS2 = "2021-06-01T12:00:00.000000" + + +def _make(entries=()): + """Return LegacyArchives with minimal mocks; entries = [(name, id, ts_str), ...].""" + repo = MagicMock() + manifest = MagicMock() + la = LegacyArchives(repo, manifest) + for name, id_, ts in entries: + la._archives[name] = {"id": id_, "time": ts} + return la, repo, manifest + + +def _archive_meta(name, id_, ts=TS, *, username="", hostname="", tags=()): + return { + "id": id_, + "name": name, + "time": ts, + "exists": True, + "username": username, + "hostname": hostname, + "size": 0, + "nfiles": 0, + "comment": "", + "tags": tags, + } + + +def _info(name, id_, ts=TS, *, username="", hostname="", tags=()): + from ..helpers.time import parse_timestamp + + return ArchiveInfo(name=name, id=id_, ts=parse_timestamp(ts), tags=tags, user=username, host=hostname) + + +def _make_list_target(infos): + """LegacyArchives with _info_tuples replaced so callers get controlled data.""" + la, repo, manifest = _make([(i.name, i.id, TS) for i in infos]) + la._info_tuples = lambda deleted=False: iter(infos) + return la + + +# ── init / raw-dict operations ─────────────────────────────────────────────────── + + +def test_init(): + la, repo, manifest = _make() + assert la._archives == {} + assert la.repository is repo + assert la.manifest is manifest + + +def test_set_raw_dict_and_get_raw_dict(): + la, _, _ = _make() + d = {"a": {"id": _id(1), "time": TS}} + la._set_raw_dict(d) + assert la._get_raw_dict() == d + + +def test_prepare(): + la, repo, manifest = _make() + m = MagicMock() + m.archives = {"x": {"id": _id(5), "time": TS}} + la.prepare(manifest, m) + assert la._archives == {"x": {"id": _id(5), "time": TS}} + + +def test_finish(): + la, _, manifest = _make([("a", _id(1), TS)]) + result = la.finish(manifest) + assert result == {"a": {"id": _id(1), "time": TS}} + + +def test_ids(): + la, _, _ = _make([("a", _id(1), TS), ("b", _id(2), TS)]) + assert list(la.ids()) == [_id(1), _id(2)] + + +def test_count(): + la, _, _ = _make([("a", _id(1), TS), ("b", _id(2), TS)]) + assert la.count() == 2 + + +def test_names(): + la, _, _ = _make([("a", _id(1), TS), ("b", _id(2), TS)]) + assert list(la.names()) == ["a", "b"] + + +def test_exists_true(): + la, _, _ = _make([("a", _id(1), TS)]) + assert la.exists("a") is True + + +def test_exists_false(): + la, _, _ = _make() + assert la.exists("missing") is False + + +# ── create ─────────────────────────────────────────────────────────────────────── + + +def test_create_with_str_ts(): + la, _, _ = _make() + la.create("a", _id(1), TS) + assert la._archives["a"] == {"id": _id(1), "time": TS} + + +def test_create_with_datetime_ts(): + la, _, _ = _make() + dt = datetime(2020, 6, 1, 12, 0, 0, tzinfo=timezone.utc) + la.create("a", _id(1), dt) + assert la._archives["a"]["time"] == dt.isoformat(timespec="microseconds") + + +def test_create_raises_if_exists(): + la, _, _ = _make([("a", _id(1), TS)]) + with pytest.raises(KeyError, match="already exists"): + la.create("a", _id(2), TS) + + +def test_create_overwrite(): + la, _, _ = _make([("a", _id(1), TS)]) + la.create("a", _id(2), TS, overwrite=True) + assert la._archives["a"]["id"] == _id(2) + + +# ── get / get_by_id ─────────────────────────────────────────────────────────────── + + +def test_get_missing_returns_none(): + la, _, _ = _make() + assert la.get("nope") is None + + +def test_get_returns_archive_info(): + la, _, _ = _make([("a", _id(1), TS)]) + info = la.get("a") + assert isinstance(info, ArchiveInfo) + assert info.name == "a" + assert info.id == _id(1) + + +def test_get_raw(): + la, _, _ = _make([("a", _id(1), TS)]) + result = la.get("a", raw=True) + assert result == {"name": "a", "id": _id(1), "time": TS} + + +def test_get_by_id_missing_returns_none(): + la, _, _ = _make() + assert la.get_by_id(_id(99)) is None + + +def test_get_by_id_returns_archive_info(): + la, _, _ = _make([("a", _id(1), TS)]) + info = la.get_by_id(_id(1)) + assert isinstance(info, ArchiveInfo) + assert info.name == "a" + + +def test_get_by_id_raw(): + la, _, _ = _make([("a", _id(1), TS)]) + result = la.get_by_id(_id(1), raw=True) + assert result == {"name": "a", "id": _id(1), "time": TS} + + +# ── NotImplementedError stubs ────────────────────────────────────────────────────── + + +def test_exists_id_not_implemented(): + la, _, _ = _make() + with pytest.raises(NotImplementedError): + la.exists_id(_id(1)) + + +def test_exists_name_and_id_not_implemented(): + la, _, _ = _make() + with pytest.raises(NotImplementedError): + la.exists_name_and_id("a", _id(1)) + + +def test_exists_name_and_ts_not_implemented(): + la, _, _ = _make() + with pytest.raises(NotImplementedError): + la.exists_name_and_ts("a", datetime.now()) + + +def test_delete_by_id_not_implemented(): + la, _, _ = _make() + with pytest.raises(NotImplementedError): + la.delete_by_id(_id(1)) + + +def test_undelete_by_id_not_implemented(): + la, _, _ = _make() + with pytest.raises(NotImplementedError): + la.undelete_by_id(_id(1)) + + +def test_nuke_by_id_not_implemented(): + la, _, _ = _make() + with pytest.raises(NotImplementedError): + la.nuke_by_id(_id(1)) + + +# ── _get_archive_meta ──────────────────────────────────────────────────────────── + + +def test_get_archive_meta_object_not_found(): + la, repo, _ = _make() + repo.get.side_effect = LegacyRepository.ObjectNotFound(_id(1), "/fake/path") + result = la._get_archive_meta(_id(1)) + assert result["exists"] is False + assert result["name"] == "archive-does-not-exist" + assert result["id"] == _id(1) + assert result["tags"] == () + + +def test_get_archive_meta_success(): + la, _, manifest = _make() + manifest.repo_objs.parse.return_value = (None, b"data") + manifest.key.unpack_archive.return_value = {} + + with patch("borg.legacy.archives.ArchiveItem") as mock_ai: + item = MagicMock() + item.version = 2 + item.name = "myarchive" + item.time = "2021-03-15T10:00:00.000000" + item.username = "alice" + item.hostname = "myhost" + item.get.side_effect = lambda k, d=None: d + mock_ai.return_value = item + + result = la._get_archive_meta(_id(1)) + + assert result["exists"] is True + assert result["name"] == "myarchive" + assert result["username"] == "alice" + assert result["hostname"] == "myhost" + + +def test_get_archive_meta_bad_version(): + la, _, manifest = _make() + manifest.repo_objs.parse.return_value = (None, b"data") + manifest.key.unpack_archive.return_value = {} + + with patch("borg.legacy.archives.ArchiveItem") as mock_ai: + item = MagicMock() + item.version = 99 + mock_ai.return_value = item + + with pytest.raises(Exception, match="Unknown archive metadata version"): + la._get_archive_meta(_id(1)) + + +# ── _infos / _info_tuples ──────────────────────────────────────────────────────── + + +def test_infos_and_info_tuples(): + la, _, _ = _make([("a", _id(1), TS)]) + la._get_archive_meta = lambda id_: _archive_meta("a", _id(1)) + infos = list(la._infos()) + assert len(infos) == 1 + assert infos[0]["name"] == "a" + tuples = list(la._info_tuples()) + assert len(tuples) == 1 + assert isinstance(tuples[0], ArchiveInfo) + assert tuples[0].name == "a" + + +# ── list ────────────────────────────────────────────────────────────────────────── + + +def test_list_no_filters(): + info = _info("a", _id(1)) + la = _make_list_target([info]) + assert la.list() == [info] + + +def test_list_sort_by_str_raises(): + la = _make_list_target([_info("a", _id(1))]) + with pytest.raises(TypeError, match="sequence"): + la.list(sort_by="name") + + +def test_list_sort_by(): + i1 = _info("b", _id(2), TS2) + i2 = _info("a", _id(1), TS) + la = _make_list_target([i1, i2]) + result = la.list(sort_by=["name"]) + assert result == [i2, i1] + + +def test_list_reverse(): + i1 = _info("a", _id(1)) + i2 = _info("b", _id(2)) + la = _make_list_target([i1, i2]) + assert la.list(reverse=True) == [i2, i1] + + +def test_list_first(): + infos = [_info(f"a{i}", _id(i + 1)) for i in range(5)] + la = _make_list_target(infos) + assert la.list(first=3) == infos[:3] + + +def test_list_last(): + infos = [_info(f"a{i}", _id(i + 1)) for i in range(5)] + la = _make_list_target(infos) + assert la.list(last=2) == infos[-2:] + + +def test_list_date_filter(): + i1 = _info("a", _id(1)) + la = _make_list_target([i1]) + with patch("borg.legacy.archives.filter_archives_by_date", return_value=[i1]) as mock_filter: + result = la.list(older="1d") + assert result == [i1] + assert mock_filter.called + + +def test_list_match_name(): + i1 = _info("archive-a", _id(1)) + i2 = _info("archive-b", _id(2)) + la = _make_list_target([i1, i2]) + result = la.list(match=["archive-a"]) + assert result == [i1] + + +def test_list_match_name_prefix(): + i1 = _info("archive-a", _id(1)) + i2 = _info("other", _id(2)) + la = _make_list_target([i1, i2]) + result = la.list(match=["name:archive-a"]) + assert result == [i1] + + +def test_list_match_user(): + i1 = _info("a", _id(1), username="alice") + i2 = _info("b", _id(2), username="bob") + la = _make_list_target([i1, i2]) + assert la.list(match=["user:alice"]) == [i1] + + +def test_list_match_host(): + i1 = _info("a", _id(1), hostname="laptop") + i2 = _info("b", _id(2), hostname="server") + la = _make_list_target([i1, i2]) + assert la.list(match=["host:laptop"]) == [i1] + + +def test_list_match_tags(): + i1 = _info("a", _id(1), tags=("prod", "db")) + i2 = _info("b", _id(2), tags=("dev",)) + la = _make_list_target([i1, i2]) + assert la.list(match=["tags:prod"]) == [i1] + + +def test_list_match_aid(): + from ..helpers.parseformat import bin_to_hex + + i1 = _info("a", _id(1)) + la = _make_list_target([i1]) + prefix = bin_to_hex(_id(1))[:4] + assert la.list(match=[f"aid:{prefix}"]) == [i1] + + +def test_list_match_aid_ambiguous(): + from ..helpers.parseformat import bin_to_hex + + i1 = _info("a", _id(1)) + i2 = _info("b", _id(1)) + la = _make_list_target([i1, i2]) + prefix = bin_to_hex(_id(1))[:4] + with pytest.raises(CommandError): + la.list(match=[f"aid:{prefix}"]) + + +# ── get_one ─────────────────────────────────────────────────────────────────────── + + +def test_get_one_exact_match(): + i1 = _info("backup", _id(1)) + la = _make_list_target([i1]) + assert la.get_one(["backup"]) == i1 + + +def test_get_one_no_match_raises(): + la = _make_list_target([]) + with pytest.raises(CommandError, match="matched 0"): + la.get_one(["missing"]) + + +def test_get_one_multiple_matches_raises(): + i1 = _info("a", _id(1)) + i2 = _info("a", _id(2)) + la = _make_list_target([i1, i2]) + with pytest.raises(CommandError, match="matched 2"): + la.get_one(["a"]) + + +# ── list_considering ────────────────────────────────────────────────────────────── + + +def test_list_considering_raises_if_name_set(): + la, _, _ = _make() + args = MagicMock() + args.name = "archive" + with pytest.raises(Error): + la.list_considering(args) + + +def test_list_considering_delegates(): + i1 = _info("a", _id(1)) + la = _make_list_target([i1]) + args = Namespace( + name=None, + sort_by="name", + match_archives=None, + first=None, + last=None, + older=None, + newer=None, + oldest=None, + newest=None, + deleted=False, + ) + result = la.list_considering(args) + assert result == [i1] + + +# ── ArchivesInterface Protocol / Manifest dispatch ──────────────────────────────── + + +def test_legacy_archives_satisfies_archives_interface(): + la, _, _ = _make() + assert isinstance(la, ArchivesInterface) + + +class _FakeLegacyRepo(LegacyRepository): + def __init__(self): + pass + + +def test_manifest_creates_legacy_archives_for_legacy_repo(): + repo = _FakeLegacyRepo() + key = PlaintextKey(repo) + manifest = Manifest(key, repo) + assert isinstance(manifest.archives, LegacyArchives) From fdc00d9e8e54147d39ff29596fc3adacbd48d190 Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 15 May 2026 22:35:09 +0530 Subject: [PATCH 2/3] testsuite: add Archives unit tests, refs #9556 --- src/borg/testsuite/archives_test.py | 611 ++++++++++++++++++++++++++++ 1 file changed, 611 insertions(+) create mode 100644 src/borg/testsuite/archives_test.py diff --git a/src/borg/testsuite/archives_test.py b/src/borg/testsuite/archives_test.py new file mode 100644 index 0000000000..de8045d92d --- /dev/null +++ b/src/borg/testsuite/archives_test.py @@ -0,0 +1,611 @@ +from argparse import Namespace +from datetime import datetime, timezone +from unittest.mock import Mock, patch + +import pytest + +from borgstore.store import ItemInfo, ObjectNotFound as StoreObjectNotFound + +from ..helpers.errors import CommandError, Error +from ..helpers.parseformat import bin_to_hex +from ..helpers.time import parse_timestamp +from ..manifest import Archives, ArchiveInfo, ArchivesInterface +from ..repository import Repository + + +def _id(n): + return bytes([n]) * 32 + + +TS = "2020-06-01T12:00:00.000000" +TS2 = "2021-06-01T12:00:00.000000" + + +def _item(id_bytes): + return ItemInfo(name=bin_to_hex(id_bytes), exists=True, size=0, directory=False) + + +def _make(): + repo = Mock() + repo.store_list.return_value = [] + manifest = Mock() + return Archives(repo, manifest), repo, manifest + + +def _archive_meta(name, id_, ts=TS, *, username="", hostname="", tags=()): + return { + "id": id_, + "name": name, + "time": ts, + "exists": True, + "username": username, + "hostname": hostname, + "size": 0, + "nfiles": 0, + "comment": "", + "tags": tags, + } + + +def _info(name, id_, ts=TS, *, username="", hostname="", tags=()): + return ArchiveInfo(name=name, id=id_, ts=parse_timestamp(ts), tags=tags, user=username, host=hostname) + + +def _stub_matching_info_tuples(infos): + ar, _, _ = _make() + ar._matching_info_tuples = Mock(side_effect=lambda match_patterns, match_end, deleted=False: list(infos)) + return ar + + +def _stub_info_tuples(infos): + ar, _, _ = _make() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter(infos)) + return ar + + +def test_archives_satisfies_archives_interface(): + ar, _, _ = _make() + assert isinstance(ar, ArchivesInterface) + + +def test_prepare_is_noop(): + ar, repo, manifest = _make() + m = Mock() + ar.prepare(manifest, m) + repo.assert_not_called() + manifest.assert_not_called() + m.assert_not_called() + + +def test_finish_returns_empty_dict(): + ar, _, manifest = _make() + assert ar.finish(manifest) == {} + + +def test_ids_empty(): + ar, _, _ = _make() + assert list(ar.ids()) == [] + + +def test_ids_returns_binary_ids(): + ar, repo, _ = _make() + repo.store_list.return_value = [_item(_id(1)), _item(_id(2))] + assert list(ar.ids()) == [_id(1), _id(2)] + + +def test_ids_store_object_not_found_gives_empty(): + ar, repo, _ = _make() + repo.store_list.side_effect = StoreObjectNotFound("archives") + assert list(ar.ids()) == [] + + +def test_ids_passes_deleted_flag(): + ar, repo, _ = _make() + repo.store_list.return_value = [_item(_id(1))] + result = list(ar.ids(deleted=True)) + assert result == [_id(1)] + repo.store_list.assert_called_once_with("archives", deleted=True) + + +def test_count_empty(): + ar, _, _ = _make() + assert ar.count() == 0 + + +def test_count(): + ar, repo, _ = _make() + repo.store_list.return_value = [_item(_id(1)), _item(_id(2))] + assert ar.count() == 2 + + +def test_names(): + ar, _, _ = _make() + metas = [_archive_meta("a", _id(1)), _archive_meta("b", _id(2))] + ar._infos = Mock(side_effect=lambda deleted=False: iter(metas)) + assert list(ar.names()) == ["a", "b"] + + +def test_exists_true(): + ar, _, _ = _make() + ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) + assert ar.exists("a") is True + + +def test_exists_false(): + ar, _, _ = _make() + ar._infos = Mock(side_effect=lambda deleted=False: iter([])) + assert ar.exists("missing") is False + + +def test_exists_id_true(): + ar, repo, _ = _make() + repo.store_list.return_value = [_item(_id(1))] + assert ar.exists_id(_id(1)) is True + + +def test_exists_id_false(): + ar, repo, _ = _make() + repo.store_list.return_value = [] + assert ar.exists_id(_id(99)) is False + + +def test_exists_id_deleted(): + ar, repo, _ = _make() + repo.store_list.return_value = [_item(_id(1))] + assert ar.exists_id(_id(1), deleted=True) is True + repo.store_list.assert_called_with("archives", deleted=True) + + +def test_exists_name_and_id_true(): + ar, _, _ = _make() + ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) + assert ar.exists_name_and_id("a", _id(1)) is True + + +def test_exists_name_and_id_false_wrong_name(): + ar, _, _ = _make() + ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) + assert ar.exists_name_and_id("b", _id(1)) is False + + +def test_exists_name_and_id_false_wrong_id(): + ar, _, _ = _make() + ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) + assert ar.exists_name_and_id("a", _id(2)) is False + + +def test_exists_name_and_ts_true(): + ar, _, _ = _make() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_info("a", _id(1))])) + assert ar.exists_name_and_ts("a", parse_timestamp(TS)) is True + + +def test_exists_name_and_ts_false_wrong_ts(): + ar, _, _ = _make() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_info("a", _id(1))])) + assert ar.exists_name_and_ts("a", parse_timestamp(TS2)) is False + + +def test_exists_name_and_ts_false_wrong_name(): + ar, _, _ = _make() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_info("a", _id(1))])) + assert ar.exists_name_and_ts("b", parse_timestamp(TS)) is False + + +def test_get_archive_meta_object_not_found(): + ar, repo, _ = _make() + repo.get.side_effect = Repository.ObjectNotFound(_id(1), "/fake/path") + result = ar._get_archive_meta(_id(1)) + assert result == { + "id": _id(1), + "name": "archive-does-not-exist", + "time": "1970-01-01T00:00:00.000000", + "exists": False, + "username": "", + "hostname": "", + "tags": (), + } + + +def test_get_archive_meta_success(): + ar, _, manifest = _make() + manifest.repo_objs.parse.return_value = (None, b"data") + manifest.key.unpack_archive.return_value = { + "version": 2, + "name": "myarchive", + "time": "2021-03-15T10:00:00.000000", + "username": "alice", + "hostname": "myhost", + "size": 1024, + "nfiles": 3, + "comment": "weekly", + } + + result = ar._get_archive_meta(_id(1)) + + assert result["exists"] is True + assert result["id"] == _id(1) + assert result["name"] == "myarchive" + assert result["time"] == "2021-03-15T10:00:00.000000" + assert result["username"] == "alice" + assert result["hostname"] == "myhost" + assert result["size"] == 1024 + assert result["nfiles"] == 3 + assert result["comment"] == "weekly" + assert result["tags"] == () + + +def test_get_archive_meta_success_with_tags(): + ar, _, manifest = _make() + manifest.repo_objs.parse.return_value = (None, b"data") + manifest.key.unpack_archive.return_value = { + "version": 2, + "name": "tagged", + "time": TS, + "username": "", + "hostname": "", + "tags": ["beta", "alpha"], + } + + result = ar._get_archive_meta(_id(1)) + + assert result["tags"] == ("alpha", "beta") + assert result["size"] == 0 + assert result["nfiles"] == 0 + assert result["comment"] == "" + + +def test_get_archive_meta_bad_version(): + ar, _, manifest = _make() + manifest.repo_objs.parse.return_value = (None, b"data") + manifest.key.unpack_archive.return_value = {"version": 99} + + with pytest.raises(Exception, match="Unknown archive metadata version"): + ar._get_archive_meta(_id(1)) + + +def test_get_missing_returns_none(): + ar, _, _ = _make() + ar._infos = Mock(side_effect=lambda deleted=False: iter([])) + assert ar.get("nope") is None + + +def test_get_returns_archive_info(): + ar, _, _ = _make() + ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) + info = ar.get("a") + assert isinstance(info, ArchiveInfo) + assert info.name == "a" + assert info.id == _id(1) + + +def test_get_raw(): + ar, _, _ = _make() + ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) + result = ar.get("a", raw=True) + assert result["name"] == "a" + assert result["id"] == _id(1) + assert result["time"] == TS + assert result["exists"] is True + + +def test_get_by_id_missing_returns_none(): + ar, repo, _ = _make() + repo.store_list.return_value = [] + assert ar.get_by_id(_id(99)) is None + + +@pytest.mark.parametrize("raw", [False, True]) +def test_get_by_id(raw): + ar, repo, _ = _make() + repo.store_list.return_value = [_item(_id(1))] + ar._get_archive_meta = Mock(side_effect=lambda id_: _archive_meta("a", _id(1))) + result = ar.get_by_id(_id(1), raw=raw) + if raw: + assert result["name"] == "a" + assert result["id"] == _id(1) + assert result["time"] == TS + assert result["exists"] is True + else: + assert isinstance(result, ArchiveInfo) + assert result.name == "a" + assert result.id == _id(1) + + +def test_get_by_id_exists_false_returns_none(): + ar, repo, _ = _make() + repo.store_list.return_value = [_item(_id(1))] + meta = _archive_meta("a", _id(1)) + meta["exists"] = False + ar._get_archive_meta = Mock(side_effect=lambda id_: meta) + assert ar.get_by_id(_id(1)) is None + + +def test_get_by_id_deleted(): + ar, repo, _ = _make() + repo.store_list.return_value = [_item(_id(1))] + ar._get_archive_meta = Mock(side_effect=lambda id_: _archive_meta("a", _id(1))) + info = ar.get_by_id(_id(1), deleted=True) + assert isinstance(info, ArchiveInfo) + repo.store_list.assert_called_with("archives", deleted=True) + + +def test_create_calls_store_store(): + ar, repo, _ = _make() + ar.create("a", _id(1), TS) + repo.store_store.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", b"") + + +def test_create_with_datetime_ts(): + ar, repo, _ = _make() + dt = datetime(2020, 6, 1, 12, 0, 0, tzinfo=timezone.utc) + ar.create("a", _id(1), dt) + repo.store_store.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", b"") + + +def test_create_overwrite_kwarg_ignored(): + # borgstore store_store is ID-addressed and idempotent; overwrite is an ArchivesInterface + # compatibility parameter that Archives intentionally ignores (unlike LegacyArchives). + ar, repo, _ = _make() + ar.create("a", _id(1), TS, overwrite=True) + repo.store_store.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", b"") + + +def test_delete_by_id(): + ar, repo, _ = _make() + ar.delete_by_id(_id(1)) + repo.store_move.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", delete=True) + + +def test_undelete_by_id(): + ar, repo, _ = _make() + ar.undelete_by_id(_id(1)) + repo.store_move.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", undelete=True) + + +def test_nuke_by_id(): + ar, repo, _ = _make() + ar.nuke_by_id(_id(1)) + repo.store_delete.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", deleted=True) + + +def test_list_no_filters(): + info = _info("a", _id(1)) + ar = _stub_matching_info_tuples([info]) + assert ar.list() == [info] + + +def test_list_sort_by_str_raises(): + ar, _, _ = _make() + with pytest.raises(TypeError, match="sort_by must be a sequence"): + ar.list(sort_by="name") + + +def test_list_sort_generator_not_materialised_regression(): + # _matching_info_tuples must materialise _info_tuples() via list() before returning; + # if that list() call is removed, the raw generator reaches .sort() and raises AttributeError. + ar = _stub_info_tuples([]) + assert ar.list(sort_by=["name"]) == [] + + +def test_list_sort_by(): + i1 = _info("b", _id(2), TS2) + i2 = _info("a", _id(1), TS) + ar = _stub_matching_info_tuples([i1, i2]) + result = ar.list(sort_by=["name"]) + assert result == [i2, i1] + + +def test_list_reverse(): + i1 = _info("a", _id(1)) + i2 = _info("b", _id(2)) + ar = _stub_matching_info_tuples([i1, i2]) + assert ar.list(reverse=True) == [i2, i1] + + +def test_list_first(): + infos = [_info(f"a{i}", _id(i + 1)) for i in range(5)] + ar = _stub_matching_info_tuples(infos) + assert ar.list(first=3) == infos[:3] + + +def test_list_last(): + infos = [_info(f"a{i}", _id(i + 1)) for i in range(5)] + ar = _stub_matching_info_tuples(infos) + assert ar.list(last=2) == infos[-2:] + + +def test_list_first_zero(): + infos = [_info(f"a{i}", _id(i + 1)) for i in range(3)] + ar = _stub_matching_info_tuples(infos) + assert ar.list(first=0) == infos + + +def test_list_last_zero(): + infos = [_info(f"a{i}", _id(i + 1)) for i in range(3)] + ar = _stub_matching_info_tuples(infos) + assert ar.list(last=0) == infos + + +def test_list_date_filter(): + i1 = _info("a", _id(1)) + ar = _stub_matching_info_tuples([i1]) + with patch("borg.manifest.filter_archives_by_date", return_value=[i1]) as mock_filter: + result = ar.list(older="1d") + assert result == [i1] + mock_filter.assert_called_once_with([i1], oldest=None, newest=None, newer=None, older="1d") + + +def test_list_deleted_passes_flag(): + ar, _, _ = _make() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([])) + ar.list(deleted=True) + ar._info_tuples.assert_called_once_with(deleted=True) + + +def test_list_match_name(): + i1 = _info("archive-a", _id(1)) + i2 = _info("archive-b", _id(2)) + ar = _stub_info_tuples([i1, i2]) + assert ar.list(match=["archive-a"]) == [i1] + + +def test_list_match_name_prefix(): + i1 = _info("archive-a", _id(1)) + i2 = _info("other", _id(2)) + ar = _stub_info_tuples([i1, i2]) + assert ar.list(match=["name:archive-a"]) == [i1] + + +def test_list_match_user(): + i1 = _info("a", _id(1), username="alice") + i2 = _info("b", _id(2), username="bob") + ar = _stub_info_tuples([i1, i2]) + assert ar.list(match=["user:alice"]) == [i1] + + +def test_list_match_host(): + i1 = _info("a", _id(1), hostname="laptop") + i2 = _info("b", _id(2), hostname="server") + ar = _stub_info_tuples([i1, i2]) + assert ar.list(match=["host:laptop"]) == [i1] + + +def test_list_match_tags(): + i1 = _info("a", _id(1), tags=("prod", "db")) + i2 = _info("b", _id(2), tags=("dev",)) + ar = _stub_info_tuples([i1, i2]) + assert ar.list(match=["tags:prod"]) == [i1] + + +def test_list_match_aid(): + i1 = _info("a", _id(1)) + ar = _stub_info_tuples([i1]) + prefix = bin_to_hex(_id(1))[:4] + assert ar.list(match=[f"aid:{prefix}"]) == [i1] + + +def test_list_match_aid_ambiguous(): + # Two distinct IDs that share the same leading byte — a realistic prefix collision. + id1 = bytes([0x01, 0x00]) + bytes(30) + id2 = bytes([0x01, 0x01]) + bytes(30) + i1 = _info("a", id1) + i2 = _info("b", id2) + ar = _stub_info_tuples([i1, i2]) + prefix = bin_to_hex(id1)[:2] # "01" — matches both IDs + with pytest.raises(CommandError, match=r"precisely one"): + ar.list(match=[f"aid:{prefix}"]) + + +def test_list_match_multiple_patterns(): + i1 = _info("archive-a", _id(1), username="alice", hostname="laptop") + i2 = _info("archive-b", _id(2), username="alice", hostname="server") + i3 = _info("archive-c", _id(3), username="bob", hostname="laptop") + ar = _stub_info_tuples([i1, i2, i3]) + result = ar.list(match=["user:alice", "host:laptop"]) + assert result == [i1] + + +def test_list_match_end_custom(): + i1 = _info("archive-a", _id(1)) + i2 = _info("other", _id(2)) + ar = _stub_info_tuples([i1, i2]) + result = ar.list(match=["archive"], match_end="") + assert result == [i1] + + +def test_get_one_exact_match(): + i1 = _info("backup", _id(1)) + ar = _stub_info_tuples([i1]) + assert ar.get_one(["backup"]) == i1 + + +def test_get_one_no_match_raises(): + ar = _stub_info_tuples([]) + with pytest.raises(CommandError, match=r"matched 0\."): + ar.get_one(["missing"]) + + +def test_get_one_multiple_matches_raises(): + i1 = _info("a", _id(1)) + i2 = _info("a", _id(2)) + ar = _stub_info_tuples([i1, i2]) + with pytest.raises(CommandError, match=r"matched 2\."): + ar.get_one(["a"]) + + +def test_get_one_deleted_passes_flag(): + i1 = _info("a", _id(1)) + ar, _, _ = _make() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([i1])) + ar.get_one(["a"], deleted=True) + ar._info_tuples.assert_called_once_with(deleted=True) + + +def test_list_considering_raises_if_name_set(): + ar, _, _ = _make() + args = Mock() + args.name = "archive" + with pytest.raises(Error): + ar.list_considering(args) + + +def test_list_considering_delegates(): + i1 = _info("b", _id(2), TS2) + i2 = _info("a", _id(1), TS) + ar = _stub_matching_info_tuples([i1, i2]) + args = Namespace( + name=None, + sort_by="name", + match_archives=None, + first=None, + last=None, + older=None, + newer=None, + oldest=None, + newest=None, + deleted=False, + ) + result = ar.list_considering(args) + assert result == [i2, i1] + + +def test_list_considering_with_match_archives(): + i1 = _info("archive-a", _id(1)) + i2 = _info("other", _id(2)) + ar = _stub_info_tuples([i1, i2]) + args = Namespace( + name=None, + sort_by="name", + match_archives=["archive-a"], + first=None, + last=None, + older=None, + newer=None, + oldest=None, + newest=None, + deleted=False, + ) + result = ar.list_considering(args) + assert result == [i1] + + +def test_list_considering_multi_key_sort(): + i1 = _info("b", _id(1), TS2) + i2 = _info("a", _id(2), TS2) + i3 = _info("c", _id(3), TS) + ar = _stub_matching_info_tuples([i1, i2, i3]) + args = Namespace( + name=None, + sort_by="ts,name", + match_archives=None, + first=None, + last=None, + older=None, + newer=None, + oldest=None, + newest=None, + deleted=False, + ) + result = ar.list_considering(args) + assert result == [i3, i2, i1] From 6cb6ac43b53637ea2e6c691bd1c85fdee6f0652d Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 22 May 2026 16:52:39 +0530 Subject: [PATCH 3/3] testsuite: rename _item to _iteminfo, _make to _archives, _info to _archiveinfo for clarity, refs #9556 --- src/borg/testsuite/archives_test.py | 196 ++++++++++----------- src/borg/testsuite/legacy_archives_test.py | 124 ++++++------- 2 files changed, 160 insertions(+), 160 deletions(-) diff --git a/src/borg/testsuite/archives_test.py b/src/borg/testsuite/archives_test.py index de8045d92d..59c4d33896 100644 --- a/src/borg/testsuite/archives_test.py +++ b/src/borg/testsuite/archives_test.py @@ -21,11 +21,11 @@ def _id(n): TS2 = "2021-06-01T12:00:00.000000" -def _item(id_bytes): +def _iteminfo(id_bytes): return ItemInfo(name=bin_to_hex(id_bytes), exists=True, size=0, directory=False) -def _make(): +def _archives(): repo = Mock() repo.store_list.return_value = [] manifest = Mock() @@ -47,29 +47,29 @@ def _archive_meta(name, id_, ts=TS, *, username="", hostname="", tags=()): } -def _info(name, id_, ts=TS, *, username="", hostname="", tags=()): +def _archiveinfo(name, id_, ts=TS, *, username="", hostname="", tags=()): return ArchiveInfo(name=name, id=id_, ts=parse_timestamp(ts), tags=tags, user=username, host=hostname) def _stub_matching_info_tuples(infos): - ar, _, _ = _make() + ar, _, _ = _archives() ar._matching_info_tuples = Mock(side_effect=lambda match_patterns, match_end, deleted=False: list(infos)) return ar def _stub_info_tuples(infos): - ar, _, _ = _make() + ar, _, _ = _archives() ar._info_tuples = Mock(side_effect=lambda deleted=False: iter(infos)) return ar def test_archives_satisfies_archives_interface(): - ar, _, _ = _make() + ar, _, _ = _archives() assert isinstance(ar, ArchivesInterface) def test_prepare_is_noop(): - ar, repo, manifest = _make() + ar, repo, manifest = _archives() m = Mock() ar.prepare(manifest, m) repo.assert_not_called() @@ -78,122 +78,122 @@ def test_prepare_is_noop(): def test_finish_returns_empty_dict(): - ar, _, manifest = _make() + ar, _, manifest = _archives() assert ar.finish(manifest) == {} def test_ids_empty(): - ar, _, _ = _make() + ar, _, _ = _archives() assert list(ar.ids()) == [] def test_ids_returns_binary_ids(): - ar, repo, _ = _make() - repo.store_list.return_value = [_item(_id(1)), _item(_id(2))] + ar, repo, _ = _archives() + repo.store_list.return_value = [_iteminfo(_id(1)), _iteminfo(_id(2))] assert list(ar.ids()) == [_id(1), _id(2)] def test_ids_store_object_not_found_gives_empty(): - ar, repo, _ = _make() + ar, repo, _ = _archives() repo.store_list.side_effect = StoreObjectNotFound("archives") assert list(ar.ids()) == [] def test_ids_passes_deleted_flag(): - ar, repo, _ = _make() - repo.store_list.return_value = [_item(_id(1))] + ar, repo, _ = _archives() + repo.store_list.return_value = [_iteminfo(_id(1))] result = list(ar.ids(deleted=True)) assert result == [_id(1)] repo.store_list.assert_called_once_with("archives", deleted=True) def test_count_empty(): - ar, _, _ = _make() + ar, _, _ = _archives() assert ar.count() == 0 def test_count(): - ar, repo, _ = _make() - repo.store_list.return_value = [_item(_id(1)), _item(_id(2))] + ar, repo, _ = _archives() + repo.store_list.return_value = [_iteminfo(_id(1)), _iteminfo(_id(2))] assert ar.count() == 2 def test_names(): - ar, _, _ = _make() + ar, _, _ = _archives() metas = [_archive_meta("a", _id(1)), _archive_meta("b", _id(2))] ar._infos = Mock(side_effect=lambda deleted=False: iter(metas)) assert list(ar.names()) == ["a", "b"] def test_exists_true(): - ar, _, _ = _make() + ar, _, _ = _archives() ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) assert ar.exists("a") is True def test_exists_false(): - ar, _, _ = _make() + ar, _, _ = _archives() ar._infos = Mock(side_effect=lambda deleted=False: iter([])) assert ar.exists("missing") is False def test_exists_id_true(): - ar, repo, _ = _make() - repo.store_list.return_value = [_item(_id(1))] + ar, repo, _ = _archives() + repo.store_list.return_value = [_iteminfo(_id(1))] assert ar.exists_id(_id(1)) is True def test_exists_id_false(): - ar, repo, _ = _make() + ar, repo, _ = _archives() repo.store_list.return_value = [] assert ar.exists_id(_id(99)) is False def test_exists_id_deleted(): - ar, repo, _ = _make() - repo.store_list.return_value = [_item(_id(1))] + ar, repo, _ = _archives() + repo.store_list.return_value = [_iteminfo(_id(1))] assert ar.exists_id(_id(1), deleted=True) is True repo.store_list.assert_called_with("archives", deleted=True) def test_exists_name_and_id_true(): - ar, _, _ = _make() + ar, _, _ = _archives() ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) assert ar.exists_name_and_id("a", _id(1)) is True def test_exists_name_and_id_false_wrong_name(): - ar, _, _ = _make() + ar, _, _ = _archives() ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) assert ar.exists_name_and_id("b", _id(1)) is False def test_exists_name_and_id_false_wrong_id(): - ar, _, _ = _make() + ar, _, _ = _archives() ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) assert ar.exists_name_and_id("a", _id(2)) is False def test_exists_name_and_ts_true(): - ar, _, _ = _make() - ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_info("a", _id(1))])) + ar, _, _ = _archives() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_archiveinfo("a", _id(1))])) assert ar.exists_name_and_ts("a", parse_timestamp(TS)) is True def test_exists_name_and_ts_false_wrong_ts(): - ar, _, _ = _make() - ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_info("a", _id(1))])) + ar, _, _ = _archives() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_archiveinfo("a", _id(1))])) assert ar.exists_name_and_ts("a", parse_timestamp(TS2)) is False def test_exists_name_and_ts_false_wrong_name(): - ar, _, _ = _make() - ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_info("a", _id(1))])) + ar, _, _ = _archives() + ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([_archiveinfo("a", _id(1))])) assert ar.exists_name_and_ts("b", parse_timestamp(TS)) is False def test_get_archive_meta_object_not_found(): - ar, repo, _ = _make() + ar, repo, _ = _archives() repo.get.side_effect = Repository.ObjectNotFound(_id(1), "/fake/path") result = ar._get_archive_meta(_id(1)) assert result == { @@ -208,7 +208,7 @@ def test_get_archive_meta_object_not_found(): def test_get_archive_meta_success(): - ar, _, manifest = _make() + ar, _, manifest = _archives() manifest.repo_objs.parse.return_value = (None, b"data") manifest.key.unpack_archive.return_value = { "version": 2, @@ -236,7 +236,7 @@ def test_get_archive_meta_success(): def test_get_archive_meta_success_with_tags(): - ar, _, manifest = _make() + ar, _, manifest = _archives() manifest.repo_objs.parse.return_value = (None, b"data") manifest.key.unpack_archive.return_value = { "version": 2, @@ -256,7 +256,7 @@ def test_get_archive_meta_success_with_tags(): def test_get_archive_meta_bad_version(): - ar, _, manifest = _make() + ar, _, manifest = _archives() manifest.repo_objs.parse.return_value = (None, b"data") manifest.key.unpack_archive.return_value = {"version": 99} @@ -265,13 +265,13 @@ def test_get_archive_meta_bad_version(): def test_get_missing_returns_none(): - ar, _, _ = _make() + ar, _, _ = _archives() ar._infos = Mock(side_effect=lambda deleted=False: iter([])) assert ar.get("nope") is None -def test_get_returns_archive_info(): - ar, _, _ = _make() +def test_get_returns_archive_archiveinfo(): + ar, _, _ = _archives() ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) info = ar.get("a") assert isinstance(info, ArchiveInfo) @@ -280,7 +280,7 @@ def test_get_returns_archive_info(): def test_get_raw(): - ar, _, _ = _make() + ar, _, _ = _archives() ar._infos = Mock(side_effect=lambda deleted=False: iter([_archive_meta("a", _id(1))])) result = ar.get("a", raw=True) assert result["name"] == "a" @@ -290,15 +290,15 @@ def test_get_raw(): def test_get_by_id_missing_returns_none(): - ar, repo, _ = _make() + ar, repo, _ = _archives() repo.store_list.return_value = [] assert ar.get_by_id(_id(99)) is None @pytest.mark.parametrize("raw", [False, True]) def test_get_by_id(raw): - ar, repo, _ = _make() - repo.store_list.return_value = [_item(_id(1))] + ar, repo, _ = _archives() + repo.store_list.return_value = [_iteminfo(_id(1))] ar._get_archive_meta = Mock(side_effect=lambda id_: _archive_meta("a", _id(1))) result = ar.get_by_id(_id(1), raw=raw) if raw: @@ -313,8 +313,8 @@ def test_get_by_id(raw): def test_get_by_id_exists_false_returns_none(): - ar, repo, _ = _make() - repo.store_list.return_value = [_item(_id(1))] + ar, repo, _ = _archives() + repo.store_list.return_value = [_iteminfo(_id(1))] meta = _archive_meta("a", _id(1)) meta["exists"] = False ar._get_archive_meta = Mock(side_effect=lambda id_: meta) @@ -322,8 +322,8 @@ def test_get_by_id_exists_false_returns_none(): def test_get_by_id_deleted(): - ar, repo, _ = _make() - repo.store_list.return_value = [_item(_id(1))] + ar, repo, _ = _archives() + repo.store_list.return_value = [_iteminfo(_id(1))] ar._get_archive_meta = Mock(side_effect=lambda id_: _archive_meta("a", _id(1))) info = ar.get_by_id(_id(1), deleted=True) assert isinstance(info, ArchiveInfo) @@ -331,13 +331,13 @@ def test_get_by_id_deleted(): def test_create_calls_store_store(): - ar, repo, _ = _make() + ar, repo, _ = _archives() ar.create("a", _id(1), TS) repo.store_store.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", b"") def test_create_with_datetime_ts(): - ar, repo, _ = _make() + ar, repo, _ = _archives() dt = datetime(2020, 6, 1, 12, 0, 0, tzinfo=timezone.utc) ar.create("a", _id(1), dt) repo.store_store.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", b"") @@ -346,37 +346,37 @@ def test_create_with_datetime_ts(): def test_create_overwrite_kwarg_ignored(): # borgstore store_store is ID-addressed and idempotent; overwrite is an ArchivesInterface # compatibility parameter that Archives intentionally ignores (unlike LegacyArchives). - ar, repo, _ = _make() + ar, repo, _ = _archives() ar.create("a", _id(1), TS, overwrite=True) repo.store_store.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", b"") def test_delete_by_id(): - ar, repo, _ = _make() + ar, repo, _ = _archives() ar.delete_by_id(_id(1)) repo.store_move.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", delete=True) def test_undelete_by_id(): - ar, repo, _ = _make() + ar, repo, _ = _archives() ar.undelete_by_id(_id(1)) repo.store_move.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", undelete=True) def test_nuke_by_id(): - ar, repo, _ = _make() + ar, repo, _ = _archives() ar.nuke_by_id(_id(1)) repo.store_delete.assert_called_once_with(f"archives/{bin_to_hex(_id(1))}", deleted=True) def test_list_no_filters(): - info = _info("a", _id(1)) + info = _archiveinfo("a", _id(1)) ar = _stub_matching_info_tuples([info]) assert ar.list() == [info] def test_list_sort_by_str_raises(): - ar, _, _ = _make() + ar, _, _ = _archives() with pytest.raises(TypeError, match="sort_by must be a sequence"): ar.list(sort_by="name") @@ -389,46 +389,46 @@ def test_list_sort_generator_not_materialised_regression(): def test_list_sort_by(): - i1 = _info("b", _id(2), TS2) - i2 = _info("a", _id(1), TS) + i1 = _archiveinfo("b", _id(2), TS2) + i2 = _archiveinfo("a", _id(1), TS) ar = _stub_matching_info_tuples([i1, i2]) result = ar.list(sort_by=["name"]) assert result == [i2, i1] def test_list_reverse(): - i1 = _info("a", _id(1)) - i2 = _info("b", _id(2)) + i1 = _archiveinfo("a", _id(1)) + i2 = _archiveinfo("b", _id(2)) ar = _stub_matching_info_tuples([i1, i2]) assert ar.list(reverse=True) == [i2, i1] def test_list_first(): - infos = [_info(f"a{i}", _id(i + 1)) for i in range(5)] + infos = [_archiveinfo(f"a{i}", _id(i + 1)) for i in range(5)] ar = _stub_matching_info_tuples(infos) assert ar.list(first=3) == infos[:3] def test_list_last(): - infos = [_info(f"a{i}", _id(i + 1)) for i in range(5)] + infos = [_archiveinfo(f"a{i}", _id(i + 1)) for i in range(5)] ar = _stub_matching_info_tuples(infos) assert ar.list(last=2) == infos[-2:] def test_list_first_zero(): - infos = [_info(f"a{i}", _id(i + 1)) for i in range(3)] + infos = [_archiveinfo(f"a{i}", _id(i + 1)) for i in range(3)] ar = _stub_matching_info_tuples(infos) assert ar.list(first=0) == infos def test_list_last_zero(): - infos = [_info(f"a{i}", _id(i + 1)) for i in range(3)] + infos = [_archiveinfo(f"a{i}", _id(i + 1)) for i in range(3)] ar = _stub_matching_info_tuples(infos) assert ar.list(last=0) == infos def test_list_date_filter(): - i1 = _info("a", _id(1)) + i1 = _archiveinfo("a", _id(1)) ar = _stub_matching_info_tuples([i1]) with patch("borg.manifest.filter_archives_by_date", return_value=[i1]) as mock_filter: result = ar.list(older="1d") @@ -437,49 +437,49 @@ def test_list_date_filter(): def test_list_deleted_passes_flag(): - ar, _, _ = _make() + ar, _, _ = _archives() ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([])) ar.list(deleted=True) ar._info_tuples.assert_called_once_with(deleted=True) def test_list_match_name(): - i1 = _info("archive-a", _id(1)) - i2 = _info("archive-b", _id(2)) + i1 = _archiveinfo("archive-a", _id(1)) + i2 = _archiveinfo("archive-b", _id(2)) ar = _stub_info_tuples([i1, i2]) assert ar.list(match=["archive-a"]) == [i1] def test_list_match_name_prefix(): - i1 = _info("archive-a", _id(1)) - i2 = _info("other", _id(2)) + i1 = _archiveinfo("archive-a", _id(1)) + i2 = _archiveinfo("other", _id(2)) ar = _stub_info_tuples([i1, i2]) assert ar.list(match=["name:archive-a"]) == [i1] def test_list_match_user(): - i1 = _info("a", _id(1), username="alice") - i2 = _info("b", _id(2), username="bob") + i1 = _archiveinfo("a", _id(1), username="alice") + i2 = _archiveinfo("b", _id(2), username="bob") ar = _stub_info_tuples([i1, i2]) assert ar.list(match=["user:alice"]) == [i1] def test_list_match_host(): - i1 = _info("a", _id(1), hostname="laptop") - i2 = _info("b", _id(2), hostname="server") + i1 = _archiveinfo("a", _id(1), hostname="laptop") + i2 = _archiveinfo("b", _id(2), hostname="server") ar = _stub_info_tuples([i1, i2]) assert ar.list(match=["host:laptop"]) == [i1] def test_list_match_tags(): - i1 = _info("a", _id(1), tags=("prod", "db")) - i2 = _info("b", _id(2), tags=("dev",)) + i1 = _archiveinfo("a", _id(1), tags=("prod", "db")) + i2 = _archiveinfo("b", _id(2), tags=("dev",)) ar = _stub_info_tuples([i1, i2]) assert ar.list(match=["tags:prod"]) == [i1] def test_list_match_aid(): - i1 = _info("a", _id(1)) + i1 = _archiveinfo("a", _id(1)) ar = _stub_info_tuples([i1]) prefix = bin_to_hex(_id(1))[:4] assert ar.list(match=[f"aid:{prefix}"]) == [i1] @@ -489,8 +489,8 @@ def test_list_match_aid_ambiguous(): # Two distinct IDs that share the same leading byte — a realistic prefix collision. id1 = bytes([0x01, 0x00]) + bytes(30) id2 = bytes([0x01, 0x01]) + bytes(30) - i1 = _info("a", id1) - i2 = _info("b", id2) + i1 = _archiveinfo("a", id1) + i2 = _archiveinfo("b", id2) ar = _stub_info_tuples([i1, i2]) prefix = bin_to_hex(id1)[:2] # "01" — matches both IDs with pytest.raises(CommandError, match=r"precisely one"): @@ -498,24 +498,24 @@ def test_list_match_aid_ambiguous(): def test_list_match_multiple_patterns(): - i1 = _info("archive-a", _id(1), username="alice", hostname="laptop") - i2 = _info("archive-b", _id(2), username="alice", hostname="server") - i3 = _info("archive-c", _id(3), username="bob", hostname="laptop") + i1 = _archiveinfo("archive-a", _id(1), username="alice", hostname="laptop") + i2 = _archiveinfo("archive-b", _id(2), username="alice", hostname="server") + i3 = _archiveinfo("archive-c", _id(3), username="bob", hostname="laptop") ar = _stub_info_tuples([i1, i2, i3]) result = ar.list(match=["user:alice", "host:laptop"]) assert result == [i1] def test_list_match_end_custom(): - i1 = _info("archive-a", _id(1)) - i2 = _info("other", _id(2)) + i1 = _archiveinfo("archive-a", _id(1)) + i2 = _archiveinfo("other", _id(2)) ar = _stub_info_tuples([i1, i2]) result = ar.list(match=["archive"], match_end="") assert result == [i1] def test_get_one_exact_match(): - i1 = _info("backup", _id(1)) + i1 = _archiveinfo("backup", _id(1)) ar = _stub_info_tuples([i1]) assert ar.get_one(["backup"]) == i1 @@ -527,23 +527,23 @@ def test_get_one_no_match_raises(): def test_get_one_multiple_matches_raises(): - i1 = _info("a", _id(1)) - i2 = _info("a", _id(2)) + i1 = _archiveinfo("a", _id(1)) + i2 = _archiveinfo("a", _id(2)) ar = _stub_info_tuples([i1, i2]) with pytest.raises(CommandError, match=r"matched 2\."): ar.get_one(["a"]) def test_get_one_deleted_passes_flag(): - i1 = _info("a", _id(1)) - ar, _, _ = _make() + i1 = _archiveinfo("a", _id(1)) + ar, _, _ = _archives() ar._info_tuples = Mock(side_effect=lambda deleted=False: iter([i1])) ar.get_one(["a"], deleted=True) ar._info_tuples.assert_called_once_with(deleted=True) def test_list_considering_raises_if_name_set(): - ar, _, _ = _make() + ar, _, _ = _archives() args = Mock() args.name = "archive" with pytest.raises(Error): @@ -551,8 +551,8 @@ def test_list_considering_raises_if_name_set(): def test_list_considering_delegates(): - i1 = _info("b", _id(2), TS2) - i2 = _info("a", _id(1), TS) + i1 = _archiveinfo("b", _id(2), TS2) + i2 = _archiveinfo("a", _id(1), TS) ar = _stub_matching_info_tuples([i1, i2]) args = Namespace( name=None, @@ -571,8 +571,8 @@ def test_list_considering_delegates(): def test_list_considering_with_match_archives(): - i1 = _info("archive-a", _id(1)) - i2 = _info("other", _id(2)) + i1 = _archiveinfo("archive-a", _id(1)) + i2 = _archiveinfo("other", _id(2)) ar = _stub_info_tuples([i1, i2]) args = Namespace( name=None, @@ -591,9 +591,9 @@ def test_list_considering_with_match_archives(): def test_list_considering_multi_key_sort(): - i1 = _info("b", _id(1), TS2) - i2 = _info("a", _id(2), TS2) - i3 = _info("c", _id(3), TS) + i1 = _archiveinfo("b", _id(1), TS2) + i2 = _archiveinfo("a", _id(2), TS2) + i3 = _archiveinfo("c", _id(3), TS) ar = _stub_matching_info_tuples([i1, i2, i3]) args = Namespace( name=None, diff --git a/src/borg/testsuite/legacy_archives_test.py b/src/borg/testsuite/legacy_archives_test.py index c17be15740..4e7d70880c 100644 --- a/src/borg/testsuite/legacy_archives_test.py +++ b/src/borg/testsuite/legacy_archives_test.py @@ -24,7 +24,7 @@ def _id(n): TS2 = "2021-06-01T12:00:00.000000" -def _make(entries=()): +def _archives(entries=()): """Return LegacyArchives with minimal mocks; entries = [(name, id, ts_str), ...].""" repo = MagicMock() manifest = MagicMock() @@ -49,7 +49,7 @@ def _archive_meta(name, id_, ts=TS, *, username="", hostname="", tags=()): } -def _info(name, id_, ts=TS, *, username="", hostname="", tags=()): +def _archiveinfo(name, id_, ts=TS, *, username="", hostname="", tags=()): from ..helpers.time import parse_timestamp return ArchiveInfo(name=name, id=id_, ts=parse_timestamp(ts), tags=tags, user=username, host=hostname) @@ -57,7 +57,7 @@ def _info(name, id_, ts=TS, *, username="", hostname="", tags=()): def _make_list_target(infos): """LegacyArchives with _info_tuples replaced so callers get controlled data.""" - la, repo, manifest = _make([(i.name, i.id, TS) for i in infos]) + la, repo, manifest = _archives([(i.name, i.id, TS) for i in infos]) la._info_tuples = lambda deleted=False: iter(infos) return la @@ -66,21 +66,21 @@ def _make_list_target(infos): def test_init(): - la, repo, manifest = _make() + la, repo, manifest = _archives() assert la._archives == {} assert la.repository is repo assert la.manifest is manifest def test_set_raw_dict_and_get_raw_dict(): - la, _, _ = _make() + la, _, _ = _archives() d = {"a": {"id": _id(1), "time": TS}} la._set_raw_dict(d) assert la._get_raw_dict() == d def test_prepare(): - la, repo, manifest = _make() + la, repo, manifest = _archives() m = MagicMock() m.archives = {"x": {"id": _id(5), "time": TS}} la.prepare(manifest, m) @@ -88,33 +88,33 @@ def test_prepare(): def test_finish(): - la, _, manifest = _make([("a", _id(1), TS)]) + la, _, manifest = _archives([("a", _id(1), TS)]) result = la.finish(manifest) assert result == {"a": {"id": _id(1), "time": TS}} def test_ids(): - la, _, _ = _make([("a", _id(1), TS), ("b", _id(2), TS)]) + la, _, _ = _archives([("a", _id(1), TS), ("b", _id(2), TS)]) assert list(la.ids()) == [_id(1), _id(2)] def test_count(): - la, _, _ = _make([("a", _id(1), TS), ("b", _id(2), TS)]) + la, _, _ = _archives([("a", _id(1), TS), ("b", _id(2), TS)]) assert la.count() == 2 def test_names(): - la, _, _ = _make([("a", _id(1), TS), ("b", _id(2), TS)]) + la, _, _ = _archives([("a", _id(1), TS), ("b", _id(2), TS)]) assert list(la.names()) == ["a", "b"] def test_exists_true(): - la, _, _ = _make([("a", _id(1), TS)]) + la, _, _ = _archives([("a", _id(1), TS)]) assert la.exists("a") is True def test_exists_false(): - la, _, _ = _make() + la, _, _ = _archives() assert la.exists("missing") is False @@ -122,26 +122,26 @@ def test_exists_false(): def test_create_with_str_ts(): - la, _, _ = _make() + la, _, _ = _archives() la.create("a", _id(1), TS) assert la._archives["a"] == {"id": _id(1), "time": TS} def test_create_with_datetime_ts(): - la, _, _ = _make() + la, _, _ = _archives() dt = datetime(2020, 6, 1, 12, 0, 0, tzinfo=timezone.utc) la.create("a", _id(1), dt) assert la._archives["a"]["time"] == dt.isoformat(timespec="microseconds") def test_create_raises_if_exists(): - la, _, _ = _make([("a", _id(1), TS)]) + la, _, _ = _archives([("a", _id(1), TS)]) with pytest.raises(KeyError, match="already exists"): la.create("a", _id(2), TS) def test_create_overwrite(): - la, _, _ = _make([("a", _id(1), TS)]) + la, _, _ = _archives([("a", _id(1), TS)]) la.create("a", _id(2), TS, overwrite=True) assert la._archives["a"]["id"] == _id(2) @@ -150,12 +150,12 @@ def test_create_overwrite(): def test_get_missing_returns_none(): - la, _, _ = _make() + la, _, _ = _archives() assert la.get("nope") is None -def test_get_returns_archive_info(): - la, _, _ = _make([("a", _id(1), TS)]) +def test_get_returns_archive_archiveinfo(): + la, _, _ = _archives([("a", _id(1), TS)]) info = la.get("a") assert isinstance(info, ArchiveInfo) assert info.name == "a" @@ -163,25 +163,25 @@ def test_get_returns_archive_info(): def test_get_raw(): - la, _, _ = _make([("a", _id(1), TS)]) + la, _, _ = _archives([("a", _id(1), TS)]) result = la.get("a", raw=True) assert result == {"name": "a", "id": _id(1), "time": TS} def test_get_by_id_missing_returns_none(): - la, _, _ = _make() + la, _, _ = _archives() assert la.get_by_id(_id(99)) is None -def test_get_by_id_returns_archive_info(): - la, _, _ = _make([("a", _id(1), TS)]) +def test_get_by_id_returns_archive_archiveinfo(): + la, _, _ = _archives([("a", _id(1), TS)]) info = la.get_by_id(_id(1)) assert isinstance(info, ArchiveInfo) assert info.name == "a" def test_get_by_id_raw(): - la, _, _ = _make([("a", _id(1), TS)]) + la, _, _ = _archives([("a", _id(1), TS)]) result = la.get_by_id(_id(1), raw=True) assert result == {"name": "a", "id": _id(1), "time": TS} @@ -190,37 +190,37 @@ def test_get_by_id_raw(): def test_exists_id_not_implemented(): - la, _, _ = _make() + la, _, _ = _archives() with pytest.raises(NotImplementedError): la.exists_id(_id(1)) def test_exists_name_and_id_not_implemented(): - la, _, _ = _make() + la, _, _ = _archives() with pytest.raises(NotImplementedError): la.exists_name_and_id("a", _id(1)) def test_exists_name_and_ts_not_implemented(): - la, _, _ = _make() + la, _, _ = _archives() with pytest.raises(NotImplementedError): la.exists_name_and_ts("a", datetime.now()) def test_delete_by_id_not_implemented(): - la, _, _ = _make() + la, _, _ = _archives() with pytest.raises(NotImplementedError): la.delete_by_id(_id(1)) def test_undelete_by_id_not_implemented(): - la, _, _ = _make() + la, _, _ = _archives() with pytest.raises(NotImplementedError): la.undelete_by_id(_id(1)) def test_nuke_by_id_not_implemented(): - la, _, _ = _make() + la, _, _ = _archives() with pytest.raises(NotImplementedError): la.nuke_by_id(_id(1)) @@ -229,7 +229,7 @@ def test_nuke_by_id_not_implemented(): def test_get_archive_meta_object_not_found(): - la, repo, _ = _make() + la, repo, _ = _archives() repo.get.side_effect = LegacyRepository.ObjectNotFound(_id(1), "/fake/path") result = la._get_archive_meta(_id(1)) assert result["exists"] is False @@ -239,7 +239,7 @@ def test_get_archive_meta_object_not_found(): def test_get_archive_meta_success(): - la, _, manifest = _make() + la, _, manifest = _archives() manifest.repo_objs.parse.return_value = (None, b"data") manifest.key.unpack_archive.return_value = {} @@ -262,7 +262,7 @@ def test_get_archive_meta_success(): def test_get_archive_meta_bad_version(): - la, _, manifest = _make() + la, _, manifest = _archives() manifest.repo_objs.parse.return_value = (None, b"data") manifest.key.unpack_archive.return_value = {} @@ -279,7 +279,7 @@ def test_get_archive_meta_bad_version(): def test_infos_and_info_tuples(): - la, _, _ = _make([("a", _id(1), TS)]) + la, _, _ = _archives([("a", _id(1), TS)]) la._get_archive_meta = lambda id_: _archive_meta("a", _id(1)) infos = list(la._infos()) assert len(infos) == 1 @@ -294,46 +294,46 @@ def test_infos_and_info_tuples(): def test_list_no_filters(): - info = _info("a", _id(1)) + info = _archiveinfo("a", _id(1)) la = _make_list_target([info]) assert la.list() == [info] def test_list_sort_by_str_raises(): - la = _make_list_target([_info("a", _id(1))]) + la = _make_list_target([_archiveinfo("a", _id(1))]) with pytest.raises(TypeError, match="sequence"): la.list(sort_by="name") def test_list_sort_by(): - i1 = _info("b", _id(2), TS2) - i2 = _info("a", _id(1), TS) + i1 = _archiveinfo("b", _id(2), TS2) + i2 = _archiveinfo("a", _id(1), TS) la = _make_list_target([i1, i2]) result = la.list(sort_by=["name"]) assert result == [i2, i1] def test_list_reverse(): - i1 = _info("a", _id(1)) - i2 = _info("b", _id(2)) + i1 = _archiveinfo("a", _id(1)) + i2 = _archiveinfo("b", _id(2)) la = _make_list_target([i1, i2]) assert la.list(reverse=True) == [i2, i1] def test_list_first(): - infos = [_info(f"a{i}", _id(i + 1)) for i in range(5)] + infos = [_archiveinfo(f"a{i}", _id(i + 1)) for i in range(5)] la = _make_list_target(infos) assert la.list(first=3) == infos[:3] def test_list_last(): - infos = [_info(f"a{i}", _id(i + 1)) for i in range(5)] + infos = [_archiveinfo(f"a{i}", _id(i + 1)) for i in range(5)] la = _make_list_target(infos) assert la.list(last=2) == infos[-2:] def test_list_date_filter(): - i1 = _info("a", _id(1)) + i1 = _archiveinfo("a", _id(1)) la = _make_list_target([i1]) with patch("borg.legacy.archives.filter_archives_by_date", return_value=[i1]) as mock_filter: result = la.list(older="1d") @@ -342,38 +342,38 @@ def test_list_date_filter(): def test_list_match_name(): - i1 = _info("archive-a", _id(1)) - i2 = _info("archive-b", _id(2)) + i1 = _archiveinfo("archive-a", _id(1)) + i2 = _archiveinfo("archive-b", _id(2)) la = _make_list_target([i1, i2]) result = la.list(match=["archive-a"]) assert result == [i1] def test_list_match_name_prefix(): - i1 = _info("archive-a", _id(1)) - i2 = _info("other", _id(2)) + i1 = _archiveinfo("archive-a", _id(1)) + i2 = _archiveinfo("other", _id(2)) la = _make_list_target([i1, i2]) result = la.list(match=["name:archive-a"]) assert result == [i1] def test_list_match_user(): - i1 = _info("a", _id(1), username="alice") - i2 = _info("b", _id(2), username="bob") + i1 = _archiveinfo("a", _id(1), username="alice") + i2 = _archiveinfo("b", _id(2), username="bob") la = _make_list_target([i1, i2]) assert la.list(match=["user:alice"]) == [i1] def test_list_match_host(): - i1 = _info("a", _id(1), hostname="laptop") - i2 = _info("b", _id(2), hostname="server") + i1 = _archiveinfo("a", _id(1), hostname="laptop") + i2 = _archiveinfo("b", _id(2), hostname="server") la = _make_list_target([i1, i2]) assert la.list(match=["host:laptop"]) == [i1] def test_list_match_tags(): - i1 = _info("a", _id(1), tags=("prod", "db")) - i2 = _info("b", _id(2), tags=("dev",)) + i1 = _archiveinfo("a", _id(1), tags=("prod", "db")) + i2 = _archiveinfo("b", _id(2), tags=("dev",)) la = _make_list_target([i1, i2]) assert la.list(match=["tags:prod"]) == [i1] @@ -381,7 +381,7 @@ def test_list_match_tags(): def test_list_match_aid(): from ..helpers.parseformat import bin_to_hex - i1 = _info("a", _id(1)) + i1 = _archiveinfo("a", _id(1)) la = _make_list_target([i1]) prefix = bin_to_hex(_id(1))[:4] assert la.list(match=[f"aid:{prefix}"]) == [i1] @@ -390,8 +390,8 @@ def test_list_match_aid(): def test_list_match_aid_ambiguous(): from ..helpers.parseformat import bin_to_hex - i1 = _info("a", _id(1)) - i2 = _info("b", _id(1)) + i1 = _archiveinfo("a", _id(1)) + i2 = _archiveinfo("b", _id(1)) la = _make_list_target([i1, i2]) prefix = bin_to_hex(_id(1))[:4] with pytest.raises(CommandError): @@ -402,7 +402,7 @@ def test_list_match_aid_ambiguous(): def test_get_one_exact_match(): - i1 = _info("backup", _id(1)) + i1 = _archiveinfo("backup", _id(1)) la = _make_list_target([i1]) assert la.get_one(["backup"]) == i1 @@ -414,8 +414,8 @@ def test_get_one_no_match_raises(): def test_get_one_multiple_matches_raises(): - i1 = _info("a", _id(1)) - i2 = _info("a", _id(2)) + i1 = _archiveinfo("a", _id(1)) + i2 = _archiveinfo("a", _id(2)) la = _make_list_target([i1, i2]) with pytest.raises(CommandError, match="matched 2"): la.get_one(["a"]) @@ -425,7 +425,7 @@ def test_get_one_multiple_matches_raises(): def test_list_considering_raises_if_name_set(): - la, _, _ = _make() + la, _, _ = _archives() args = MagicMock() args.name = "archive" with pytest.raises(Error): @@ -433,7 +433,7 @@ def test_list_considering_raises_if_name_set(): def test_list_considering_delegates(): - i1 = _info("a", _id(1)) + i1 = _archiveinfo("a", _id(1)) la = _make_list_target([i1]) args = Namespace( name=None, @@ -455,7 +455,7 @@ def test_list_considering_delegates(): def test_legacy_archives_satisfies_archives_interface(): - la, _, _ = _make() + la, _, _ = _archives() assert isinstance(la, ArchivesInterface)