Differential D6625 Diff 24116 swh/loader/package/tests/test_loader.py

Changeset View

Standalone View

swh/loader/package/tests/test_loader.py

Show First 20 Lines • Show All 52 Lines • ▼ Show 20 Lines	class StubPackageInfo(BasePackageInfo):
pass		pass


class StubPackageLoader(PackageLoader[StubPackageInfo]):		class StubPackageLoader(PackageLoader[StubPackageInfo]):
def get_versions(self):		def get_versions(self):
return ["v1.0", "v2.0", "v3.0", "v4.0"]		return ["v1.0", "v2.0", "v3.0", "v4.0"]

def get_package_info(self, version):		def get_package_info(self, version):
p_info = StubPackageInfo("http://example.org", f"example-{version}.tar")		p_info = StubPackageInfo(
		"http://example.org", f"example-{version}.tar", version=version
		)
extid_type = "extid-type1" if version in ("v1.0", "v2.0") else "extid-type2"		extid_type = "extid-type1" if version in ("v1.0", "v2.0") else "extid-type2"
# Versions 1.0 and 2.0 have an extid of a given type, v3.0 has an extid		# Versions 1.0 and 2.0 have an extid of a given type, v3.0 has an extid
# of a different type		# of a different type
patch.object(		patch.object(
p_info,		p_info,
"extid",		"extid",
return_value=(extid_type, f"extid-of-{version}".encode()),		return_value=(extid_type, f"extid-of-{version}".encode()),
autospec=True,		autospec=True,
).start()		).start()
yield (f"branch-{version}", p_info)		yield (f"branch-{version}", p_info)

def _load_release(self, version, p_info, origin):		def _load_release(self, p_info, origin):
return None		return None


def test_loader_origin_visit_failure(swh_storage):		def test_loader_origin_visit_failure(swh_storage):
"""Failure to add origin or origin visit should failed immediately		"""Failure to add origin or origin visit should failed immediately

"""		"""
loader = PackageLoader(swh_storage, "some-url")		loader = PackageLoader(swh_storage, "some-url")
loader.storage = FakeStorage()		loader.storage = FakeStorage()

actual_load_status = loader.load()		actual_load_status = loader.load()
assert actual_load_status == {"status": "failed"}		assert actual_load_status == {"status": "failed"}

loader.storage = FakeStorage2()		loader.storage = FakeStorage2()

actual_load_status2 = loader.load()		actual_load_status2 = loader.load()
assert actual_load_status2 == {"status": "failed"}		assert actual_load_status2 == {"status": "failed"}


def test_resolve_object_from_extids() -> None:		def test_resolve_object_from_extids() -> None:
loader = PackageLoader(None, None) # type: ignore		loader = PackageLoader(None, None) # type: ignore

p_info = Mock(wraps=BasePackageInfo(None, None)) # type: ignore		p_info = Mock(wraps=BasePackageInfo(None, None, None)) # type: ignore

# The PackageInfo does not support extids		# The PackageInfo does not support extids
p_info.extid.return_value = None		p_info.extid.return_value = None
known_extids = {		known_extids = {
("extid-type", b"extid-of-aaaa"): [		("extid-type", b"extid-of-aaaa"): [
CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20),		CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20),
]		]
}		}
▲ Show 20 Lines • Show All 107 Lines • ▼ Show 20 Lines	def test_load_extids() -> None:
).start()		).start()

loader.load()		loader.load()

