Differential D3627 Diff 12814 swh/storage/tests/test_storage.py

Changeset View

Standalone View

swh/storage/tests/test_storage.py

Show All 35 Lines	from swh.model.model import (
Release,		Release,
Revision,		Revision,
Snapshot,		Snapshot,
)		)
from swh.model.hypothesis_strategies import objects		from swh.model.hypothesis_strategies import objects
from swh.storage import get_storage		from swh.storage import get_storage
from swh.storage.converters import origin_url_to_sha1 as sha1		from swh.storage.converters import origin_url_to_sha1 as sha1
from swh.storage.exc import HashCollision, StorageArgumentException		from swh.storage.exc import HashCollision, StorageArgumentException
from swh.storage.interface import StorageInterface		from swh.storage.interface import StorageInterface, PagedResult # noqa
from swh.storage.utils import content_hex_hashes, now		from swh.storage.utils import content_hex_hashes, now


@contextmanager		@contextmanager
def db_transaction(storage):		def db_transaction(storage):
with storage.db() as db:		with storage.db() as db:
with db.transaction() as cur:		with db.transaction() as cur:
yield db, cur		yield db, cur
▲ Show 20 Lines • Show All 1,195 Lines • ▼ Show 20 Lines	def _generate_random_visits(self, nb_visits=100, start=0, end=7):
days = random.randint(0, 28)		days = random.randint(0, 28)
weeks = random.randint(start, end)		weeks = random.randint(start, end)
date_visit = today - timedelta(		date_visit = today - timedelta(
weeks=weeks, hours=hours, minutes=minutes, seconds=seconds, days=days		weeks=weeks, hours=hours, minutes=minutes, seconds=seconds, days=days
)		)
visits.append(date_visit)		visits.append(date_visit)
return visits		return visits

		def test_origin_visit_get__unknown_origin(self, swh_storage):
		actual_page = swh_storage.origin_visit_get("foo")
		assert actual_page.next_page_token is None
		assert actual_page.results == []
		assert actual_page == PagedResult()

		def test_origin_visit_get__validation_failure(self, swh_storage, sample_data):
		origin = sample_data.origin
		swh_storage.origin_add([origin])
		with pytest.raises(
		StorageArgumentException, match="page_token must be a string"
		):
		swh_storage.origin_visit_get(origin.url, page_token=10) # not bytes

		with pytest.raises(
		StorageArgumentException, match="order must be one of asc, desc"
		):
		swh_storage.origin_visit_get(origin.url, order="foobar") # wrong order

def test_origin_visit_get_all(self, swh_storage, sample_data):		def test_origin_visit_get_all(self, swh_storage, sample_data):
origin = sample_data.origin		origin = sample_data.origin
swh_storage.origin_add([origin])		swh_storage.origin_add([origin])
visits = swh_storage.origin_visit_add(		ov1, ov2, ov3 = swh_storage.origin_visit_add(
[		[
OriginVisit(		OriginVisit(
origin=origin.url,		origin=origin.url,
date=sample_data.date_visit1,		date=sample_data.date_visit1,
type=sample_data.type_visit1,		type=sample_data.type_visit1,
),		),
OriginVisit(		OriginVisit(
origin=origin.url,		origin=origin.url,
date=sample_data.date_visit2,		date=sample_data.date_visit2,
type=sample_data.type_visit2,		type=sample_data.type_visit2,
),		),
OriginVisit(		OriginVisit(
origin=origin.url,		origin=origin.url,
date=sample_data.date_visit2,		date=sample_data.date_visit2,
type=sample_data.type_visit2,		type=sample_data.type_visit2,
),		),
]		]
)		)
ov1, ov2, ov3 = [
{**v.to_dict(), "status": "created", "snapshot": None, "metadata": None,}
for v in visits
]

# order asc, no pagination, no limit		# order asc, no pagination, no limit
all_visits = list(swh_storage.origin_visit_get(origin.url))		actual_page = swh_storage.origin_visit_get(origin.url)
assert all_visits == [ov1, ov2, ov3]		assert actual_page.next_page_token is None
		assert actual_page == PagedResult(results=[ov1, ov2, ov3])

