Differential D7029 Diff 25524 swh/loader/package/tests/test_loader.py

Changeset View

Standalone View

swh/loader/package/tests/test_loader.py

Show First 20 Lines • Show All 87 Lines • ▼ Show 20 Lines	def test_loader_origin_visit_failure(swh_storage):

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		storage = get_storage("memory")
		target = b"\x01" * 20
		rel1 = Release(
		name=b"aaaa",
		message=b"aaaa",
		target=target,
		target_type=ModelObjectType.DIRECTORY,
		synthetic=False,
		)
		rel2 = Release(
		name=b"bbbb",
		message=b"bbbb",
		target=target,
		target_type=ModelObjectType.DIRECTORY,
		synthetic=False,
		)
		storage.release_add([rel1, rel2])

		loader = PackageLoader(storage, "http://example.org/") # 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", 0, b"extid-of-aaaa"): [rel1.swhid()]}
("extid-type", 0, b"extid-of-aaaa"): [
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", 0, 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", 0, 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 = {rel1.id}
assert loader.resolve_object_from_extids(		assert (
known_extids, p_info, whitelist		loader.resolve_object_from_extids(known_extids, p_info, whitelist)
) == CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20)		== rel1.swhid()
		)

# 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 = {rel1.id}
known_extids = {		known_extids = {("extid-type", 0, b"extid-of-aaaa"): [rel2.swhid(), rel1.swhid()]}
("extid-type", 0, b"extid-of-aaaa"): [		assert (
CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"b" * 20),		loader.resolve_object_from_extids(known_extids, p_info, whitelist)
CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20),		== rel1.swhid()
]		)
}
assert loader.resolve_object_from_extids(
known_extids, p_info, whitelist		def test_resolve_object_from_extids_missing_target() -> None:
) == CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20)		storage = get_storage("memory")

		target = b"\x01" * 20
		rel = Release(
		name=b"aaaa",
		message=b"aaaa",
		target=target,
		target_type=ModelObjectType.DIRECTORY,
		synthetic=False,
		)

		loader = PackageLoader(storage, "http://example.org/") # type: ignore

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

		known_extids = {("extid-type", 0, b"extid-of-aaaa"): [rel.swhid()]}
		p_info.extid.return_value = ("extid-type", 0, b"extid-of-aaaa")
		whitelist = {rel.id}

		# Targeted release is missing from the storage
		assert loader.resolve_object_from_extids(known_extids, p_info, whitelist) is None

		storage.release_add([rel])

		# Targeted release now exists
		assert (
		loader.resolve_object_from_extids(known_extids, p_info, whitelist)
		== rel.swhid()
		)


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")

Show All 9 Lines	def test_load_get_known_extids() -> None:
)		)


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"""
storage = get_storage("memory")		storage = get_storage("memory")

origin = "http://example.org"
rel1_swhid = CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"a" * 20)
rel2_swhid = CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"b" * 20)
rel3_swhid = CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"c" * 20)
rel4_swhid = CoreSWHID(object_type=ObjectType.RELEASE, object_id=b"d" * 20)
dir_swhid = CoreSWHID(object_type=ObjectType.DIRECTORY, object_id=b"e" * 20)		dir_swhid = CoreSWHID(object_type=ObjectType.DIRECTORY, object_id=b"e" * 20)

		rels = [
		Release(
		name=f"v{i}.0".encode(),
		message=b"blah\n",
		target=dir_swhid.object_id,
		target_type=ModelObjectType.DIRECTORY,
		synthetic=True,
		)
		for i in (1, 2, 3, 4)
		]
		storage.release_add(rels[0:3])

		origin = "http://example.org"
		rel1_swhid = rels[0].swhid()
		rel2_swhid = rels[1].swhid()
		rel3_swhid = rels[2].swhid()
		rel4_swhid = rels[3].swhid()

# Results of a previous load		# Results of a previous load
storage.extid_add(		storage.extid_add(
[		[
ExtID("extid-type1", b"extid-of-v1.0", rel1_swhid),		ExtID("extid-type1", b"extid-of-v1.0", rel1_swhid),
ExtID("extid-type2", b"extid-of-v2.0", rel2_swhid),		ExtID("extid-type2", b"extid-of-v2.0", rel2_swhid),
]		]
)		)
last_snapshot = Snapshot(		last_snapshot = Snapshot(
▲ Show 20 Lines • Show All 275 Lines • Show Last 20 Lines