Differential D3311 Diff 11728 swh/storage/tests/test_storage.py

Changeset View

Standalone View

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 1,788 Lines • ▼ Show 20 Lines	def test_origin_visit_status_add(self, swh_storage):
assert origin_visit2["metadata"] == {"intrinsic": "something"}		assert origin_visit2["metadata"] == {"intrinsic": "something"}

actual_objects = list(swh_storage.journal_writer.journal.objects)		actual_objects = list(swh_storage.journal_writer.journal.objects)

expected_origins = [origin1, origin2]		expected_origins = [origin1, origin2]
expected_visits = [ov1, ov2]		expected_visits = [ov1, ov2]
expected_visit_statuses = []		expected_visit_statuses = []
for visit in expected_visits: # out of origin-visit-add calls		for visit in expected_visits: # out of origin-visit-add calls
visit_status = visit.to_dict()		visit_status = visit.to_dict()
		ardumontAuthorUnsubmitted Done Inline Actions mmmph, wrong location, that should go in the test around origin-visit-get-latest (we are in the origin-visit-status-add test here ;) ardumont: mmmph, wrong location, that should go in the test around origin-visit-get-latest (we are in the…
visit_status.pop("type")		visit_status.pop("type")
expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status))		expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status))

# out of origin-visit-status add calls		# out of origin-visit-status add calls
expected_visit_statuses += [visit_status1, visit_status2]		expected_visit_statuses += [visit_status1, visit_status2]
expected_objects = (		expected_objects = (
[("origin", o) for o in expected_origins]		[("origin", o) for o in expected_origins]
+ [("origin_visit", v) for v in expected_visits]		+ [("origin_visit", v) for v in expected_visits]
▲ Show 20 Lines • Show All 189 Lines • ▼ Show 20 Lines
def test_origin_visit_get_by__unknown_origin(self, swh_storage):		def test_origin_visit_get_by__unknown_origin(self, swh_storage):
assert swh_storage.origin_visit_get_by("foo", 10) is None		assert swh_storage.origin_visit_get_by("foo", 10) is None

def test_origin_visit_get_by_no_result(self, swh_storage):		def test_origin_visit_get_by_no_result(self, swh_storage):
swh_storage.origin_add([data.origin])		swh_storage.origin_add([data.origin])
actual_origin_visit = swh_storage.origin_visit_get_by(data.origin["url"], 999)		actual_origin_visit = swh_storage.origin_visit_get_by(data.origin["url"], 999)
assert actual_origin_visit is None		assert actual_origin_visit is None

		def test_origin_visit_get_latest_none(self, swh_storage):
		"""Origin visit get latest on unknown objects should return nothing

		"""
		# unknown origin so no result
		assert swh_storage.origin_visit_get_latest("unknown-origin") is None

		# unknown type
		origin = Origin.from_dict(data.origin)
		swh_storage.origin_add_one(origin)
		assert swh_storage.origin_visit_get_latest(origin.url, type="unknown") is None

		def test_origin_visit_get_latest_filter_type(self, swh_storage):
		"""Filtering origin visit get latest with filter type should be ok

		"""
		origin = Origin.from_dict(data.origin)
		swh_storage.origin_add_one(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_visit2,
		status="ongoing",
		snapshot=None,
		)
		assert data.type_visit1 != data.type_visit2
		assert data.date_visit1 < data.date_visit2

		ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])
		origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit)
		origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit)

		assert data.type_visit1 != data.type_visit2

		# Check type filter is ok
		actual_ov1 = swh_storage.origin_visit_get_latest(
		origin.url, type=data.type_visit1,
		)
		assert actual_ov1 == origin_visit1

		actual_ov3 = swh_storage.origin_visit_get_latest(
		origin.url, type=data.type_visit2,
		)
		assert actual_ov3 == origin_visit3

		new_type = "npm"
		assert new_type not in [data.type_visit1, data.type_visit2]

		assert (
		swh_storage.origin_visit_get_latest(
		origin.url, type=new_type, # no visit matching that type
		)
		is None
		)

def test_origin_visit_get_latest(self, swh_storage):		def test_origin_visit_get_latest(self, swh_storage):
		olasdUnsubmitted Not Done Inline Actions This test is now 180 lines long. Could we consider splitting it into more readable chunks? I think comparing the results for various filters should be split into separate tests, instead of done sequentially in a single huge test (that I have a hard time following). olasd: This test is now 180 lines long. Could we consider splitting it into more readable chunks? I…
		ardumontAuthorUnsubmitted Done Inline Actions granted! I have a hard time myself. ardumont: granted! I have a hard time myself.
		ardumontAuthorUnsubmitted Done Inline Actions I'll start by getting the type filtering test in another test for that diff as that makes sense here. ardumont: I'll start by getting the type filtering test in another test for that diff as that makes sense…
origin_url = swh_storage.origin_add_one(data.origin)		origin = Origin.from_dict(data.origin)
		swh_storage.origin_add_one(origin)
visit1 = OriginVisit(		visit1 = OriginVisit(
origin=origin_url,		origin=origin.url,
date=data.date_visit1,		date=data.date_visit1,
type=data.type_visit1,		type=data.type_visit1,
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)
visit2 = OriginVisit(		visit2 = OriginVisit(
origin=origin_url,		origin=origin.url,
date=data.date_visit2,		date=data.date_visit2,
type=data.type_visit2,		type=data.type_visit2,
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)
# Add a visit with the same date as the previous one		# Add a visit with the same date as the previous one
visit3 = OriginVisit(		visit3 = OriginVisit(
origin=origin_url,		origin=origin.url,
date=data.date_visit2,		date=data.date_visit2,
type=data.type_visit2,		type=data.type_visit2,
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)

ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])		ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])