# order asc, no pagination, limit		# order asc, no pagination, limit
all_visits2 = list(swh_storage.origin_visit_get(origin.url, limit=2))		actual_page = swh_storage.origin_visit_get(origin.url, limit=2)
		vlorentzUnsubmitted Not Done Inline Actions No need to test `next_page_token`, it's opaque. Just check it's not None. (same comment for other occurences below) vlorentz: No need to test `next_page_token`, it's opaque. Just check it's not None. (same comment for…
assert all_visits2 == [ov1, ov2]		next_page_token = actual_page.next_page_token
		assert next_page_token is not None
		assert actual_page.results == [ov1, ov2]

# order asc, pagination, no limit		# order asc, pagination, no limit
all_visits3 = list(		actual_page = swh_storage.origin_visit_get(
		vlorentzUnsubmitted Not Done Inline Actions reuse `next_page_token` from the previous call here. (same comment for other occurences below) vlorentz: reuse `next_page_token` from the previous call here. (same comment for other occurences below)
		ardumontAuthorUnsubmitted Done Inline Actions Well, i'd like to keep the existing tests consistent with what we have already. completely make this opaque will kinda drop the values in that test, making it more error prone... ardumont: Well, i'd like to keep the existing tests consistent with what we have already. completely make…
swh_storage.origin_visit_get(origin.url, last_visit=ov1["visit"])		origin.url, page_token=next_page_token
)		)
assert all_visits3 == [ov2, ov3]		assert actual_page.next_page_token is None
		assert actual_page.results == [ov3]
		assert actual_page == PagedResult(results=[ov3])
		vlorentzUnsubmitted Not Done Inline Actions nope (should be opaque) vlorentz: nope (should be opaque)
		ardumontAuthorUnsubmitted Done Inline Actions well, yes but i also said that i wanted to keep the existing tests comparable with the existing ones... That test is intended to cover the 8 cases of pagination. If i'm starting to use the opaque identifier here, it will become unreadable... ardumont: well, yes but i also said that i wanted to keep the existing tests comparable with the existing…
		vlorentzUnsubmitted Not Done Inline Actions meh. vlorentz: meh.

		next_page_token = str(ov1.visit)
		actual_page = swh_storage.origin_visit_get(
		origin.url, page_token=next_page_token
		)
		assert actual_page.next_page_token is None
		assert actual_page == PagedResult(results=[ov2, ov3])

# order asc, pagination, limit		# order asc, pagination, limit
all_visits4 = list(		actual_page = swh_storage.origin_visit_get(
swh_storage.origin_visit_get(origin.url, last_visit=ov2["visit"], limit=1)		origin.url, page_token=next_page_token, limit=2
		)
		assert actual_page.next_page_token is None
		assert actual_page.results == [ov2, ov3]
		vlorentzUnsubmitted Not Done Inline Actions same vlorentz: same
		assert actual_page == PagedResult(results=[ov2, ov3])

		next_page_token = str(ov2.visit)
		actual_page = swh_storage.origin_visit_get(
		origin.url, page_token=next_page_token, limit=1
)		)
assert all_visits4 == [ov3]		assert actual_page.next_page_token is None
		assert actual_page == PagedResult(results=[ov3])

# order desc, no pagination, no limit		# order desc, no pagination, no limit
all_visits5 = list(swh_storage.origin_visit_get(origin.url, order="desc"))		actual_page = swh_storage.origin_visit_get(origin.url, order="desc")
assert all_visits5 == [ov3, ov2, ov1]		assert actual_page.next_page_token is None
		assert actual_page == PagedResult(results=[ov3, ov2, ov1])

# order desc, no pagination, limit		# order desc, no pagination, limit
all_visits6 = list(		actual_page = swh_storage.origin_visit_get(origin.url, limit=2, order="desc")
swh_storage.origin_visit_get(origin.url, limit=2, order="desc")		next_page_token = actual_page.next_page_token
)		assert next_page_token is not None
assert all_visits6 == [ov3, ov2]		assert actual_page.results == [ov3, ov2]

		actual_page = swh_storage.origin_visit_get(
		origin.url, page_token=next_page_token, order="desc"
		)
		assert actual_page.next_page_token is None
		assert actual_page.results == [ov1]
		assert actual_page == PagedResult(results=[ov1])

		vlorentzUnsubmitted Not Done Inline Actions get the next page of results and test it vlorentz: get the next page of results and test it
# order desc, pagination, no limit		# order desc, pagination, no limit
all_visits7 = list(		next_page_token = str(ov3.visit)
swh_storage.origin_visit_get(		actual_page = swh_storage.origin_visit_get(
origin.url, last_visit=ov3["visit"], order="desc"		origin.url, page_token=next_page_token, order="desc"
		vlorentzUnsubmitted Not Done Inline Actions same vlorentz: same
)		)
)		assert actual_page.next_page_token is None
assert all_visits7 == [ov2, ov1]		assert actual_page == PagedResult(results=[ov2, ov1])

		vlorentzUnsubmitted Not Done Inline Actions assert actual_page.next_page_token is None vlorentz: assert actual_page.next_page_token is None
		ardumontAuthorUnsubmitted Done Inline Actions I thought it enough already. ardumont: I thought it enough already.
# order desc, pagination, limit		# order desc, pagination, limit
all_visits8 = list(		actual_page = swh_storage.origin_visit_get(
swh_storage.origin_visit_get(		origin.url, page_token=next_page_token, order="desc", limit=1
origin.url, last_visit=ov3["visit"], order="desc", limit=1
)
)		)
assert all_visits8 == [ov2]		next_page_token = actual_page.next_page_token
		assert next_page_token is not None
		assert actual_page.results == [ov2]

def test_origin_visit_get__unknown_origin(self, swh_storage):		actual_page = swh_storage.origin_visit_get(
assert [] == list(swh_storage.origin_visit_get("foo"))		origin.url, page_token=next_page_token, order="desc"
		)
		assert actual_page == PagedResult(results=[ov1])

		vlorentzUnsubmitted Done Inline Actions you should get the next page of results and test it too vlorentz: you should get the next page of results and test it too
def test_origin_visit_status_get_random(self, swh_storage, sample_data):		def test_origin_visit_status_get_random(self, swh_storage, sample_data):
origins = sample_data.origins[:2]		origins = sample_data.origins[:2]
swh_storage.origin_add(origins)		swh_storage.origin_add(origins)

# Add some random visits within the selection range		# Add some random visits within the selection range
visits = self._generate_random_visits()		visits = self._generate_random_visits()
visit_type = "git"		visit_type = "git"

▲ Show 20 Lines • Show All 217 Lines • ▼ Show 20 Lines	def test_origin_visit_add(self, swh_storage, sample_data):
ovs2 = OriginVisitStatus(		ovs2 = OriginVisitStatus(
origin=origin1.url,		origin=origin1.url,
visit=ov2.visit,		visit=ov2.visit,
date=date_visit2,		date=date_visit2,
status="created",		status="created",
snapshot=None,		snapshot=None,
)		)

actual_origin_visits = list(swh_storage.origin_visit_get(origin1.url))		actual_visits = swh_storage.origin_visit_get(origin1.url).results
expected_visits = [		expected_visits = [ov1, ov2]
{**ovs1.to_dict(), "type": ov1.type},		assert len(expected_visits) == len(actual_visits)
{**ovs2.to_dict(), "type": ov2.type},
]

assert len(expected_visits) == len(actual_origin_visits)

for visit in expected_visits:		for visit in expected_visits:
		vlorentzUnsubmitted Not Done Inline Actions equivalent to: assert set(actual_visits) == {ov1, ov2} vlorentz: equivalent to: ``` assert set(actual_visits) == {ov1, ov2} ```
		vlorentzUnsubmitted Not Done Inline Actions you forgot this comment vlorentz: you forgot this comment
assert visit in actual_origin_visits		assert visit in actual_visits

actual_objects = list(swh_storage.journal_writer.journal.objects)		actual_objects = list(swh_storage.journal_writer.journal.objects)
expected_objects = list(		expected_objects = list(
[("origin", origin1)]		[("origin", origin1)]
+ [("origin_visit", visit) for visit in [ov1, ov2]] * 2		+ [("origin_visit", visit) for visit in expected_visits] * 2
+ [("origin_visit_status", ovs) for ovs in [ovs1, ovs2]]		+ [("origin_visit_status", ovs) for ovs in [ovs1, ovs2]]
)		)

for obj in expected_objects:		for obj in expected_objects:
assert obj in actual_objects		assert obj in actual_objects

def test_origin_visit_add_validation(self, swh_storage, sample_data):		def test_origin_visit_add_validation(self, swh_storage, sample_data):
"""Unknown origin when adding visits should raise"""		"""Unknown origin when adding visits should raise"""
▲ Show 20 Lines • Show All 126 Lines • ▼ Show 20 Lines	def test_origin_visit_status_add_twice(self, swh_storage, sample_data):

ovs1 = OriginVisitStatus(		ovs1 = OriginVisitStatus(
origin=origin1.url,		origin=origin1.url,
visit=ov1.visit,		visit=ov1.visit,
date=sample_data.date_visit1,		date=sample_data.date_visit1,
status="created",		status="created",
snapshot=None,		snapshot=None,
)		)
date_visit_now = now()		date_visit_now = round_to_milliseconds(now())
visit_status1 = OriginVisitStatus(		visit_status1 = OriginVisitStatus(
origin=ov1.origin,		origin=ov1.origin,
visit=ov1.visit,		visit=ov1.visit,
date=date_visit_now,		date=date_visit_now,
status="full",		status="full",
snapshot=snapshot.id,		snapshot=snapshot.id,
)		)

swh_storage.origin_visit_status_add([visit_status1])		swh_storage.origin_visit_status_add([visit_status1])
# second call will ignore existing entries (will send to storage though)		# second call will ignore existing entries (will send to storage though)
swh_storage.origin_visit_status_add([visit_status1])		swh_storage.origin_visit_status_add([visit_status1])

origin_visits = list(swh_storage.origin_visit_get(ov1.origin))		visit_status = swh_storage.origin_visit_status_get_latest(ov1.origin, ov1.visit)
		assert visit_status == visit_status1
assert len(origin_visits) == 1
origin_visit1 = origin_visits[0]
assert origin_visit1
assert origin_visit1["status"] == "full"
assert origin_visit1["snapshot"] == snapshot.id

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

expected_origins = [origin1]		expected_origins = [origin1]
expected_visits = [ov1]		expected_visits = [ov1]
expected_visit_statuses = [ovs1, visit_status1, visit_status1]		expected_visit_statuses = [ovs1, visit_status1, visit_status1]

# write twice in the journal		# write twice in the journal
▲ Show 20 Lines • Show All 2,465 Lines • Show Last 20 Lines