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