Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/migrate_extrinsic_metadata/test_debian.py
Show First 20 Lines • Show All 208 Lines • ▼ Show 20 Lines | def test_debian_origins_from_row__no_result(): | ||||
snapshot = Snapshot(id=b"42" * 10, branches={},) | snapshot = Snapshot(id=b"42" * 10, branches={},) | ||||
# no branch | # no branch | ||||
with patch("iter_origin_visits", return_value=[visit]): | with patch("iter_origin_visits", return_value=[visit]): | ||||
with patch("iter_origin_visit_statuses", return_value=[status]): | with patch("iter_origin_visit_statuses", return_value=[status]): | ||||
with patch("snapshot_get_all_branches", return_value=snapshot): | with patch("snapshot_get_all_branches", return_value=snapshot): | ||||
assert debian_origins_from_row(revision_row, storage) == [] | assert debian_origins_from_row(revision_row, storage) == [] | ||||
snapshot = attr.evolve( | |||||
ardumont: missing_revision_id = b"21" * 10
...
SnapshotBranch(target_type=TargetType.REVISION… | |||||
snapshot, | |||||
branches={ | |||||
b"foo": SnapshotBranch(target_type=TargetType.REVISION, target=b"21" * 10,) | |||||
}, | |||||
) | |||||
storage.revision_get.return_value = [None] | |||||
# branch points to unknown revision | |||||
with patch("iter_origin_visits", return_value=[visit]): | |||||
with patch("iter_origin_visit_statuses", return_value=[status]): | |||||
with patch("snapshot_get_all_branches", return_value=snapshot): | |||||
assert debian_origins_from_row(revision_row, storage) == [] | |||||
assert storage.method_calls == [ | |||||
call.revision_get([b"21212121212121212121"]), | |||||
call.revision_get([b"21212121212121212121"]), | |||||
call.revision_get([b"21212121212121212121"]), | |||||
] | |||||
storage.reset_mock() | |||||
revision = Revision( | revision = Revision( | ||||
id=b"21" * 10, | id=b"21" * 10, | ||||
message=b"foo", | message=b"foo", | ||||
author=Person.from_fullname(b"foo"), | author=Person.from_fullname(b"foo"), | ||||
committer=Person.from_fullname(b"foo"), | committer=Person.from_fullname(b"foo"), | ||||
date=TimestampWithTimezone( | date=TimestampWithTimezone( | ||||
timestamp=Timestamp(seconds=1580076204, microseconds=0), | timestamp=Timestamp(seconds=1580076204, microseconds=0), | ||||
offset=60, | offset=60, | ||||
Show All 15 Lines | def test_debian_origins_from_row__no_result(): | ||||
storage.revision_get.return_value = [revision] | storage.revision_get.return_value = [revision] | ||||
# no matching branch | # no matching branch | ||||
with patch("iter_origin_visits", return_value=[visit]): | with patch("iter_origin_visits", return_value=[visit]): | ||||
with patch("iter_origin_visit_statuses", return_value=[status]): | with patch("iter_origin_visit_statuses", return_value=[status]): | ||||
with patch("snapshot_get_all_branches", return_value=snapshot): | with patch("snapshot_get_all_branches", return_value=snapshot): | ||||
assert debian_origins_from_row(revision_row, storage) == [] | assert debian_origins_from_row(revision_row, storage) == [] | ||||
assert storage.method_calls == [] | assert storage.method_calls == [ | ||||
call.revision_get([b"21212121212121212121"]), | |||||
call.revision_get([b"21212121212121212121"]), | |||||
call.revision_get([b"21212121212121212121"]), | |||||
] | |||||
def test_debian_origins_from_row__check_revisions(): | def test_debian_origins_from_row__check_revisions(): | ||||
"""Tests debian_origins_from_row errors when the revision at the head | """Tests debian_origins_from_row errors when the revision at the head | ||||
of a branch is a DSC and has no parents | of a branch is a DSC and has no parents | ||||
""" | """ | ||||
storage = Mock() | storage = Mock() | ||||
▲ Show 20 Lines • Show All 319 Lines • Show Last 20 Lines |
missing_revision_id = b"21" * 10
...
SnapshotBranch(target_type=TargetType.REVISION, target=missing_revision_id,)
...
call.revision_get([missing_revision_id]),
...