Changeset View
Changeset View
Standalone View
Standalone View
swh/indexer/tests/utils.py
Show First 20 Lines • Show All 504 Lines • ▼ Show 20 Lines | |||||
def fill_storage(storage): | def fill_storage(storage): | ||||
visit_types = {} | visit_types = {} | ||||
for visit in ORIGIN_VISITS: | for visit in ORIGIN_VISITS: | ||||
storage.origin_add_one({"url": visit["url"]}) | storage.origin_add_one({"url": visit["url"]}) | ||||
visit_types[visit["url"]] = visit["type"] | visit_types[visit["url"]] = visit["type"] | ||||
for snap in SNAPSHOTS: | for snap in SNAPSHOTS: | ||||
origin_url = snap["origin"] | origin_url = snap["origin"] | ||||
visit = storage.origin_visit_add( | visit = storage.origin_visit_add( | ||||
[ | [OriginVisit(origin=origin_url, date=now(), type=visit_types[origin_url],)] | ||||
OriginVisit( | |||||
origin=origin_url, | |||||
date=now(), | |||||
type=visit_types[origin_url], | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | |||||
] | |||||
)[0] | )[0] | ||||
snap_id = snap.get("id") or bytes([random.randint(0, 255) for _ in range(32)]) | snap_id = snap.get("id") or bytes([random.randint(0, 255) for _ in range(32)]) | ||||
storage.snapshot_add([{"id": snap_id, "branches": snap["branches"]}]) | storage.snapshot_add([{"id": snap_id, "branches": snap["branches"]}]) | ||||
visit_status = OriginVisitStatus( | visit_status = OriginVisitStatus( | ||||
origin=origin_url, | origin=origin_url, | ||||
visit=visit.visit, | visit=visit.visit, | ||||
date=now(), | date=now(), | ||||
status="full", | status="full", | ||||
▲ Show 20 Lines • Show All 211 Lines • Show Last 20 Lines |