Page MenuHomeSoftware Heritage
Differential D529 Diff 1641 swh/web/tests/common/test_service.py

Changeset View

Standalone View

View Options

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
from nose.tools import istest
from unittest.mock import MagicMock, patch, call from unittest.mock import MagicMock, patch, call
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 SWHWebTestCase from swh.web.tests.testcase import SWHWebTestCase
▲ Show 20 Lines • Show All 94 Lines • ▼ Show 20 Lines def setUp(self):
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') @patch('swh.web.common.service.storage')
@istest def test_lookup_multiple_hashes_ball_missing(self, mock_storage):
def lookup_multiple_hashes_ball_missing(self, mock_storage):
# given # given
mock_storage.content_missing_per_sha1 = MagicMock(return_value=[]) mock_storage.content_missing_per_sha1 = MagicMock(return_value=[])
# when # when
actual_lookup = service.lookup_multiple_hashes( actual_lookup = service.lookup_multiple_hashes(
[{'filename': 'a', [{'filename': 'a',
'sha1': '456caf10e9535160d90e874b45aa426de762f19f'}, 'sha1': '456caf10e9535160d90e874b45aa426de762f19f'},
{'filename': 'b', {'filename': 'b',
'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}]) 'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}])
# then # then
self.assertEquals(actual_lookup, [ self.assertEquals(actual_lookup, [
{'filename': 'a', {'filename': 'a',
'sha1': '456caf10e9535160d90e874b45aa426de762f19f', 'sha1': '456caf10e9535160d90e874b45aa426de762f19f',
'found': True}, 'found': True},
{'filename': 'b', {'filename': 'b',
'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865', 'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865',
'found': True} 'found': True}
]) ])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_multiple_hashes_some_missing(self, mock_storage):
def lookup_multiple_hashes_some_missing(self, mock_storage):
# given # given
mock_storage.content_missing_per_sha1 = MagicMock(return_value=[ mock_storage.content_missing_per_sha1 = MagicMock(return_value=[
hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f') hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')
]) ])
# when # when
actual_lookup = service.lookup_multiple_hashes( actual_lookup = service.lookup_multiple_hashes(
[{'filename': 'a', [{'filename': 'a',
'sha1': '456caf10e9535160d90e874b45aa426de762f19f'}, 'sha1': '456caf10e9535160d90e874b45aa426de762f19f'},
{'filename': 'b', {'filename': 'b',
'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}]) 'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}])
# then # then
self.assertEquals(actual_lookup, [ self.assertEquals(actual_lookup, [
{'filename': 'a', {'filename': 'a',
'sha1': '456caf10e9535160d90e874b45aa426de762f19f', 'sha1': '456caf10e9535160d90e874b45aa426de762f19f',
'found': False}, 'found': False},
{'filename': 'b', {'filename': 'b',
'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865', 'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865',
'found': True} 'found': True}
]) ])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_hash_does_not_exist(self, mock_storage):
def lookup_hash_does_not_exist(self, mock_storage):
# given # given
mock_storage.content_find = MagicMock(return_value=None) mock_storage.content_find = MagicMock(return_value=None)
# when # when
actual_lookup = service.lookup_hash( actual_lookup = service.lookup_hash(
'sha1_git:123caf10e9535160d90e874b45aa426de762f19f') 'sha1_git:123caf10e9535160d90e874b45aa426de762f19f')
# then # then
self.assertEquals({'found': None, self.assertEquals({'found': None,
'algo': 'sha1_git'}, actual_lookup) 'algo': 'sha1_git'}, actual_lookup)
# check the function has been called with parameters # check the function has been called with parameters
mock_storage.content_find.assert_called_with( mock_storage.content_find.assert_called_with(
{'sha1_git': {'sha1_git':
hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')}) hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_hash_exist(self, mock_storage):
def lookup_hash_exist(self, mock_storage):
# given # given
stub_content = { stub_content = {
'sha1': hash_to_bytes( 'sha1': hash_to_bytes(
'456caf10e9535160d90e874b45aa426de762f19f') '456caf10e9535160d90e874b45aa426de762f19f')
} }
mock_storage.content_find = MagicMock(return_value=stub_content) mock_storage.content_find = MagicMock(return_value=stub_content)
# when # when
actual_lookup = service.lookup_hash( actual_lookup = service.lookup_hash(
'sha1:456caf10e9535160d90e874b45aa426de762f19f') 'sha1:456caf10e9535160d90e874b45aa426de762f19f')
# then # then
self.assertEquals({'found': stub_content, self.assertEquals({'found': stub_content,
'algo': 'sha1'}, actual_lookup) 'algo': 'sha1'}, actual_lookup)
mock_storage.content_find.assert_called_with( mock_storage.content_find.assert_called_with(
{'sha1': {'sha1':
hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')} hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')}
) )
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_search_hash_does_not_exist(self, mock_storage):
def search_hash_does_not_exist(self, mock_storage):
# given # given
mock_storage.content_find = MagicMock(return_value=None) mock_storage.content_find = MagicMock(return_value=None)
# when # when
actual_lookup = service.search_hash( actual_lookup = service.search_hash(
'sha1_git:123caf10e9535160d90e874b45aa426de762f19f') 'sha1_git:123caf10e9535160d90e874b45aa426de762f19f')
# then # then
self.assertEquals({'found': False}, actual_lookup) self.assertEquals({'found': False}, actual_lookup)
# check the function has been called with parameters # check the function has been called with parameters
mock_storage.content_find.assert_called_with( mock_storage.content_find.assert_called_with(
{'sha1_git': {'sha1_git':
hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')}) hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_search_hash_exist(self, mock_storage):
def search_hash_exist(self, mock_storage):
# given # given
stub_content = { stub_content = {
'sha1': hash_to_bytes( 'sha1': hash_to_bytes(
'456caf10e9535160d90e874b45aa426de762f19f') '456caf10e9535160d90e874b45aa426de762f19f')
} }
mock_storage.content_find = MagicMock(return_value=stub_content) mock_storage.content_find = MagicMock(return_value=stub_content)
# when # when
actual_lookup = service.search_hash( actual_lookup = service.search_hash(
'sha1:456caf10e9535160d90e874b45aa426de762f19f') 'sha1:456caf10e9535160d90e874b45aa426de762f19f')
# then # then
self.assertEquals({'found': True}, actual_lookup) self.assertEquals({'found': True}, actual_lookup)
mock_storage.content_find.assert_called_with( mock_storage.content_find.assert_called_with(
{'sha1': {'sha1':
hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')}, hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')},
) )
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@istest def test_lookup_content_ctags(self, mock_idx_storage):
def lookup_content_ctags(self, mock_idx_storage):
# given # given
mock_idx_storage.content_ctags_get = MagicMock( mock_idx_storage.content_ctags_get = MagicMock(
return_value=[{ return_value=[{
'id': hash_to_bytes( 'id': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f'), '123caf10e9535160d90e874b45aa426de762f19f'),
'line': 100, 'line': 100,
'name': 'hello', 'name': 'hello',
'kind': 'function', 'kind': 'function',
Show All 15 Lines def test_lookup_content_ctags(self, mock_idx_storage):
# then # then
self.assertEqual(actual_ctags, expected_ctags) self.assertEqual(actual_ctags, expected_ctags)
mock_idx_storage.content_ctags_get.assert_called_with( mock_idx_storage.content_ctags_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')]) [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@istest def test_lookup_content_ctags_no_hash(self, mock_idx_storage):
def lookup_content_ctags_no_hash(self, mock_idx_storage):
# given # given
mock_idx_storage.content_ctags_get = MagicMock(return_value=[]) mock_idx_storage.content_ctags_get = MagicMock(return_value=[])
# when # when
actual_ctags = list(service.lookup_content_ctags( actual_ctags = list(service.lookup_content_ctags(
'sha1:123caf10e9535160d90e874b45aa426de762f19f')) 'sha1:123caf10e9535160d90e874b45aa426de762f19f'))
# then # then
self.assertEqual(actual_ctags, []) self.assertEqual(actual_ctags, [])
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@istest def test_lookup_content_filetype(self, mock_idx_storage):
def lookup_content_filetype(self, mock_idx_storage):
# given # given
mock_idx_storage.content_mimetype_get = MagicMock( mock_idx_storage.content_mimetype_get = MagicMock(
return_value=[{ return_value=[{
'id': hash_to_bytes( 'id': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f'), '123caf10e9535160d90e874b45aa426de762f19f'),
'mimetype': b'text/x-c++', 'mimetype': b'text/x-c++',
'encoding': b'us-ascii', 'encoding': b'us-ascii',
}]) }])
Show All 10 Lines def test_lookup_content_filetype(self, mock_idx_storage):
# then # then
self.assertEqual(actual_filetype, expected_filetype) self.assertEqual(actual_filetype, expected_filetype)
mock_idx_storage.content_mimetype_get.assert_called_with( mock_idx_storage.content_mimetype_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')]) [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_filetype_2(self, mock_storage, mock_idx_storage):
def lookup_content_filetype_2(self, mock_storage, mock_idx_storage):
# given # given
mock_storage.content_find = MagicMock( mock_storage.content_find = MagicMock(
return_value={ return_value={
'sha1': hash_to_bytes( 'sha1': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f') '123caf10e9535160d90e874b45aa426de762f19f')
} }
) )
mock_idx_storage.content_mimetype_get = MagicMock( mock_idx_storage.content_mimetype_get = MagicMock(
Show All 20 Lines def test_lookup_content_filetype_2(self, mock_storage, mock_idx_storage):
mock_storage.content_find( mock_storage.content_find(
'sha1_git', hash_to_bytes( 'sha1_git', hash_to_bytes(
'456caf10e9535160d90e874b45aa426de762f19f') '456caf10e9535160d90e874b45aa426de762f19f')
) )
mock_idx_storage.content_mimetype_get.assert_called_with( mock_idx_storage.content_mimetype_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')]) [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@istest def test_lookup_content_language(self, mock_idx_storage):
def lookup_content_language(self, mock_idx_storage):
# given # given
mock_idx_storage.content_language_get = MagicMock( mock_idx_storage.content_language_get = MagicMock(
return_value=[{ return_value=[{
'id': hash_to_bytes( 'id': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f'), '123caf10e9535160d90e874b45aa426de762f19f'),
'lang': 'python', 'lang': 'python',
}]) }])
expected_language = { expected_language = {
'id': '123caf10e9535160d90e874b45aa426de762f19f', 'id': '123caf10e9535160d90e874b45aa426de762f19f',
'lang': 'python', 'lang': 'python',
} }
# when # when
actual_language = service.lookup_content_language( actual_language = service.lookup_content_language(
'sha1:123caf10e9535160d90e874b45aa426de762f19f') 'sha1:123caf10e9535160d90e874b45aa426de762f19f')
# then # then
self.assertEqual(actual_language, expected_language) self.assertEqual(actual_language, expected_language)
mock_idx_storage.content_language_get.assert_called_with( mock_idx_storage.content_language_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')]) [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_language_2(self, mock_storage, mock_idx_storage):
def lookup_content_language_2(self, mock_storage, mock_idx_storage):
# given # given
mock_storage.content_find = MagicMock( mock_storage.content_find = MagicMock(
return_value={ return_value={
'sha1': hash_to_bytes( 'sha1': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f') '123caf10e9535160d90e874b45aa426de762f19f')
} }
) )
mock_idx_storage.content_language_get = MagicMock( mock_idx_storage.content_language_get = MagicMock(
Show All 18 Lines def test_lookup_content_language_2(self, mock_storage, mock_idx_storage):
mock_storage.content_find( mock_storage.content_find(
'sha1_git', hash_to_bytes( 'sha1_git', hash_to_bytes(
'456caf10e9535160d90e874b45aa426de762f19f') '456caf10e9535160d90e874b45aa426de762f19f')
) )
mock_idx_storage.content_language_get.assert_called_with( mock_idx_storage.content_language_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')]) [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@istest def test_lookup_expression(self, mock_idx_storage):
def lookup_expression(self, mock_idx_storage):
# given # given
mock_idx_storage.content_ctags_search = MagicMock( mock_idx_storage.content_ctags_search = MagicMock(
return_value=[{ return_value=[{
'id': hash_to_bytes( 'id': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f'), '123caf10e9535160d90e874b45aa426de762f19f'),
'name': 'foobar', 'name': 'foobar',
'kind': 'variable', 'kind': 'variable',
'lang': 'C', 'lang': 'C',
Show All 13 Lines def test_lookup_expression(self, mock_idx_storage):
# then # then
self.assertEqual(actual_ctags, expected_ctags) self.assertEqual(actual_ctags, expected_ctags)
mock_idx_storage.content_ctags_search.assert_called_with( mock_idx_storage.content_ctags_search.assert_called_with(
'foobar', last_sha1='hash', limit=10) 'foobar', last_sha1='hash', limit=10)
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@istest def test_lookup_expression_no_result(self, mock_idx_storage):
def lookup_expression_no_result(self, mock_idx_storage):
# given # given
mock_idx_storage.content_ctags_search = MagicMock( mock_idx_storage.content_ctags_search = MagicMock(
return_value=[]) return_value=[])
expected_ctags = [] expected_ctags = []
# when # when
actual_ctags = list(service.lookup_expression( actual_ctags = list(service.lookup_expression(
'barfoo', last_sha1='hash', per_page=10)) 'barfoo', last_sha1='hash', per_page=10))
# then # then
self.assertEqual(actual_ctags, expected_ctags) self.assertEqual(actual_ctags, expected_ctags)
mock_idx_storage.content_ctags_search.assert_called_with( mock_idx_storage.content_ctags_search.assert_called_with(
'barfoo', last_sha1='hash', limit=10) 'barfoo', last_sha1='hash', limit=10)
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@istest def test_lookup_content_license(self, mock_idx_storage):
def lookup_content_license(self, mock_idx_storage):
# given # given
mock_idx_storage.content_fossology_license_get = MagicMock( mock_idx_storage.content_fossology_license_get = MagicMock(
return_value=[{ return_value=[{
hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f'): [{ hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f'): [{
'licenses': ['GPL-3.0+'], 'licenses': ['GPL-3.0+'],
'tool': {} 'tool': {}
}] }]
}]) }])
Show All 12 Lines def test_lookup_content_license(self, mock_idx_storage):
# then # then
self.assertEqual(actual_license, expected_license) self.assertEqual(actual_license, expected_license)
mock_idx_storage.content_fossology_license_get.assert_called_with( mock_idx_storage.content_fossology_license_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')]) [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
@patch('swh.web.common.service.idx_storage') @patch('swh.web.common.service.idx_storage')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_license_2(self, mock_storage, mock_idx_storage):
def lookup_content_license_2(self, mock_storage, mock_idx_storage):
# given # given
mock_storage.content_find = MagicMock( mock_storage.content_find = MagicMock(
return_value={ return_value={
'sha1': hash_to_bytes( 'sha1': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f') '123caf10e9535160d90e874b45aa426de762f19f')
} }
) )
mock_idx_storage.content_fossology_license_get = MagicMock( mock_idx_storage.content_fossology_license_get = MagicMock(
Show All 23 Lines def test_lookup_content_license_2(self, mock_storage, mock_idx_storage):
mock_storage.content_find( mock_storage.content_find(
'sha1_git', hash_to_bytes( 'sha1_git', hash_to_bytes(
'456caf10e9535160d90e874b45aa426de762f19f') '456caf10e9535160d90e874b45aa426de762f19f')
) )
mock_idx_storage.content_fossology_license_get.assert_called_with( mock_idx_storage.content_fossology_license_get.assert_called_with(
[hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')]) [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_provenance(self, mock_storage):
def lookup_content_provenance(self, mock_storage):
# given # given
mock_storage.content_find_provenance = MagicMock( mock_storage.content_find_provenance = MagicMock(
return_value=(p for p in [{ return_value=(p for p in [{
'content': hash_to_bytes( 'content': hash_to_bytes(
'123caf10e9535160d90e874b45aa426de762f19f'), '123caf10e9535160d90e874b45aa426de762f19f'),
'revision': hash_to_bytes( 'revision': hash_to_bytes(
'456caf10e9535160d90e874b45aa426de762f19f'), '456caf10e9535160d90e874b45aa426de762f19f'),
'origin': 100, 'origin': 100,
Show All 15 Lines def test_lookup_content_provenance(self, mock_storage):
# then # then
self.assertEqual(list(actual_provenances), expected_provenances) self.assertEqual(list(actual_provenances), expected_provenances)
mock_storage.content_find_provenance.assert_called_with( mock_storage.content_find_provenance.assert_called_with(
{'sha1_git': {'sha1_git':
hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')}) hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_provenance_not_found(self, mock_storage):
def lookup_content_provenance_not_found(self, mock_storage):
# given # given
mock_storage.content_find_provenance = MagicMock(return_value=None) mock_storage.content_find_provenance = MagicMock(return_value=None)
# when # when
actual_provenances = service.lookup_content_provenance( actual_provenances = service.lookup_content_provenance(
'sha1_git:456caf10e9535160d90e874b45aa426de762f19f') 'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')
# then # then
self.assertIsNone(actual_provenances) self.assertIsNone(actual_provenances)
mock_storage.content_find_provenance.assert_called_with( mock_storage.content_find_provenance.assert_called_with(
{'sha1_git': {'sha1_git':
hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')}) hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_stat_counters(self, mock_storage):
def stat_counters(self, mock_storage):
# given # given
input_stats = { input_stats = {
"content": 1770830, "content": 1770830,
"directory": 211683, "directory": 211683,
"directory_entry_dir": 209167, "directory_entry_dir": 209167,
"directory_entry_file": 1807094, "directory_entry_file": 1807094,
"directory_entry_rev": 0, "directory_entry_rev": 0,
"entity": 0, "entity": 0,
Show All 12 Lines def test_stat_counters(self, mock_storage):
# then # then
expected_stats = input_stats expected_stats = input_stats
self.assertEqual(actual_stats, expected_stats) self.assertEqual(actual_stats, expected_stats)
mock_storage.stat_counters.assert_called_with() mock_storage.stat_counters.assert_called_with()
@patch('swh.web.common.service._lookup_origin_visits') @patch('swh.web.common.service._lookup_origin_visits')
@istest def test_lookup_origin_visits(self, mock_lookup_visits):
def lookup_origin_visits(self, mock_lookup_visits):
# given # given
date_origin_visit2 = datetime.datetime( date_origin_visit2 = datetime.datetime(
2013, 7, 1, 20, 0, 0, 2013, 7, 1, 20, 0, 0,
tzinfo=datetime.timezone.utc) tzinfo=datetime.timezone.utc)
date_origin_visit3 = datetime.datetime( date_origin_visit3 = datetime.datetime(
2015, 1, 1, 21, 0, 0, 2015, 1, 1, 21, 0, 0,
tzinfo=datetime.timezone.utc) tzinfo=datetime.timezone.utc)
Show All 36 Lines def test_lookup_origin_visits(self, mock_lookup_visits):
# then # then
self.assertEqual(list(actual_origin_visits), expected_origin_visits) self.assertEqual(list(actual_origin_visits), expected_origin_visits)
mock_lookup_visits.assert_called_once_with( mock_lookup_visits.assert_called_once_with(
6, last_visit=None, limit=10) 6, last_visit=None, limit=10)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_origin_visit(self, mock_storage):
def lookup_origin_visit(self, mock_storage):
# given # given
stub_result = self.origin_visit1 stub_result = self.origin_visit1
mock_storage.origin_visit_get_by.return_value = stub_result mock_storage.origin_visit_get_by.return_value = stub_result
expected_origin_visit = { expected_origin_visit = {
'date': self.origin_visit1['date'].isoformat(), 'date': self.origin_visit1['date'].isoformat(),
'origin': self.origin_visit1['origin'], 'origin': self.origin_visit1['origin'],
'visit': self.origin_visit1['visit'] 'visit': self.origin_visit1['visit']
} }
# when # when
actual_origin_visit = service.lookup_origin_visit(1, 1) actual_origin_visit = service.lookup_origin_visit(1, 1)
# then # then
self.assertEqual(actual_origin_visit, expected_origin_visit) self.assertEqual(actual_origin_visit, expected_origin_visit)
mock_storage.origin_visit_get_by.assert_called_once_with(1, 1) mock_storage.origin_visit_get_by.assert_called_once_with(1, 1)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_origin(self, mock_storage):
def lookup_origin(self, mock_storage):
# given # given
mock_storage.origin_get = MagicMock(return_value={ mock_storage.origin_get = MagicMock(return_value={
'id': 'origin-id', 'id': 'origin-id',
'url': 'ftp://some/url/to/origin', 'url': 'ftp://some/url/to/origin',
'type': 'ftp'}) 'type': 'ftp'})
# when # when
actual_origin = service.lookup_origin({'id': 'origin-id'}) actual_origin = service.lookup_origin({'id': 'origin-id'})
# then # then
self.assertEqual(actual_origin, {'id': 'origin-id', self.assertEqual(actual_origin, {'id': 'origin-id',
'url': 'ftp://some/url/to/origin', 'url': 'ftp://some/url/to/origin',
'type': 'ftp'}) 'type': 'ftp'})
mock_storage.origin_get.assert_called_with({'id': 'origin-id'}) mock_storage.origin_get.assert_called_with({'id': 'origin-id'})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_release_ko_id_checksum_not_a_sha1(self, mock_storage):
def lookup_release_ko_id_checksum_not_ok_because_not_a_sha1(self,
mock_storage):
# given # given
mock_storage.release_get = MagicMock() mock_storage.release_get = MagicMock()
with self.assertRaises(BadInputExc) as cm: with self.assertRaises(BadInputExc) as cm:
# when # when
service.lookup_release('not-a-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 mock_storage.release_get.called = False
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_release_ko_id_checksum_too_long(self, mock_storage):
def lookup_release_ko_id_checksum_ok_but_not_a_sha1(self, mock_storage):
# given # given
mock_storage.release_get = MagicMock() mock_storage.release_get = MagicMock()
# when # when
with self.assertRaises(BadInputExc) as cm: with self.assertRaises(BadInputExc) as cm:
service.lookup_release( service.lookup_release(
'13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4daf5' '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4daf5'
'1aea892abe') '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 mock_storage.release_get.called = False
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_directory_with_path_not_found(self, mock_storage):
def lookup_directory_with_path_not_found(self, mock_storage):
# given # given
mock_storage.lookup_directory_with_path = MagicMock(return_value=None) mock_storage.lookup_directory_with_path = MagicMock(return_value=None)
sha1_git = '65a55bbdf3629f916219feb3dcc7393ded1bc8db' sha1_git = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
# when # when
actual_directory = mock_storage.lookup_directory_with_path( actual_directory = mock_storage.lookup_directory_with_path(
sha1_git, 'some/path/here') sha1_git, 'some/path/here')
self.assertIsNone(actual_directory) self.assertIsNone(actual_directory)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_directory_with_path_found(self, mock_storage):
def lookup_directory_with_path_found(self, mock_storage):
# given # given
sha1_git = '65a55bbdf3629f916219feb3dcc7393ded1bc8db' sha1_git = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
entry = {'id': 'dir-id', entry = {'id': 'dir-id',
'type': 'dir', 'type': 'dir',
'name': 'some/path/foo'} 'name': 'some/path/foo'}
mock_storage.lookup_directory_with_path = MagicMock(return_value=entry) mock_storage.lookup_directory_with_path = MagicMock(return_value=entry)
# when # when
actual_directory = mock_storage.lookup_directory_with_path( actual_directory = mock_storage.lookup_directory_with_path(
sha1_git, 'some/path/here') sha1_git, 'some/path/here')
self.assertEqual(entry, actual_directory) self.assertEqual(entry, actual_directory)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_release(self, mock_storage):
def lookup_release(self, mock_storage):
# given # given
mock_storage.release_get = MagicMock(return_value=[{ mock_storage.release_get = MagicMock(return_value=[{
'id': hash_to_bytes('65a55bbdf3629f916219feb3dcc7393ded1bc8db'), 'id': hash_to_bytes('65a55bbdf3629f916219feb3dcc7393ded1bc8db'),
'target': None, 'target': None,
'date': { 'date': {
'timestamp': datetime.datetime( 'timestamp': datetime.datetime(
2015, 1, 1, 22, 0, 0, 2015, 1, 1, 22, 0, 0,
tzinfo=datetime.timezone.utc).timestamp(), tzinfo=datetime.timezone.utc).timestamp(),
Show All 17 Lines def test_lookup_release(self, mock_storage):
'name': 'v0.0.1', 'name': 'v0.0.1',
'message': 'synthetic release', 'message': 'synthetic release',
'synthetic': True, 'synthetic': True,
}) })
mock_storage.release_get.assert_called_with( mock_storage.release_get.assert_called_with(
[hash_to_bytes('65a55bbdf3629f916219feb3dcc7393ded1bc8db')]) [hash_to_bytes('65a55bbdf3629f916219feb3dcc7393ded1bc8db')])
@istest def test_lookup_revision_with_context_ko_not_a_sha1_1(self):
def lookup_revision_with_context_ko_not_a_sha1_1(self):
# given # given
sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4' \ sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4' \
'daf51aea892abe' 'daf51aea892abe'
sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db' sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
# when # 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])
@istest def test_lookup_revision_with_context_ko_not_a_sha1_2(self):
def lookup_revision_with_context_ko_not_a_sha1_2(self):
# given # given
sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db' sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f6' \ sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f6' \
'2d4daf51aea892abe' '2d4daf51aea892abe'
# when # 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') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_with_context_ko_sha1_git_does_not_exist(
def lookup_revision_with_context_ko_sha1_git_does_not_exist(
self, self,
mock_storage): mock_storage):
# given # given
sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db' sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
sha1_git = '777777bdf3629f916219feb3dcc7393ded1bc8db' sha1_git = '777777bdf3629f916219feb3dcc7393ded1bc8db'
sha1_git_bin = hash_to_bytes(sha1_git) sha1_git_bin = hash_to_bytes(sha1_git)
mock_storage.revision_get.return_value = None mock_storage.revision_get.return_value = None
# when # 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 777777bdf3629f916219feb3dcc7393ded1bc8db'
' not found', cm.exception.args[0]) ' not found', cm.exception.args[0])
mock_storage.revision_get.assert_called_once_with( mock_storage.revision_get.assert_called_once_with(
[sha1_git_bin]) [sha1_git_bin])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_with_context_ko_root_sha1_git_does_not_exist(
def lookup_revision_with_context_ko_root_sha1_git_does_not_exist(
self, self,
mock_storage): mock_storage):
# given # given
sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db' sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
sha1_git = '777777bdf3629f916219feb3dcc7393ded1bc8db' sha1_git = '777777bdf3629f916219feb3dcc7393ded1bc8db'
sha1_git_root_bin = hash_to_bytes(sha1_git_root) sha1_git_root_bin = hash_to_bytes(sha1_git_root)
sha1_git_bin = hash_to_bytes(sha1_git) sha1_git_bin = hash_to_bytes(sha1_git)
mock_storage.revision_get.side_effect = ['foo', None] mock_storage.revision_get.side_effect = ['foo', None]
# when # 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 65a55bbdf3629f916219feb3dcc7393ded1bc8db'
' not found', cm.exception.args[0]) ' not found', cm.exception.args[0])
mock_storage.revision_get.assert_has_calls([call([sha1_git_bin]), mock_storage.revision_get.assert_has_calls([call([sha1_git_bin]),
call([sha1_git_root_bin])]) call([sha1_git_root_bin])])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_revision_with_context(self, mock_query, mock_storage):
def lookup_revision_with_context(self, mock_query, mock_storage):
# given # given
sha1_git_root = '666' sha1_git_root = '666'
sha1_git = '883' sha1_git = '883'
sha1_git_root_bin = b'666' sha1_git_root_bin = b'666'
sha1_git_bin = b'883' sha1_git_bin = b'883'
sha1_git_root_dict = { sha1_git_root_dict = {
▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines def test_lookup_revision_with_context(self, mock_query, mock_storage):
[call(sha1_git, ['sha1'], 'Only sha1_git is supported.'), [call(sha1_git, ['sha1'], 'Only sha1_git is supported.'),
call(sha1_git_root, ['sha1'], 'Only sha1_git is supported.')]) call(sha1_git_root, ['sha1'], 'Only sha1_git is supported.')])
mock_storage.revision_log.assert_called_with( mock_storage.revision_log.assert_called_with(
[sha1_git_root_bin], 100) [sha1_git_root_bin], 100)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_revision_with_context_retrieved_as_dict(
def lookup_revision_with_context_sha1_git_root_already_retrieved_as_dict(
self, mock_query, mock_storage): self, mock_query, mock_storage):
# given # given
sha1_git = '883' sha1_git = '883'
sha1_git_root_bin = b'666' sha1_git_root_bin = b'666'
sha1_git_bin = b'883' sha1_git_bin = b'883'
sha1_git_root_dict = { sha1_git_root_dict = {
▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Lines def test_lookup_revision_with_context_retrieved_as_dict(
mock_storage.revision_get.assert_called_once_with([sha1_git_bin]) mock_storage.revision_get.assert_called_once_with([sha1_git_bin])
mock_storage.revision_log.assert_called_with( mock_storage.revision_log.assert_called_with(
[sha1_git_root_bin], 100) [sha1_git_root_bin], 100)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_directory_with_revision_not_found(self,
def lookup_directory_with_revision_ko_revision_not_found(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',
b'123') b'123')
mock_storage.revision_get.return_value = None mock_storage.revision_get.return_value = None
# when # when
with self.assertRaises(NotFoundExc) as cm: with self.assertRaises(NotFoundExc) as cm:
service.lookup_directory_with_revision('123') service.lookup_directory_with_revision('123')
self.assertIn('Revision 123 not found', cm.exception.args[0]) self.assertIn('Revision 123 not found', cm.exception.args[0])
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'])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_directory_with_revision_ko_revision_with_path_to_nowhere(
def lookup_directory_with_revision_ko_revision_with_path_to_nowhere(
self, 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',
b'123') b'123')
dir_id = b'dir-id-as-sha1' dir_id = b'dir-id-as-sha1'
Show All 17 Lines def test_lookup_directory_with_revision_ko_revision_with_path_to_nowhere(
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'path', b'to', b'something', b'unknown']) b'dir-id-as-sha1', [b'path', b'to', b'something', b'unknown'])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_directory_with_revision_ko_type_not_implemented(
def lookup_directory_with_revision_ko_type_not_implemented(
self, 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',
b'123') b'123')
Show All 27 Lines def test_lookup_directory_with_revision_ko_type_not_implemented(
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'rev']) b'dir-id-as-sha1', [b'some', b'path', b'to', b'rev'])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_directory_with_revision_revision_without_path(
def lookup_directory_with_revision_revision_without_path(self, self, mock_query, mock_storage,
mock_query, ):
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',
b'123') b'123')
dir_id = b'dir-id-as-sha1' dir_id = b'dir-id-as-sha1'
mock_storage.revision_get.return_value = [{ mock_storage.revision_get.return_value = [{
'directory': dir_id, 'directory': dir_id,
}] }]
Show All 18 Lines ):
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_ls.assert_called_once_with(dir_id) 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')
@istest def test_lookup_directory_with_revision_with_path_to_dir(self,
def lookup_directory_with_revision_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',
b'123') b'123')
dir_id = b'dir-id-as-sha1' dir_id = b'dir-id-as-sha1'
mock_storage.revision_get.return_value = [{ mock_storage.revision_get.return_value = [{
'directory': dir_id, 'directory': dir_id,
}] }]
Show All 29 Lines def test_lookup_directory_with_revision_with_path_to_dir(self,
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(
dir_id, dir_id,
[b'some', b'path']) [b'some', b'path'])
mock_storage.directory_ls.assert_called_once_with(b'456') mock_storage.directory_ls.assert_called_once_with(b'456')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_directory_with_revision_with_path_to_file_wo_data(
def lookup_directory_with_revision_revision_with_path_to_file_without_data(
self, 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',
b'123') b'123')
Show All 29 Lines def test_lookup_directory_with_revision_with_path_to_file_wo_data(
('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'})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_directory_with_revision_with_path_to_file_w_data(
def lookup_directory_with_revision_revision_with_path_to_file_with_data(
self, 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',
b'123') b'123')
▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Lines def test_lookup_directory_with_revision_with_path_to_file_w_data(
('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') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision(self, mock_storage):
def lookup_revision(self, mock_storage):
# given # given
mock_storage.revision_get = MagicMock( mock_storage.revision_get = MagicMock(
return_value=[self.SAMPLE_REVISION_RAW]) return_value=[self.SAMPLE_REVISION_RAW])
# when # when
actual_revision = service.lookup_revision( actual_revision = service.lookup_revision(
self.SHA1_SAMPLE) self.SHA1_SAMPLE)
# then # then
self.assertEqual(actual_revision, self.SAMPLE_REVISION) self.assertEqual(actual_revision, self.SAMPLE_REVISION)
mock_storage.revision_get.assert_called_with( mock_storage.revision_get.assert_called_with(
[self.SHA1_SAMPLE_BIN]) [self.SHA1_SAMPLE_BIN])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_invalid_msg(self, mock_storage):
def 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
expected_revision['message'] = None expected_revision['message'] = None
expected_revision['message_decoding_failed'] = True expected_revision['message_decoding_failed'] = True
mock_storage.revision_get = MagicMock(return_value=[stub_rev]) mock_storage.revision_get = MagicMock(return_value=[stub_rev])
# when # when
actual_revision = service.lookup_revision( actual_revision = service.lookup_revision(
self.SHA1_SAMPLE) self.SHA1_SAMPLE)
# then # then
self.assertEqual(actual_revision, expected_revision) self.assertEqual(actual_revision, expected_revision)
mock_storage.revision_get.assert_called_with( mock_storage.revision_get.assert_called_with(
[self.SHA1_SAMPLE_BIN]) [self.SHA1_SAMPLE_BIN])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_msg_ok(self, mock_storage):
def lookup_revision_msg_ok(self, mock_storage):
# given # given
mock_storage.revision_get.return_value = [self.SAMPLE_REVISION_RAW] mock_storage.revision_get.return_value = [self.SAMPLE_REVISION_RAW]
# when # when
rv = service.lookup_revision_message( rv = service.lookup_revision_message(
self.SHA1_SAMPLE) self.SHA1_SAMPLE)
# then # then
self.assertEquals(rv, {'message': self.SAMPLE_MESSAGE_BIN}) self.assertEquals(rv, {'message': self.SAMPLE_MESSAGE_BIN})
mock_storage.revision_get.assert_called_with( mock_storage.revision_get.assert_called_with(
[self.SHA1_SAMPLE_BIN]) [self.SHA1_SAMPLE_BIN])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_msg_absent(self, mock_storage):
def lookup_revision_msg_absent(self, mock_storage):
# given # given
stub_revision = self.SAMPLE_REVISION_RAW stub_revision = self.SAMPLE_REVISION_RAW
del stub_revision['message'] del stub_revision['message']
mock_storage.revision_get.return_value = stub_revision mock_storage.revision_get.return_value = stub_revision
# when # when
with self.assertRaises(NotFoundExc) as cm: with self.assertRaises(NotFoundExc) as cm:
service.lookup_revision_message( service.lookup_revision_message(
self.SHA1_SAMPLE) self.SHA1_SAMPLE)
# then # then
mock_storage.revision_get.assert_called_with( mock_storage.revision_get.assert_called_with(
[self.SHA1_SAMPLE_BIN]) [self.SHA1_SAMPLE_BIN])
self.assertEqual( self.assertEqual(
cm.exception.args[0], cm.exception.args[0],
'No message for revision with sha1_git %s.' % self.SHA1_SAMPLE, 'No message for revision with sha1_git %s.' % self.SHA1_SAMPLE,
) )
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_msg_norev(self, mock_storage):
def lookup_revision_msg_norev(self, mock_storage):
# given # given
mock_storage.revision_get.return_value = None mock_storage.revision_get.return_value = None
# when # when
with self.assertRaises(NotFoundExc) as cm: with self.assertRaises(NotFoundExc) as cm:
service.lookup_revision_message( service.lookup_revision_message(
self.SHA1_SAMPLE) self.SHA1_SAMPLE)
# then # then
mock_storage.revision_get.assert_called_with( mock_storage.revision_get.assert_called_with(
[self.SHA1_SAMPLE_BIN]) [self.SHA1_SAMPLE_BIN])
self.assertEqual( self.assertEqual(
cm.exception.args[0], cm.exception.args[0],
'Revision with sha1_git %s not found.' % self.SHA1_SAMPLE, 'Revision with sha1_git %s not found.' % self.SHA1_SAMPLE,
) )
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_multiple(self, mock_storage):
def lookup_revision_multiple(self, mock_storage):
# given # given
sha1 = self.SHA1_SAMPLE sha1 = self.SHA1_SAMPLE
sha1_other = 'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc' sha1_other = 'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'
stub_revisions = [ stub_revisions = [
self.SAMPLE_REVISION_RAW, self.SAMPLE_REVISION_RAW,
{ {
'id': hash_to_bytes(sha1_other), 'id': hash_to_bytes(sha1_other),
▲ Show 20 Lines • Show All 64 Lines • ▼ Show 20 Lines def test_lookup_revision_multiple(self, mock_storage):
]) ])
self.assertEqual( self.assertEqual(
list(mock_storage.revision_get.call_args[0][0]), list(mock_storage.revision_get.call_args[0][0]),
[hash_to_bytes(sha1), [hash_to_bytes(sha1),
hash_to_bytes(sha1_other)]) hash_to_bytes(sha1_other)])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_multiple_none_found(self, mock_storage):
def lookup_revision_multiple_none_found(self, mock_storage):
# given # given
sha1_bin = self.SHA1_SAMPLE sha1_bin = self.SHA1_SAMPLE
sha1_other = 'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc' sha1_other = 'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'
mock_storage.revision_get.return_value = [] mock_storage.revision_get.return_value = []
# then # then
actual_revisions = service.lookup_revision_multiple( actual_revisions = service.lookup_revision_multiple(
[sha1_bin, sha1_other]) [sha1_bin, sha1_other])
self.assertEqual(list(actual_revisions), []) self.assertEqual(list(actual_revisions), [])
self.assertEqual( self.assertEqual(
list(mock_storage.revision_get.call_args[0][0]), list(mock_storage.revision_get.call_args[0][0]),
[hash_to_bytes(self.SHA1_SAMPLE), [hash_to_bytes(self.SHA1_SAMPLE),
hash_to_bytes(sha1_other)]) hash_to_bytes(sha1_other)])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_log(self, mock_storage):
def lookup_revision_log(self, mock_storage):
# given # given
stub_revision_log = [self.SAMPLE_REVISION_RAW] stub_revision_log = [self.SAMPLE_REVISION_RAW]
mock_storage.revision_log = MagicMock(return_value=stub_revision_log) mock_storage.revision_log = MagicMock(return_value=stub_revision_log)
# when # when
actual_revision = service.lookup_revision_log( actual_revision = service.lookup_revision_log(
'abcdbe353ed3480476f032475e7c233eff7371d5', 'abcdbe353ed3480476f032475e7c233eff7371d5',
limit=25) limit=25)
# then # then
self.assertEqual(list(actual_revision), [self.SAMPLE_REVISION]) self.assertEqual(list(actual_revision), [self.SAMPLE_REVISION])
mock_storage.revision_log.assert_called_with( mock_storage.revision_log.assert_called_with(
[hash_to_bytes('abcdbe353ed3480476f032475e7c233eff7371d5')], 25) [hash_to_bytes('abcdbe353ed3480476f032475e7c233eff7371d5')], 25)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_log_by(self, mock_storage):
def lookup_revision_log_by(self, mock_storage):
# given # given
stub_revision_log = [self.SAMPLE_REVISION_RAW] stub_revision_log = [self.SAMPLE_REVISION_RAW]
mock_storage.revision_log_by = MagicMock( mock_storage.revision_log_by = MagicMock(
return_value=stub_revision_log) return_value=stub_revision_log)
# when # when
actual_log = service.lookup_revision_log_by( actual_log = service.lookup_revision_log_by(
1, 'refs/heads/master', None, limit=100) 1, 'refs/heads/master', None, limit=100)
# then # then
self.assertEqual(list(actual_log), [self.SAMPLE_REVISION]) self.assertEqual(list(actual_log), [self.SAMPLE_REVISION])
mock_storage.revision_log_by.assert_called_with( mock_storage.revision_log_by.assert_called_with(
1, 'refs/heads/master', None, limit=100) 1, 'refs/heads/master', None, limit=100)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_log_by_nolog(self, mock_storage):
def lookup_revision_log_by_nolog(self, mock_storage):
# given # given
mock_storage.revision_log_by = MagicMock(return_value=None) mock_storage.revision_log_by = MagicMock(return_value=None)
# when # when
res = service.lookup_revision_log_by( res = service.lookup_revision_log_by(
1, 'refs/heads/master', None, limit=100) 1, 'refs/heads/master', None, limit=100)
# then # then
self.assertEquals(res, None) self.assertEquals(res, None)
mock_storage.revision_log_by.assert_called_with( mock_storage.revision_log_by.assert_called_with(
1, 'refs/heads/master', None, limit=100) 1, 'refs/heads/master', None, limit=100)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_raw_not_found(self, mock_storage):
def lookup_content_raw_not_found(self, mock_storage):
# given # given
mock_storage.content_find = MagicMock(return_value=None) mock_storage.content_find = MagicMock(return_value=None)
# when # when
with self.assertRaises(NotFoundExc) as cm: with self.assertRaises(NotFoundExc) as cm:
service.lookup_content_raw('sha1:' + self.SHA1_SAMPLE) service.lookup_content_raw('sha1:' + self.SHA1_SAMPLE)
self.assertIn(cm.exception.args[0], self.assertIn(cm.exception.args[0],
'Content with %s checksum equals to %s not found!' % 'Content with %s checksum equals to %s not found!' %
('sha1', self.SHA1_SAMPLE)) ('sha1', self.SHA1_SAMPLE))
mock_storage.content_find.assert_called_with( mock_storage.content_find.assert_called_with(
{'sha1': hash_to_bytes(self.SHA1_SAMPLE)}) {'sha1': hash_to_bytes(self.SHA1_SAMPLE)})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_raw(self, mock_storage):
def lookup_content_raw(self, mock_storage):
# given # given
mock_storage.content_find = MagicMock(return_value={ mock_storage.content_find = MagicMock(return_value={
'sha1': self.SHA1_SAMPLE, 'sha1': self.SHA1_SAMPLE,
}) })
mock_storage.content_get = MagicMock(return_value=[{ mock_storage.content_get = MagicMock(return_value=[{
'data': b'binary data'}]) 'data': b'binary data'}])
# when # when
actual_content = service.lookup_content_raw( actual_content = service.lookup_content_raw(
'sha256:%s' % self.SHA256_SAMPLE) 'sha256:%s' % self.SHA256_SAMPLE)
# then # then
self.assertEquals(actual_content, {'data': b'binary data'}) self.assertEquals(actual_content, {'data': b'binary data'})
mock_storage.content_find.assert_called_once_with( mock_storage.content_find.assert_called_once_with(
{'sha256': self.SHA256_SAMPLE_BIN}) {'sha256': self.SHA256_SAMPLE_BIN})
mock_storage.content_get.assert_called_once_with( mock_storage.content_get.assert_called_once_with(
[self.SHA1_SAMPLE]) [self.SHA1_SAMPLE])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_not_found(self, mock_storage):
def lookup_content_not_found(self, mock_storage):
# given # given
mock_storage.content_find = MagicMock(return_value=None) mock_storage.content_find = MagicMock(return_value=None)
# when # when
with self.assertRaises(NotFoundExc) as cm: with self.assertRaises(NotFoundExc) as cm:
# then # then
service.lookup_content('sha1:%s' % self.SHA1_SAMPLE) service.lookup_content('sha1:%s' % self.SHA1_SAMPLE)
self.assertIn(cm.exception.args[0], self.assertIn(cm.exception.args[0],
'Content with %s checksum equals to %s not found!' % 'Content with %s checksum equals to %s not found!' %
('sha1', self.SHA1_SAMPLE)) ('sha1', self.SHA1_SAMPLE))
mock_storage.content_find.assert_called_with( mock_storage.content_find.assert_called_with(
{'sha1': self.SHA1_SAMPLE_BIN}) {'sha1': self.SHA1_SAMPLE_BIN})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_with_sha1(self, mock_storage):
def lookup_content_with_sha1(self, mock_storage):
# given # given
mock_storage.content_find = MagicMock( mock_storage.content_find = MagicMock(
return_value=self.SAMPLE_CONTENT_RAW) return_value=self.SAMPLE_CONTENT_RAW)
# when # when
actual_content = service.lookup_content( actual_content = service.lookup_content(
'sha1:%s' % self.SHA1_SAMPLE) 'sha1:%s' % self.SHA1_SAMPLE)
# then # then
self.assertEqual(actual_content, self.SAMPLE_CONTENT) self.assertEqual(actual_content, self.SAMPLE_CONTENT)
mock_storage.content_find.assert_called_with( mock_storage.content_find.assert_called_with(
{'sha1': hash_to_bytes(self.SHA1_SAMPLE)}) {'sha1': hash_to_bytes(self.SHA1_SAMPLE)})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_content_with_sha256(self, mock_storage):
def lookup_content_with_sha256(self, mock_storage):
# given # given
stub_content = self.SAMPLE_CONTENT_RAW stub_content = self.SAMPLE_CONTENT_RAW
stub_content['status'] = 'visible' stub_content['status'] = 'visible'
expected_content = self.SAMPLE_CONTENT expected_content = self.SAMPLE_CONTENT
expected_content['status'] = 'visible' expected_content['status'] = 'visible'
mock_storage.content_find = MagicMock( mock_storage.content_find = MagicMock(
return_value=stub_content) return_value=stub_content)
# when # when
actual_content = service.lookup_content( actual_content = service.lookup_content(
'sha256:%s' % self.SHA256_SAMPLE) 'sha256:%s' % self.SHA256_SAMPLE)
# then # then
self.assertEqual(actual_content, expected_content) self.assertEqual(actual_content, expected_content)
mock_storage.content_find.assert_called_with( mock_storage.content_find.assert_called_with(
{'sha256': self.SHA256_SAMPLE_BIN}) {'sha256': self.SHA256_SAMPLE_BIN})
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_person(self, mock_storage):
def lookup_person(self, mock_storage):
# given # given
mock_storage.person_get = MagicMock(return_value=[{ mock_storage.person_get = MagicMock(return_value=[{
'id': 'person_id', 'id': 'person_id',
'name': b'some_name', 'name': b'some_name',
'email': b'some-email', 'email': b'some-email',
}]) }])
# when # when
actual_person = service.lookup_person('person_id') actual_person = service.lookup_person('person_id')
# then # then
self.assertEqual(actual_person, { self.assertEqual(actual_person, {
'id': 'person_id', 'id': 'person_id',
'name': 'some_name', 'name': 'some_name',
'email': 'some-email', 'email': 'some-email',
}) })
mock_storage.person_get.assert_called_with(['person_id']) mock_storage.person_get.assert_called_with(['person_id'])
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_directory_bad_checksum(self, mock_storage):
def lookup_directory_bad_checksum(self, mock_storage):
# given # given
mock_storage.directory_ls = MagicMock() mock_storage.directory_ls = MagicMock()
# when # when
with self.assertRaises(BadInputExc): with self.assertRaises(BadInputExc):
service.lookup_directory('directory_id') service.lookup_directory('directory_id')
# then # then
mock_storage.directory_ls.called = False mock_storage.directory_ls.called = False
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_directory_not_found(self, mock_query, mock_storage):
def lookup_directory_not_found(self, mock_query, mock_storage):
# given # given
mock_query.parse_hash_with_algorithms_or_throws.return_value = ( mock_query.parse_hash_with_algorithms_or_throws.return_value = (
'sha1', 'sha1',
'directory-id-bin') 'directory-id-bin')
mock_storage.directory_ls.return_value = [] mock_storage.directory_ls.return_value = []
# when # when
with self.assertRaises(NotFoundExc) as cm: with self.assertRaises(NotFoundExc) as cm:
service.lookup_directory('directory_id') service.lookup_directory('directory_id')
self.assertIn('Directory with sha1_git directory_id not found', self.assertIn('Directory with sha1_git directory_id not found',
cm.exception.args[0]) cm.exception.args[0])
# then # then
mock_query.parse_hash_with_algorithms_or_throws.assert_called_with( mock_query.parse_hash_with_algorithms_or_throws.assert_called_with(
'directory_id', ['sha1'], 'Only sha1_git is supported.') 'directory_id', ['sha1'], 'Only sha1_git is supported.')
mock_storage.directory_ls.assert_called_with('directory-id-bin') mock_storage.directory_ls.assert_called_with('directory-id-bin')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_directory(self, mock_query, mock_storage):
def lookup_directory(self, mock_query, mock_storage):
mock_query.parse_hash_with_algorithms_or_throws.return_value = ( mock_query.parse_hash_with_algorithms_or_throws.return_value = (
'sha1', 'sha1',
'directory-sha1-bin') 'directory-sha1-bin')
# given # given
stub_dir_entries = [{ stub_dir_entries = [{
'sha1': self.SHA1_SAMPLE_BIN, 'sha1': self.SHA1_SAMPLE_BIN,
'sha256': self.SHA256_SAMPLE_BIN, 'sha256': self.SHA256_SAMPLE_BIN,
Show All 29 Lines def test_lookup_directory(self, mock_query, mock_storage):
self.assertEqual(actual_directory_ls, expected_dir_entries) self.assertEqual(actual_directory_ls, expected_dir_entries)
mock_query.parse_hash_with_algorithms_or_throws.assert_called_with( mock_query.parse_hash_with_algorithms_or_throws.assert_called_with(
'directory-sha1', ['sha1'], 'Only sha1_git is supported.') 'directory-sha1', ['sha1'], 'Only sha1_git is supported.')
mock_storage.directory_ls.assert_called_with( mock_storage.directory_ls.assert_called_with(
'directory-sha1-bin') 'directory-sha1-bin')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_directory_empty(self, mock_storage):
def lookup_directory_empty(self, mock_storage):
empty_dir_sha1 = '4b825dc642cb6eb9a060e54bf8d69288fbee4904' empty_dir_sha1 = '4b825dc642cb6eb9a060e54bf8d69288fbee4904'
mock_storage.directory_ls.return_value = [] mock_storage.directory_ls.return_value = []
# when # when
actual_directory_ls = list(service.lookup_directory(empty_dir_sha1)) actual_directory_ls = list(service.lookup_directory(empty_dir_sha1))
# then # then
self.assertEqual(actual_directory_ls, []) self.assertEqual(actual_directory_ls, [])
self.assertFalse(mock_storage.directory_ls.called) self.assertFalse(mock_storage.directory_ls.called)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_by_nothing_found(self, mock_storage):
def lookup_revision_by_nothing_found(self, mock_storage):
# given # given
mock_storage.revision_get_by.return_value = None mock_storage.revision_get_by.return_value = None
# when # when
with self.assertRaises(NotFoundExc): with self.assertRaises(NotFoundExc):
service.lookup_revision_by(1) service.lookup_revision_by(1)
# then # then
mock_storage.revision_get_by.assert_called_with(1, 'refs/heads/master', # noqa mock_storage.revision_get_by.assert_called_with(1, 'refs/heads/master', # noqa
limit=1, limit=1,
timestamp=None) timestamp=None)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_by(self, mock_storage):
def lookup_revision_by(self, mock_storage):
# given # given
stub_rev = self.SAMPLE_REVISION_RAW stub_rev = self.SAMPLE_REVISION_RAW
expected_rev = self.SAMPLE_REVISION expected_rev = self.SAMPLE_REVISION
mock_storage.revision_get_by.return_value = [stub_rev] mock_storage.revision_get_by.return_value = [stub_rev]
# when # when
actual_revision = service.lookup_revision_by(10, 'master2', 'some-ts') actual_revision = service.lookup_revision_by(10, 'master2', 'some-ts')
# then # then
self.assertEquals(actual_revision, expected_rev) self.assertEquals(actual_revision, expected_rev)
mock_storage.revision_get_by.assert_called_with(10, 'master2', mock_storage.revision_get_by.assert_called_with(10, 'master2',
limit=1, limit=1,
timestamp='some-ts') timestamp='some-ts')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_by_nomerge(self, mock_storage):
def lookup_revision_by_nomerge(self, mock_storage):
# given # given
stub_rev = self.SAMPLE_REVISION_RAW stub_rev = self.SAMPLE_REVISION_RAW
stub_rev['parents'] = [ stub_rev['parents'] = [
hash_to_bytes('adc83b19e793491b1c6ea0fd8b46cd9f32e592fc')] hash_to_bytes('adc83b19e793491b1c6ea0fd8b46cd9f32e592fc')]
expected_rev = self.SAMPLE_REVISION expected_rev = self.SAMPLE_REVISION
expected_rev['parents'] = ['adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'] expected_rev['parents'] = ['adc83b19e793491b1c6ea0fd8b46cd9f32e592fc']
mock_storage.revision_get_by.return_value = [stub_rev] mock_storage.revision_get_by.return_value = [stub_rev]
# when # when
actual_revision = service.lookup_revision_by(10, 'master2', 'some-ts') actual_revision = service.lookup_revision_by(10, 'master2', 'some-ts')
# then # then
self.assertEquals(actual_revision, expected_rev) self.assertEquals(actual_revision, expected_rev)
mock_storage.revision_get_by.assert_called_with(10, 'master2', mock_storage.revision_get_by.assert_called_with(10, 'master2',
limit=1, limit=1,
timestamp='some-ts') timestamp='some-ts')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_by_merge(self, mock_storage):
def lookup_revision_by_merge(self, mock_storage):
# given # given
stub_rev = self.SAMPLE_REVISION_RAW stub_rev = self.SAMPLE_REVISION_RAW
stub_rev['parents'] = [ stub_rev['parents'] = [
hash_to_bytes('adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'), hash_to_bytes('adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'),
hash_to_bytes('ffff3b19e793491b1c6db0fd8b46cd9f32e592fc') hash_to_bytes('ffff3b19e793491b1c6db0fd8b46cd9f32e592fc')
] ]
expected_rev = self.SAMPLE_REVISION expected_rev = self.SAMPLE_REVISION
Show All 11 Lines def test_lookup_revision_by_merge(self, mock_storage):
# then # then
self.assertEquals(actual_revision, expected_rev) self.assertEquals(actual_revision, expected_rev)
mock_storage.revision_get_by.assert_called_with(10, 'master2', mock_storage.revision_get_by.assert_called_with(10, 'master2',
limit=1, limit=1,
timestamp='some-ts') timestamp='some-ts')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_with_context_by_ko(self, mock_storage):
def lookup_revision_with_context_by_ko(self, mock_storage):
# given # given
mock_storage.revision_get_by.return_value = None mock_storage.revision_get_by.return_value = None
# when # when
origin_id = 1 origin_id = 1
branch_name = 'master3' branch_name = 'master3'
ts = None ts = None
with self.assertRaises(NotFoundExc) as cm: with self.assertRaises(NotFoundExc) as cm:
service.lookup_revision_with_context_by(origin_id, branch_name, ts, service.lookup_revision_with_context_by(origin_id, branch_name, ts,
'sha1') 'sha1')
# then # then
self.assertIn( self.assertIn(
'Revision with (origin_id: %s, branch_name: %s' 'Revision with (origin_id: %s, branch_name: %s'
', ts: %s) not found.' % (origin_id, ', ts: %s) not found.' % (origin_id,
branch_name, branch_name,
ts), cm.exception.args[0]) ts), cm.exception.args[0])
mock_storage.revision_get_by.assert_called_once_with( mock_storage.revision_get_by.assert_called_once_with(
origin_id, branch_name, limit=1, timestamp=ts) origin_id, branch_name, limit=1, timestamp=ts)
@patch('swh.web.common.service.lookup_revision_with_context') @patch('swh.web.common.service.lookup_revision_with_context')
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@istest def test_lookup_revision_with_context_by(
def lookup_revision_with_context_by(self, mock_storage, self, mock_storage, mock_lookup_revision_with_context,
mock_lookup_revision_with_context): ):
# given # given
stub_root_rev = {'id': 'root-rev-id'} stub_root_rev = {'id': 'root-rev-id'}
mock_storage.revision_get_by.return_value = [{'id': 'root-rev-id'}] mock_storage.revision_get_by.return_value = [{'id': 'root-rev-id'}]
stub_rev = {'id': 'rev-found'} stub_rev = {'id': 'rev-found'}
mock_lookup_revision_with_context.return_value = stub_rev mock_lookup_revision_with_context.return_value = stub_rev
# when # when
origin_id = 1 origin_id = 1
Show All 9 Lines ):
mock_storage.revision_get_by.assert_called_once_with( mock_storage.revision_get_by.assert_called_once_with(
origin_id, branch_name, limit=1, timestamp=ts) origin_id, branch_name, limit=1, timestamp=ts)
mock_lookup_revision_with_context.assert_called_once_with( mock_lookup_revision_with_context.assert_called_once_with(
stub_root_rev, sha1_git, 100) stub_root_rev, sha1_git, 100)
@patch('swh.web.common.service.storage') @patch('swh.web.common.service.storage')
@patch('swh.web.common.service.query') @patch('swh.web.common.service.query')
@istest def test_lookup_entity_by_uuid(self, mock_query, mock_storage):
def lookup_entity_by_uuid(self, mock_query, mock_storage):
# given # given
uuid_test = 'correct-uuid' uuid_test = 'correct-uuid'
mock_query.parse_uuid4.return_value = uuid_test mock_query.parse_uuid4.return_value = uuid_test
stub_entities = [{'uuid': uuid_test}] stub_entities = [{'uuid': uuid_test}]
mock_storage.entity_get.return_value = stub_entities mock_storage.entity_get.return_value = stub_entities
# when # when
actual_entities = list(service.lookup_entity_by_uuid(uuid_test)) actual_entities = list(service.lookup_entity_by_uuid(uuid_test))
# then # then
self.assertEquals(actual_entities, stub_entities) self.assertEquals(actual_entities, stub_entities)
mock_query.parse_uuid4.assert_called_once_with(uuid_test) mock_query.parse_uuid4.assert_called_once_with(uuid_test)
mock_storage.entity_get.assert_called_once_with(uuid_test) mock_storage.entity_get.assert_called_once_with(uuid_test)
@istest def test_lookup_revision_through_ko_not_implemented(self):
def lookup_revision_through_ko_not_implemented(self):
# then # then
with self.assertRaises(NotImplementedError): with self.assertRaises(NotImplementedError):
service.lookup_revision_through({ service.lookup_revision_through({
'something-unknown': 10, 'something-unknown': 10,
}) })
@patch('swh.web.common.service.lookup_revision_with_context_by') @patch('swh.web.common.service.lookup_revision_with_context_by')
@istest def test_lookup_revision_through_with_context_by(self, mock_lookup):
def lookup_revision_through_with_context_by(self, mock_lookup):
# given # given
stub_rev = {'id': 'rev'} stub_rev = {'id': 'rev'}
mock_lookup.return_value = stub_rev mock_lookup.return_value = stub_rev
# when # when
actual_revision = service.lookup_revision_through({ actual_revision = service.lookup_revision_through({
'origin_id': 1, 'origin_id': 1,
'branch_name': 'master', 'branch_name': 'master',
'ts': None, 'ts': None,
'sha1_git': 'sha1-git' 'sha1_git': 'sha1-git'
}, limit=1000) }, limit=1000)
# then # then
self.assertEquals(actual_revision, stub_rev) self.assertEquals(actual_revision, stub_rev)
mock_lookup.assert_called_once_with( mock_lookup.assert_called_once_with(
1, 'master', None, 'sha1-git', 1000) 1, 'master', None, 'sha1-git', 1000)
@patch('swh.web.common.service.lookup_revision_by') @patch('swh.web.common.service.lookup_revision_by')
@istest def test_lookup_revision_through_with_revision_by(self, mock_lookup):
def lookup_revision_through_with_revision_by(self, mock_lookup):
# given # given
stub_rev = {'id': 'rev'} stub_rev = {'id': 'rev'}
mock_lookup.return_value = stub_rev mock_lookup.return_value = stub_rev
# when # when
actual_revision = service.lookup_revision_through({ actual_revision = service.lookup_revision_through({
'origin_id': 2, 'origin_id': 2,
'branch_name': 'master2', 'branch_name': 'master2',
'ts': 'some-ts', 'ts': 'some-ts',
}, limit=10) }, limit=10)
# then # then
self.assertEquals(actual_revision, stub_rev) self.assertEquals(actual_revision, stub_rev)
mock_lookup.assert_called_once_with( mock_lookup.assert_called_once_with(
2, 'master2', 'some-ts') 2, 'master2', 'some-ts')
@patch('swh.web.common.service.lookup_revision_with_context') @patch('swh.web.common.service.lookup_revision_with_context')
@istest def test_lookup_revision_through_with_context(self, mock_lookup):
def lookup_revision_through_with_context(self, mock_lookup):
# given # given
stub_rev = {'id': 'rev'} stub_rev = {'id': 'rev'}
mock_lookup.return_value = stub_rev mock_lookup.return_value = stub_rev
# when # when
actual_revision = service.lookup_revision_through({ actual_revision = service.lookup_revision_through({
'sha1_git_root': 'some-sha1-root', 'sha1_git_root': 'some-sha1-root',
'sha1_git': 'some-sha1', 'sha1_git': 'some-sha1',
}) })
# then # then
self.assertEquals(actual_revision, stub_rev) self.assertEquals(actual_revision, stub_rev)
mock_lookup.assert_called_once_with( mock_lookup.assert_called_once_with(
'some-sha1-root', 'some-sha1', 100) 'some-sha1-root', 'some-sha1', 100)
@patch('swh.web.common.service.lookup_revision') @patch('swh.web.common.service.lookup_revision')
@istest def test_lookup_revision_through_with_revision(self, mock_lookup):
def lookup_revision_through_with_revision(self, mock_lookup):
# given # given
stub_rev = {'id': 'rev'} stub_rev = {'id': 'rev'}
mock_lookup.return_value = stub_rev mock_lookup.return_value = stub_rev
# when # when
actual_revision = service.lookup_revision_through({ actual_revision = service.lookup_revision_through({
'sha1_git': 'some-sha1', 'sha1_git': 'some-sha1',
}) })
# then # then
self.assertEquals(actual_revision, stub_rev) self.assertEquals(actual_revision, stub_rev)
mock_lookup.assert_called_once_with( mock_lookup.assert_called_once_with(
'some-sha1') 'some-sha1')
@patch('swh.web.common.service.lookup_revision_through') @patch('swh.web.common.service.lookup_revision_through')
@istest def test_lookup_directory_through_revision_ko_not_found(
def lookup_directory_through_revision_ko_not_found(
self, mock_lookup_rev): self, mock_lookup_rev):
# given # given
mock_lookup_rev.return_value = None mock_lookup_rev.return_value = None
# when # when
with self.assertRaises(NotFoundExc): with self.assertRaises(NotFoundExc):
service.lookup_directory_through_revision( service.lookup_directory_through_revision(
{'id': 'rev'}, 'some/path', 100) {'id': 'rev'}, 'some/path', 100)
mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 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_revision_through')
@patch('swh.web.common.service.lookup_directory_with_revision') @patch('swh.web.common.service.lookup_directory_with_revision')
@istest def test_lookup_directory_through_revision_ok_with_data(
def lookup_directory_through_revision_ok_with_data(
self, mock_lookup_dir, mock_lookup_rev): self, mock_lookup_dir, mock_lookup_rev):
# given # given
mock_lookup_rev.return_value = {'id': 'rev-id'} mock_lookup_rev.return_value = {'id': 'rev-id'}
mock_lookup_dir.return_value = {'type': 'dir', mock_lookup_dir.return_value = {'type': 'dir',
'content': []} 'content': []}
# when # when
rev_id, dir_result = service.lookup_directory_through_revision( rev_id, dir_result = service.lookup_directory_through_revision(
{'id': 'rev'}, 'some/path', 100) {'id': 'rev'}, 'some/path', 100)
# then # then
self.assertEquals(rev_id, 'rev-id') self.assertEquals(rev_id, 'rev-id')
self.assertEquals(dir_result, {'type': 'dir', self.assertEquals(dir_result, {'type': 'dir',
'content': []}) 'content': []})
mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 100) mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 100)
mock_lookup_dir.assert_called_once_with('rev-id', 'some/path', False) 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_revision_through')
@patch('swh.web.common.service.lookup_directory_with_revision') @patch('swh.web.common.service.lookup_directory_with_revision')
@istest def test_lookup_directory_through_revision_ok_with_content(
def lookup_directory_through_revision_ok_with_content(
self, mock_lookup_dir, mock_lookup_rev): self, mock_lookup_dir, mock_lookup_rev):
# given # given
mock_lookup_rev.return_value = {'id': 'rev-id'} mock_lookup_rev.return_value = {'id': 'rev-id'}
stub_result = {'type': 'file', stub_result = {'type': 'file',
'revision': 'rev-id', 'revision': 'rev-id',
'content': {'data': b'blah', 'content': {'data': b'blah',
'sha1': 'sha1'}} 'sha1': 'sha1'}}
mock_lookup_dir.return_value = stub_result mock_lookup_dir.return_value = stub_result
Show All 10 Lines
About · Developer information · Legal