Differential D627 Diff 1930 swh/web/tests/algos/test_revisions_walker.py

Changeset View

Standalone View

swh/web/tests/algos/test_revisions_walker.py

This file was added.

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

				import heapq

				from collections import deque
				from unittest.mock import patch

				from swh.web.algos.revisions_walker import get_revisions_walker
				from swh.web.common.utils import parse_timestamp
				from swh.web.tests.testcase import SWHWebTestCase

				from .data.revisions_walker_data import test_revisions


				def _get_rev(rev_id):
				return test_revisions[rev_id]


				def _get_rev_log(rev_id, limit=1):
				return [_get_rev(rev_id)]


				def _get_timestamp(rev):
				return parse_timestamp(rev['committer_date']).timestamp()


				_rev_start = 'bab5c80b211035739997ebd361a679fa85b39465'
				_max_revs = 20


				@patch('swh.web.algos.revisions_walker.service')
				class RevisionsWalkerTest(SWHWebTestCase):

				def test_revisions_walker_committer_date(self, mock_service):
				mock_service.lookup_revision_log.side_effect = _get_rev_log
				revs_walker = get_revisions_walker('committer_date', _rev_start,
				max_revs=_max_revs)

				revs_queue = []
				revs_committer_date = []
				rev_start = _get_rev(_rev_start)
				commit_time = _get_timestamp(rev_start)
				heapq.heappush(revs_queue, (-commit_time, _rev_start))
				while len(revs_committer_date) < _max_revs:
				_, rev_id = heapq.heappop(revs_queue)
				rev = _get_rev(rev_id)
				revs_committer_date.append(rev)
				for p_rev_id in rev['parents']:
				commit_time = _get_timestamp(_get_rev(p_rev_id))
				heapq.heappush(revs_queue, (-commit_time, p_rev_id))

				self.assertEqual(list(revs_walker), revs_committer_date[0:_max_revs])

				def test_revisions_walker_bfs(self, mock_service):
				mock_service.lookup_revision_log.side_effect = _get_rev_log
				revs_walker = get_revisions_walker('bfs', _rev_start,
				max_revs=_max_revs)

				revs_queue = deque([_get_rev(_rev_start)])
				revs_bfs = []
				while len(revs_bfs) < _max_revs:
				rev = revs_queue.popleft()
				revs_bfs.append(rev)
				for p_rev_id in rev['parents']:
				revs_queue.append(_get_rev(p_rev_id))

				self.assertEqual(list(revs_walker), revs_bfs[0:_max_revs])

				def test_revisions_walker_dfs(self, mock_service):
				mock_service.lookup_revision_log.side_effect = _get_rev_log
				revs_walker = get_revisions_walker('dfs', _rev_start,
				max_revs=_max_revs)

				revs_stack = [_get_rev(_rev_start)]
				revs_dfs = []
				while len(revs_dfs) < _max_revs:
				rev = revs_stack.pop()
				revs_dfs.append(rev)
				for p_rev_id in reversed(rev['parents']):
				revs_stack.append(_get_rev(p_rev_id))

				self.assertEqual(list(revs_walker), revs_dfs[0:_max_revs])

				def test_revisions_walker_dfs_post(self, mock_service):
				mock_service.lookup_revision_log.side_effect = _get_rev_log
				revs_walker = get_revisions_walker('dfs_post', _rev_start,
				max_revs=_max_revs)

				revs_stack = [_get_rev(_rev_start)]
				revs_dfs_post = []
				while len(revs_dfs_post) < _max_revs:
				rev = revs_stack.pop()
				revs_dfs_post.append(rev)
				for p_rev_id in rev['parents']:
				revs_stack.append(_get_rev(p_rev_id))

				self.assertEqual(list(revs_walker), revs_dfs_post[0:_max_revs])