Differential D8504 Diff 30639 swh/loader/package/tests/test_loader.py

Changeset View

Standalone View

swh/loader/package/tests/test_loader.py

Show First 20 Lines • Show All 109 Lines • ▼ Show 20 Lines	def test_loader_origin_visit_success(swh_storage, requests_mock_datadir):

loader = StubPackageLoader(swh_storage, ORIGIN_URL)		loader = StubPackageLoader(swh_storage, ORIGIN_URL)

assert loader.load() == {		assert loader.load() == {
"snapshot_id": "dcb9ecef64af73f2cdac7f5463cb6dece6b1db61",		"snapshot_id": "dcb9ecef64af73f2cdac7f5463cb6dece6b1db61",
"status": "eventful",		"status": "eventful",
}		}

		assert loader.load_status() == {"status": "eventful"}
		assert loader.visit_status() == "full"

assert set(loader.last_snapshot().branches.keys()) == {		assert set(loader.last_snapshot().branches.keys()) == {
f"branch-{version}".encode() for version in loader.get_versions()		f"branch-{version}".encode() for version in loader.get_versions()
}		}


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 = StubPackageLoader(swh_storage, "some-url")		loader = StubPackageLoader(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"}

		assert loader.load_status() == {"status": "failed"}
		assert loader.visit_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"}

		assert loader.load_status() == {"status": "failed"}
		assert loader.visit_status() == "failed"


def test_resolve_object_from_extids() -> None:		def test_resolve_object_from_extids() -> None:
storage = get_storage("memory")		storage = get_storage("memory")
target = b"\x01" * 20		target = b"\x01" * 20
rel1 = Release(		rel1 = Release(
name=b"aaaa",		name=b"aaaa",
message=b"aaaa",		message=b"aaaa",
target=target,		target=target,
▲ Show 20 Lines • Show All 164 Lines • ▼ Show 20 Lines	patch.object(
loader,		loader,
"_load_release",		"_load_release",
return_value=(rel4_swhid.object_id, dir_swhid.object_id),		return_value=(rel4_swhid.object_id, dir_swhid.object_id),
autospec=True,		autospec=True,
).start()		).start()

loader.load()		loader.load()

		assert loader.load_status() == {"status": "eventful"}
		assert loader.visit_status() == "full"

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(		call(
StubPackageInfo(		StubPackageInfo(
f"{origin}/example-v2.0.tar.gz", "example-v2.0.tar.gz", "v2.0"		f"{origin}/example-v2.0.tar.gz", "example-v2.0.tar.gz", "v2.0"
),		),
▲ Show 20 Lines • Show All 154 Lines • ▼ Show 20 Lines	patch.object(
return_value=["v1.0", "v2.0", "v3.0"],		return_value=["v1.0", "v2.0", "v3.0"],
autospec=True,		autospec=True,
).start()		).start()

caplog.set_level(logging.ERROR)		caplog.set_level(logging.ERROR)

loader.load()		loader.load()

		assert loader.load_status() == {"status": "eventful"}
		assert loader.visit_status() == "full"

assert len(caplog.records) == 1		assert len(caplog.records) == 1
(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
▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Lines	class StubPackageLoaderWithError(StubPackageLoader):
def get_versions(self, args, *kwargs):		def get_versions(self, args, *kwargs):
raise Exception("error")		raise Exception("error")


def test_loader_sentry_tags_on_error(swh_storage, sentry_events):		def test_loader_sentry_tags_on_error(swh_storage, sentry_events):
origin_url = ORIGIN_URL		origin_url = ORIGIN_URL
loader = StubPackageLoaderWithError(swh_storage, origin_url)		loader = StubPackageLoaderWithError(swh_storage, origin_url)
loader.load()		loader.load()
		assert loader.load_status() == {"status": "failed"}
		assert loader.visit_status() == "failed"
sentry_tags = sentry_events[0]["tags"]		sentry_tags = sentry_events[0]["tags"]
assert sentry_tags.get(SENTRY_ORIGIN_URL_TAG_NAME) == origin_url		assert sentry_tags.get(SENTRY_ORIGIN_URL_TAG_NAME) == origin_url
assert (		assert (
sentry_tags.get(SENTRY_VISIT_TYPE_TAG_NAME)		sentry_tags.get(SENTRY_VISIT_TYPE_TAG_NAME)
== StubPackageLoaderWithError.visit_type		== StubPackageLoaderWithError.visit_type
)		)