Differential D3314 Diff 11746 swh/storage/tests/test_storage.py

Changeset View

Standalone View

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 2,857 Lines • ▼ Show 20 Lines	def test_snapshot_add_twice__by_origin_visit(self, swh_storage):
(		(
"origin_visit",		"origin_visit",
OriginVisit.from_dict({**data3, "type": data.type_visit2}),		OriginVisit.from_dict({**data3, "type": data.type_visit2}),
),		),
("origin_visit_status", OriginVisitStatus.from_dict(data3)),		("origin_visit_status", OriginVisitStatus.from_dict(data3)),
("origin_visit_status", OriginVisitStatus.from_dict(data4),),		("origin_visit_status", OriginVisitStatus.from_dict(data4),),
]		]

def test_snapshot_get_latest(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)
visit1 = OriginVisit(
origin=origin_url,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
)
visit2 = OriginVisit(
origin=origin_url,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)
# Add a visit with the same date as the previous one
visit3 = OriginVisit(
origin=origin_url,
date=data.date_visit2,
type=data.type_visit3,
status="ongoing",
snapshot=None,
)
ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])
# Two visits, both with no snapshot: latest snapshot is None
assert swh_storage.snapshot_get_latest(origin_url) is None

# Add snapshot to visit1, latest snapshot = visit 1 snapshot
swh_storage.snapshot_add([data.complete_snapshot])
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=ov1.visit,
date=now(),
status="ongoing",
snapshot=data.complete_snapshot["id"],
)
]
)
assert {
**data.complete_snapshot,
"next_branch": None,
} == swh_storage.snapshot_get_latest(origin_url)

# Status filter: all three visits are status=ongoing, so no snapshot
# returned
assert (
swh_storage.snapshot_get_latest(origin_url, allowed_statuses=["full"])
is None
)

# Mark the first visit as completed and check status filter again
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=ov1.visit,
date=now(),
status="full",
snapshot=data.complete_snapshot["id"],
)
]
)
assert {
**data.complete_snapshot,
"next_branch": None,
} == swh_storage.snapshot_get_latest(origin_url, allowed_statuses=["full"])

# Add snapshot to visit2 and check that the new snapshot is returned
swh_storage.snapshot_add([data.empty_snapshot])
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=ov2.visit,
date=now(),
status="ongoing",
snapshot=data.empty_snapshot["id"],
)
]
)
assert {
**data.empty_snapshot,
"next_branch": None,
} == swh_storage.snapshot_get_latest(origin_url)

# Check that the status filter is still working
assert {
**data.complete_snapshot,
"next_branch": None,
} == swh_storage.snapshot_get_latest(origin_url, allowed_statuses=["full"])

# Add snapshot to visit3 (same date as visit2) and check that
# the new snapshot is returned
swh_storage.snapshot_add([data.complete_snapshot])
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=ov3.visit,
date=now(),
status="ongoing",
snapshot=data.complete_snapshot["id"],
)
]
)
assert {
**data.complete_snapshot,
"next_branch": None,
} == swh_storage.snapshot_get_latest(origin_url)

def test_snapshot_get_latest__missing_snapshot(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)
assert swh_storage.snapshot_get_latest(origin_url) is None
visit1 = OriginVisit(
origin=origin_url,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
)
visit2 = OriginVisit(
origin=origin_url,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)
ov1, ov2 = swh_storage.origin_visit_add([visit1, visit2])

# Two visits, both with no snapshot: latest snapshot is None
assert swh_storage.snapshot_get_latest(origin_url) is None

# Add unknown snapshot to visit1, check that the inconsistency is
# detected
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=ov1.visit,
date=now(),
status="ongoing",
snapshot=data.complete_snapshot["id"],
)
]
)
with pytest.raises(Exception):
# XXX: should the exception be more specific than this?
swh_storage.snapshot_get_latest(origin_url)

# Status filter: both visits are status=ongoing, so no snapshot
# returned
assert (
swh_storage.snapshot_get_latest(origin_url, allowed_statuses=["full"])
is None
)

# Mark the first visit as completed and check status filter again
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=ov1.visit,
date=now(),
status="full",
snapshot=data.complete_snapshot["id"],
)
]
)

with pytest.raises(Exception):
# XXX: should the exception be more specific than this?
swh_storage.snapshot_get_latest(origin_url, allowed_statuses=["full"]),

# Actually add the snapshot and check status filter again
swh_storage.snapshot_add([data.complete_snapshot])
assert {
**data.complete_snapshot,
"next_branch": None,
} == swh_storage.snapshot_get_latest(origin_url)

# Add unknown snapshot to visit2 and check that the inconsistency
# is detected
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=ov2.visit,
date=now(),
status="ongoing",
snapshot=data.snapshot["id"],
)
]
)
with pytest.raises(Exception):
# XXX: should the exception be more specific than this?
swh_storage.snapshot_get_latest(origin_url)

# Actually add that snapshot and check that the new one is returned
swh_storage.snapshot_add([data.snapshot])
assert {
**data.snapshot,
"next_branch": None,
} == swh_storage.snapshot_get_latest(origin_url)

def test_snapshot_get_random(self, swh_storage):		def test_snapshot_get_random(self, swh_storage):
swh_storage.snapshot_add(		swh_storage.snapshot_add(
[data.snapshot, data.empty_snapshot, data.complete_snapshot]		[data.snapshot, data.empty_snapshot, data.complete_snapshot]
)		)

assert swh_storage.snapshot_get_random() in {		assert swh_storage.snapshot_get_random() in {
data.snapshot["id"],		data.snapshot["id"],
data.empty_snapshot["id"],		data.empty_snapshot["id"],
▲ Show 20 Lines • Show All 1,087 Lines • Show Last 20 Lines