Differential D3296 Diff 11696 swh/storage/tests/test_storage.py

Changeset View

Standalone View

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 93 Lines • ▼ Show 20 Lines	):

"""		"""
for k in keys_to_check:		for k in keys_to_check:
expected_list = set([c.get(k) for c in expected_contents])		expected_list = set([c.get(k) for c in expected_contents])
actual_list = set([c.get(k) for c in actual_contents])		actual_list = set([c.get(k) for c in actual_contents])
assert actual_list == expected_list, k		assert actual_list == expected_list, k


		def round_to_milliseconds(date):
		"""Round datetime to milliseconds before insertion, so equality doesn't fail after a
		round-trip through a DB (eg. Cassandra)

		"""
		return date.replace(microsecond=round(date.microsecond, -3))


class LazyContent(Content):		class LazyContent(Content):
def with_data(self):		def with_data(self):
return Content.from_dict({**self.to_dict(), "data": data.cont["data"]})		return Content.from_dict({**self.to_dict(), "data": data.cont["data"]})


class TestStorage:		class TestStorage:
"""Main class for Storage testing.		"""Main class for Storage testing.

▲ Show 20 Lines • Show All 1,512 Lines • ▼ Show 20 Lines

def test_origin_visit_add(self, swh_storage):		def test_origin_visit_add(self, swh_storage):
origin1 = Origin.from_dict(data.origin2)		origin1 = Origin.from_dict(data.origin2)
swh_storage.origin_add_one(origin1)		swh_storage.origin_add_one(origin1)

date_visit = now()		date_visit = now()
date_visit2 = date_visit + datetime.timedelta(minutes=1)		date_visit2 = date_visit + datetime.timedelta(minutes=1)

# Round to milliseconds before insertion, so equality doesn't fail		date_visit = round_to_milliseconds(date_visit)
# after a round-trip through a DB (eg. Cassandra)		date_visit2 = round_to_milliseconds(date_visit2)
date_visit = date_visit.replace(microsecond=round(date_visit.microsecond, -3))
date_visit2 = date_visit2.replace(
microsecond=round(date_visit2.microsecond, -3)
)

visit1 = OriginVisit(		visit1 = OriginVisit(
origin=origin1.url,		origin=origin1.url,
date=date_visit,		date=date_visit,
type=data.type_visit1,		type=data.type_visit1,
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)
▲ Show 20 Lines • Show All 487 Lines • ▼ Show 20 Lines	def test_origin_visit_get_latest(self, swh_storage):
"snapshot": data.complete_snapshot["id"],		"snapshot": data.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": data.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):
		origin1 = Origin.from_dict(data.origin)
		swh_storage.origin_add_one(data.origin)

		# to have some reference visits

		ov1, ov2 = swh_storage.origin_visit_add(
		[
		OriginVisit(
		origin=origin1.url,
		date=data.date_visit1,
		type=data.type_visit1,
		status="ongoing",
		snapshot=None,
		),
		OriginVisit(
		origin=origin1.url,
		date=data.date_visit2,
		type=data.type_visit2,
		status="ongoing",
		snapshot=None,
		),
		]
		)

		snapshot = Snapshot.from_dict(data.complete_snapshot)
		swh_storage.snapshot_add([snapshot])

		date_now = now()
		date_now = round_to_milliseconds(date_now)
		assert data.date_visit1 < data.date_visit2
		assert data.date_visit2 < date_now

		ovs1 = OriginVisitStatus(
		origin=origin1.url,
		visit=ov1.visit,
		date=data.date_visit1,
		status="partial",
		snapshot=None,
		)
		ovs2 = OriginVisitStatus(
		origin=origin1.url,
		visit=ov1.visit,
		date=data.date_visit2,
		status="ongoing",
		snapshot=None,
		)
		ovs3 = OriginVisitStatus(
		origin=origin1.url,
		visit=ov2.visit,
		date=data.date_visit2,
		status="ongoing",
		snapshot=None,
		)
		ovs4 = OriginVisitStatus(
		origin=origin1.url,
		visit=ov2.visit,
		date=date_now,
		status="full",
		snapshot=snapshot.id,
		metadata={"something": "wicked"},
		)

		swh_storage.origin_visit_status_add([ovs1, ovs2, ovs3, ovs4])

		# unknown origin so no result
		actual_origin_visit = swh_storage.origin_visit_status_get_latest(
		"unknown-origin", ov1.visit
		)
		assert actual_origin_visit is None

		# unknown visit so no result
		actual_origin_visit = swh_storage.origin_visit_status_get_latest(
		ov1.origin, ov1.visit + 10
		)
		assert actual_origin_visit is None

		# Two visits, both with no snapshot, take the most recent
		actual_origin_visit2 = swh_storage.origin_visit_status_get_latest(
		origin1.url, ov1.visit
		)
		assert isinstance(actual_origin_visit2, OriginVisitStatus)
		assert actual_origin_visit2 == ovs2
		assert ovs2.origin == origin1.url
		assert ovs2.visit == ov1.visit

		actual_origin_visit = swh_storage.origin_visit_status_get_latest(
		origin1.url, ov1.visit, require_snapshot=True
		)
		# there is no visit with snapshot yet for that visit
		assert actual_origin_visit is None

		actual_origin_visit2 = swh_storage.origin_visit_status_get_latest(
		origin1.url, ov1.visit, allowed_statuses=["partial", "ongoing"]
		)
		# visit status with partial status visit elected
		assert actual_origin_visit2 == ovs2
		assert actual_origin_visit2.status == "ongoing"

		actual_origin_visit4 = swh_storage.origin_visit_status_get_latest(
		origin1.url, ov2.visit, require_snapshot=True
		)
		assert actual_origin_visit4 == ovs4
		assert actual_origin_visit4.snapshot == snapshot.id

		actual_origin_visit = swh_storage.origin_visit_status_get_latest(
		origin1.url, ov2.visit, require_snapshot=True, allowed_statuses=["ongoing"]
		)
		# nothing matches so nothing
		assert actual_origin_visit is None # there is no visit with status full

		actual_origin_visit3 = swh_storage.origin_visit_status_get_latest(
		origin1.url, ov2.visit, allowed_statuses=["ongoing"]
		)
		assert actual_origin_visit3 == ovs3

def test_person_fullname_unicity(self, swh_storage):		def test_person_fullname_unicity(self, swh_storage):
# given (person injection through revisions for example)		# given (person injection through revisions for example)
revision = data.revision		revision = data.revision

# create a revision with same committer fullname but wo name and email		# create a revision with same committer fullname but wo name and email
revision2 = copy.deepcopy(data.revision2)		revision2 = copy.deepcopy(data.revision2)
revision2["committer"] = dict(revision["committer"])		revision2["committer"] = dict(revision["committer"])
revision2["committer"]["email"] = None		revision2["committer"]["email"] = None
▲ Show 20 Lines • Show All 1,820 Lines • Show Last 20 Lines