Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/algos/test_origin.py
Show First 20 Lines • Show All 78 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_visit_status_none(swh_storage, sample_data_model): | def test_origin_get_latest_visit_status_none(swh_storage, sample_data): | ||||
"""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 = sample_data_model["origin"][0] | origin = sample_data["origin"][0] | ||||
origin_visit = sample_data_model["origin_visit"][0] | origin_visit = sample_data["origin_visit"][0] | ||||
assert origin_visit.origin == origin.url | assert origin_visit.origin == origin.url | ||||
swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
swh_storage.origin_visit_add([origin_visit])[0] | swh_storage.origin_visit_add([origin_visit])[0] | ||||
assert origin_visit.type != "unknown" | assert origin_visit.type != "unknown" | ||||
actual_origin_visit = origin_get_latest_visit_status( | actual_origin_visit = origin_get_latest_visit_status( | ||||
swh_storage, origin.url, type="unknown" | swh_storage, origin.url, type="unknown" | ||||
) | ) | ||||
assert actual_origin_visit is None | assert actual_origin_visit is None | ||||
actual_origin_visit = origin_get_latest_visit_status( | actual_origin_visit = origin_get_latest_visit_status( | ||||
swh_storage, origin.url, require_snapshot=True | swh_storage, origin.url, require_snapshot=True | ||||
) | ) | ||||
assert actual_origin_visit is None | assert actual_origin_visit is None | ||||
actual_origin_visit = origin_get_latest_visit_status( | actual_origin_visit = origin_get_latest_visit_status( | ||||
swh_storage, origin.url, allowed_statuses=["unknown"] | swh_storage, origin.url, allowed_statuses=["unknown"] | ||||
) | ) | ||||
assert actual_origin_visit is None | assert actual_origin_visit is None | ||||
def init_storage_with_origin_visits(swh_storage, sample_data_model): | def init_storage_with_origin_visits(swh_storage, sample_data): | ||||
"""Initialize storage with origin/origin-visit/origin-visit-status | """Initialize storage with origin/origin-visit/origin-visit-status | ||||
""" | """ | ||||
snapshot = sample_data_model["snapshot"][2] | snapshot = sample_data["snapshot"][2] | ||||
origin1, origin2 = sample_data_model["origin"][:2] | origin1, origin2 = sample_data["origin"][:2] | ||||
swh_storage.origin_add([origin1, origin2]) | swh_storage.origin_add([origin1, origin2]) | ||||
ov1, ov2 = swh_storage.origin_visit_add( | ov1, ov2 = swh_storage.origin_visit_add( | ||||
[ | [ | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin1.url, date=data.date_visit1, type=data.type_visit1, | origin=origin1.url, date=data.date_visit1, type=data.type_visit1, | ||||
), | ), | ||||
OriginVisit( | OriginVisit( | ||||
▲ Show 20 Lines • Show All 46 Lines • ▼ Show 20 Lines | def init_storage_with_origin_visits(swh_storage, sample_data): | ||||
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_visit_status_filter_type(swh_storage, sample_data_model): | def test_origin_get_latest_visit_status_filter_type(swh_storage, sample_data): | ||||
"""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, sample_data_model) | objects = init_storage_with_origin_visits(swh_storage, sample_data) | ||||
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 visit for origin1 url with type_visit2 | # no visit for origin1 url with type_visit2 | ||||
assert ( | assert ( | ||||
origin_get_latest_visit_status(swh_storage, origin1.url, type=data.type_visit2) | origin_get_latest_visit_status(swh_storage, origin1.url, type=data.type_visit2) | ||||
is None | is None | ||||
Show All 23 Lines | def test_origin_get_latest_visit_status_filter_type(swh_storage, sample_data): | ||||
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_visit_status_filter_status(swh_storage, sample_data_model): | def test_origin_get_latest_visit_status_filter_status(swh_storage, sample_data): | ||||
objects = init_storage_with_origin_visits(swh_storage, sample_data_model) | objects = init_storage_with_origin_visits(swh_storage, sample_data) | ||||
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( | ||||
swh_storage, origin2.url, allowed_statuses=["failed"] | swh_storage, origin2.url, allowed_statuses=["failed"] | ||||
Show All 31 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_visit_status_filter_snapshot(swh_storage, sample_data_model): | def test_origin_get_latest_visit_status_filter_snapshot(swh_storage, sample_data): | ||||
objects = init_storage_with_origin_visits(swh_storage, sample_data_model) | objects = init_storage_with_origin_visits(swh_storage, sample_data) | ||||
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 | ||||
Show All 26 Lines |