mirror of
https://github.com/borgbackup/borg.git
synced 2026-05-23 18:45:53 -04:00
testsuite: rename _item to _iteminfo, _make to _archives, _info to _archiveinfo for clarity, refs #9556
This commit is contained in:
Mrityunjay Raj
parent
fdc00d9e8e
commit
6cb6ac43b5
2 changed files with 160 additions and 160 deletions
|
|
@ -21,11 +21,11 @@ TS = "2020-06-01T12:00:00.000000"
|
|||
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,
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ TS = "2020-06-01T12:00:00.000000"
|
|||
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)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue