Differential D3648 Diff 12849 swh/storage/tests/algos/test_origin.py

Changeset View

Standalone View

swh/storage/tests/algos/test_origin.py

# Copyright (C) 2019-2020 The Software Heritage developers		# Copyright (C) 2019-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

		import copy
		import datetime
import pytest		import pytest

from unittest.mock import patch		from unittest.mock import patch

from swh.model.model import Origin, OriginVisit, OriginVisitStatus		from swh.model.model import Origin, OriginVisit, OriginVisitStatus

from swh.storage.algos.origin import iter_origins, origin_get_latest_visit_status		from swh.storage.algos.origin import (
		iter_origins,
		origin_get_latest_visit_status,
		iter_origin_visits,
		iter_origin_visit_statuses,
		)
		from swh.storage.interface import ListOrder
from swh.storage.utils import now		from swh.storage.utils import now

from swh.storage.tests.test_storage import round_to_milliseconds		from swh.storage.tests.test_storage import round_to_milliseconds


def assert_list_eq(left, right, msg=None):		def assert_list_eq(left, right, msg=None):
assert list(left) == list(right), msg		assert list(left) == list(right), msg

▲ Show 20 Lines • Show All 293 Lines • ▼ Show 20 Lines	def test_origin_get_latest_visit_status_filter_snapshot(swh_storage, sample_data):
# Requiring the latest visit with a snapshot, we still find the previous visit		# Requiring the latest visit with a snapshot, we still find the previous visit
ov2, ovs22 = origin_get_latest_visit_status(		ov2, ovs22 = origin_get_latest_visit_status(
swh_storage, origin2.url, require_snapshot=True		swh_storage, origin2.url, require_snapshot=True
)		)
assert actual_ov2.origin == ov2.origin		assert actual_ov2.origin == ov2.origin
assert actual_ov2.visit == ov2.visit		assert actual_ov2.visit == ov2.visit
assert actual_ov2.type == ov2.type		assert actual_ov2.type == ov2.type
assert actual_ovs22 == ovs22		assert actual_ovs22 == ovs22


		def test_iter_origin_visits(swh_storage, sample_data):
		"""Iter over origin visits for an origin returns all visits"""
		origin1, origin2 = sample_data.origins[:2]
		swh_storage.origin_add([origin1, origin2])

		date_past = now() - datetime.timedelta(weeks=20)

		new_visits = []
		for visit_id in range(20, 0, -1): # will trigger pagination
		visit = OriginVisit(
		origin=origin1.url,
		date=date_past - datetime.timedelta(days=visit_id),
		type="git",
		)
		new_visits.append(visit)
		visits = swh_storage.origin_visit_add(new_visits)
		reverse_visits = copy.deepcopy(visits)
		reverse_visits.reverse()
		vlorentzUnsubmitted Not Done Inline Actions The order is confusing, because "reverse_visits" is the list in chronological order. Do this: for visit_id in range(20): visit = OriginVisit( ... date=date_past + datetime.timedelta(days=visit_id), ) visits = swh_storage.origin_visit_add(new_visits) reversed_visits = list(reversed(visits)) And swap visits and reversed_visits in the following lines vlorentz: The order is confusing, because "reverse_visits" is the list in chronological order. Do this…

		# no limit, order asc
		actual_visits = list(iter_origin_visits(swh_storage, origin1.url))
		assert actual_visits == visits

		# no limit, order desc
		actual_visits = list(
		iter_origin_visits(swh_storage, origin1.url, order=ListOrder.DESC)
		)
		assert actual_visits == reverse_visits

		# no result
		actual_visits = list(iter_origin_visits(swh_storage, origin2.url))
		assert actual_visits == []


		def test_iter_origin_visit_status(swh_storage, sample_data):
		origin1, origin2 = sample_data.origins[:2]
		swh_storage.origin_add([origin1])

		ov1 = swh_storage.origin_visit_add([sample_data.origin_visit])[0]
		assert ov1.origin == origin1.url

		date_past = now() - datetime.timedelta(weeks=20)

		ovs1 = OriginVisitStatus(
		origin=origin1.url,
		visit=ov1.visit,
		date=ov1.date,
		status="created",
		snapshot=None,
		)
		new_visit_statuses = [ovs1]
		for i in range(20, 0, -1): # will trigger pagination
		status_date = date_past - datetime.timedelta(days=i)
		visit_status = OriginVisitStatus(
		origin=origin1.url,
		visit=ov1.visit,
		date=status_date,
		status="created",
		snapshot=None,
		)
		new_visit_statuses.append(visit_status)

		visit_statuses = swh_storage.origin_visit_add(new_visit_statuses)
		reverse_visit_statuses = copy.deepcopy(visit_statuses)
		reverse_visit_statuses.reverse()

		# order asc
		actual_visit_statuses = list(
		iter_origin_visit_statuses(swh_storage, ov1.origin, ov1.visit)
		)
		assert actual_visit_statuses == visit_statuses

		# order desc
		actual_visit_statuses = list(
		iter_origin_visit_statuses(
		swh_storage, ov1.origin, ov1.visit, order=ListOrder.DESC
		)
		)
		assert actual_visit_statuses == reverse_visit_statuses

		# no result
		actual_visit_statuses = list(
		iter_origin_visit_statuses(swh_storage, origin2.url, ov1.visit)
		)
		assert actual_visit_statuses == []