assert loader._load_release.mock_calls == [ # type: ignore		assert loader._load_release.mock_calls == [ # type: ignore
# v1.0: not loaded because there is already its (extid_type, extid, rel)		# v1.0: not loaded because there is already its (extid_type, extid, rel)
# in the storage.		# in the storage.
# v2.0: loaded, because there is already a similar extid, but different type		# v2.0: loaded, because there is already a similar extid, but different type
call("v2.0", StubPackageInfo(origin, "example-v2.0.tar"), Origin(url=origin)),		call(StubPackageInfo(origin, "example-v2.0.tar", "v2.0"), Origin(url=origin),),
# v3.0: loaded despite having an (extid_type, extid) in storage, because		# v3.0: loaded despite having an (extid_type, extid) in storage, because
# the target of the extid is not in the previous snapshot		# the target of the extid is not in the previous snapshot
call("v3.0", StubPackageInfo(origin, "example-v3.0.tar"), Origin(url=origin)),		call(StubPackageInfo(origin, "example-v3.0.tar", "v3.0"), Origin(url=origin),),
# v4.0: loaded, because there isn't its extid		# v4.0: loaded, because there isn't its extid
call("v4.0", StubPackageInfo(origin, "example-v4.0.tar"), Origin(url=origin)),		call(StubPackageInfo(origin, "example-v4.0.tar", "v4.0"), Origin(url=origin),),
]		]

# then check the snapshot has all the branches.		# then check the snapshot has all the branches.
# versions 2.0 to 4.0 all point to rel4_swhid (instead of the value of the last		# versions 2.0 to 4.0 all point to rel4_swhid (instead of the value of the last
# snapshot), because they had to be loaded (mismatched extid), and the mocked		# snapshot), because they had to be loaded (mismatched extid), and the mocked
# _load_release always returns rel4_swhid.		# _load_release always returns rel4_swhid.
snapshot = Snapshot(		snapshot = Snapshot(
branches={		branches={
▲ Show 20 Lines • Show All 135 Lines • ▼ Show 20 Lines	def test_load_upgrade_from_revision_extids(caplog):
(record,) = caplog.records		(record,) = caplog.records
assert record.levelname == "ERROR"		assert record.levelname == "ERROR"
assert "Failed to upgrade branch branch-v2.0" in record.message		assert "Failed to upgrade branch branch-v2.0" in record.message

assert loader._load_release.mock_calls == [		assert loader._load_release.mock_calls == [
# v1.0: not loaded because there is already a revision matching it		# v1.0: not loaded because there is already a revision matching it
# v2.0: loaded, as the revision is missing from the storage even though there		# v2.0: loaded, as the revision is missing from the storage even though there
# is an extid		# is an extid
call("v2.0", StubPackageInfo(origin, "example-v2.0.tar"), Origin(url=origin)),		call(StubPackageInfo(origin, "example-v2.0.tar", "v2.0"), Origin(url=origin)),
# v3.0: loaded (did not exist yet)		# v3.0: loaded (did not exist yet)
call("v3.0", StubPackageInfo(origin, "example-v3.0.tar"), Origin(url=origin)),		call(StubPackageInfo(origin, "example-v3.0.tar", "v3.0"), Origin(url=origin)),
]		]

snapshot = Snapshot(		snapshot = Snapshot(
branches={		branches={
b"branch-v1.0": SnapshotBranch(		b"branch-v1.0": SnapshotBranch(
target_type=TargetType.RELEASE, target=rel1_swhid.object_id		target_type=TargetType.RELEASE, target=rel1_swhid.object_id
),		),
b"branch-v2.0": SnapshotBranch(		b"branch-v2.0": SnapshotBranch(
Show All 23 Lines	def test_manifest_extid():
"""		"""

@attr.s		@attr.s
class TestPackageInfo(BasePackageInfo):		class TestPackageInfo(BasePackageInfo):
a = attr.ib()		a = attr.ib()
b = attr.ib()		b = attr.ib()
length = attr.ib()		length = attr.ib()
filename = attr.ib()		filename = attr.ib()
version = attr.ib()

MANIFEST_FORMAT = string.Template("$a $b")		MANIFEST_FORMAT = string.Template("$a $b")

p_info = TestPackageInfo(		p_info = TestPackageInfo(
url="http://example.org/",		url="http://example.org/",
a=1,		a=1,
b=2,		b=2,
length=221837,		length=221837,
Show All 24 Lines