test_service.py
No OneTemporary
Actions

Size

69 KB

Subscribers

None

test_service.py
View Options

	# Copyright (C) 2015-2017 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 datetime
	import unittest

	from nose.tools import istest
	from unittest.mock import MagicMock, patch, call

	from swh.model.hashutil import hash_to_bytes, hash_to_hex

	from swh.web.common import service
	from swh.web.common.exc import BadInputExc, NotFoundExc


	class ServiceTestCase(unittest.TestCase):

	def setUp(self):
	self.BLAKE2S256_SAMPLE = ('685395c5dc57cada459364f0946d3dd45b'
	'ad5fcbabc1048edb44380f1d31d0aa')
	self.BLAKE2S256_SAMPLE_BIN = hash_to_bytes(self.BLAKE2S256_SAMPLE)
	self.SHA1_SAMPLE = '40e71b8614fcd89ccd17ca2b1d9e66c5b00a6d03'
	self.SHA1_SAMPLE_BIN = hash_to_bytes(self.SHA1_SAMPLE)
	self.SHA256_SAMPLE = ('8abb0aa566452620ecce816eecdef4792d77a'
	'293ad8ea82a4d5ecb4d36f7e560')
	self.SHA256_SAMPLE_BIN = hash_to_bytes(self.SHA256_SAMPLE)
	self.SHA1GIT_SAMPLE = '25d1a2e8f32937b0f498a5ca87f823d8df013c01'
	self.SHA1GIT_SAMPLE_BIN = hash_to_bytes(self.SHA1GIT_SAMPLE)
	self.DIRECTORY_ID = '7834ef7e7c357ce2af928115c6c6a42b7e2a44e6'
	self.DIRECTORY_ID_BIN = hash_to_bytes(self.DIRECTORY_ID)
	self.AUTHOR_ID_BIN = {
	'name': b'author',
	'email': b'author@company.org',
	}
	self.AUTHOR_ID = {
	'name': 'author',
	'email': 'author@company.org',
	}
	self.COMMITTER_ID_BIN = {
	'name': b'committer',
	'email': b'committer@corp.org',
	}
	self.COMMITTER_ID = {
	'name': 'committer',
	'email': 'committer@corp.org',
	}
	self.SAMPLE_DATE_RAW = {
	'timestamp': datetime.datetime(
	2000, 1, 17, 11, 23, 54,
	tzinfo=datetime.timezone.utc,
	).timestamp(),
	'offset': 0,
	'negative_utc': False,
	}
	self.SAMPLE_DATE = '2000-01-17T11:23:54+00:00'
	self.SAMPLE_MESSAGE_BIN = b'elegant fix for bug 31415957'
	self.SAMPLE_MESSAGE = 'elegant fix for bug 31415957'

	self.SAMPLE_REVISION = {
	'id': self.SHA1_SAMPLE,
	'directory': self.DIRECTORY_ID,
	'author': self.AUTHOR_ID,
	'committer': self.COMMITTER_ID,
	'message': self.SAMPLE_MESSAGE,
	'date': self.SAMPLE_DATE,
	'committer_date': self.SAMPLE_DATE,
	'synthetic': False,
	'type': 'git',
	'parents': [],
	'metadata': {},
	'merge': False
	}
	self.SAMPLE_REVISION_RAW = {
	'id': self.SHA1_SAMPLE_BIN,
	'directory': self.DIRECTORY_ID_BIN,
	'author': self.AUTHOR_ID_BIN,
	'committer': self.COMMITTER_ID_BIN,
	'message': self.SAMPLE_MESSAGE_BIN,
	'date': self.SAMPLE_DATE_RAW,
	'committer_date': self.SAMPLE_DATE_RAW,
	'synthetic': False,
	'type': 'git',
	'parents': [],
	'metadata': [],
	}

	self.SAMPLE_CONTENT = {
	'checksums': {
	'blake2s256': self.BLAKE2S256_SAMPLE,
	'sha1': self.SHA1_SAMPLE,
	'sha256': self.SHA256_SAMPLE,
	'sha1_git': self.SHA1GIT_SAMPLE,
	},
	'length': 190,
	'status': 'absent'
	}
	self.SAMPLE_CONTENT_RAW = {
	'blake2s256': self.BLAKE2S256_SAMPLE_BIN,
	'sha1': self.SHA1_SAMPLE_BIN,
	'sha256': self.SHA256_SAMPLE_BIN,
	'sha1_git': self.SHA1GIT_SAMPLE_BIN,
	'length': 190,
	'status': 'hidden'
	}

	self.date_origin_visit1 = datetime.datetime(
	2015, 1, 1, 22, 0, 0,
	tzinfo=datetime.timezone.utc)

	self.origin_visit1 = {
	'date': self.date_origin_visit1,
	'origin': 1,
	'visit': 1
	}

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_multiple_hashes_ball_missing(self, mock_storage):
	# given
	mock_storage.content_missing_per_sha1 = MagicMock(return_value=[])

	# when
	actual_lookup = service.lookup_multiple_hashes(
	[{'filename': 'a',
	'sha1': '456caf10e9535160d90e874b45aa426de762f19f'},
	{'filename': 'b',
	'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}])

	# then
	self.assertEquals(actual_lookup, [
	{'filename': 'a',
	'sha1': '456caf10e9535160d90e874b45aa426de762f19f',
	'found': True},
	{'filename': 'b',
	'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865',
	'found': True}
	])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_multiple_hashes_some_missing(self, mock_storage):
	# given
	mock_storage.content_missing_per_sha1 = MagicMock(return_value=[
	hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')
	])

	# when
	actual_lookup = service.lookup_multiple_hashes(
	[{'filename': 'a',
	'sha1': '456caf10e9535160d90e874b45aa426de762f19f'},
	{'filename': 'b',
	'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}])

	# then
	self.assertEquals(actual_lookup, [
	{'filename': 'a',
	'sha1': '456caf10e9535160d90e874b45aa426de762f19f',
	'found': False},
	{'filename': 'b',
	'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865',
	'found': True}
	])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_hash_does_not_exist(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(return_value=None)

	# when
	actual_lookup = service.lookup_hash(
	'sha1_git:123caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEquals({'found': None,
	'algo': 'sha1_git'}, actual_lookup)

	# check the function has been called with parameters
	mock_storage.content_find.assert_called_with(
	{'sha1_git':
	hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')})

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_hash_exist(self, mock_storage):
	# given
	stub_content = {
	'sha1': hash_to_bytes(
	'456caf10e9535160d90e874b45aa426de762f19f')
	}
	mock_storage.content_find = MagicMock(return_value=stub_content)

	# when
	actual_lookup = service.lookup_hash(
	'sha1:456caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEquals({'found': stub_content,
	'algo': 'sha1'}, actual_lookup)

	mock_storage.content_find.assert_called_with(
	{'sha1':
	hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')}
	)

	@patch('swh.web.common.service.storage')
	@istest
	def search_hash_does_not_exist(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(return_value=None)

	# when
	actual_lookup = service.search_hash(
	'sha1_git:123caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEquals({'found': False}, actual_lookup)

	# check the function has been called with parameters
	mock_storage.content_find.assert_called_with(
	{'sha1_git':
	hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')})

	@patch('swh.web.common.service.storage')
	@istest
	def search_hash_exist(self, mock_storage):
	# given
	stub_content = {
	'sha1': hash_to_bytes(
	'456caf10e9535160d90e874b45aa426de762f19f')
	}
	mock_storage.content_find = MagicMock(return_value=stub_content)

	# when
	actual_lookup = service.search_hash(
	'sha1:456caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEquals({'found': True}, actual_lookup)

	mock_storage.content_find.assert_called_with(
	{'sha1':
	hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')},
	)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_ctags(self, mock_storage):
	# given
	mock_storage.content_ctags_get = MagicMock(
	return_value=[{
	'id': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'line': 100,
	'name': 'hello',
	'kind': 'function',
	'tool_name': 'ctags',
	'tool_version': 'some-version',
	}])
	expected_ctags = [{
	'id': '123caf10e9535160d90e874b45aa426de762f19f',
	'line': 100,
	'name': 'hello',
	'kind': 'function',
	'tool_name': 'ctags',
	'tool_version': 'some-version',
	}]

	# when
	actual_ctags = list(service.lookup_content_ctags(
	'sha1:123caf10e9535160d90e874b45aa426de762f19f'))

	# then
	self.assertEqual(actual_ctags, expected_ctags)

	mock_storage.content_ctags_get.assert_called_with(
	[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_ctags_no_hash(self, mock_storage):
	# given
	mock_storage.content_find.return_value = None
	mock_storage.content_ctags_get = MagicMock(
	return_value=None)

	# when
	actual_ctags = list(service.lookup_content_ctags(
	'sha1_git:123caf10e9535160d90e874b45aa426de762f19f'))

	# then
	self.assertEqual(actual_ctags, [])

	mock_storage.content_find.assert_called_once_with(
	{'sha1_git': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f')})

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_filetype(self, mock_storage):
	# given
	mock_storage.content_mimetype_get = MagicMock(
	return_value=[{
	'id': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'mimetype': b'text/x-c++',
	'encoding': b'us-ascii',
	}])
	expected_filetype = {
	'id': '123caf10e9535160d90e874b45aa426de762f19f',
	'mimetype': 'text/x-c++',
	'encoding': 'us-ascii',
	}

	# when
	actual_filetype = service.lookup_content_filetype(
	'sha1:123caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEqual(actual_filetype, expected_filetype)

	mock_storage.content_mimetype_get.assert_called_with(
	[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_filetype_2(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(
	return_value={
	'sha1': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f')
	}
	)
	mock_storage.content_mimetype_get = MagicMock(
	return_value=[{
	'id': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'mimetype': b'text/x-python',
	'encoding': b'us-ascii',
	}]
	)
	expected_filetype = {
	'id': '123caf10e9535160d90e874b45aa426de762f19f',
	'mimetype': 'text/x-python',
	'encoding': 'us-ascii',
	}

	# when
	actual_filetype = service.lookup_content_filetype(
	'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEqual(actual_filetype, expected_filetype)

	mock_storage.content_find(
	'sha1_git', hash_to_bytes(
	'456caf10e9535160d90e874b45aa426de762f19f')
	)
	mock_storage.content_mimetype_get.assert_called_with(
	[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_language(self, mock_storage):
	# given
	mock_storage.content_language_get = MagicMock(
	return_value=[{
	'id': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'lang': 'python',
	}])
	expected_language = {
	'id': '123caf10e9535160d90e874b45aa426de762f19f',
	'lang': 'python',
	}

	# when
	actual_language = service.lookup_content_language(
	'sha1:123caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEqual(actual_language, expected_language)

	mock_storage.content_language_get.assert_called_with(
	[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_language_2(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(
	return_value={
	'sha1': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f')
	}
	)
	mock_storage.content_language_get = MagicMock(
	return_value=[{
	'id': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'lang': 'haskell',
	}]
	)
	expected_language = {
	'id': '123caf10e9535160d90e874b45aa426de762f19f',
	'lang': 'haskell',
	}

	# when
	actual_language = service.lookup_content_language(
	'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEqual(actual_language, expected_language)

	mock_storage.content_find(
	'sha1_git', hash_to_bytes(
	'456caf10e9535160d90e874b45aa426de762f19f')
	)
	mock_storage.content_language_get.assert_called_with(
	[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_expression(self, mock_storage):
	# given
	mock_storage.content_ctags_search = MagicMock(
	return_value=[{
	'id': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'name': 'foobar',
	'kind': 'variable',
	'lang': 'C',
	'line': 10
	}])
	expected_ctags = [{
	'sha1': '123caf10e9535160d90e874b45aa426de762f19f',
	'name': 'foobar',
	'kind': 'variable',
	'lang': 'C',
	'line': 10
	}]

	# when
	actual_ctags = list(service.lookup_expression(
	'foobar', last_sha1='hash', per_page=10))

	# then
	self.assertEqual(actual_ctags, expected_ctags)

	mock_storage.content_ctags_search.assert_called_with(
	'foobar', last_sha1='hash', limit=10)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_expression_no_result(self, mock_storage):
	# given
	mock_storage.content_ctags_search = MagicMock(
	return_value=[])
	expected_ctags = []

	# when
	actual_ctags = list(service.lookup_expression(
	'barfoo', last_sha1='hash', per_page=10))

	# then
	self.assertEqual(actual_ctags, expected_ctags)

	mock_storage.content_ctags_search.assert_called_with(
	'barfoo', last_sha1='hash', limit=10)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_license(self, mock_storage):
	# given
	mock_storage.content_fossology_license_get = MagicMock(
	return_value=[{
	'id': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'lang': 'python',
	}])
	expected_license = {
	'id': '123caf10e9535160d90e874b45aa426de762f19f',
	'lang': 'python',
	}

	# when
	actual_license = service.lookup_content_license(
	'sha1:123caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEqual(actual_license, expected_license)

	mock_storage.content_fossology_license_get.assert_called_with(
	[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_license_2(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(
	return_value={
	'sha1': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f')
	}
	)
	mock_storage.content_fossology_license_get = MagicMock(
	return_value=[{
	'id': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'lang': 'haskell',
	}]
	)
	expected_license = {
	'id': '123caf10e9535160d90e874b45aa426de762f19f',
	'lang': 'haskell',
	}

	# when
	actual_license = service.lookup_content_license(
	'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEqual(actual_license, expected_license)

	mock_storage.content_find(
	'sha1_git', hash_to_bytes(
	'456caf10e9535160d90e874b45aa426de762f19f')
	)
	mock_storage.content_fossology_license_get.assert_called_with(
	[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_provenance(self, mock_storage):
	# given
	mock_storage.content_find_provenance = MagicMock(
	return_value=(p for p in [{
	'content': hash_to_bytes(
	'123caf10e9535160d90e874b45aa426de762f19f'),
	'revision': hash_to_bytes(
	'456caf10e9535160d90e874b45aa426de762f19f'),
	'origin': 100,
	'visit': 1,
	'path': b'octavio-3.4.0/octave.html/doc_002dS_005fISREG.html'
	}]))
	expected_provenances = [{
	'content': '123caf10e9535160d90e874b45aa426de762f19f',
	'revision': '456caf10e9535160d90e874b45aa426de762f19f',
	'origin': 100,
	'visit': 1,
	'path': 'octavio-3.4.0/octave.html/doc_002dS_005fISREG.html'
	}]

	# when
	actual_provenances = service.lookup_content_provenance(
	'sha1_git:123caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertEqual(list(actual_provenances), expected_provenances)

	mock_storage.content_find_provenance.assert_called_with(
	{'sha1_git':
	hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')})

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_provenance_not_found(self, mock_storage):
	# given
	mock_storage.content_find_provenance = MagicMock(return_value=None)

	# when
	actual_provenances = service.lookup_content_provenance(
	'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')

	# then
	self.assertIsNone(actual_provenances)

	mock_storage.content_find_provenance.assert_called_with(
	{'sha1_git':
	hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')})

	@patch('swh.web.common.service.storage')
	@istest
	def stat_counters(self, mock_storage):
	# given
	input_stats = {
	"content": 1770830,
	"directory": 211683,
	"directory_entry_dir": 209167,
	"directory_entry_file": 1807094,
	"directory_entry_rev": 0,
	"entity": 0,
	"entity_history": 0,
	"occurrence": 0,
	"occurrence_history": 19600,
	"origin": 1096,
	"person": 0,
	"release": 8584,
	"revision": 7792,
	"revision_history": 0,
	"skipped_content": 0
	}
	mock_storage.stat_counters = MagicMock(return_value=input_stats)

	# when
	actual_stats = service.stat_counters()

	# then
	expected_stats = input_stats
	self.assertEqual(actual_stats, expected_stats)

	mock_storage.stat_counters.assert_called_with()

	@patch('swh.web.common.service._lookup_origin_visits')
	@istest
	def lookup_origin_visits(self, mock_lookup_visits):
	# given
	date_origin_visit2 = datetime.datetime(
	2013, 7, 1, 20, 0, 0,
	tzinfo=datetime.timezone.utc)

	date_origin_visit3 = datetime.datetime(
	2015, 1, 1, 21, 0, 0,
	tzinfo=datetime.timezone.utc)
	stub_result = [self.origin_visit1, {
	'date': date_origin_visit2,
	'origin': 1,
	'visit': 2,
	'target': hash_to_bytes(
	'65a55bbdf3629f916219feb3dcc7393ded1bc8db'),
	'branch': b'master',
	'target_type': 'release',
	'metadata': None,
	}, {
	'date': date_origin_visit3,
	'origin': 1,
	'visit': 3
	}]
	mock_lookup_visits.return_value = stub_result

	# when
	expected_origin_visits = [{
	'date': self.origin_visit1['date'].isoformat(),
	'origin': self.origin_visit1['origin'],
	'visit': self.origin_visit1['visit']
	}, {
	'date': date_origin_visit2.isoformat(),
	'origin': 1,
	'visit': 2,
	'target': '65a55bbdf3629f916219feb3dcc7393ded1bc8db',
	'branch': 'master',
	'target_type': 'release',
	'metadata': {},
	}, {
	'date': date_origin_visit3.isoformat(),
	'origin': 1,
	'visit': 3
	}]

	actual_origin_visits = service.lookup_origin_visits(6)

	# then
	self.assertEqual(list(actual_origin_visits), expected_origin_visits)

	mock_lookup_visits.assert_called_once_with(
	6, last_visit=None, limit=10)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_origin_visit(self, mock_storage):
	# given
	stub_result = self.origin_visit1
	mock_storage.origin_visit_get_by.return_value = stub_result

	expected_origin_visit = {
	'date': self.origin_visit1['date'].isoformat(),
	'origin': self.origin_visit1['origin'],
	'visit': self.origin_visit1['visit']
	}

	# when
	actual_origin_visit = service.lookup_origin_visit(1, 1)

	# then
	self.assertEqual(actual_origin_visit, expected_origin_visit)

	mock_storage.origin_visit_get_by.assert_called_once_with(1, 1)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_origin(self, mock_storage):
	# given
	mock_storage.origin_get = MagicMock(return_value={
	'id': 'origin-id',
	'lister': 'uuid-lister',
	'project': 'uuid-project',
	'url': 'ftp://some/url/to/origin',
	'type': 'ftp'})

	# when
	actual_origin = service.lookup_origin({'id': 'origin-id'})

	# then
	self.assertEqual(actual_origin, {'id': 'origin-id',
	'lister': 'uuid-lister',
	'project': 'uuid-project',
	'url': 'ftp://some/url/to/origin',
	'type': 'ftp'})

	mock_storage.origin_get.assert_called_with({'id': 'origin-id'})

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_release_ko_id_checksum_not_ok_because_not_a_sha1(self,
	mock_storage):
	# given
	mock_storage.release_get = MagicMock()

	with self.assertRaises(BadInputExc) as cm:
	# when
	service.lookup_release('not-a-sha1')
	self.assertIn('invalid checksum', cm.exception.args[0])

	mock_storage.release_get.called = False

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_release_ko_id_checksum_ok_but_not_a_sha1(self, mock_storage):
	# given
	mock_storage.release_get = MagicMock()

	# when
	with self.assertRaises(BadInputExc) as cm:
	service.lookup_release(
	'13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4daf5'
	'1aea892abe')
	self.assertIn('sha1_git supported', cm.exception.args[0])

	mock_storage.release_get.called = False

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_directory_with_path_not_found(self, mock_storage):
	# given
	mock_storage.lookup_directory_with_path = MagicMock(return_value=None)

	sha1_git = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'

	# when
	actual_directory = mock_storage.lookup_directory_with_path(
	sha1_git, 'some/path/here')

	self.assertIsNone(actual_directory)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_directory_with_path_found(self, mock_storage):
	# given
	sha1_git = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
	entry = {'id': 'dir-id',
	'type': 'dir',
	'name': 'some/path/foo'}

	mock_storage.lookup_directory_with_path = MagicMock(return_value=entry)

	# when
	actual_directory = mock_storage.lookup_directory_with_path(
	sha1_git, 'some/path/here')

	self.assertEqual(entry, actual_directory)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_release(self, mock_storage):
	# given
	mock_storage.release_get = MagicMock(return_value=[{
	'id': hash_to_bytes('65a55bbdf3629f916219feb3dcc7393ded1bc8db'),
	'target': None,
	'date': {
	'timestamp': datetime.datetime(
	2015, 1, 1, 22, 0, 0,
	tzinfo=datetime.timezone.utc).timestamp(),
	'offset': 0,
	'negative_utc': True,
	},
	'name': b'v0.0.1',
	'message': b'synthetic release',
	'synthetic': True,
	}])

	# when
	actual_release = service.lookup_release(
	'65a55bbdf3629f916219feb3dcc7393ded1bc8db')

	# then
	self.assertEqual(actual_release, {
	'id': '65a55bbdf3629f916219feb3dcc7393ded1bc8db',
	'target': None,
	'date': '2015-01-01T22:00:00-00:00',
	'name': 'v0.0.1',
	'message': 'synthetic release',
	'synthetic': True,
	})

	mock_storage.release_get.assert_called_with(
	[hash_to_bytes('65a55bbdf3629f916219feb3dcc7393ded1bc8db')])

	@istest
	def lookup_revision_with_context_ko_not_a_sha1_1(self):
	# given
	sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4' \
	'daf51aea892abe'
	sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'

	# when
	with self.assertRaises(BadInputExc) as cm:
	service.lookup_revision_with_context(sha1_git_root, sha1_git)
	self.assertIn('Only sha1_git is supported', cm.exception.args[0])

	@istest
	def lookup_revision_with_context_ko_not_a_sha1_2(self):
	# given
	sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
	sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f6' \
	'2d4daf51aea892abe'

	# when
	with self.assertRaises(BadInputExc) as cm:
	service.lookup_revision_with_context(sha1_git_root, sha1_git)
	self.assertIn('Only sha1_git is supported', cm.exception.args[0])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_with_context_ko_sha1_git_does_not_exist(
	self,
	mock_storage):
	# given
	sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
	sha1_git = '777777bdf3629f916219feb3dcc7393ded1bc8db'

	sha1_git_bin = hash_to_bytes(sha1_git)

	mock_storage.revision_get.return_value = None

	# when
	with self.assertRaises(NotFoundExc) as cm:
	service.lookup_revision_with_context(sha1_git_root, sha1_git)
	self.assertIn('Revision 777777bdf3629f916219feb3dcc7393ded1bc8db'
	' not found', cm.exception.args[0])

	mock_storage.revision_get.assert_called_once_with(
	[sha1_git_bin])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_with_context_ko_root_sha1_git_does_not_exist(
	self,
	mock_storage):
	# given
	sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
	sha1_git = '777777bdf3629f916219feb3dcc7393ded1bc8db'

	sha1_git_root_bin = hash_to_bytes(sha1_git_root)
	sha1_git_bin = hash_to_bytes(sha1_git)

	mock_storage.revision_get.side_effect = ['foo', None]

	# when
	with self.assertRaises(NotFoundExc) as cm:
	service.lookup_revision_with_context(sha1_git_root, sha1_git)
	self.assertIn('Revision 65a55bbdf3629f916219feb3dcc7393ded1bc8db'
	' not found', cm.exception.args[0])

	mock_storage.revision_get.assert_has_calls([call([sha1_git_bin]),
	call([sha1_git_root_bin])])

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_revision_with_context(self, mock_query, mock_storage):
	# given
	sha1_git_root = '666'
	sha1_git = '883'

	sha1_git_root_bin = b'666'
	sha1_git_bin = b'883'

	sha1_git_root_dict = {
	'id': sha1_git_root_bin,
	'parents': [b'999'],
	}
	sha1_git_dict = {
	'id': sha1_git_bin,
	'parents': [],
	'directory': b'278',
	}

	stub_revisions = [
	sha1_git_root_dict,
	{
	'id': b'999',
	'parents': [b'777', b'883', b'888'],
	},
	{
	'id': b'777',
	'parents': [b'883'],
	},
	sha1_git_dict,
	{
	'id': b'888',
	'parents': [b'889'],
	},
	{
	'id': b'889',
	'parents': [],
	},
	]

	# inputs ok
	mock_query.parse_hash_with_algorithms_or_throws.side_effect = [
	('sha1', sha1_git_bin),
	('sha1', sha1_git_root_bin)
	]

	# lookup revision first 883, then 666 (both exists)
	mock_storage.revision_get.return_value = [
	sha1_git_dict,
	sha1_git_root_dict
	]

	mock_storage.revision_log = MagicMock(
	return_value=stub_revisions)

	# when

	actual_revision = service.lookup_revision_with_context(
	sha1_git_root,
	sha1_git)

	# then
	self.assertEquals(actual_revision, {
	'id': hash_to_hex(sha1_git_bin),
	'parents': [],
	'children': [hash_to_hex(b'999'), hash_to_hex(b'777')],
	'directory': hash_to_hex(b'278'),
	'merge': False
	})

	mock_query.parse_hash_with_algorithms_or_throws.assert_has_calls(
	[call(sha1_git, ['sha1'], 'Only sha1_git is supported.'),
	call(sha1_git_root, ['sha1'], 'Only sha1_git is supported.')])

	mock_storage.revision_log.assert_called_with(
	[sha1_git_root_bin], 100)

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_revision_with_context_sha1_git_root_already_retrieved_as_dict(
	self, mock_query, mock_storage):
	# given
	sha1_git = '883'

	sha1_git_root_bin = b'666'
	sha1_git_bin = b'883'

	sha1_git_root_dict = {
	'id': sha1_git_root_bin,
	'parents': [b'999'],
	}

	sha1_git_dict = {
	'id': sha1_git_bin,
	'parents': [],
	'directory': b'278',
	}

	stub_revisions = [
	sha1_git_root_dict,
	{
	'id': b'999',
	'parents': [b'777', b'883', b'888'],
	},
	{
	'id': b'777',
	'parents': [b'883'],
	},
	sha1_git_dict,
	{
	'id': b'888',
	'parents': [b'889'],
	},
	{
	'id': b'889',
	'parents': [],
	},
	]

	# inputs ok
	mock_query.parse_hash_with_algorithms_or_throws.return_value = (
	'sha1', sha1_git_bin)

	# lookup only on sha1
	mock_storage.revision_get.return_value = [sha1_git_dict]

	mock_storage.revision_log.return_value = stub_revisions

	# when
	actual_revision = service.lookup_revision_with_context(
	{'id': sha1_git_root_bin},
	sha1_git)

	# then
	self.assertEquals(actual_revision, {
	'id': hash_to_hex(sha1_git_bin),
	'parents': [],
	'children': [hash_to_hex(b'999'), hash_to_hex(b'777')],
	'directory': hash_to_hex(b'278'),
	'merge': False
	})

	mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with( # noqa
	sha1_git, ['sha1'], 'Only sha1_git is supported.')

	mock_storage.revision_get.assert_called_once_with([sha1_git_bin])

	mock_storage.revision_log.assert_called_with(
	[sha1_git_root_bin], 100)

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory_with_revision_ko_revision_not_found(self,
	mock_query,
	mock_storage):
	# given
	mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
	b'123')
	mock_storage.revision_get.return_value = None

	# when
	with self.assertRaises(NotFoundExc) as cm:
	service.lookup_directory_with_revision('123')
	self.assertIn('Revision 123 not found', cm.exception.args[0])

	mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
	('123', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.revision_get.assert_called_once_with([b'123'])

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory_with_revision_ko_revision_with_path_to_nowhere(
	self,
	mock_query,
	mock_storage):
	# given
	mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
	b'123')

	dir_id = b'dir-id-as-sha1'
	mock_storage.revision_get.return_value = [{
	'directory': dir_id,
	}]

	mock_storage.directory_entry_get_by_path.return_value = None

	# when
	with self.assertRaises(NotFoundExc) as cm:
	service.lookup_directory_with_revision(
	'123',
	'path/to/something/unknown')
	self.assertIn("Directory/File 'path/to/something/unknown' " +
	"pointed to by revision 123 not found",
	cm.exception.args[0])

	mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
	('123', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.revision_get.assert_called_once_with([b'123'])
	mock_storage.directory_entry_get_by_path.assert_called_once_with(
	b'dir-id-as-sha1', [b'path', b'to', b'something', b'unknown'])

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory_with_revision_ko_type_not_implemented(
	self,
	mock_query,
	mock_storage):

	# given
	mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
	b'123')

	dir_id = b'dir-id-as-sha1'
	mock_storage.revision_get.return_value = [{
	'directory': dir_id,
	}]

	mock_storage.directory_entry_get_by_path.return_value = {
	'type': 'rev',
	'name': b'some/path/to/rev',
	'target': b'456'
	}

	stub_content = {
	'id': b'12',
	'type': 'file'
	}

	mock_storage.content_get.return_value = stub_content

	# when
	with self.assertRaises(NotImplementedError) as cm:
	service.lookup_directory_with_revision(
	'123',
	'some/path/to/rev')
	self.assertIn("Entity of type 'rev' not implemented.",
	cm.exception.args[0])

	# then
	mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
	('123', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.revision_get.assert_called_once_with([b'123'])
	mock_storage.directory_entry_get_by_path.assert_called_once_with(
	b'dir-id-as-sha1', [b'some', b'path', b'to', b'rev'])

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory_with_revision_revision_without_path(self,
	mock_query,
	mock_storage):
	# given
	mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
	b'123')

	dir_id = b'dir-id-as-sha1'
	mock_storage.revision_get.return_value = [{
	'directory': dir_id,
	}]

	stub_dir_entries = [{
	'id': b'123',
	'type': 'dir'
	}, {
	'id': b'456',
	'type': 'file'
	}]

	mock_storage.directory_ls.return_value = stub_dir_entries

	# when
	actual_directory_entries = service.lookup_directory_with_revision(
	'123')

	self.assertEqual(actual_directory_entries['type'], 'dir')
	self.assertEqual(list(actual_directory_entries['content']),
	stub_dir_entries)

	mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
	('123', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.revision_get.assert_called_once_with([b'123'])
	mock_storage.directory_ls.assert_called_once_with(dir_id)

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory_with_revision_revision_with_path_to_dir(self,
	mock_query,
	mock_storage):
	# given
	mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
	b'123')

	dir_id = b'dir-id-as-sha1'
	mock_storage.revision_get.return_value = [{
	'directory': dir_id,
	}]

	stub_dir_entries = [{
	'id': b'12',
	'type': 'dir'
	}, {
	'id': b'34',
	'type': 'file'
	}]

	mock_storage.directory_entry_get_by_path.return_value = {
	'type': 'dir',
	'name': b'some/path',
	'target': b'456'
	}
	mock_storage.directory_ls.return_value = stub_dir_entries

	# when
	actual_directory_entries = service.lookup_directory_with_revision(
	'123',
	'some/path')

	self.assertEqual(actual_directory_entries['type'], 'dir')
	self.assertEqual(actual_directory_entries['revision'], '123')
	self.assertEqual(actual_directory_entries['path'], 'some/path')
	self.assertEqual(list(actual_directory_entries['content']),
	stub_dir_entries)

	mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
	('123', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.revision_get.assert_called_once_with([b'123'])
	mock_storage.directory_entry_get_by_path.assert_called_once_with(
	dir_id,
	[b'some', b'path'])
	mock_storage.directory_ls.assert_called_once_with(b'456')

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory_with_revision_revision_with_path_to_file_without_data(
	self,
	mock_query,
	mock_storage):

	# given
	mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
	b'123')

	dir_id = b'dir-id-as-sha1'
	mock_storage.revision_get.return_value = [{
	'directory': dir_id,
	}]

	mock_storage.directory_entry_get_by_path.return_value = {
	'type': 'file',
	'name': b'some/path/to/file',
	'target': b'789'
	}

	stub_content = {
	'status': 'visible',
	}

	mock_storage.content_find.return_value = stub_content

	# when
	actual_content = service.lookup_directory_with_revision(
	'123',
	'some/path/to/file')

	# then
	self.assertEqual(actual_content, {'type': 'file',
	'revision': '123',
	'path': 'some/path/to/file',
	'content': stub_content})

	mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
	('123', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.revision_get.assert_called_once_with([b'123'])
	mock_storage.directory_entry_get_by_path.assert_called_once_with(
	b'dir-id-as-sha1', [b'some', b'path', b'to', b'file'])
	mock_storage.content_find.assert_called_once_with({'sha1_git': b'789'})

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory_with_revision_revision_with_path_to_file_with_data(
	self,
	mock_query,
	mock_storage):

	# given
	mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
	b'123')

	dir_id = b'dir-id-as-sha1'
	mock_storage.revision_get.return_value = [{
	'directory': dir_id,
	}]

	mock_storage.directory_entry_get_by_path.return_value = {
	'type': 'file',
	'name': b'some/path/to/file',
	'target': b'789'
	}

	stub_content = {
	'status': 'visible',
	'sha1': b'content-sha1'
	}

	mock_storage.content_find.return_value = stub_content
	mock_storage.content_get.return_value = [{
	'sha1': b'content-sha1',
	'data': b'some raw data'
	}]

	expected_content = {
	'status': 'visible',
	'checksums': {
	'sha1': hash_to_hex(b'content-sha1'),
	},
	'data': b'some raw data'
	}

	# when
	actual_content = service.lookup_directory_with_revision(
	'123',
	'some/path/to/file',
	with_data=True)

	# then
	self.assertEqual(actual_content, {'type': 'file',
	'revision': '123',
	'path': 'some/path/to/file',
	'content': expected_content})

	mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
	('123', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.revision_get.assert_called_once_with([b'123'])
	mock_storage.directory_entry_get_by_path.assert_called_once_with(
	b'dir-id-as-sha1', [b'some', b'path', b'to', b'file'])
	mock_storage.content_find.assert_called_once_with({'sha1_git': b'789'})
	mock_storage.content_get.assert_called_once_with([b'content-sha1'])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision(self, mock_storage):
	# given
	mock_storage.revision_get = MagicMock(
	return_value=[self.SAMPLE_REVISION_RAW])

	# when
	actual_revision = service.lookup_revision(
	self.SHA1_SAMPLE)

	# then
	self.assertEqual(actual_revision, self.SAMPLE_REVISION)

	mock_storage.revision_get.assert_called_with(
	[self.SHA1_SAMPLE_BIN])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_invalid_msg(self, mock_storage):
	# given
	stub_rev = self.SAMPLE_REVISION_RAW
	stub_rev['message'] = b'elegant fix for bug \xff'

	expected_revision = self.SAMPLE_REVISION
	expected_revision['message'] = None
	expected_revision['message_decoding_failed'] = True
	mock_storage.revision_get = MagicMock(return_value=[stub_rev])

	# when
	actual_revision = service.lookup_revision(
	self.SHA1_SAMPLE)

	# then
	self.assertEqual(actual_revision, expected_revision)

	mock_storage.revision_get.assert_called_with(
	[self.SHA1_SAMPLE_BIN])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_msg_ok(self, mock_storage):
	# given
	mock_storage.revision_get.return_value = [self.SAMPLE_REVISION_RAW]

	# when
	rv = service.lookup_revision_message(
	self.SHA1_SAMPLE)

	# then
	self.assertEquals(rv, {'message': self.SAMPLE_MESSAGE_BIN})
	mock_storage.revision_get.assert_called_with(
	[self.SHA1_SAMPLE_BIN])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_msg_absent(self, mock_storage):
	# given
	stub_revision = self.SAMPLE_REVISION_RAW
	del stub_revision['message']
	mock_storage.revision_get.return_value = stub_revision

	# when
	with self.assertRaises(NotFoundExc) as cm:
	service.lookup_revision_message(
	self.SHA1_SAMPLE)

	# then
	mock_storage.revision_get.assert_called_with(
	self.SHA1_SAMPLE_BIN)
	self.assertEqual(cm.exception.args[0], 'No message for revision '
	'with sha1_git '
	'18d8be353ed3480476f032475e7c233eff7371d5.')

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_msg_norev(self, mock_storage):
	# given
	mock_storage.revision_get.return_value = None

	# when
	with self.assertRaises(NotFoundExc) as cm:
	service.lookup_revision_message(
	self.SHA1_SAMPLE)

	# then
	mock_storage.revision_get.assert_called_with(
	self.SHA1_SAMPLE_BIN)
	self.assertEqual(cm.exception.args[0], 'Revision with sha1_git '
	'18d8be353ed3480476f032475e7c233eff7371d5 '
	'not found.')

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_multiple(self, mock_storage):
	# given
	sha1 = self.SHA1_SAMPLE
	sha1_other = 'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'

	stub_revisions = [
	self.SAMPLE_REVISION_RAW,
	{
	'id': hash_to_bytes(sha1_other),
	'directory': 'abcdbe353ed3480476f032475e7c233eff7371d5',
	'author': {
	'name': b'name',
	'email': b'name@surname.org',
	},
	'committer': {
	'name': b'name',
	'email': b'name@surname.org',
	},
	'message': b'ugly fix for bug 42',
	'date': {
	'timestamp': datetime.datetime(
	2000, 1, 12, 5, 23, 54,
	tzinfo=datetime.timezone.utc).timestamp(),
	'offset': 0,
	'negative_utc': False
	},
	'date_offset': 0,
	'committer_date': {
	'timestamp': datetime.datetime(
	2000, 1, 12, 5, 23, 54,
	tzinfo=datetime.timezone.utc).timestamp(),
	'offset': 0,
	'negative_utc': False
	},
	'committer_date_offset': 0,
	'synthetic': False,
	'type': 'git',
	'parents': [],
	'metadata': [],
	}
	]

	mock_storage.revision_get.return_value = stub_revisions

	# when
	actual_revisions = service.lookup_revision_multiple(
	[sha1, sha1_other])

	# then
	self.assertEqual(list(actual_revisions), [
	self.SAMPLE_REVISION,
	{
	'id': sha1_other,
	'directory': 'abcdbe353ed3480476f032475e7c233eff7371d5',
	'author': {
	'name': 'name',
	'email': 'name@surname.org',
	},
	'committer': {
	'name': 'name',
	'email': 'name@surname.org',
	},
	'message': 'ugly fix for bug 42',
	'date': '2000-01-12T05:23:54+00:00',
	'date_offset': 0,
	'committer_date': '2000-01-12T05:23:54+00:00',
	'committer_date_offset': 0,
	'synthetic': False,
	'type': 'git',
	'parents': [],
	'metadata': {},
	'merge': False
	}
	])

	self.assertEqual(
	list(mock_storage.revision_get.call_args[0][0]),
	[hash_to_bytes(sha1),
	hash_to_bytes(sha1_other)])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_multiple_none_found(self, mock_storage):
	# given
	sha1_bin = self.SHA1_SAMPLE
	sha1_other = 'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'

	mock_storage.revision_get.return_value = []

	# then
	actual_revisions = service.lookup_revision_multiple(
	[sha1_bin, sha1_other])

	self.assertEqual(list(actual_revisions), [])

	self.assertEqual(
	list(mock_storage.revision_get.call_args[0][0]),
	[hash_to_bytes(self.SHA1_SAMPLE),
	hash_to_bytes(sha1_other)])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_log(self, mock_storage):
	# given
	stub_revision_log = [self.SAMPLE_REVISION_RAW]
	mock_storage.revision_log = MagicMock(return_value=stub_revision_log)

	# when
	actual_revision = service.lookup_revision_log(
	'abcdbe353ed3480476f032475e7c233eff7371d5',
	limit=25)

	# then
	self.assertEqual(list(actual_revision), [self.SAMPLE_REVISION])

	mock_storage.revision_log.assert_called_with(
	[hash_to_bytes('abcdbe353ed3480476f032475e7c233eff7371d5')], 25)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_log_by(self, mock_storage):
	# given
	stub_revision_log = [self.SAMPLE_REVISION_RAW]
	mock_storage.revision_log_by = MagicMock(
	return_value=stub_revision_log)

	# when
	actual_log = service.lookup_revision_log_by(
	1, 'refs/heads/master', None, limit=100)
	# then
	self.assertEqual(list(actual_log), [self.SAMPLE_REVISION])

	mock_storage.revision_log_by.assert_called_with(
	1, 'refs/heads/master', None, limit=100)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_log_by_nolog(self, mock_storage):
	# given
	mock_storage.revision_log_by = MagicMock(return_value=None)

	# when
	res = service.lookup_revision_log_by(
	1, 'refs/heads/master', None, limit=100)
	# then
	self.assertEquals(res, None)
	mock_storage.revision_log_by.assert_called_with(
	1, 'refs/heads/master', None, limit=100)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_raw_not_found(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(return_value=None)

	# when
	with self.assertRaises(NotFoundExc) as cm:
	service.lookup_content_raw('sha1:' + self.SHA1_SAMPLE)
	self.assertIn(cm.exc.args[0],
	'Content with %s checksum equals to %s not found!' %
	('sha1', self.SHA1_SAMPLE))

	mock_storage.content_find.assert_called_with(
	{'sha1': hash_to_bytes(self.SHA1_SAMPLE)})

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_raw(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(return_value={
	'sha1': self.SHA1_SAMPLE,
	})
	mock_storage.content_get = MagicMock(return_value=[{
	'data': b'binary data'}])

	# when
	actual_content = service.lookup_content_raw(
	'sha256:%s' % self.SHA256_SAMPLE)

	# then
	self.assertEquals(actual_content, {'data': b'binary data'})

	mock_storage.content_find.assert_called_once_with(
	{'sha256': self.SHA256_SAMPLE_BIN})
	mock_storage.content_get.assert_called_once_with(
	[self.SHA1_SAMPLE])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_not_found(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(return_value=None)

	# when
	actual_content = service.lookup_content(
	'sha1:%s' % self.SHA1_SAMPLE)

	# then
	self.assertIsNone(actual_content)

	mock_storage.content_find.assert_called_with(
	{'sha1': self.SHA1_SAMPLE_BIN})

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_with_sha1(self, mock_storage):
	# given
	mock_storage.content_find = MagicMock(
	return_value=self.SAMPLE_CONTENT_RAW)

	# when
	actual_content = service.lookup_content(
	'sha1:%s' % self.SHA1_SAMPLE)

	# then
	self.assertEqual(actual_content, self.SAMPLE_CONTENT)

	mock_storage.content_find.assert_called_with(
	{'sha1': hash_to_bytes(self.SHA1_SAMPLE)})

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_content_with_sha256(self, mock_storage):
	# given
	stub_content = self.SAMPLE_CONTENT_RAW
	stub_content['status'] = 'visible'

	expected_content = self.SAMPLE_CONTENT
	expected_content['status'] = 'visible'
	mock_storage.content_find = MagicMock(
	return_value=stub_content)

	# when
	actual_content = service.lookup_content(
	'sha256:%s' % self.SHA256_SAMPLE)

	# then
	self.assertEqual(actual_content, expected_content)

	mock_storage.content_find.assert_called_with(
	{'sha256': self.SHA256_SAMPLE_BIN})

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_person(self, mock_storage):
	# given
	mock_storage.person_get = MagicMock(return_value=[{
	'id': 'person_id',
	'name': b'some_name',
	'email': b'some-email',
	}])

	# when
	actual_person = service.lookup_person('person_id')

	# then
	self.assertEqual(actual_person, {
	'id': 'person_id',
	'name': 'some_name',
	'email': 'some-email',
	})

	mock_storage.person_get.assert_called_with(['person_id'])

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_directory_bad_checksum(self, mock_storage):
	# given
	mock_storage.directory_ls = MagicMock()

	# when
	with self.assertRaises(BadInputExc):
	service.lookup_directory('directory_id')

	# then
	mock_storage.directory_ls.called = False

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory_not_found(self, mock_query, mock_storage):
	# given
	mock_query.parse_hash_with_algorithms_or_throws.return_value = (
	'sha1',
	'directory-id-bin')
	mock_storage.directory_get.return_value = None

	# when
	with self.assertRaises(NotFoundExc) as cm:
	service.lookup_directory('directory_id')
	self.assertIn('Directory with sha1_git directory_id not found',
	cm.exception.args[0])

	# then
	mock_query.parse_hash_with_algorithms_or_throws.assert_called_with(
	'directory_id', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.directory_get.assert_called_with(['directory-id-bin'])
	mock_storage.directory_ls.called = False

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_directory(self, mock_query, mock_storage):
	mock_query.parse_hash_with_algorithms_or_throws.return_value = (
	'sha1',
	'directory-sha1-bin')

	# something that exists is all that matters here
	mock_storage.directory_get.return_value = {'id': b'directory-sha1-bin'}

	# given
	stub_dir_entries = [{
	'sha1': self.SHA1_SAMPLE_BIN,
	'sha256': self.SHA256_SAMPLE_BIN,
	'sha1_git': self.SHA1GIT_SAMPLE_BIN,
	'blake2s256': self.BLAKE2S256_SAMPLE_BIN,
	'target': hash_to_bytes(
	'40e71b8614fcd89ccd17ca2b1d9e66c5b00a6d03'),
	'dir_id': self.DIRECTORY_ID_BIN,
	'name': b'bob',
	'type': 10,
	}]

	expected_dir_entries = [{
	'checksums': {
	'sha1': self.SHA1_SAMPLE,
	'sha256': self.SHA256_SAMPLE,
	'sha1_git': self.SHA1GIT_SAMPLE,
	'blake2s256': self.BLAKE2S256_SAMPLE
	},
	'target': '40e71b8614fcd89ccd17ca2b1d9e66c5b00a6d03',
	'dir_id': self.DIRECTORY_ID,
	'name': 'bob',
	'type': 10,
	}]

	mock_storage.directory_ls.return_value = stub_dir_entries

	# when
	actual_directory_ls = list(service.lookup_directory(
	'directory-sha1'))

	# then
	self.assertEqual(actual_directory_ls, expected_dir_entries)

	mock_query.parse_hash_with_algorithms_or_throws.assert_called_with(
	'directory-sha1', ['sha1'], 'Only sha1_git is supported.')
	mock_storage.directory_ls.assert_called_with(
	'directory-sha1-bin')

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_by_nothing_found(self, mock_storage):
	# given
	mock_storage.revision_get_by.return_value = None

	# when
	actual_revisions = service.lookup_revision_by(1)

	# then
	self.assertIsNone(actual_revisions)

	mock_storage.revision_get_by.assert_called_with(1, 'refs/heads/master',
	limit=1,
	timestamp=None)

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_by(self, mock_storage):
	# given
	stub_rev = self.SAMPLE_REVISION_RAW

	expected_rev = self.SAMPLE_REVISION

	mock_storage.revision_get_by.return_value = [stub_rev]

	# when
	actual_revision = service.lookup_revision_by(10, 'master2', 'some-ts')

	# then
	self.assertEquals(actual_revision, expected_rev)

	mock_storage.revision_get_by.assert_called_with(10, 'master2',
	limit=1,
	timestamp='some-ts')

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_by_nomerge(self, mock_storage):
	# given
	stub_rev = self.SAMPLE_REVISION_RAW
	stub_rev['parents'] = [
	hash_to_bytes('adc83b19e793491b1c6ea0fd8b46cd9f32e592fc')]

	expected_rev = self.SAMPLE_REVISION
	expected_rev['parents'] = ['adc83b19e793491b1c6ea0fd8b46cd9f32e592fc']
	mock_storage.revision_get_by.return_value = [stub_rev]

	# when
	actual_revision = service.lookup_revision_by(10, 'master2', 'some-ts')

	# then
	self.assertEquals(actual_revision, expected_rev)

	mock_storage.revision_get_by.assert_called_with(10, 'master2',
	limit=1,
	timestamp='some-ts')

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_by_merge(self, mock_storage):
	# given
	stub_rev = self.SAMPLE_REVISION_RAW
	stub_rev['parents'] = [
	hash_to_bytes('adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'),
	hash_to_bytes('ffff3b19e793491b1c6db0fd8b46cd9f32e592fc')
	]

	expected_rev = self.SAMPLE_REVISION
	expected_rev['parents'] = [
	'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc',
	'ffff3b19e793491b1c6db0fd8b46cd9f32e592fc'
	]
	expected_rev['merge'] = True

	mock_storage.revision_get_by.return_value = [stub_rev]

	# when
	actual_revision = service.lookup_revision_by(10, 'master2', 'some-ts')

	# then
	self.assertEquals(actual_revision, expected_rev)

	mock_storage.revision_get_by.assert_called_with(10, 'master2',
	limit=1,
	timestamp='some-ts')

	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_with_context_by_ko(self, mock_storage):
	# given
	mock_storage.revision_get_by.return_value = None

	# when
	with self.assertRaises(NotFoundExc) as cm:
	origin_id = 1
	branch_name = 'master3'
	ts = None
	service.lookup_revision_with_context_by(origin_id, branch_name, ts,
	'sha1')
	# then
	self.assertIn(
	'Revision with (origin_id: %s, branch_name: %s'
	', ts: %s) not found.' % (origin_id,
	branch_name,
	ts), cm.exception.args[0])

	mock_storage.revision_get_by.assert_called_once_with(
	origin_id, branch_name, ts)

	@patch('swh.web.common.service.lookup_revision_with_context')
	@patch('swh.web.common.service.storage')
	@istest
	def lookup_revision_with_context_by(self, mock_storage,
	mock_lookup_revision_with_context):
	# given
	stub_root_rev = {'id': 'root-rev-id'}
	mock_storage.revision_get_by.return_value = [{'id': 'root-rev-id'}]
	stub_rev = {'id': 'rev-found'}
	mock_lookup_revision_with_context.return_value = stub_rev

	# when
	origin_id = 1
	branch_name = 'master3'
	ts = None
	sha1_git = 'sha1'
	actual_root_rev, actual_rev = service.lookup_revision_with_context_by(
	origin_id, branch_name, ts, sha1_git)

	# then
	self.assertEquals(actual_root_rev, stub_root_rev)
	self.assertEquals(actual_rev, stub_rev)

	mock_storage.revision_get_by.assert_called_once_with(
	origin_id, branch_name, limit=1, timestamp=ts)
	mock_lookup_revision_with_context.assert_called_once_with(
	stub_root_rev, sha1_git, 100)

	@patch('swh.web.common.service.storage')
	@patch('swh.web.common.service.query')
	@istest
	def lookup_entity_by_uuid(self, mock_query, mock_storage):
	# given
	uuid_test = 'correct-uuid'
	mock_query.parse_uuid4.return_value = uuid_test
	stub_entities = [{'uuid': uuid_test}]

	mock_storage.entity_get.return_value = stub_entities

	# when
	actual_entities = list(service.lookup_entity_by_uuid(uuid_test))

	# then
	self.assertEquals(actual_entities, stub_entities)

	mock_query.parse_uuid4.assert_called_once_with(uuid_test)
	mock_storage.entity_get.assert_called_once_with(uuid_test)

	@istest
	def lookup_revision_through_ko_not_implemented(self):
	# then
	with self.assertRaises(NotImplementedError):
	service.lookup_revision_through({
	'something-unknown': 10,
	})

	@patch('swh.web.common.service.lookup_revision_with_context_by')
	@istest
	def lookup_revision_through_with_context_by(self, mock_lookup):
	# given
	stub_rev = {'id': 'rev'}
	mock_lookup.return_value = stub_rev

	# when
	actual_revision = service.lookup_revision_through({
	'origin_id': 1,
	'branch_name': 'master',
	'ts': None,
	'sha1_git': 'sha1-git'
	}, limit=1000)

	# then
	self.assertEquals(actual_revision, stub_rev)

	mock_lookup.assert_called_once_with(
	1, 'master', None, 'sha1-git', 1000)

	@patch('swh.web.common.service.lookup_revision_by')
	@istest
	def lookup_revision_through_with_revision_by(self, mock_lookup):
	# given
	stub_rev = {'id': 'rev'}
	mock_lookup.return_value = stub_rev

	# when
	actual_revision = service.lookup_revision_through({
	'origin_id': 2,
	'branch_name': 'master2',
	'ts': 'some-ts',
	}, limit=10)

	# then
	self.assertEquals(actual_revision, stub_rev)

	mock_lookup.assert_called_once_with(
	2, 'master2', 'some-ts')

	@patch('swh.web.common.service.lookup_revision_with_context')
	@istest
	def lookup_revision_through_with_context(self, mock_lookup):
	# given
	stub_rev = {'id': 'rev'}
	mock_lookup.return_value = stub_rev

	# when
	actual_revision = service.lookup_revision_through({
	'sha1_git_root': 'some-sha1-root',
	'sha1_git': 'some-sha1',
	})

	# then
	self.assertEquals(actual_revision, stub_rev)

	mock_lookup.assert_called_once_with(
	'some-sha1-root', 'some-sha1', 100)

	@patch('swh.web.common.service.lookup_revision')
	@istest
	def lookup_revision_through_with_revision(self, mock_lookup):
	# given
	stub_rev = {'id': 'rev'}
	mock_lookup.return_value = stub_rev

	# when
	actual_revision = service.lookup_revision_through({
	'sha1_git': 'some-sha1',
	})

	# then
	self.assertEquals(actual_revision, stub_rev)

	mock_lookup.assert_called_once_with(
	'some-sha1')

	@patch('swh.web.common.service.lookup_revision_through')
	@istest
	def lookup_directory_through_revision_ko_not_found(
	self, mock_lookup_rev):
	# given
	mock_lookup_rev.return_value = None

	# when
	with self.assertRaises(NotFoundExc):
	service.lookup_directory_through_revision(
	{'id': 'rev'}, 'some/path', 100)

	mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 100)

	@patch('swh.web.common.service.lookup_revision_through')
	@patch('swh.web.common.service.lookup_directory_with_revision')
	@istest
	def lookup_directory_through_revision_ok_with_data(
	self, mock_lookup_dir, mock_lookup_rev):
	# given
	mock_lookup_rev.return_value = {'id': 'rev-id'}
	mock_lookup_dir.return_value = {'type': 'dir',
	'content': []}

	# when
	rev_id, dir_result = service.lookup_directory_through_revision(
	{'id': 'rev'}, 'some/path', 100)
	# then
	self.assertEquals(rev_id, 'rev-id')
	self.assertEquals(dir_result, {'type': 'dir',
	'content': []})

	mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 100)
	mock_lookup_dir.assert_called_once_with('rev-id', 'some/path', False)

	@patch('swh.web.common.service.lookup_revision_through')
	@patch('swh.web.common.service.lookup_directory_with_revision')
	@istest
	def lookup_directory_through_revision_ok_with_content(
	self, mock_lookup_dir, mock_lookup_rev):
	# given
	mock_lookup_rev.return_value = {'id': 'rev-id'}
	stub_result = {'type': 'file',
	'revision': 'rev-id',
	'content': {'data': b'blah',
	'sha1': 'sha1'}}
	mock_lookup_dir.return_value = stub_result

	# when
	rev_id, dir_result = service.lookup_directory_through_revision(
	{'id': 'rev'}, 'some/path', 10, with_data=True)
	# then
	self.assertEquals(rev_id, 'rev-id')
	self.assertEquals(dir_result, stub_result)

	mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 10)
	mock_lookup_dir.assert_called_once_with('rev-id', 'some/path', True)

File Metadata

Mime Type: text/x-python
Expires: Sat, Jun 21, 5:17 PM (1 w, 3 d ago)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 3321943

test_service.pyNo OneTemporaryActions

test_service.pyView Options

File Metadata

Event Timeline

test_service.py
No OneTemporary
Actions

test_service.py
View Options