Differential D6751 Diff 24539 swh/loader/package/tests/test_loader.py

Changeset View

Standalone View

swh/loader/package/tests/test_loader.py

Show First 20 Lines • Show All 61 Lines • ▼ Show 20 Lines	def get_package_info(self, version):
"http://example.org", f"example-{version}.tar", version=version		"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, 0, 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, p_info, origin):		def _load_release(self, p_info, origin):
return None		return None


Show All 16 Lines
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, 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", 0, b"extid-of-aaaa"): [
CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20),		CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20),
]		]
}		}
whitelist = {b"unused"}		whitelist = {b"unused"}
assert loader.resolve_object_from_extids(known_extids, p_info, whitelist) is None		assert loader.resolve_object_from_extids(known_extids, p_info, whitelist) is None

# Some known extid, and the PackageInfo is not one of them (ie. cache miss)		# Some known extid, and the PackageInfo is not one of them (ie. cache miss)
p_info.extid.return_value = ("extid-type", b"extid-of-cccc")		p_info.extid.return_value = ("extid-type", 0, b"extid-of-cccc")
assert loader.resolve_object_from_extids(known_extids, p_info, whitelist) is None		assert loader.resolve_object_from_extids(known_extids, p_info, whitelist) is None

# Some known extid, and the PackageInfo is one of them (ie. cache hit),		# Some known extid, and the PackageInfo is one of them (ie. cache hit),
# but the target release was not in the previous snapshot		# but the target release was not in the previous snapshot
p_info.extid.return_value = ("extid-type", b"extid-of-aaaa")		p_info.extid.return_value = ("extid-type", 0, b"extid-of-aaaa")
assert loader.resolve_object_from_extids(known_extids, p_info, whitelist) is None		assert loader.resolve_object_from_extids(known_extids, p_info, whitelist) is None

# Some known extid, and the PackageInfo is one of them (ie. cache hit),		# Some known extid, and the PackageInfo is one of them (ie. cache hit),
# and the target release was in the previous snapshot		# and the target release was in the previous snapshot
whitelist = {b"a" * 20}		whitelist = {b"a" * 20}
assert loader.resolve_object_from_extids(		assert loader.resolve_object_from_extids(
known_extids, p_info, whitelist		known_extids, p_info, whitelist
) == CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20)		) == CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20)

# Same as before, but there is more than one extid, and only one is an allowed		# Same as before, but there is more than one extid, and only one is an allowed
# release		# release
whitelist = {b"a" * 20}		whitelist = {b"a" * 20}
known_extids = {		known_extids = {
("extid-type", b"extid-of-aaaa"): [		("extid-type", 0, b"extid-of-aaaa"): [
CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"b" * 20),		CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"b" * 20),
CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20),		CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20),
]		]
}		}
assert loader.resolve_object_from_extids(		assert loader.resolve_object_from_extids(
known_extids, p_info, whitelist		known_extids, p_info, whitelist
) == CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20)		) == CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20)


def test_load_get_known_extids() -> None:		def test_load_get_known_extids() -> None:
"""Checks PackageLoader.load() fetches known extids efficiently"""		"""Checks PackageLoader.load() fetches known extids efficiently"""
storage = Mock(wraps=get_storage("memory"))		storage = Mock(wraps=get_storage("memory"))

loader = StubPackageLoader(storage, "http://example.org")		loader = StubPackageLoader(storage, "http://example.org")

loader.load()		loader.load()

# Calls should be grouped by extid type		# Calls should be grouped by extid type
storage.extid_get_from_extid.assert_has_calls(		storage.extid_get_from_extid.assert_has_calls(
[		[
call("extid-type1", [b"extid-of-v1.0", b"extid-of-v2.0"]),		call("extid-type1", [b"extid-of-v1.0", b"extid-of-v2.0"], version=0),
call("extid-type2", [b"extid-of-v3.0", b"extid-of-v4.0"]),		call("extid-type2", [b"extid-of-v3.0", b"extid-of-v4.0"], version=0),
],		],
any_order=True,		any_order=True,
)		)


def test_load_extids() -> None:		def test_load_extids() -> None:
"""Checks PackageLoader.load() skips iff it should, and writes (only)		"""Checks PackageLoader.load() skips iff it should, and writes (only)
the new ExtIDs"""		the new ExtIDs"""
▲ Show 20 Lines • Show All 153 Lines • ▼ Show 20 Lines	def test_load_upgrade_from_revision_extids(caplog):
rev1_swhid = rev1.swhid()		rev1_swhid = rev1.swhid()
rel1_swhid = rel1.swhid()		rel1_swhid = rel1.swhid()
rev2_swhid = CoreSWHID(object_type=ObjectType.REVISION, object_id=b"b" * 20)		rev2_swhid = CoreSWHID(object_type=ObjectType.REVISION, object_id=b"b" * 20)
rel2_swhid = CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"c" * 20)		rel2_swhid = CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"c" * 20)

# Results of a previous load		# Results of a previous load
storage.extid_add(		storage.extid_add(
[		[
ExtID("extid-type1", b"extid-of-v1.0", rev1_swhid),		ExtID("extid-type1", b"extid-of-v1.0", rev1_swhid, 0),
ExtID("extid-type1", b"extid-of-v2.0", rev2_swhid),		ExtID("extid-type1", b"extid-of-v2.0", rev2_swhid, 0),
]		]
)		)
storage.revision_add([rev1])		storage.revision_add([rev1])
last_snapshot = Snapshot(		last_snapshot = Snapshot(
branches={		branches={
b"v1.0": SnapshotBranch(		b"v1.0": SnapshotBranch(
target_type=TargetType.REVISION, target=rev1_swhid.object_id		target_type=TargetType.REVISION, target=rev1_swhid.object_id
),		),
▲ Show 20 Lines • Show All 94 Lines • ▼ Show 20 Lines	p_info = TestPackageInfo(
a=1,		a=1,
b=2,		b=2,
length=221837,		length=221837,
filename="8sync-0.1.0.tar.gz",		filename="8sync-0.1.0.tar.gz",
version="0.1.0",		version="0.1.0",
)		)

actual_id = p_info.extid()		actual_id = p_info.extid()
assert actual_id == ("package-manifest-sha256", hashlib.sha256(b"1 2").digest())		assert actual_id == ("package-manifest-sha256", 0, hashlib.sha256(b"1 2").digest())


def test_no_env_swh_config_filename_raise(monkeypatch):		def test_no_env_swh_config_filename_raise(monkeypatch):
"""No SWH_CONFIG_FILENAME environment variable makes package loader init raise		"""No SWH_CONFIG_FILENAME environment variable makes package loader init raise

"""		"""

class DummyPackageLoader(PackageLoader):		class DummyPackageLoader(PackageLoader):
Show All 10 Lines