origin_visit1 = swh_storage.origin_visit_get_by(origin_url, ov1.visit)		origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit)
origin_visit2 = swh_storage.origin_visit_get_by(origin_url, ov2.visit)		origin_visit2 = swh_storage.origin_visit_get_by(origin.url, ov2.visit)
origin_visit3 = swh_storage.origin_visit_get_by(origin_url, ov3.visit)		origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit)
# Two visits, both with no snapshot		# Two visits, both with no snapshot
assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url)		assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url)
assert (		assert (
swh_storage.origin_visit_get_latest(origin_url, require_snapshot=True)		swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)
is None		is None
)		)

# Add snapshot to visit1; require_snapshot=True makes it return		# Add snapshot to visit1; require_snapshot=True makes it return
# visit1 and require_snapshot=False still returns visit2		# visit1 and require_snapshot=False still returns visit2
swh_storage.snapshot_add([data.complete_snapshot])		complete_snapshot = Snapshot.from_dict(data.complete_snapshot)
		swh_storage.snapshot_add([complete_snapshot])
swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin.url,
visit=ov1.visit,		visit=ov1.visit,
date=now(),		date=now(),
status="ongoing",		status="ongoing",
snapshot=data.complete_snapshot["id"],		snapshot=complete_snapshot.id,
)		)
]		]
)		)
assert {		assert {
**origin_visit1,		**origin_visit1,
"snapshot": data.complete_snapshot["id"],		"snapshot": complete_snapshot.id,
} == swh_storage.origin_visit_get_latest(origin_url, require_snapshot=True)		} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)

assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url)		assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url)

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

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

assert {		assert {
**origin_visit1,		**origin_visit1,
"snapshot": data.complete_snapshot["id"],		"snapshot": complete_snapshot.id,
"status": "full",		"status": "full",
} == swh_storage.origin_visit_get_latest(origin_url, allowed_statuses=["full"])		} == swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"])

assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url)		assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url)

# Add snapshot to visit2 and check that the new snapshot is returned		# Add snapshot to visit2 and check that the new snapshot is returned
swh_storage.snapshot_add([data.empty_snapshot])		empty_snapshot = Snapshot.from_dict(data.empty_snapshot)
		swh_storage.snapshot_add([empty_snapshot])

swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin.url,
visit=ov2.visit,		visit=ov2.visit,
date=now(),		date=now(),
status="ongoing",		status="ongoing",
snapshot=data.empty_snapshot["id"],		snapshot=empty_snapshot.id,
)		)
]		]
)		)
assert {		assert {
**origin_visit2,		**origin_visit2,
"snapshot": data.empty_snapshot["id"],		"snapshot": empty_snapshot.id,
} == swh_storage.origin_visit_get_latest(origin_url, require_snapshot=True)		} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)

assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url)		assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url)

# Check that the status filter is still working		# Check that the status filter is still working
assert {		assert {
**origin_visit1,		**origin_visit1,
"snapshot": data.complete_snapshot["id"],		"snapshot": complete_snapshot.id,
"status": "full",		"status": "full",
} == swh_storage.origin_visit_get_latest(origin_url, allowed_statuses=["full"])		} == swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"])

# Add snapshot to visit3 (same date as visit2)		# Add snapshot to visit3 (same date as visit2)
swh_storage.snapshot_add([data.complete_snapshot])

swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin.url,
visit=ov3.visit,		visit=ov3.visit,
date=now(),		date=now(),
status="ongoing",		status="ongoing",
snapshot=data.complete_snapshot["id"],		snapshot=complete_snapshot.id,
)		)
]		]
)		)
assert {		assert {
**origin_visit1,		**origin_visit1,
"snapshot": data.complete_snapshot["id"],		"snapshot": complete_snapshot.id,
"status": "full",		"status": "full",
} == swh_storage.origin_visit_get_latest(origin_url, allowed_statuses=["full"])		} == swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"])
assert {		assert {
**origin_visit1,		**origin_visit1,
"snapshot": data.complete_snapshot["id"],		"snapshot": complete_snapshot.id,
"status": "full",		"status": "full",
} == swh_storage.origin_visit_get_latest(		} == swh_storage.origin_visit_get_latest(
origin_url, allowed_statuses=["full"], require_snapshot=True		origin.url, allowed_statuses=["full"], require_snapshot=True
)		)
assert {		assert {
**origin_visit3,		**origin_visit3,
"snapshot": data.complete_snapshot["id"],		"snapshot": complete_snapshot.id,
} == swh_storage.origin_visit_get_latest(origin_url)		} == swh_storage.origin_visit_get_latest(origin.url)

assert {		assert {
**origin_visit3,		**origin_visit3,
"snapshot": data.complete_snapshot["id"],		"snapshot": complete_snapshot.id,
} == swh_storage.origin_visit_get_latest(origin_url, require_snapshot=True)		} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)

def test_origin_visit_status_get_latest(self, swh_storage):		def test_origin_visit_status_get_latest(self, swh_storage):
origin1 = Origin.from_dict(data.origin)		origin1 = Origin.from_dict(data.origin)
swh_storage.origin_add_one(data.origin)		swh_storage.origin_add_one(data.origin)

# to have some reference visits		# to have some reference visits

ov1, ov2 = swh_storage.origin_visit_add(		ov1, ov2 = swh_storage.origin_visit_add(
▲ Show 20 Lines • Show All 1,937 Lines • Show Last 20 Lines