Differential D3473 Diff 12316 swh/loader/tests/test_init.py

Changeset View

Standalone View

swh/loader/tests/test_init.py

Show First 20 Lines • Show All 171 Lines • ▼ Show 20 Lines	def test_prepare_repository_from_archive_no_filename(datadir, tmp_path):
assert os.path.exists(expected_uncompressed_archive_path)		assert os.path.exists(expected_uncompressed_archive_path)


def test_decode_target_edge():		def test_decode_target_edge():
assert not decode_target(None)		assert not decode_target(None)


def test_decode_target():		def test_decode_target():
actual_alias_decode_target = decode_target(		for target_alias in ["something", b"something"]:
{"target_type": "alias", "target": b"something",}		target = {
)		"target_type": "alias",
		"target": target_alias,
		}
		actual_alias_decode_target = decode_target(target)
assert actual_alias_decode_target == {		assert actual_alias_decode_target == {
"target_type": "alias",		"target_type": "alias",
"target": "something",		"target": "something",
}		}

actual_decode_target = decode_target(		for hash_ in [hash_hex, hash_to_bytes(hash_hex)]:
{"target_type": "revision", "target": hash_to_bytes(hash_hex),}		target = {"target_type": "revision", "target": hash_}
)		actual_decode_target = decode_target(target)

assert actual_decode_target == {		assert actual_decode_target == {
"target_type": "revision",		"target_type": "revision",
"target": hash_hex,		"target": hash_hex,
}		}


def test_check_snapshot(swh_storage):		def test_check_snapshot(swh_storage):
snap_id = "2498dbf535f882bc7f9a18fb16c9ad27fda7bab7"		"""Check snapshot should not raise when everything is fine"""
snapshot = Snapshot(		snapshot = Snapshot(
id=hash_to_bytes(snap_id),		id=hash_to_bytes("2498dbf535f882bc7f9a18fb16c9ad27fda7bab7"),
branches={		branches={
b"master": SnapshotBranch(		b"master": SnapshotBranch(
target=hash_to_bytes(hash_hex), target_type=TargetType.REVISION,		target=hash_to_bytes(hash_hex), target_type=TargetType.REVISION,
),		),
},		},
)		)

s = swh_storage.snapshot_add([snapshot])		s = swh_storage.snapshot_add([snapshot])
assert s == {		assert s == {
"snapshot:add": 1,		"snapshot:add": 1,
}		}

expected_snapshot = {		for snap in [snapshot, snapshot.to_dict()]:
"id": snap_id,		check_snapshot(snap, swh_storage)
"branches": {"master": {"target": hash_hex, "target_type": "revision",}},
}
check_snapshot(expected_snapshot, swh_storage)


def test_check_snapshot_failure(swh_storage):		def test_check_snapshot_failure(swh_storage):
		"""check_snapshot should raise if something goes wrong"""
		snap_id_hex = "2498dbf535f882bc7f9a18fb16c9ad27fda7bab7"
snapshot = Snapshot(		snapshot = Snapshot(
id=hash_to_bytes("2498dbf535f882bc7f9a18fb16c9ad27fda7bab7"),		id=hash_to_bytes(snap_id_hex),
branches={		branches={
b"master": SnapshotBranch(		b"master": SnapshotBranch(
target=hash_to_bytes(hash_hex), target_type=TargetType.REVISION,		target=hash_to_bytes(hash_hex), target_type=TargetType.REVISION,
),		),
},		},
)		)

s = swh_storage.snapshot_add([snapshot])		s = swh_storage.snapshot_add([snapshot])
assert s == {		assert s == {
"snapshot:add": 1,		"snapshot:add": 1,
}		}

unexpected_snapshot = {		unexpected_snapshot = {
"id": "2498dbf535f882bc7f9a18fb16c9ad27fda7bab7", # id is correct		"id": "2498dbf535f882bc7f9a18fb16c9ad27fda7bab7", # id is correct
"branches": {		"branches": {
"master": {"target": hash_hex, "target_type": "release",} # wrong branch		"master": {"target": hash_hex, "target_type": "release",} # wrong branch
},		},
}		}

		# id is correct, the branch is wrong, that should raise nonetheless
		for snap_id in [snap_id_hex, snapshot.id]:
with pytest.raises(AssertionError, match="Differing items"):		with pytest.raises(AssertionError, match="Differing items"):
		unexpected_snapshot["id"] = snap_id
check_snapshot(unexpected_snapshot, swh_storage)		check_snapshot(unexpected_snapshot, swh_storage)

# snapshot id which does not exist		# snapshot id which does not exist
unexpected_snapshot["id"] = "999666f535f882bc7f9a18fb16c9ad27fda7bab7"		wrong_snap_id_hex = "999666f535f882bc7f9a18fb16c9ad27fda7bab7"
		for snap_id in [wrong_snap_id_hex, hash_to_bytes(wrong_snap_id_hex)]:
		unexpected_snapshot["id"] = wrong_snap_id_hex
with pytest.raises(AssertionError, match="is not found"):		with pytest.raises(AssertionError, match="is not found"):
check_snapshot(unexpected_snapshot, swh_storage)		check_snapshot(unexpected_snapshot, swh_storage)

		# not a Snapshot object, raise!
		with pytest.raises(AssertionError, match="variable 'snapshot' must be a snapshot"):
		check_snapshot(ORIGIN_VISIT, swh_storage)