Differential D829 Diff 2738 swh/web/tests/common/test_service.py

Changeset View

Standalone View

swh/web/tests/common/test_service.py

# Copyright (C) 2015-2018 The Software Heritage developers		# Copyright (C) 2015-2018 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 Affero General Public License version 3, or any later version		# License: GNU Affero 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 datetime		import datetime
		import itertools
		import pytest
		import random

from unittest.mock import MagicMock, patch, call		from hypothesis import given
		from unittest.mock import MagicMock, patch

from swh.model.hashutil import hash_to_bytes, hash_to_hex		from swh.model.hashutil import hash_to_bytes, hash_to_hex

from swh.web.common import service		from swh.web.common import service
from swh.web.common.exc import BadInputExc, NotFoundExc		from swh.web.common.exc import BadInputExc, NotFoundExc
from swh.web.tests.testcase import WebTestCase		from swh.web.tests.strategies import (
		content, contents, unknown_content, unknown_contents,
		contents_with_ctags, origin, visit_dates, directory,
		release, revision, unknown_revision, ancestor_revisions,
		non_ancestor_revisions, invalid_sha1, sha256,
		revision_with_submodules
		)
		from swh.web.tests.testcase import (
		WebTestCase, ctags_json_missing, fossology_missing
		)


class ServiceTestCase(WebTestCase):		class ServiceTestCase(WebTestCase):

def setUp(self):		def setUp(self):
self.BLAKE2S256_SAMPLE = ('685395c5dc57cada459364f0946d3dd45b'		self.BLAKE2S256_SAMPLE = ('685395c5dc57cada459364f0946d3dd45b'
'ad5fcbabc1048edb44380f1d31d0aa')		'ad5fcbabc1048edb44380f1d31d0aa')
self.BLAKE2S256_SAMPLE_BIN = hash_to_bytes(self.BLAKE2S256_SAMPLE)		self.BLAKE2S256_SAMPLE_BIN = hash_to_bytes(self.BLAKE2S256_SAMPLE)
▲ Show 20 Lines • Show All 86 Lines • ▼ Show 20 Lines	def setUp(self):
tzinfo=datetime.timezone.utc)		tzinfo=datetime.timezone.utc)

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

@patch('swh.web.common.service.storage')		@given(contents())
def test_lookup_multiple_hashes_ball_missing(self, mock_storage):		def test_lookup_multiple_hashes_all_present(self, contents):
# given		input_data = []
mock_storage.content_missing_per_sha1 = MagicMock(return_value=[])		expected_output = []
		for cnt in contents:
