Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/migrate_extrinsic_metadata/test_debian.py
Show First 20 Lines • Show All 156 Lines • ▼ Show 20 Lines | |||||
def test_debian_origins_from_row__no_result(): | def test_debian_origins_from_row__no_result(): | ||||
"""Tests debian_origins_from_row when there's no origin, visit, status, | """Tests debian_origins_from_row when there's no origin, visit, status, | ||||
snapshot, branch, or matching branch. | snapshot, branch, or matching branch. | ||||
""" | """ | ||||
storage = Mock() | storage = Mock() | ||||
origin_url = "deb://Debian/packages/kalgebra" | origin_url = "deb://Debian/packages/kalgebra" | ||||
snapshot_id = b"42424242424242424242" | |||||
revision_id = b"21212121212121212121" | |||||
revision_row = { | revision_row = { | ||||
"id": b"\x00\x00\x03l1\x1e\xf3:(\x1b\x05h\x8fn\xad\xcf\xc0\x94:\xee", | "id": b"\x00\x00\x03l1\x1e\xf3:(\x1b\x05h\x8fn\xad\xcf\xc0\x94:\xee", | ||||
"metadata": {"original_artifact": [{"filename": "kalgebra_19.12.1-1.dsc",},]}, | "metadata": {"original_artifact": [{"filename": "kalgebra_19.12.1-1.dsc",},]}, | ||||
} | } | ||||
# no visit | # no visit | ||||
with patch("iter_origin_visits", return_value=[]): | with patch("iter_origin_visits", return_value=[]): | ||||
Show All 26 Lines | def test_debian_origins_from_row__no_result(): | ||||
# no snapshot | # no snapshot | ||||
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]): | ||||
assert debian_origins_from_row(revision_row, storage) == [] | assert debian_origins_from_row(revision_row, storage) == [] | ||||
assert storage.method_calls == [] | assert storage.method_calls == [] | ||||
status = attr.evolve(status, snapshot=b"42" * 10) | status = attr.evolve(status, snapshot=snapshot_id) | ||||
snapshot = Snapshot(id=b"42" * 10, branches={},) | snapshot = Snapshot(id=snapshot_id, 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) == [] | ||||
ardumont: missing_revision_id = b"21" * 10
...
SnapshotBranch(target_type=TargetType.REVISION… | |||||
snapshot = attr.evolve( | |||||
snapshot, | |||||
branches={ | |||||
b"foo": SnapshotBranch(target_type=TargetType.REVISION, target=revision_id,) | |||||
}, | |||||
) | |||||
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([revision_id]), | |||||
call.revision_get([revision_id]), | |||||
call.revision_get([revision_id]), | |||||
] | |||||
storage.reset_mock() | |||||
revision = Revision( | revision = Revision( | ||||
id=b"21" * 10, | id=revision_id, | ||||
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, | ||||
negative_utc=False, | negative_utc=False, | ||||
), | ), | ||||
Show All 13 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([revision_id]), | |||||
call.revision_get([revision_id]), | |||||
call.revision_get([revision_id]), | |||||
] | |||||
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]),
...