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] 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)