# when		input_data.append({'sha1': cnt['sha1']})
actual_lookup = service.lookup_multiple_hashes(		expected_output.append({'sha1': cnt['sha1'],
[{'filename': 'a',		'found': True})
'sha1': '456caf10e9535160d90e874b45aa426de762f19f'},
{'filename': 'b',		self.assertEqual(service.lookup_multiple_hashes(input_data),
'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}])		expected_output)

# then		@given(contents(), unknown_contents())
self.assertEqual(actual_lookup, [		def test_lookup_multiple_hashes_some_missing(self, contents,
{'filename': 'a',		unknown_contents):
'sha1': '456caf10e9535160d90e874b45aa426de762f19f',		input_contents = list(itertools.chain(contents, unknown_contents))
'found': True},		random.shuffle(input_contents)
{'filename': 'b',
'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865',		input_data = []
'found': True}		expected_output = []
])		for cnt in input_contents:
		input_data.append({'sha1': cnt['sha1']})
@patch('swh.web.common.service.storage')		expected_output.append({'sha1': cnt['sha1'],
def test_lookup_multiple_hashes_some_missing(self, mock_storage):		'found': cnt in contents})
# given
mock_storage.content_missing_per_sha1 = MagicMock(return_value=[		self.assertEqual(service.lookup_multiple_hashes(input_data),
hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')		expected_output)
])
		@given(unknown_content())
# when		def test_lookup_hash_does_not_exist(self, unknown_content):
actual_lookup = service.lookup_multiple_hashes(
[{'filename': 'a',		actual_lookup = service.lookup_hash('sha1_git:%s' %
'sha1': '456caf10e9535160d90e874b45aa426de762f19f'},		unknown_content['sha1_git'])
{'filename': 'b',
'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}])		self.assertEqual(actual_lookup, {'found': None,
		'algo': 'sha1_git'})
# then
self.assertEqual(actual_lookup, [		@given(content())
{'filename': 'a',		def test_lookup_hash_exist(self, content):
'sha1': '456caf10e9535160d90e874b45aa426de762f19f',
'found': False},		actual_lookup = service.lookup_hash('sha1:%s' % content['sha1'])
{'filename': 'b',
'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865',		content_metadata = self.content_get_metadata(content['sha1'])
'found': True}
])

@patch('swh.web.common.service.storage')
def test_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.assertEqual({'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')
def test_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.assertEqual({'found': content_metadata,
self.assertEqual({		'algo': 'sha1'}, actual_lookup)
'found': {
'checksums': {
'sha1': '456caf10e9535160d90e874b45aa426de762f19f'
}
},
'algo': 'sha1'
}, actual_lookup)

mock_storage.content_find.assert_called_with(		@given(unknown_content())
{'sha1':		def test_search_hash_does_not_exist(self, content):
hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')}
)

@patch('swh.web.common.service.storage')		actual_lookup = service.search_hash('sha1_git:%s' %
def test_search_hash_does_not_exist(self, mock_storage):		content['sha1_git'])
# given
mock_storage.content_find = MagicMock(return_value=None)

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

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

# check the function has been called with parameters		@given(content())
mock_storage.content_find.assert_called_with(		def test_search_hash_exist(self, content):
{'sha1_git':
hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')})

@patch('swh.web.common.service.storage')		actual_lookup = service.search_hash('sha1:%s' % content['sha1'])
def test_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.assertEqual({'found': True}, actual_lookup)		self.assertEqual({'found': True}, actual_lookup)

mock_storage.content_find.assert_called_with(		@pytest.mark.skipif(ctags_json_missing,
{'sha1':		reason="requires ctags with json output support")
hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')},		@given(contents_with_ctags())
)		def test_lookup_content_ctags(self, contents_with_ctags):

@patch('swh.web.common.service.idx_storage')		content_sha1 = random.choice(contents_with_ctags['sha1s'])
def test_lookup_content_ctags(self, mock_idx_storage):		self.content_add_ctags(content_sha1)
# given		actual_ctags = \
mock_idx_storage.content_ctags_get = MagicMock(		list(service.lookup_content_ctags('sha1:%s' % content_sha1))
return_value=[{
'id': hash_to_bytes(		expected_data = list(self.content_get_ctags(content_sha1))
'123caf10e9535160d90e874b45aa426de762f19f'),		for ctag in expected_data:
'line': 100,		ctag['id'] = content_sha1
'name': 'hello',
'kind': 'function',		self.assertEqual(actual_ctags, expected_data)
'tool_name': 'ctags',
'tool_version': 'some-version',		@given(unknown_content())
}])		def test_lookup_content_ctags_no_hash(self, unknown_content):
expected_ctags = [{
'id': '123caf10e9535160d90e874b45aa426de762f19f',		actual_ctags = \
'line': 100,		list(service.lookup_content_ctags('sha1:%s' %
'name': 'hello',		unknown_content['sha1']))
'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_idx_storage.content_ctags_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

@patch('swh.web.common.service.idx_storage')
def test_lookup_content_ctags_no_hash(self, mock_idx_storage):
# given
mock_idx_storage.content_ctags_get = MagicMock(return_value=[])

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

# then
self.assertEqual(actual_ctags, [])		self.assertEqual(actual_ctags, [])

@patch('swh.web.common.service.idx_storage')		@given(content())
def test_lookup_content_filetype(self, mock_idx_storage):		def test_lookup_content_filetype(self, content):
# given
mock_idx_storage.content_mimetype_get = MagicMock(
return_value=[{
'id': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f'),
'mimetype': 'text/x-c++',
'encoding': '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_idx_storage.content_mimetype_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

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

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

# then		expected_filetype = self.content_get_mimetype(content['sha1'])
self.assertEqual(actual_filetype, expected_filetype)		self.assertEqual(actual_filetype, expected_filetype)

mock_storage.content_find(		@given(content())
'sha1_git', hash_to_bytes(		def test_lookup_content_language(self, content):
'456caf10e9535160d90e874b45aa426de762f19f')
)
mock_idx_storage.content_mimetype_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

@patch('swh.web.common.service.idx_storage')
def test_lookup_content_language(self, mock_idx_storage):
# given
mock_idx_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_idx_storage.content_language_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

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

# when		self.content_add_language(content['sha1'])
actual_language = service.lookup_content_language(		actual_language = service.lookup_content_language(content['sha1'])
'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')

# then		expected_language = self.content_get_language(content['sha1'])
self.assertEqual(actual_language, expected_language)		self.assertEqual(actual_language, expected_language)

mock_storage.content_find(		@given(contents_with_ctags())
'sha1_git', hash_to_bytes(		def test_lookup_expression(self, contents_with_ctags):
'456caf10e9535160d90e874b45aa426de762f19f')
)
mock_idx_storage.content_language_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

@patch('swh.web.common.service.idx_storage')		per_page = 10
def test_lookup_expression(self, mock_idx_storage):		expected_ctags = []
# given
mock_idx_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		for content_sha1 in contents_with_ctags['sha1s']:
actual_ctags = list(service.lookup_expression(		if len(expected_ctags) == per_page:
'foobar', last_sha1='hash', per_page=10))		break
		self.content_add_ctags(content_sha1)
		for ctag in self.content_get_ctags(content_sha1):
		if len(expected_ctags) == per_page:
		break
		if ctag['name'] == contents_with_ctags['symbol_name']:
		del ctag['id']
		ctag['sha1'] = content_sha1
		expected_ctags.append(ctag)

		actual_ctags = \
		list(service.lookup_expression(contents_with_ctags['symbol_name'],
		last_sha1=None, per_page=10))

# then
self.assertEqual(actual_ctags, expected_ctags)		self.assertEqual(actual_ctags, expected_ctags)

mock_idx_storage.content_ctags_search.assert_called_with(		def test_lookup_expression_no_result(self):
'foobar', last_sha1='hash', limit=10)

@patch('swh.web.common.service.idx_storage')
def test_lookup_expression_no_result(self, mock_idx_storage):
# given
mock_idx_storage.content_ctags_search = MagicMock(
return_value=[])
expected_ctags = []		expected_ctags = []

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

# then
self.assertEqual(actual_ctags, expected_ctags)		self.assertEqual(actual_ctags, expected_ctags)

mock_idx_storage.content_ctags_search.assert_called_with(		@pytest.mark.skipif(fossology_missing,
'barfoo', last_sha1='hash', limit=10)		reason="requires fossology-nomossa installed")
		@given(content())
@patch('swh.web.common.service.idx_storage')		def test_lookup_content_license(self, content):
def test_lookup_content_license(self, mock_idx_storage):
# given
mock_idx_storage.content_fossology_license_get = MagicMock(
return_value=[{
hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f'): [{
'licenses': ['GPL-3.0+'],
'tool': {}
}]
}])
expected_license = {
'id': '123caf10e9535160d90e874b45aa426de762f19f',
'facts': [{
'licenses': ['GPL-3.0+'],
'tool': {}
}]
}

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

# then
self.assertEqual(actual_license, expected_license)

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

@patch('swh.web.common.service.idx_storage')
@patch('swh.web.common.service.storage')
def test_lookup_content_license_2(self, mock_storage, mock_idx_storage):
# given
mock_storage.content_find = MagicMock(
return_value={
'sha1': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f')
}
)
mock_idx_storage.content_fossology_license_get = MagicMock(
return_value=[{
hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f'): [{
'licenses': ['BSD-2-Clause'],
'tool': {}
}]

}]		self.content_add_license(content['sha1'])
)		actual_license = service.lookup_content_license(content['sha1'])
expected_license = {
'id': '123caf10e9535160d90e874b45aa426de762f19f',
'facts': [{
'licenses': ['BSD-2-Clause'],
'tool': {}
}]
}

# when		expected_license = self.content_get_license(content['sha1'])
actual_license = service.lookup_content_license(
'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')

# then
self.assertEqual(actual_license, expected_license)		self.assertEqual(actual_license, expected_license)

mock_storage.content_find(		def test_stat_counters(self):
'sha1_git', hash_to_bytes(
'456caf10e9535160d90e874b45aa426de762f19f')
)
mock_idx_storage.content_fossology_license_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])

@patch('swh.web.common.service.storage')
def test_stat_counters(self, mock_storage):
# given
input_stats = {
"content": 1770830,
"directory": 211683,
"directory_entry_dir": 209167,
"directory_entry_file": 1807094,
"directory_entry_rev": 0,
"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()		actual_stats = service.stat_counters()
		self.assertEqual(actual_stats, self.storage.stat_counters())

# then		@given(origin(), visit_dates())
expected_stats = input_stats		def test_lookup_origin_visits(self, origin, visit_dates):
self.assertEqual(actual_stats, expected_stats)

mock_storage.stat_counters.assert_called_with()

@patch('swh.web.common.service._lookup_origin_visits')
def test_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		origin_id = self.storage.origin_add_one(origin)
expected_origin_visits = [{		for ts in visit_dates:
'date': self.origin_visit1['date'].isoformat(),		self.storage.origin_visit_add(origin_id, ts)
'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)		actual_origin_visits = list(service.lookup_origin_visits(origin_id))

# then		expected_visits = list(self.storage.origin_visit_get(origin_id))
self.assertEqual(list(actual_origin_visits), expected_origin_visits)		for visit in expected_visits:
		visit['date'] = visit['date'].isoformat()
		visit['metadata'] = {}

mock_lookup_visits.assert_called_once_with(		self.assertEqual(actual_origin_visits, expected_visits)
6, last_visit=None, limit=10)

@patch('swh.web.common.service.storage')		@given(origin(), visit_dates())
def test_lookup_origin_visit(self, mock_storage):		def test_lookup_origin_visit(self, origin, visit_dates):
# given		origin_id = self.storage.origin_add_one(origin)
stub_result = self.origin_visit1		visits = []
mock_storage.origin_visit_get_by.return_value = stub_result		for ts in visit_dates:
		visits.append(self.storage.origin_visit_add(origin_id, ts))

expected_origin_visit = {		visit = random.choice(visits)['visit']
'date': self.origin_visit1['date'].isoformat(),		actual_origin_visit = service.lookup_origin_visit(origin_id, visit)
'origin': self.origin_visit1['origin'],
'visit': self.origin_visit1['visit']
}

# when		expected_visit = dict(self.storage.origin_visit_get_by(origin_id,
actual_origin_visit = service.lookup_origin_visit(1, 1)		visit))
		expected_visit['date'] = expected_visit['date'].isoformat()
		expected_visit['metadata'] = {}

# then		self.assertEqual(actual_origin_visit, expected_visit)
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')
def test_lookup_origin(self, mock_storage):
# given
mock_storage.origin_get = MagicMock(return_value={
'id': 'origin-id',
'url': 'ftp://some/url/to/origin',
'type': 'ftp'})

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

# then		@given(origin())
self.assertEqual(actual_origin, {'id': 'origin-id',		def test_lookup_origin(self, origin):
'url': 'ftp://some/url/to/origin',		origin_id = self.storage.origin_add_one(origin)
'type': 'ftp'})

mock_storage.origin_get.assert_called_with({'id': 'origin-id'})		actual_origin = service.lookup_origin({'id': origin_id})
		expected_origin = self.storage.origin_get({'id': origin_id})
		self.assertEqual(actual_origin, expected_origin)

@patch('swh.web.common.service.storage')		actual_origin = service.lookup_origin({'type': origin['type'],
def test_lookup_release_ko_id_checksum_not_a_sha1(self, mock_storage):		'url': origin['url']})
# given		expected_origin = self.storage.origin_get({'type': origin['type'],
mock_storage.release_get = MagicMock()		'url': origin['url']})
		self.assertEqual(actual_origin, expected_origin)

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

mock_storage.release_get.called = False		@given(sha256())
		def test_lookup_release_ko_id_checksum_too_long(self, sha256):
@patch('swh.web.common.service.storage')
def test_lookup_release_ko_id_checksum_too_long(self, mock_storage):
# given
mock_storage.release_get = MagicMock()

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

mock_storage.release_get.called = False		@given(directory())
		def test_lookup_directory_with_path_not_found(self, directory):
@patch('swh.web.common.service.storage')		path = 'some/invalid/path/here'
def test_lookup_directory_with_path_not_found(self, mock_storage):		with self.assertRaises(NotFoundExc) as cm:
# given		service.lookup_directory_with_path(directory, path)
mock_storage.lookup_directory_with_path = MagicMock(return_value=None)		self.assertEqual('Directory entry with path %s from %s '
		'not found' % (path, directory),
sha1_git = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'		cm.exception.args[0])

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

self.assertIsNone(actual_directory)

@patch('swh.web.common.service.storage')
def test_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')
def test_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')])

def test_lookup_revision_with_context_ko_not_a_sha1_1(self):		@given(directory())
# given		def test_lookup_directory_with_path_found(self, directory):
sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4' \		directory_content = self.directory_ls(directory)
'daf51aea892abe'		directory_entry = random.choice(directory_content)
sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'		path = directory_entry['name']
		actual_result = service.lookup_directory_with_path(directory, path)
		self.assertEqual(actual_result, directory_entry)

		@given(release())
		def test_lookup_release(self, release):
		actual_release = service.lookup_release(release)

		self.assertEqual(actual_release,
		self.release_get(release))

		@given(revision(), invalid_sha1(), sha256())
		def test_lookup_revision_with_context_ko_not_a_sha1(self, revision,
		invalid_sha1,
		sha256):
		sha1_git_root = revision
		sha1_git = invalid_sha1

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

def test_lookup_revision_with_context_ko_not_a_sha1_2(self):		sha1_git = sha256
# given
sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f6' \
'2d4daf51aea892abe'

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

@patch('swh.web.common.service.storage')		@given(revision(), unknown_revision())
def test_lookup_revision_with_context_ko_sha1_git_does_not_exist(		def test_lookup_revision_with_context_ko_sha1_git_does_not_exist(
self,		self, revision, unknown_revision):
mock_storage):		sha1_git_root = revision
# given		sha1_git = unknown_revision
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:		with self.assertRaises(NotFoundExc) as cm:
service.lookup_revision_with_context(sha1_git_root, sha1_git)		service.lookup_revision_with_context(sha1_git_root, sha1_git)
self.assertIn('Revision 777777bdf3629f916219feb3dcc7393ded1bc8db'		self.assertIn('Revision %s not found' % sha1_git, cm.exception.args[0])
' not found', cm.exception.args[0])

mock_storage.revision_get.assert_called_once_with(
[sha1_git_bin])

@patch('swh.web.common.service.storage')		@given(revision(), unknown_revision())
def test_lookup_revision_with_context_ko_root_sha1_git_does_not_exist(		def test_lookup_revision_with_context_ko_root_sha1_git_does_not_exist(
self,		self, revision, unknown_revision):
mock_storage):		sha1_git_root = unknown_revision
# given		sha1_git = revision
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:		with self.assertRaises(NotFoundExc) as cm:
service.lookup_revision_with_context(sha1_git_root, sha1_git)		service.lookup_revision_with_context(sha1_git_root, sha1_git)
self.assertIn('Revision root 65a55bbdf3629f916219feb3dcc7393ded1bc8db'		self.assertIn('Revision root %s not found' % sha1_git_root,
' not found', cm.exception.args[0])		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')
def test_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.assertEqual(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')
def test_lookup_revision_with_context_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		@given(ancestor_revisions())
actual_revision = service.lookup_revision_with_context(		def test_lookup_revision_with_context(self, ancestor_revisions):
{'id': sha1_git_root_bin},		sha1_git = ancestor_revisions['sha1_git']
		root_sha1_git = ancestor_revisions['sha1_git_root']
		for sha1_git_root in (root_sha1_git,
		{'id': hash_to_bytes(root_sha1_git)}):
		actual_revision = \
		service.lookup_revision_with_context(sha1_git_root,
sha1_git)		sha1_git)

# then		children = []
self.assertEqual(actual_revision, {		for rev in self.revision_log(root_sha1_git):
'id': hash_to_hex(sha1_git_bin),		for p_rev in rev['parents']:
'parents': [],		p_rev_hex = hash_to_hex(p_rev)
'children': [hash_to_hex(b'999'), hash_to_hex(b'777')],		if p_rev_hex == sha1_git:
'directory': hash_to_hex(b'278'),		children.append(rev['id'])
'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])		expected_revision = self.revision_get(sha1_git)
		expected_revision['children'] = children
mock_storage.revision_log.assert_called_with(		self.assertEqual(actual_revision, expected_revision)
[sha1_git_root_bin], 100)

@patch('swh.web.common.service.storage')		@given(non_ancestor_revisions())
@patch('swh.web.common.service.query')		def test_lookup_revision_with_context_ko(self, non_ancestor_revisions):
def test_lookup_directory_with_revision_not_found(self,		sha1_git = non_ancestor_revisions['sha1_git']
mock_query,		root_sha1_git = non_ancestor_revisions['sha1_git_root']
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:		with self.assertRaises(NotFoundExc) as cm:
service.lookup_directory_with_revision('123')		service.lookup_revision_with_context(root_sha1_git, sha1_git)
self.assertIn('Revision 123 not found', cm.exception.args[0])		self.assertIn('Revision %s is not an ancestor of %s' %
		(sha1_git, root_sha1_git), cm.exception.args[0])

mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with		@given(unknown_revision())
('123', ['sha1'], 'Only sha1_git is supported.')		def test_lookup_directory_with_revision_not_found(self, unknown_revision):
mock_storage.revision_get.assert_called_once_with([b'123'])

@patch('swh.web.common.service.storage')		with self.assertRaises(NotFoundExc) as cm:
@patch('swh.web.common.service.query')		service.lookup_directory_with_revision(unknown_revision)
def test_lookup_directory_with_revision_ko_revision_with_path_to_nowhere(		self.assertIn('Revision %s not found' % unknown_revision,
self,		cm.exception.args[0])
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		@given(revision())
		def test_lookup_directory_with_revision_ko_path_to_nowhere(self, revision):

# when		invalid_path = 'path/to/something/unknown'
with self.assertRaises(NotFoundExc) as cm:		with self.assertRaises(NotFoundExc) as cm:
service.lookup_directory_with_revision(		service.lookup_directory_with_revision(revision, invalid_path)
'123',
'path/to/something/unknown')
exception_text = cm.exception.args[0].lower()		exception_text = cm.exception.args[0].lower()
self.assertIn('directory or file', exception_text)		self.assertIn('directory or file', exception_text)
self.assertIn('path/to/something/unknown', exception_text)		self.assertIn(invalid_path, exception_text)
self.assertIn('revision 123', exception_text)		self.assertIn('revision %s' % revision, exception_text)
self.assertIn('not found', exception_text)		self.assertIn('not found', exception_text)

mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with		@given(revision_with_submodules())
('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')
def test_lookup_directory_with_revision_ko_type_not_implemented(		def test_lookup_directory_with_revision_ko_type_not_implemented(
self,		self, revision_with_submodules):
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:		with self.assertRaises(NotImplementedError) as cm:
service.lookup_directory_with_revision(		service.lookup_directory_with_revision(
'123',		revision_with_submodules['rev_sha1_git'],
'some/path/to/rev')		revision_with_submodules['rev_dir_rev_path'])
self.assertIn("Entity of type rev not implemented.",		self.assertIn("Entity of type rev not implemented.",
cm.exception.args[0])		cm.exception.args[0])

# then		@given(revision())
mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with		def test_lookup_directory_with_revision_without_path(self, revision):
('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')		actual_directory_entries = \
@patch('swh.web.common.service.query')		service.lookup_directory_with_revision(revision)
def test_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'		revision_data = self.revision_get(revision)
mock_storage.revision_get.return_value = [{		expected_directory_entries = \
'directory': dir_id,		self.directory_ls(revision_data['directory'])
}]

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(actual_directory_entries['type'], 'dir')
self.assertEqual(list(actual_directory_entries['content']),		self.assertEqual(actual_directory_entries['content'],
stub_dir_entries)		expected_directory_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.storage')
@patch('swh.web.common.service.query')		@patch('swh.web.common.service.query')
def test_lookup_directory_with_revision_with_path_to_dir(self,		def test_lookup_directory_with_revision_with_path_to_dir(self,
mock_query,		mock_query,
mock_storage):		mock_storage):
# given		# given
mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',		mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
▲ Show 20 Lines • Show All 140 Lines • ▼ Show 20 Lines	def test_lookup_directory_with_revision_with_path_to_file_w_data(
mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with		mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
('123', ['sha1'], 'Only sha1_git is supported.')		('123', ['sha1'], 'Only sha1_git is supported.')
mock_storage.revision_get.assert_called_once_with([b'123'])		mock_storage.revision_get.assert_called_once_with([b'123'])
mock_storage.directory_entry_get_by_path.assert_called_once_with(		mock_storage.directory_entry_get_by_path.assert_called_once_with(
b'dir-id-as-sha1', [b'some', b'path', b'to', b'file'])		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_find.assert_called_once_with({'sha1_git': b'789'})
mock_storage.content_get.assert_called_once_with([b'content-sha1'])		mock_storage.content_get.assert_called_once_with([b'content-sha1'])

@patch('swh.web.common.service.storage')		@given(revision())
def test_lookup_revision(self, mock_storage):		def test_lookup_revision(self, revision):
# given		actual_revision = service.lookup_revision(revision)
mock_storage.revision_get = MagicMock(		self.assertEqual(actual_revision, self.revision_get(revision))
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')		@patch('swh.web.common.service.storage')
def test_lookup_revision_invalid_msg(self, mock_storage):		def test_lookup_revision_invalid_msg(self, mock_storage):
# given		# given
stub_rev = self.SAMPLE_REVISION_RAW		stub_rev = self.SAMPLE_REVISION_RAW
stub_rev['message'] = b'elegant fix for bug \xff'		stub_rev['message'] = b'elegant fix for bug \xff'

expected_revision = self.SAMPLE_REVISION		expected_revision = self.SAMPLE_REVISION
▲ Show 20 Lines • Show All 649 Lines • Show Last 20 Lines