Differential D3330 Diff 11795 swh/storage/tests/algos/test_origin.py

Changeset View

Standalone View

swh/storage/tests/algos/test_origin.py

Show First 20 Lines • Show All 71 Lines • ▼ Show 20 Lines	def test_iter_origins_batch_size(mock_origin_get_range, swh_storage):

list(iter_origins(swh_storage))		list(iter_origins(swh_storage))
mock_origin_get_range.assert_called_with(origin_from=1, origin_count=10000)		mock_origin_get_range.assert_called_with(origin_from=1, origin_count=10000)

list(iter_origins(swh_storage, batch_size=42))		list(iter_origins(swh_storage, batch_size=42))
mock_origin_get_range.assert_called_with(origin_from=1, origin_count=42)		mock_origin_get_range.assert_called_with(origin_from=1, origin_count=42)


def test_origin_get_latest_status_none(swh_storage):		def test_origin_get_latest_visit_status_none(swh_storage):
"""Looking up unknown objects should return nothing		"""Looking up unknown objects should return nothing

"""		"""
# unknown origin so no result		# unknown origin so no result
assert origin_get_latest_visit_status(swh_storage, "unknown-origin") is None		assert origin_get_latest_visit_status(swh_storage, "unknown-origin") is None

# unknown type so no result		# unknown type so no result
origin = Origin.from_dict(data.origin)		origin = Origin.from_dict(data.origin)
▲ Show 20 Lines • Show All 97 Lines • ▼ Show 20 Lines	def init_storage_with_origin_visits(swh_storage):
swh_storage.origin_visit_status_add([ovs11, ovs12, ovs21, ovs22])		swh_storage.origin_visit_status_add([ovs11, ovs12, ovs21, ovs22])
return {		return {
"origin": [origin1, origin2],		"origin": [origin1, origin2],
"origin_visit": [ov1, ov2],		"origin_visit": [ov1, ov2],
"origin_visit_status": [ovs11, ovs12, ovs21, ovs22],		"origin_visit_status": [ovs11, ovs12, ovs21, ovs22],
}		}


def test_origin_get_latest_status_filter_type(swh_storage):		def test_origin_get_latest_visit_status_filter_type(swh_storage):
"""Filtering origin visit per types should yield consistent results		"""Filtering origin visit per types should yield consistent results

"""		"""
objects = init_storage_with_origin_visits(swh_storage)		objects = init_storage_with_origin_visits(swh_storage)
origin1, origin2 = objects["origin"]		origin1, origin2 = objects["origin"]
ov1, ov2 = objects["origin_visit"]		ov1, ov2 = objects["origin_visit"]
ovs11, ovs12, _, ovs22 = objects["origin_visit_status"]		ovs11, ovs12, _, ovs22 = objects["origin_visit_status"]

Show All 27 Lines	def test_origin_get_latest_visit_status_filter_type(swh_storage):
assert isinstance(actual_ov2, OriginVisit)		assert isinstance(actual_ov2, OriginVisit)
assert isinstance(actual_ovs22, OriginVisitStatus)		assert isinstance(actual_ovs22, OriginVisitStatus)
assert actual_ov2.origin == ov2.origin		assert actual_ov2.origin == ov2.origin
assert actual_ov2.visit == ov2.visit		assert actual_ov2.visit == ov2.visit
assert actual_ov2.type == data.type_visit2		assert actual_ov2.type == data.type_visit2
assert actual_ovs22 == ovs22		assert actual_ovs22 == ovs22


def test_origin_get_latest_status_filter_status(swh_storage):		def test_origin_get_latest_visit_status_filter_status(swh_storage):
objects = init_storage_with_origin_visits(swh_storage)		objects = init_storage_with_origin_visits(swh_storage)
origin1, origin2 = objects["origin"]		origin1, origin2 = objects["origin"]
ov1, ov2 = objects["origin_visit"]		ov1, ov2 = objects["origin_visit"]
ovs11, ovs12, _, ovs22 = objects["origin_visit_status"]		ovs11, ovs12, _, ovs22 = objects["origin_visit_status"]

# no failed status for that visit		# no failed status for that visit
assert (		assert (
origin_get_latest_visit_status(		origin_get_latest_visit_status(
Show All 32 Lines	actual_ov2, actual_ovs22 = origin_get_latest_visit_status(
swh_storage, origin2.url, allowed_statuses=["full"]		swh_storage, origin2.url, allowed_statuses=["full"]
)		)
assert actual_ov2.origin == ov2.origin		assert actual_ov2.origin == ov2.origin
assert actual_ov2.visit == ov2.visit		assert actual_ov2.visit == ov2.visit
assert actual_ov2.type == data.type_visit2		assert actual_ov2.type == data.type_visit2
assert actual_ovs22 == ovs22		assert actual_ovs22 == ovs22


def test_origin_get_latest_status_filter_snapshot(swh_storage):		def test_origin_get_latest_visit_status_filter_snapshot(swh_storage):
objects = init_storage_with_origin_visits(swh_storage)		objects = init_storage_with_origin_visits(swh_storage)
origin1, origin2 = objects["origin"]		origin1, origin2 = objects["origin"]
_, ov2 = objects["origin_visit"]		_, ov2 = objects["origin_visit"]
_, _, _, ovs22 = objects["origin_visit_status"]		_, _, _, ovs22 = objects["origin_visit_status"]

# there is no visit with snapshot yet for that visit		# there is no visit with snapshot yet for that visit
assert (		assert (
origin_get_latest_visit_status(swh_storage, origin1.url, require_snapshot=True)		origin_get_latest_visit_status(swh_storage, origin1.url, require_snapshot=True)
is None		is None
)		)

# visit status with partial status visit elected		# visit status with partial status visit elected
actual_ov2, actual_ovs22 = origin_get_latest_visit_status(		actual_ov2, actual_ovs22 = origin_get_latest_visit_status(
swh_storage, origin2.url, require_snapshot=True		swh_storage, origin2.url, require_snapshot=True
)		)
assert actual_ov2.origin == ov2.origin		assert actual_ov2.origin == ov2.origin
assert actual_ov2.visit == ov2.visit		assert actual_ov2.visit == ov2.visit
assert actual_ov2.type == ov2.type		assert actual_ov2.type == ov2.type
assert actual_ovs22 == ovs22		assert actual_ovs22 == ovs22

		date_now = now()

		# Add another visit
		swh_storage.origin_visit_add(
		[
		OriginVisit(
		origin=origin2.url,
		date=date_now,
		type=data.type_visit2,
		status="ongoing",
		snapshot=None,
		),
		]
		)

		# Requiring the latest visit with a snapshot, we still find the previous visit
		ov2, ovs22 = origin_get_latest_visit_status(
		swh_storage, origin2.url, require_snapshot=True
		)
		assert actual_ov2.origin == ov2.origin
		assert actual_ov2.visit == ov2.visit
		assert actual_ov2.type == ov2.type
		assert actual_ovs22 == ovs22