diff --git a/debian/control b/debian/control
index c797c40e9..9d9f43dc8 100644
--- a/debian/control
+++ b/debian/control
@@ -1,48 +1,48 @@
Source: swh-web
Maintainer: Software Heritage developers <swh-devel@inria.fr>
Section: python
Priority: optional
Build-Depends: curl,
debhelper (>= 9),
dh-python (>= 2),
python3-all,
python3-bs4,
python3-django (>= 1.10.7~),
python3-djangorestframework (>= 3.4.0~),
python3-django-webpack-loader,
python3-django-js-reverse,
python3-docutils,
python3-htmlmin,
python3-hypothesis (>= 3.11.0~),
python3-magic (>= 0.3.0~),
python3-lxml,
python3-pytest,
python3-pytest-django,
python3-pygments,
python3-pypandoc,
python3-requests,
python3-setuptools,
python3-sphinx,
python3-sphinxcontrib.httpdomain,
python3-yaml,
python3-swh.core (>= 0.0.40~),
python3-swh.model (>= 0.0.25~),
- python3-swh.storage (>= 0.0.115~),
+ python3-swh.storage (>= 0.0.118~),
python3-swh.indexer (>= 0.0.120~),
python3-swh.vault (>= 0.0.20~),
python3-swh.scheduler (>= 0.0.31~),
python3-swh.loader.git (>= 0.0.47~)
Standards-Version: 3.9.6
Homepage: https://forge.softwareheritage.org/diffusion/DWUI/
Package: python3-swh.web
Architecture: all
Depends: python3-swh.core (>= 0.0.40~),
python3-swh.model (>= 0.0.25~),
- python3-swh.storage (>= 0.0.115~),
+ python3-swh.storage (>= 0.0.118~),
python3-swh.indexer.storage (>= 0.0.120~),
python3-swh.vault (>= 0.0.20~),
python3-swh.scheduler (>= 0.0.31~),
${misc:Depends},
${python3:Depends}
Description: Software Heritage Web Applications
diff --git a/requirements-swh.txt b/requirements-swh.txt
index f80614c3a..add6f9f4d 100644
--- a/requirements-swh.txt
+++ b/requirements-swh.txt
@@ -1,6 +1,6 @@
swh.core >= 0.0.40
swh.model >= 0.0.25
-swh.storage >= 0.0.115
+swh.storage >= 0.0.118
swh.vault >= 0.0.20
swh.indexer >= 0.0.120
swh.scheduler >= 0.0.31
\ No newline at end of file
diff --git a/swh/web/tests/common/test_service.py b/swh/web/tests/common/test_service.py
index 8a705df32..2a8e90fb4 100644
--- a/swh/web/tests/common/test_service.py
+++ b/swh/web/tests/common/test_service.py
@@ -1,1925 +1,864 @@
# Copyright (C) 2015-2018 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import copy
import datetime
+import itertools
+import pytest
+import random
-from unittest.mock import MagicMock, patch, call
+from collections import defaultdict
+from hypothesis import given
from swh.model.hashutil import hash_to_bytes, hash_to_hex
from swh.web.common import service
from swh.web.common.exc import BadInputExc, NotFoundExc
-from swh.web.tests.testcase import WebTestCase
+from swh.web.tests.strategies import (
+ content, contents, unknown_content, unknown_contents,
+ contents_with_ctags, origin, new_origin, visit_dates, directory,
+ release, revision, unknown_revision, revisions, unknown_revisions,
+ ancestor_revisions, non_ancestor_revisions, invalid_sha1, sha256,
+ revision_with_submodules, unknown_directory, empty_directory
+)
+from swh.web.tests.testcase import (
+ WebTestCase, ctags_json_missing, fossology_missing
+)
class ServiceTestCase(WebTestCase):
def setUp(self):
- self.BLAKE2S256_SAMPLE = ('685395c5dc57cada459364f0946d3dd45b'
- 'ad5fcbabc1048edb44380f1d31d0aa')
- self.BLAKE2S256_SAMPLE_BIN = hash_to_bytes(self.BLAKE2S256_SAMPLE)
+
self.SHA1_SAMPLE = '40e71b8614fcd89ccd17ca2b1d9e66c5b00a6d03'
self.SHA1_SAMPLE_BIN = hash_to_bytes(self.SHA1_SAMPLE)
- self.SHA256_SAMPLE = ('8abb0aa566452620ecce816eecdef4792d77a'
- '293ad8ea82a4d5ecb4d36f7e560')
- self.SHA256_SAMPLE_BIN = hash_to_bytes(self.SHA256_SAMPLE)
- self.SHA1GIT_SAMPLE = '25d1a2e8f32937b0f498a5ca87f823d8df013c01'
- self.SHA1GIT_SAMPLE_BIN = hash_to_bytes(self.SHA1GIT_SAMPLE)
+
self.DIRECTORY_ID = '7834ef7e7c357ce2af928115c6c6a42b7e2a44e6'
self.DIRECTORY_ID_BIN = hash_to_bytes(self.DIRECTORY_ID)
self.AUTHOR_ID_BIN = {
'name': b'author',
'email': b'author@company.org',
+ 'fullname': b'author <author@company.org>'
}
self.AUTHOR_ID = {
'name': 'author',
'email': 'author@company.org',
+ 'fullname': 'author <author@company.org>'
}
self.COMMITTER_ID_BIN = {
'name': b'committer',
'email': b'committer@corp.org',
+ 'fullname': b'committer <committer@corp.org>'
}
self.COMMITTER_ID = {
'name': 'committer',
'email': 'committer@corp.org',
+ 'fullname': 'committer <committer@corp.org>'
}
self.SAMPLE_DATE_RAW = {
- 'timestamp': datetime.datetime(
+ 'timestamp': int(datetime.datetime(
2000, 1, 17, 11, 23, 54,
tzinfo=datetime.timezone.utc,
- ).timestamp(),
+ ).timestamp()),
'offset': 0,
'negative_utc': False,
}
self.SAMPLE_DATE = '2000-01-17T11:23:54+00:00'
self.SAMPLE_MESSAGE_BIN = b'elegant fix for bug 31415957'
self.SAMPLE_MESSAGE = 'elegant fix for bug 31415957'
self.SAMPLE_REVISION = {
'id': self.SHA1_SAMPLE,
'directory': self.DIRECTORY_ID,
'author': self.AUTHOR_ID,
'committer': self.COMMITTER_ID,
'message': self.SAMPLE_MESSAGE,
'date': self.SAMPLE_DATE,
'committer_date': self.SAMPLE_DATE,
'synthetic': False,
'type': 'git',
'parents': [],
'metadata': {},
'merge': False
}
+
self.SAMPLE_REVISION_RAW = {
'id': self.SHA1_SAMPLE_BIN,
'directory': self.DIRECTORY_ID_BIN,
'author': self.AUTHOR_ID_BIN,
'committer': self.COMMITTER_ID_BIN,
'message': self.SAMPLE_MESSAGE_BIN,
'date': self.SAMPLE_DATE_RAW,
'committer_date': self.SAMPLE_DATE_RAW,
'synthetic': False,
'type': 'git',
'parents': [],
'metadata': [],
}
- self.SAMPLE_CONTENT = {
- 'checksums': {
- 'blake2s256': self.BLAKE2S256_SAMPLE,
- 'sha1': self.SHA1_SAMPLE,
- 'sha256': self.SHA256_SAMPLE,
- 'sha1_git': self.SHA1GIT_SAMPLE,
- },
- 'length': 190,
- 'status': 'absent'
- }
- self.SAMPLE_CONTENT_RAW = {
- 'blake2s256': self.BLAKE2S256_SAMPLE_BIN,
- 'sha1': self.SHA1_SAMPLE_BIN,
- 'sha256': self.SHA256_SAMPLE_BIN,
- 'sha1_git': self.SHA1GIT_SAMPLE_BIN,
- 'length': 190,
- 'status': 'hidden'
- }
+ @given(contents())
+ def test_lookup_multiple_hashes_all_present(self, contents):
+ input_data = []
+ expected_output = []
+ for cnt in contents:
+ input_data.append({'sha1': cnt['sha1']})
+ expected_output.append({'sha1': cnt['sha1'],
+ 'found': True})
- self.date_origin_visit1 = datetime.datetime(
- 2015, 1, 1, 22, 0, 0,
- tzinfo=datetime.timezone.utc)
+ self.assertEqual(service.lookup_multiple_hashes(input_data),
+ expected_output)
- self.origin_visit1 = {
- 'date': self.date_origin_visit1,
- 'origin': 1,
- 'visit': 1
- }
+ @given(contents(), unknown_contents())
+ def test_lookup_multiple_hashes_some_missing(self, contents,
+ unknown_contents):
+ input_contents = list(itertools.chain(contents, unknown_contents))
+ random.shuffle(input_contents)
- @patch('swh.web.common.service.storage')
- def test_lookup_multiple_hashes_ball_missing(self, mock_storage):
- # given
- mock_storage.content_missing_per_sha1 = MagicMock(return_value=[])
-
- # when
- actual_lookup = service.lookup_multiple_hashes(
- [{'filename': 'a',
- 'sha1': '456caf10e9535160d90e874b45aa426de762f19f'},
- {'filename': 'b',
- 'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}])
-
- # then
- self.assertEqual(actual_lookup, [
- {'filename': 'a',
- 'sha1': '456caf10e9535160d90e874b45aa426de762f19f',
- 'found': True},
- {'filename': 'b',
- 'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865',
- 'found': True}
- ])
-
- @patch('swh.web.common.service.storage')
- def test_lookup_multiple_hashes_some_missing(self, mock_storage):
- # given
- mock_storage.content_missing_per_sha1 = MagicMock(return_value=[
- hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')
- ])
-
- # when
- actual_lookup = service.lookup_multiple_hashes(
- [{'filename': 'a',
- 'sha1': '456caf10e9535160d90e874b45aa426de762f19f'},
- {'filename': 'b',
- 'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865'}])
-
- # then
- self.assertEqual(actual_lookup, [
- {'filename': 'a',
- 'sha1': '456caf10e9535160d90e874b45aa426de762f19f',
- 'found': False},
- {'filename': 'b',
- 'sha1': '745bab676c8f3cec8016e0c39ea61cf57e518865',
- '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': {
- 'checksums': {
- 'sha1': '456caf10e9535160d90e874b45aa426de762f19f'
- }
- },
- 'algo': 'sha1'
- }, actual_lookup)
-
- mock_storage.content_find.assert_called_with(
- {'sha1':
- hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')}
- )
+ input_data = []
+ expected_output = []
+ for cnt in input_contents:
+ input_data.append({'sha1': cnt['sha1']})
+ expected_output.append({'sha1': cnt['sha1'],
+ 'found': cnt in contents})
+
+ self.assertEqual(service.lookup_multiple_hashes(input_data),
+ expected_output)
+
+ @given(unknown_content())
+ def test_lookup_hash_does_not_exist(self, unknown_content):
+
+ actual_lookup = service.lookup_hash('sha1_git:%s' %
+ unknown_content['sha1_git'])
- @patch('swh.web.common.service.storage')
- def test_search_hash_does_not_exist(self, mock_storage):
- # given
- mock_storage.content_find = MagicMock(return_value=None)
+ self.assertEqual(actual_lookup, {'found': None,
+ 'algo': 'sha1_git'})
- # when
- actual_lookup = service.search_hash(
- 'sha1_git:123caf10e9535160d90e874b45aa426de762f19f')
+ @given(content())
+ def test_lookup_hash_exist(self, content):
+
+ actual_lookup = service.lookup_hash('sha1:%s' % content['sha1'])
+
+ content_metadata = self.content_get_metadata(content['sha1'])
+
+ self.assertEqual({'found': content_metadata,
+ 'algo': 'sha1'}, actual_lookup)
+
+ @given(unknown_content())
+ def test_search_hash_does_not_exist(self, content):
+
+ actual_lookup = service.search_hash('sha1_git:%s' %
+ content['sha1_git'])
- # then
self.assertEqual({'found': False}, actual_lookup)
- # check the function has been called with parameters
- mock_storage.content_find.assert_called_with(
- {'sha1_git':
- hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')})
-
- @patch('swh.web.common.service.storage')
- 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)
+ @given(content())
+ def test_search_hash_exist(self, content):
- mock_storage.content_find.assert_called_with(
- {'sha1':
- hash_to_bytes('456caf10e9535160d90e874b45aa426de762f19f')},
- )
+ actual_lookup = service.search_hash('sha1:%s' % content['sha1'])
- @patch('swh.web.common.service.idx_storage')
- def test_lookup_content_ctags(self, mock_idx_storage):
- # given
- mock_idx_storage.content_ctags_get = MagicMock(
- return_value=[{
- 'id': hash_to_bytes(
- '123caf10e9535160d90e874b45aa426de762f19f'),
- 'line': 100,
- 'name': 'hello',
- 'kind': 'function',
- 'tool_name': 'ctags',
- 'tool_version': 'some-version',
- }])
- expected_ctags = [{
- 'id': '123caf10e9535160d90e874b45aa426de762f19f',
- 'line': 100,
- 'name': 'hello',
- 'kind': 'function',
- 'tool_name': 'ctags',
- 'tool_version': 'some-version',
- }]
-
- # when
- actual_ctags = list(service.lookup_content_ctags(
- 'sha1:123caf10e9535160d90e874b45aa426de762f19f'))
-
- # then
- self.assertEqual(actual_ctags, expected_ctags)
+ self.assertEqual({'found': True}, actual_lookup)
- mock_idx_storage.content_ctags_get.assert_called_with(
- [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
+ @pytest.mark.skipif(ctags_json_missing,
+ reason="requires ctags with json output support")
+ @given(contents_with_ctags())
+ def test_lookup_content_ctags(self, contents_with_ctags):
- @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=[])
+ content_sha1 = random.choice(contents_with_ctags['sha1s'])
+ self.content_add_ctags(content_sha1)
+ actual_ctags = \
+ list(service.lookup_content_ctags('sha1:%s' % content_sha1))
- # when
- actual_ctags = list(service.lookup_content_ctags(
- 'sha1:123caf10e9535160d90e874b45aa426de762f19f'))
+ expected_data = list(self.content_get_ctags(content_sha1))
+ for ctag in expected_data:
+ ctag['id'] = content_sha1
- # then
- self.assertEqual(actual_ctags, [])
+ self.assertEqual(actual_ctags, expected_data)
- @patch('swh.web.common.service.idx_storage')
- def test_lookup_content_filetype(self, mock_idx_storage):
- # 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',
- }
+ @given(unknown_content())
+ def test_lookup_content_ctags_no_hash(self, unknown_content):
- # when
- actual_filetype = service.lookup_content_filetype(
- 'sha1:123caf10e9535160d90e874b45aa426de762f19f')
+ actual_ctags = \
+ list(service.lookup_content_ctags('sha1:%s' %
+ unknown_content['sha1']))
- # then
- self.assertEqual(actual_filetype, expected_filetype)
+ self.assertEqual(actual_ctags, [])
- mock_idx_storage.content_mimetype_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_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',
- }
+ @given(content())
+ def test_lookup_content_filetype(self, content):
- # when
- actual_filetype = service.lookup_content_filetype(
- 'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')
+ self.content_add_mimetype(content['sha1'])
+ actual_filetype = service.lookup_content_filetype(content['sha1'])
- # then
+ expected_filetype = self.content_get_mimetype(content['sha1'])
self.assertEqual(actual_filetype, expected_filetype)
- mock_storage.content_find(
- 'sha1_git', hash_to_bytes(
- '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',
- }
+ @given(content())
+ def test_lookup_content_language(self, content):
- # when
- actual_language = service.lookup_content_language(
- 'sha1:123caf10e9535160d90e874b45aa426de762f19f')
+ self.content_add_language(content['sha1'])
+ actual_language = service.lookup_content_language(content['sha1'])
- # then
+ expected_language = self.content_get_language(content['sha1'])
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',
- }
+ @given(contents_with_ctags())
+ def test_lookup_expression(self, contents_with_ctags):
- # when
- actual_language = service.lookup_content_language(
- 'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')
+ per_page = 10
+ expected_ctags = []
- # then
- self.assertEqual(actual_language, expected_language)
+ for content_sha1 in contents_with_ctags['sha1s']:
+ if len(expected_ctags) == per_page:
+ 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))
- mock_storage.content_find(
- 'sha1_git', hash_to_bytes(
- '456caf10e9535160d90e874b45aa426de762f19f')
- )
- mock_idx_storage.content_language_get.assert_called_with(
- [hash_to_bytes('123caf10e9535160d90e874b45aa426de762f19f')])
-
- @patch('swh.web.common.service.idx_storage')
- def test_lookup_expression(self, mock_idx_storage):
- # 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
- actual_ctags = list(service.lookup_expression(
- 'foobar', last_sha1='hash', per_page=10))
-
- # then
self.assertEqual(actual_ctags, expected_ctags)
- mock_idx_storage.content_ctags_search.assert_called_with(
- 'foobar', last_sha1='hash', limit=10)
+ def test_lookup_expression_no_result(self):
- @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 = []
- # when
- actual_ctags = list(service.lookup_expression(
- 'barfoo', last_sha1='hash', per_page=10))
-
- # then
+ actual_ctags = \
+ list(service.lookup_expression('barfoo', last_sha1=None,
+ per_page=10))
self.assertEqual(actual_ctags, expected_ctags)
- mock_idx_storage.content_ctags_search.assert_called_with(
- 'barfoo', last_sha1='hash', limit=10)
-
- @patch('swh.web.common.service.idx_storage')
- 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': {}
- }]
- }
+ @pytest.mark.skipif(fossology_missing,
+ reason="requires fossology-nomossa installed")
+ @given(content())
+ def test_lookup_content_license(self, content):
- # when
- actual_license = service.lookup_content_license(
- 'sha1:123caf10e9535160d90e874b45aa426de762f19f')
+ self.content_add_license(content['sha1'])
+ actual_license = service.lookup_content_license(content['sha1'])
- # then
+ expected_license = self.content_get_license(content['sha1'])
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': {}
- }]
-
- }]
- )
- expected_license = {
- 'id': '123caf10e9535160d90e874b45aa426de762f19f',
- 'facts': [{
- 'licenses': ['BSD-2-Clause'],
- 'tool': {}
- }]
- }
-
- # when
- actual_license = service.lookup_content_license(
- 'sha1_git:456caf10e9535160d90e874b45aa426de762f19f')
+ def test_stat_counters(self):
+ actual_stats = service.stat_counters()
+ self.assertEqual(actual_stats, self.storage.stat_counters())
- # then
- self.assertEqual(actual_license, expected_license)
+ @given(new_origin(), visit_dates())
+ def test_lookup_origin_visits(self, new_origin, visit_dates):
- mock_storage.content_find(
- '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)
+ origin_id = self.storage.origin_add_one(new_origin)
+ for ts in visit_dates:
+ self.storage.origin_visit_add(origin_id, ts)
- # when
- actual_stats = service.stat_counters()
+ actual_origin_visits = list(service.lookup_origin_visits(origin_id))
- # then
- expected_stats = input_stats
- self.assertEqual(actual_stats, expected_stats)
-
- mock_storage.stat_counters.assert_called_with()
-
- @patch('swh.web.common.service._lookup_origin_visits')
- 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
- expected_origin_visits = [{
- 'date': self.origin_visit1['date'].isoformat(),
- 'origin': self.origin_visit1['origin'],
- 'visit': self.origin_visit1['visit']
- }, {
- 'date': date_origin_visit2.isoformat(),
- 'origin': 1,
- 'visit': 2,
- 'target': '65a55bbdf3629f916219feb3dcc7393ded1bc8db',
- 'branch': 'master',
- 'target_type': 'release',
- 'metadata': {},
- }, {
- 'date': date_origin_visit3.isoformat(),
- 'origin': 1,
- 'visit': 3
- }]
-
- actual_origin_visits = service.lookup_origin_visits(6)
-
- # then
- self.assertEqual(list(actual_origin_visits), expected_origin_visits)
-
- mock_lookup_visits.assert_called_once_with(
- 6, last_visit=None, limit=10)
-
- @patch('swh.web.common.service.storage')
- def test_lookup_origin_visit(self, mock_storage):
- # given
- stub_result = self.origin_visit1
- mock_storage.origin_visit_get_by.return_value = stub_result
-
- expected_origin_visit = {
- 'date': self.origin_visit1['date'].isoformat(),
- 'origin': self.origin_visit1['origin'],
- 'visit': self.origin_visit1['visit']
- }
+ expected_visits = list(self.storage.origin_visit_get(origin_id))
+ for visit in expected_visits:
+ visit['date'] = visit['date'].isoformat()
+ visit['metadata'] = {}
- # when
- actual_origin_visit = service.lookup_origin_visit(1, 1)
+ self.assertEqual(actual_origin_visits, expected_visits)
- # then
- self.assertEqual(actual_origin_visit, expected_origin_visit)
+ @given(new_origin(), visit_dates())
+ def test_lookup_origin_visit(self, new_origin, visit_dates):
+ origin_id = self.storage.origin_add_one(new_origin)
+ visits = []
+ for ts in visit_dates:
+ visits.append(self.storage.origin_visit_add(origin_id, ts))
- mock_storage.origin_visit_get_by.assert_called_once_with(1, 1)
+ visit = random.choice(visits)['visit']
+ actual_origin_visit = service.lookup_origin_visit(origin_id, visit)
- @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'})
+ expected_visit = dict(self.storage.origin_visit_get_by(origin_id,
+ visit))
+ expected_visit['date'] = expected_visit['date'].isoformat()
+ expected_visit['metadata'] = {}
- # when
- actual_origin = service.lookup_origin({'id': 'origin-id'})
+ self.assertEqual(actual_origin_visit, expected_visit)
- # then
- self.assertEqual(actual_origin, {'id': 'origin-id',
- 'url': 'ftp://some/url/to/origin',
- 'type': 'ftp'})
+ @given(new_origin())
+ def test_lookup_origin(self, new_origin):
+ origin_id = self.storage.origin_add_one(new_origin)
- 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')
- def test_lookup_release_ko_id_checksum_not_a_sha1(self, mock_storage):
- # given
- mock_storage.release_get = MagicMock()
+ actual_origin = service.lookup_origin({'type': new_origin['type'],
+ 'url': new_origin['url']})
+ expected_origin = self.storage.origin_get({'type': new_origin['type'],
+ 'url': new_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:
- # when
- service.lookup_release('not-a-sha1')
+ service.lookup_release(invalid_sha1)
self.assertIn('invalid checksum', cm.exception.args[0].lower())
- mock_storage.release_get.called = False
-
- @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
+ @given(sha256())
+ def test_lookup_release_ko_id_checksum_too_long(self, sha256):
with self.assertRaises(BadInputExc) as cm:
- service.lookup_release(
- '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4daf5'
- '1aea892abe')
+ service.lookup_release(sha256)
self.assertEqual('Only sha1_git is supported.', cm.exception.args[0])
- mock_storage.release_get.called = False
-
- @patch('swh.web.common.service.storage')
- def test_lookup_directory_with_path_not_found(self, mock_storage):
- # given
- mock_storage.lookup_directory_with_path = MagicMock(return_value=None)
-
- sha1_git = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
-
- # when
- actual_directory = mock_storage.lookup_directory_with_path(
- sha1_git, 'some/path/here')
-
- self.assertIsNone(actual_directory)
-
- @patch('swh.web.common.service.storage')
- 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
- sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f62d4' \
- 'daf51aea892abe'
- sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
-
- # when
+ @given(directory())
+ def test_lookup_directory_with_path_not_found(self, directory):
+ path = 'some/invalid/path/here'
+ with self.assertRaises(NotFoundExc) as cm:
+ service.lookup_directory_with_path(directory, path)
+ self.assertEqual('Directory entry with path %s from %s '
+ 'not found' % (path, directory),
+ cm.exception.args[0])
+
+ @given(directory())
+ def test_lookup_directory_with_path_found(self, directory):
+ directory_content = self.directory_ls(directory)
+ directory_entry = random.choice(directory_content)
+ 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
+
with self.assertRaises(BadInputExc) as cm:
service.lookup_revision_with_context(sha1_git_root, sha1_git)
- self.assertIn('Only sha1_git is supported', cm.exception.args[0])
+ self.assertIn('Invalid checksum query string', cm.exception.args[0])
- def test_lookup_revision_with_context_ko_not_a_sha1_2(self):
- # given
- sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
- sha1_git = '13c1d34d138ec13b5ebad226dc2528dc7506c956e4646f6' \
- '2d4daf51aea892abe'
+ sha1_git = sha256
- # when
with self.assertRaises(BadInputExc) as cm:
service.lookup_revision_with_context(sha1_git_root, sha1_git)
self.assertIn('Only sha1_git is supported', cm.exception.args[0])
- @patch('swh.web.common.service.storage')
+ @given(revision(), unknown_revision())
def test_lookup_revision_with_context_ko_sha1_git_does_not_exist(
- self,
- mock_storage):
- # given
- sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
- sha1_git = '777777bdf3629f916219feb3dcc7393ded1bc8db'
-
- sha1_git_bin = hash_to_bytes(sha1_git)
+ self, revision, unknown_revision):
+ sha1_git_root = revision
+ sha1_git = unknown_revision
- mock_storage.revision_get.return_value = None
-
- # when
with self.assertRaises(NotFoundExc) as cm:
service.lookup_revision_with_context(sha1_git_root, sha1_git)
- self.assertIn('Revision 777777bdf3629f916219feb3dcc7393ded1bc8db'
- ' not found', cm.exception.args[0])
-
- mock_storage.revision_get.assert_called_once_with(
- [sha1_git_bin])
+ self.assertIn('Revision %s not found' % sha1_git, cm.exception.args[0])
- @patch('swh.web.common.service.storage')
+ @given(revision(), unknown_revision())
def test_lookup_revision_with_context_ko_root_sha1_git_does_not_exist(
- self,
- mock_storage):
- # given
- sha1_git_root = '65a55bbdf3629f916219feb3dcc7393ded1bc8db'
- sha1_git = '777777bdf3629f916219feb3dcc7393ded1bc8db'
-
- sha1_git_root_bin = hash_to_bytes(sha1_git_root)
- sha1_git_bin = hash_to_bytes(sha1_git)
+ self, revision, unknown_revision):
+ sha1_git_root = unknown_revision
+ sha1_git = revision
- mock_storage.revision_get.side_effect = ['foo', None]
-
- # when
with self.assertRaises(NotFoundExc) as cm:
service.lookup_revision_with_context(sha1_git_root, sha1_git)
- self.assertIn('Revision root 65a55bbdf3629f916219feb3dcc7393ded1bc8db'
- ' not found', cm.exception.args[0])
-
- mock_storage.revision_get.assert_has_calls([call([sha1_git_bin]),
- call([sha1_git_root_bin])])
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- 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',
- }
+ self.assertIn('Revision root %s not found' % sha1_git_root,
+ cm.exception.args[0])
- stub_revisions = [
- sha1_git_root_dict,
- {
- 'id': b'999',
- 'parents': [b'777', b'883', b'888'],
- },
- {
- 'id': b'777',
- 'parents': [b'883'],
- },
- sha1_git_dict,
- {
- 'id': b'888',
- 'parents': [b'889'],
- },
- {
- 'id': b'889',
- 'parents': [],
- },
- ]
-
- # inputs ok
- mock_query.parse_hash_with_algorithms_or_throws.return_value = (
- 'sha1', sha1_git_bin)
-
- # lookup only on sha1
- mock_storage.revision_get.return_value = [sha1_git_dict]
-
- mock_storage.revision_log.return_value = stub_revisions
-
- # when
- actual_revision = service.lookup_revision_with_context(
- {'id': sha1_git_root_bin},
- sha1_git)
-
- # then
- self.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
- })
+ @given(ancestor_revisions())
+ def test_lookup_revision_with_context(self, ancestor_revisions):
+ 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)
+
+ children = []
+ for rev in self.revision_log(root_sha1_git):
+ for p_rev in rev['parents']:
+ p_rev_hex = hash_to_hex(p_rev)
+ if p_rev_hex == sha1_git:
+ children.append(rev['id'])
+
+ expected_revision = self.revision_get(sha1_git)
+ expected_revision['children'] = children
+ self.assertEqual(actual_revision, expected_revision)
+
+ @given(non_ancestor_revisions())
+ def test_lookup_revision_with_context_ko(self, non_ancestor_revisions):
+ sha1_git = non_ancestor_revisions['sha1_git']
+ root_sha1_git = non_ancestor_revisions['sha1_git_root']
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with( # noqa
- sha1_git, ['sha1'], 'Only sha1_git is supported.')
+ with self.assertRaises(NotFoundExc) as cm:
+ service.lookup_revision_with_context(root_sha1_git, sha1_git)
+ self.assertIn('Revision %s is not an ancestor of %s' %
+ (sha1_git, root_sha1_git), cm.exception.args[0])
- mock_storage.revision_get.assert_called_once_with([sha1_git_bin])
+ @given(unknown_revision())
+ def test_lookup_directory_with_revision_not_found(self, unknown_revision):
- mock_storage.revision_log.assert_called_with(
- [sha1_git_root_bin], 100)
+ with self.assertRaises(NotFoundExc) as cm:
+ service.lookup_directory_with_revision(unknown_revision)
+ self.assertIn('Revision %s not found' % unknown_revision,
+ cm.exception.args[0])
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- def test_lookup_directory_with_revision_not_found(self,
- mock_query,
- mock_storage):
- # given
- mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
- b'123')
- mock_storage.revision_get.return_value = None
+ @given(revision())
+ def test_lookup_directory_with_revision_ko_path_to_nowhere(self, revision):
- # when
- with self.assertRaises(NotFoundExc) as cm:
- service.lookup_directory_with_revision('123')
- self.assertIn('Revision 123 not found', cm.exception.args[0])
-
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
- ('123', ['sha1'], 'Only sha1_git is supported.')
- mock_storage.revision_get.assert_called_once_with([b'123'])
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- def test_lookup_directory_with_revision_ko_revision_with_path_to_nowhere(
- self,
- mock_query,
- mock_storage):
- # given
- mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
- b'123')
-
- dir_id = b'dir-id-as-sha1'
- mock_storage.revision_get.return_value = [{
- 'directory': dir_id,
- }]
-
- mock_storage.directory_entry_get_by_path.return_value = None
-
- # when
+ invalid_path = 'path/to/something/unknown'
with self.assertRaises(NotFoundExc) as cm:
- service.lookup_directory_with_revision(
- '123',
- 'path/to/something/unknown')
+ service.lookup_directory_with_revision(revision, invalid_path)
exception_text = cm.exception.args[0].lower()
self.assertIn('directory or file', exception_text)
- self.assertIn('path/to/something/unknown', exception_text)
- self.assertIn('revision 123', exception_text)
+ self.assertIn(invalid_path, exception_text)
+ self.assertIn('revision %s' % revision, exception_text)
self.assertIn('not found', exception_text)
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
- ('123', ['sha1'], 'Only sha1_git is supported.')
- mock_storage.revision_get.assert_called_once_with([b'123'])
- mock_storage.directory_entry_get_by_path.assert_called_once_with(
- b'dir-id-as-sha1', [b'path', b'to', b'something', b'unknown'])
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
+ @given(revision_with_submodules())
def test_lookup_directory_with_revision_ko_type_not_implemented(
- self,
- mock_query,
- mock_storage):
-
- # given
- mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
- b'123')
-
- dir_id = b'dir-id-as-sha1'
- mock_storage.revision_get.return_value = [{
- 'directory': dir_id,
- }]
-
- mock_storage.directory_entry_get_by_path.return_value = {
- 'type': 'rev',
- 'name': b'some/path/to/rev',
- 'target': b'456'
- }
-
- stub_content = {
- 'id': b'12',
- 'type': 'file'
- }
+ self, revision_with_submodules):
- mock_storage.content_get.return_value = stub_content
-
- # when
with self.assertRaises(NotImplementedError) as cm:
service.lookup_directory_with_revision(
- '123',
- 'some/path/to/rev')
+ revision_with_submodules['rev_sha1_git'],
+ revision_with_submodules['rev_dir_rev_path'])
self.assertIn("Entity of type rev not implemented.",
cm.exception.args[0])
- # then
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
- ('123', ['sha1'], 'Only sha1_git is supported.')
- mock_storage.revision_get.assert_called_once_with([b'123'])
- mock_storage.directory_entry_get_by_path.assert_called_once_with(
- b'dir-id-as-sha1', [b'some', b'path', b'to', b'rev'])
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- 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'
- mock_storage.revision_get.return_value = [{
- 'directory': dir_id,
- }]
-
- stub_dir_entries = [{
- 'id': b'123',
- 'type': 'dir'
- }, {
- 'id': b'456',
- 'type': 'file'
- }]
-
- mock_storage.directory_ls.return_value = stub_dir_entries
-
- # when
- actual_directory_entries = service.lookup_directory_with_revision(
- '123')
+ @given(revision())
+ def test_lookup_directory_with_revision_without_path(self, revision):
- self.assertEqual(actual_directory_entries['type'], 'dir')
- self.assertEqual(list(actual_directory_entries['content']),
- stub_dir_entries)
-
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
- ('123', ['sha1'], 'Only sha1_git is supported.')
- mock_storage.revision_get.assert_called_once_with([b'123'])
- mock_storage.directory_ls.assert_called_once_with(dir_id)
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- def test_lookup_directory_with_revision_with_path_to_dir(self,
- mock_query,
- mock_storage):
- # given
- mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
- b'123')
-
- dir_id = b'dir-id-as-sha1'
- mock_storage.revision_get.return_value = [{
- 'directory': dir_id,
- }]
-
- stub_dir_entries = [{
- 'id': b'12',
- 'type': 'dir'
- }, {
- 'id': b'34',
- 'type': 'file'
- }]
-
- mock_storage.directory_entry_get_by_path.return_value = {
- 'type': 'dir',
- 'name': b'some/path',
- 'target': b'456'
- }
- mock_storage.directory_ls.return_value = stub_dir_entries
+ actual_directory_entries = \
+ service.lookup_directory_with_revision(revision)
- # when
- actual_directory_entries = service.lookup_directory_with_revision(
- '123',
- 'some/path')
+ revision_data = self.revision_get(revision)
+ expected_directory_entries = \
+ self.directory_ls(revision_data['directory'])
self.assertEqual(actual_directory_entries['type'], 'dir')
- self.assertEqual(actual_directory_entries['revision'], '123')
- self.assertEqual(actual_directory_entries['path'], 'some/path')
- self.assertEqual(list(actual_directory_entries['content']),
- stub_dir_entries)
-
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
- ('123', ['sha1'], 'Only sha1_git is supported.')
- mock_storage.revision_get.assert_called_once_with([b'123'])
- mock_storage.directory_entry_get_by_path.assert_called_once_with(
- dir_id,
- [b'some', b'path'])
- mock_storage.directory_ls.assert_called_once_with(b'456')
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- def test_lookup_directory_with_revision_with_path_to_file_wo_data(
- self,
- mock_query,
- mock_storage):
-
- # given
- mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
- b'123')
-
- dir_id = b'dir-id-as-sha1'
- mock_storage.revision_get.return_value = [{
- 'directory': dir_id,
- }]
-
- mock_storage.directory_entry_get_by_path.return_value = {
- 'type': 'file',
- 'name': b'some/path/to/file',
- 'target': b'789'
- }
-
- stub_content = {
- 'status': 'visible',
- }
-
- mock_storage.content_find.return_value = stub_content
-
- # when
- actual_content = service.lookup_directory_with_revision(
- '123',
- 'some/path/to/file')
-
- # then
- self.assertEqual(actual_content, {'type': 'file',
- 'revision': '123',
- 'path': 'some/path/to/file',
- 'content': stub_content})
-
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
- ('123', ['sha1'], 'Only sha1_git is supported.')
- mock_storage.revision_get.assert_called_once_with([b'123'])
- mock_storage.directory_entry_get_by_path.assert_called_once_with(
- b'dir-id-as-sha1', [b'some', b'path', b'to', b'file'])
- mock_storage.content_find.assert_called_once_with({'sha1_git': b'789'})
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- def test_lookup_directory_with_revision_with_path_to_file_w_data(
- self,
- mock_query,
- mock_storage):
-
- # given
- mock_query.parse_hash_with_algorithms_or_throws.return_value = ('sha1',
- b'123')
-
- dir_id = b'dir-id-as-sha1'
- mock_storage.revision_get.return_value = [{
- 'directory': dir_id,
- }]
-
- mock_storage.directory_entry_get_by_path.return_value = {
- 'type': 'file',
- 'name': b'some/path/to/file',
- 'target': b'789'
- }
-
- stub_content = {
- 'status': 'visible',
- 'sha1': b'content-sha1'
- }
-
- mock_storage.content_find.return_value = stub_content
- mock_storage.content_get.return_value = [{
- 'sha1': b'content-sha1',
- 'data': b'some raw data'
- }]
-
- expected_content = {
- 'status': 'visible',
- 'checksums': {
- 'sha1': hash_to_hex(b'content-sha1'),
- },
- 'data': b'some raw data'
- }
+ self.assertEqual(actual_directory_entries['content'],
+ expected_directory_entries)
+
+ @given(revision())
+ def test_lookup_directory_with_revision_with_path(self, revision):
+
+ revision_data = self.revision_get(revision)
+ dir_entries = [e for e in self.directory_ls(revision_data['directory'])
+ if e['type'] in ('file', 'dir')]
+ expected_dir_entry = random.choice(dir_entries)
+
+ actual_dir_entry = \
+ service.lookup_directory_with_revision(revision,
+ expected_dir_entry['name'])
+
+ self.assertEqual(actual_dir_entry['type'], expected_dir_entry['type'])
+ self.assertEqual(actual_dir_entry['revision'], revision)
+ self.assertEqual(actual_dir_entry['path'], expected_dir_entry['name'])
+ if actual_dir_entry['type'] == 'file':
+ del actual_dir_entry['content']['checksums']['blake2s256']
+ for key in ('checksums', 'status', 'length'):
+ self.assertEqual(actual_dir_entry['content'][key],
+ expected_dir_entry[key])
+ else:
+ sub_dir_entries = self.directory_ls(expected_dir_entry['target'])
+ self.assertEqual(actual_dir_entry['content'], sub_dir_entries)
+
+ @given(revision())
+ def test_lookup_directory_with_revision_with_path_to_file_and_data(
+ self, revision):
+
+ revision_data = self.revision_get(revision)
+ dir_entries = [e for e in self.directory_ls(revision_data['directory'])
+ if e['type'] == 'file']
+ expected_dir_entry = random.choice(dir_entries)
+ expected_data = \
+ self.content_get(expected_dir_entry['checksums']['sha1'])
+
+ actual_dir_entry = \
+ service.lookup_directory_with_revision(revision,
+ expected_dir_entry['name'],
+ with_data=True)
+
+ self.assertEqual(actual_dir_entry['type'], expected_dir_entry['type'])
+ self.assertEqual(actual_dir_entry['revision'], revision)
+ self.assertEqual(actual_dir_entry['path'], expected_dir_entry['name'])
+ del actual_dir_entry['content']['checksums']['blake2s256']
+ for key in ('checksums', 'status', 'length'):
+ self.assertEqual(actual_dir_entry['content'][key],
+ expected_dir_entry[key])
+ self.assertEqual(actual_dir_entry['content']['data'],
+ expected_data['data'])
+
+ @given(revision())
+ def test_lookup_revision(self, revision):
+ actual_revision = service.lookup_revision(revision)
+ self.assertEqual(actual_revision, self.revision_get(revision))
+
+ @given(unknown_revision())
+ def test_lookup_revision_invalid_msg(self, new_revision_id):
+
+ new_revision = copy.deepcopy(self.SAMPLE_REVISION_RAW)
+ new_revision['id'] = hash_to_bytes(new_revision_id)
+ new_revision['message'] = b'elegant fix for bug \xff'
+ self.storage.revision_add([new_revision])
+
+ revision = service.lookup_revision(new_revision_id)
+ self.assertEqual(revision['message'], None)
+ self.assertEqual(revision['message_decoding_failed'], True)
+
+ @given(unknown_revision())
+ def test_lookup_revision_msg_ok(self, new_revision_id):
+
+ new_revision = copy.deepcopy(self.SAMPLE_REVISION_RAW)
+ new_revision['id'] = hash_to_bytes(new_revision_id)
+ self.storage.revision_add([new_revision])
+
+ revision_message = service.lookup_revision_message(new_revision_id)
+
+ self.assertEqual(revision_message,
+ {'message': self.SAMPLE_MESSAGE_BIN})
+
+ @given(unknown_revision())
+ def test_lookup_revision_msg_absent(self, new_revision_id):
+
+ new_revision = copy.deepcopy(self.SAMPLE_REVISION_RAW)
+ new_revision['id'] = hash_to_bytes(new_revision_id)
+ del new_revision['message']
+ self.storage.revision_add([new_revision])
- # when
- actual_content = service.lookup_directory_with_revision(
- '123',
- 'some/path/to/file',
- with_data=True)
-
- # then
- self.assertEqual(actual_content, {'type': 'file',
- 'revision': '123',
- 'path': 'some/path/to/file',
- 'content': expected_content})
-
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_once_with
- ('123', ['sha1'], 'Only sha1_git is supported.')
- mock_storage.revision_get.assert_called_once_with([b'123'])
- mock_storage.directory_entry_get_by_path.assert_called_once_with(
- b'dir-id-as-sha1', [b'some', b'path', b'to', b'file'])
- mock_storage.content_find.assert_called_once_with({'sha1_git': b'789'})
- mock_storage.content_get.assert_called_once_with([b'content-sha1'])
-
- @patch('swh.web.common.service.storage')
- def test_lookup_revision(self, mock_storage):
- # given
- mock_storage.revision_get = MagicMock(
- return_value=[self.SAMPLE_REVISION_RAW])
-
- # when
- actual_revision = service.lookup_revision(
- self.SHA1_SAMPLE)
-
- # then
- self.assertEqual(actual_revision, self.SAMPLE_REVISION)
-
- mock_storage.revision_get.assert_called_with(
- [self.SHA1_SAMPLE_BIN])
-
- @patch('swh.web.common.service.storage')
- def test_lookup_revision_invalid_msg(self, mock_storage):
- # given
- stub_rev = self.SAMPLE_REVISION_RAW
- stub_rev['message'] = b'elegant fix for bug \xff'
-
- expected_revision = self.SAMPLE_REVISION
- expected_revision['message'] = None
- expected_revision['message_decoding_failed'] = True
- mock_storage.revision_get = MagicMock(return_value=[stub_rev])
-
- # when
- actual_revision = service.lookup_revision(
- self.SHA1_SAMPLE)
-
- # then
- self.assertEqual(actual_revision, expected_revision)
-
- mock_storage.revision_get.assert_called_with(
- [self.SHA1_SAMPLE_BIN])
-
- @patch('swh.web.common.service.storage')
- def test_lookup_revision_msg_ok(self, mock_storage):
- # given
- mock_storage.revision_get.return_value = [self.SAMPLE_REVISION_RAW]
-
- # when
- rv = service.lookup_revision_message(
- self.SHA1_SAMPLE)
-
- # then
- self.assertEqual(rv, {'message': self.SAMPLE_MESSAGE_BIN})
- mock_storage.revision_get.assert_called_with(
- [self.SHA1_SAMPLE_BIN])
-
- @patch('swh.web.common.service.storage')
- def test_lookup_revision_msg_absent(self, mock_storage):
- # given
- stub_revision = self.SAMPLE_REVISION_RAW
- del stub_revision['message']
- mock_storage.revision_get.return_value = stub_revision
-
- # when
with self.assertRaises(NotFoundExc) as cm:
- service.lookup_revision_message(
- self.SHA1_SAMPLE)
+ service.lookup_revision_message(new_revision_id)
- # then
- mock_storage.revision_get.assert_called_with(
- [self.SHA1_SAMPLE_BIN])
self.assertEqual(
cm.exception.args[0],
- 'No message for revision with sha1_git %s.' % self.SHA1_SAMPLE,
+ 'No message for revision with sha1_git %s.' % new_revision_id
)
- @patch('swh.web.common.service.storage')
- def test_lookup_revision_msg_norev(self, mock_storage):
- # given
- mock_storage.revision_get.return_value = None
+ @given(unknown_revision())
+ def test_lookup_revision_msg_no_rev(self, unknown_revision):
- # when
with self.assertRaises(NotFoundExc) as cm:
- service.lookup_revision_message(
- self.SHA1_SAMPLE)
+ service.lookup_revision_message(unknown_revision)
- # then
- mock_storage.revision_get.assert_called_with(
- [self.SHA1_SAMPLE_BIN])
self.assertEqual(
cm.exception.args[0],
- 'Revision with sha1_git %s not found.' % self.SHA1_SAMPLE,
+ 'Revision with sha1_git %s not found.' % unknown_revision
)
- @patch('swh.web.common.service.storage')
- def test_lookup_revision_multiple(self, mock_storage):
- # given
- sha1 = self.SHA1_SAMPLE
- sha1_other = 'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'
-
- stub_revisions = [
- self.SAMPLE_REVISION_RAW,
- {
- 'id': hash_to_bytes(sha1_other),
- 'directory': 'abcdbe353ed3480476f032475e7c233eff7371d5',
- 'author': {
- 'name': b'name',
- 'email': b'name@surname.org',
- },
- 'committer': {
- 'name': b'name',
- 'email': b'name@surname.org',
- },
- 'message': b'ugly fix for bug 42',
- 'date': {
- 'timestamp': datetime.datetime(
- 2000, 1, 12, 5, 23, 54,
- tzinfo=datetime.timezone.utc).timestamp(),
- 'offset': 0,
- 'negative_utc': False
- },
- 'date_offset': 0,
- 'committer_date': {
- 'timestamp': datetime.datetime(
- 2000, 1, 12, 5, 23, 54,
- tzinfo=datetime.timezone.utc).timestamp(),
- 'offset': 0,
- 'negative_utc': False
- },
- 'committer_date_offset': 0,
- 'synthetic': False,
- 'type': 'git',
- 'parents': [],
- 'metadata': [],
- }
- ]
-
- mock_storage.revision_get.return_value = stub_revisions
-
- # when
- actual_revisions = service.lookup_revision_multiple(
- [sha1, sha1_other])
-
- # then
- self.assertEqual(list(actual_revisions), [
- self.SAMPLE_REVISION,
- {
- 'id': sha1_other,
- 'directory': 'abcdbe353ed3480476f032475e7c233eff7371d5',
- 'author': {
- 'name': 'name',
- 'email': 'name@surname.org',
- },
- 'committer': {
- 'name': 'name',
- 'email': 'name@surname.org',
- },
- 'message': 'ugly fix for bug 42',
- 'date': '2000-01-12T05:23:54+00:00',
- 'date_offset': 0,
- 'committer_date': '2000-01-12T05:23:54+00:00',
- 'committer_date_offset': 0,
- 'synthetic': False,
- 'type': 'git',
- 'parents': [],
- 'metadata': {},
- 'merge': False
- }
- ])
+ @given(revisions())
+ def test_lookup_revision_multiple(self, revisions):
- self.assertEqual(
- list(mock_storage.revision_get.call_args[0][0]),
- [hash_to_bytes(sha1),
- hash_to_bytes(sha1_other)])
+ actual_revisions = list(service.lookup_revision_multiple(revisions))
- @patch('swh.web.common.service.storage')
- def test_lookup_revision_multiple_none_found(self, mock_storage):
- # given
- sha1_bin = self.SHA1_SAMPLE
- sha1_other = 'adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'
+ expected_revisions = []
+ for rev in revisions:
+ expected_revisions.append(self.revision_get(rev))
- mock_storage.revision_get.return_value = []
+ self.assertEqual(actual_revisions, expected_revisions)
- # then
- actual_revisions = service.lookup_revision_multiple(
- [sha1_bin, sha1_other])
+ @given(unknown_revisions())
+ def test_lookup_revision_multiple_none_found(self, unknown_revisions):
- self.assertEqual(list(actual_revisions), [])
+ actual_revisions = \
+ list(service.lookup_revision_multiple(unknown_revisions))
+
+ self.assertEqual(actual_revisions, [None] * len(unknown_revisions))
+
+ @given(revision())
+ def test_lookup_revision_log(self, revision):
+
+ actual_revision_log = \
+ list(service.lookup_revision_log(revision, limit=25))
+ expected_revision_log = self.revision_log(revision, limit=25)
+
+ self.assertEqual(actual_revision_log, expected_revision_log)
+
+ def _get_origin_branches(self, origin):
+ origin_visit = self.origin_visit_get(origin['id'])[0]
+ snapshot = self.snapshot_get(origin_visit['snapshot'])
+ branches = {k: v for (k, v) in snapshot['branches'].items()
+ if v['target_type'] == 'revision'}
+ return branches
+
+ @given(origin())
+ def test_lookup_revision_log_by(self, origin):
+
+ branches = self._get_origin_branches(origin)
+ branch_name = random.choice(list(branches.keys()))
+
+ actual_log = \
+ list(service.lookup_revision_log_by(origin['id'], branch_name,
+ None, limit=25))
+
+ expected_log = \
+ self.revision_log(branches[branch_name]['target'], limit=25)
+
+ self.assertEqual(actual_log, expected_log)
+
+ @given(origin())
+ def test_lookup_revision_log_by_notfound(self, origin):
- self.assertEqual(
- list(mock_storage.revision_get.call_args[0][0]),
- [hash_to_bytes(self.SHA1_SAMPLE),
- hash_to_bytes(sha1_other)])
-
- @patch('swh.web.common.service.storage')
- def test_lookup_revision_log(self, mock_storage):
- # given
- stub_revision_log = [self.SAMPLE_REVISION_RAW]
- mock_storage.revision_log = MagicMock(return_value=stub_revision_log)
-
- # when
- actual_revision = service.lookup_revision_log(
- 'abcdbe353ed3480476f032475e7c233eff7371d5',
- limit=25)
-
- # then
- self.assertEqual(list(actual_revision), [self.SAMPLE_REVISION])
-
- mock_storage.revision_log.assert_called_with(
- [hash_to_bytes('abcdbe353ed3480476f032475e7c233eff7371d5')], 25)
-
- @patch('swh.web.common.service.lookup_revision_log')
- @patch('swh.web.common.service.lookup_snapshot')
- @patch('swh.web.common.service.get_origin_visit')
- def test_lookup_revision_log_by(self, mock_get_origin_visit,
- mock_lookup_snapshot,
- mock_lookup_revision_log):
- # given
- mock_get_origin_visit.return_value = {'snapshot': self.SHA1_SAMPLE}
- mock_lookup_snapshot.return_value = \
- {
- 'branches': {
- 'refs/heads/master': {
- 'target_type': 'revision',
- 'target': self.SAMPLE_REVISION['id']
- }
- }
- }
-
- mock_lookup_revision_log.return_value = [self.SAMPLE_REVISION]
-
- # when
- actual_log = service.lookup_revision_log_by(
- 1, 'refs/heads/master', None, limit=100)
- # then
- self.assertEqual(list(actual_log), [self.SAMPLE_REVISION])
-
- @patch('swh.web.common.service.lookup_snapshot')
- @patch('swh.web.common.service.get_origin_visit')
- def test_lookup_revision_log_by_notfound(self, mock_get_origin_visit,
- mock_lookup_snapshot):
- # given
- mock_get_origin_visit.return_value = {'snapshot': self.SHA1_SAMPLE}
- mock_lookup_snapshot.return_value = {'branches': {}}
-
- # when
with self.assertRaises(NotFoundExc):
service.lookup_revision_log_by(
- 1, 'refs/heads/master', None, limit=100)
+ origin['id'], 'unknown_branch_name', None, limit=100)
- @patch('swh.web.common.service.storage')
- def test_lookup_content_raw_not_found(self, mock_storage):
- # given
- mock_storage.content_find = MagicMock(return_value=None)
+ @given(unknown_content())
+ def test_lookup_content_raw_not_found(self, unknown_content):
- # when
with self.assertRaises(NotFoundExc) as cm:
- service.lookup_content_raw('sha1:' + self.SHA1_SAMPLE)
+ service.lookup_content_raw('sha1:' + unknown_content['sha1'])
+
self.assertIn(cm.exception.args[0],
'Content with %s checksum equals to %s not found!' %
- ('sha1', self.SHA1_SAMPLE))
-
- mock_storage.content_find.assert_called_with(
- {'sha1': hash_to_bytes(self.SHA1_SAMPLE)})
+ ('sha1', unknown_content['sha1']))
- @patch('swh.web.common.service.storage')
- def test_lookup_content_raw(self, mock_storage):
- # given
- mock_storage.content_find = MagicMock(return_value={
- 'sha1': self.SHA1_SAMPLE,
- })
- mock_storage.content_get = MagicMock(return_value=[{
- 'data': b'binary data'}])
+ @given(content())
+ def test_lookup_content_raw(self, content):
- # when
actual_content = service.lookup_content_raw(
- 'sha256:%s' % self.SHA256_SAMPLE)
+ 'sha256:%s' % content['sha256'])
- # then
- self.assertEqual(actual_content, {'data': b'binary data'})
+ expected_content = self.content_get(content['sha1'])
- mock_storage.content_find.assert_called_once_with(
- {'sha256': self.SHA256_SAMPLE_BIN})
- mock_storage.content_get.assert_called_once_with(
- [hash_to_bytes(self.SHA1_SAMPLE)])
+ self.assertEqual(actual_content, expected_content)
- @patch('swh.web.common.service.storage')
- def test_lookup_content_not_found(self, mock_storage):
- # given
- mock_storage.content_find = MagicMock(return_value=None)
+ @given(unknown_content())
+ def test_lookup_content_not_found(self, unknown_content):
- # when
with self.assertRaises(NotFoundExc) as cm:
- # then
- service.lookup_content('sha1:%s' % self.SHA1_SAMPLE)
+ service.lookup_content('sha1:%s' % unknown_content['sha1'])
+
self.assertIn(cm.exception.args[0],
'Content with %s checksum equals to %s not found!' %
- ('sha1', self.SHA1_SAMPLE))
-
- mock_storage.content_find.assert_called_with(
- {'sha1': self.SHA1_SAMPLE_BIN})
+ ('sha1', unknown_content['sha1']))
- @patch('swh.web.common.service.storage')
- def test_lookup_content_with_sha1(self, mock_storage):
- # given
- mock_storage.content_find = MagicMock(
- return_value=self.SAMPLE_CONTENT_RAW)
+ @given(content())
+ def test_lookup_content_with_sha1(self, content):
- # when
actual_content = service.lookup_content(
- 'sha1:%s' % self.SHA1_SAMPLE)
+ 'sha1:%s' % content['sha1'])
- # then
- self.assertEqual(actual_content, self.SAMPLE_CONTENT)
+ expected_content = self.content_get_metadata(content['sha1'])
- mock_storage.content_find.assert_called_with(
- {'sha1': hash_to_bytes(self.SHA1_SAMPLE)})
-
- @patch('swh.web.common.service.storage')
- def test_lookup_content_with_sha256(self, mock_storage):
- # given
- stub_content = self.SAMPLE_CONTENT_RAW
- stub_content['status'] = 'visible'
+ self.assertEqual(actual_content, expected_content)
- expected_content = self.SAMPLE_CONTENT
- expected_content['status'] = 'visible'
- mock_storage.content_find = MagicMock(
- return_value=stub_content)
+ @given(content())
+ def test_lookup_content_with_sha256(self, content):
- # when
actual_content = service.lookup_content(
- 'sha256:%s' % self.SHA256_SAMPLE)
+ 'sha256:%s' % content['sha256'])
- # then
- self.assertEqual(actual_content, expected_content)
+ expected_content = self.content_get_metadata(content['sha1'])
- mock_storage.content_find.assert_called_with(
- {'sha256': self.SHA256_SAMPLE_BIN})
+ self.assertEqual(actual_content, expected_content)
- @patch('swh.web.common.service.storage')
- def test_lookup_person(self, mock_storage):
- # given
- mock_storage.person_get = MagicMock(return_value=[{
- 'id': 'person_id',
- 'name': b'some_name',
- 'email': b'some-email',
- }])
+ @given(revision())
+ def test_lookup_person(self, revision):
- # when
- actual_person = service.lookup_person('person_id')
+ rev_data = self.revision_get(revision)
- # then
- self.assertEqual(actual_person, {
- 'id': 'person_id',
- 'name': 'some_name',
- 'email': 'some-email',
- })
+ actual_person = service.lookup_person(rev_data['author']['id'])
- mock_storage.person_get.assert_called_with(['person_id'])
+ self.assertEqual(actual_person, rev_data['author'])
- @patch('swh.web.common.service.storage')
- def test_lookup_directory_bad_checksum(self, mock_storage):
- # given
- mock_storage.directory_ls = MagicMock()
+ def test_lookup_directory_bad_checksum(self):
- # when
with self.assertRaises(BadInputExc):
service.lookup_directory('directory_id')
- # then
- mock_storage.directory_ls.called = False
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- def test_lookup_directory_not_found(self, mock_query, mock_storage):
- # given
- mock_query.parse_hash_with_algorithms_or_throws.return_value = (
- 'sha1',
- 'directory-id-bin')
- mock_storage.directory_missing.return_value = ['directory-id-bin']
+ @given(unknown_directory())
+ def test_lookup_directory_not_found(self, unknown_directory):
- # when
with self.assertRaises(NotFoundExc) as cm:
- service.lookup_directory('directory_id')
+ service.lookup_directory(unknown_directory)
- self.assertIn('Directory with sha1_git directory_id not found',
- cm.exception.args[0])
+ self.assertIn('Directory with sha1_git %s not found'
+ % unknown_directory, cm.exception.args[0])
+
+ @given(directory())
+ def test_lookup_directory(self, directory):
- # then
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_with(
- 'directory_id', ['sha1'], 'Only sha1_git is supported.')
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.query')
- def test_lookup_directory(self, mock_query, mock_storage):
- mock_query.parse_hash_with_algorithms_or_throws.return_value = (
- 'sha1',
- 'directory-sha1-bin')
-
- # given
- stub_dir_entries = [{
- 'sha1': self.SHA1_SAMPLE_BIN,
- 'sha256': self.SHA256_SAMPLE_BIN,
- 'sha1_git': self.SHA1GIT_SAMPLE_BIN,
- 'blake2s256': self.BLAKE2S256_SAMPLE_BIN,
- 'target': hash_to_bytes(
- '40e71b8614fcd89ccd17ca2b1d9e66c5b00a6d03'),
- 'dir_id': self.DIRECTORY_ID_BIN,
- 'name': b'bob',
- 'type': 10,
- }]
-
- expected_dir_entries = [{
- 'checksums': {
- 'sha1': self.SHA1_SAMPLE,
- 'sha256': self.SHA256_SAMPLE,
- 'sha1_git': self.SHA1GIT_SAMPLE,
- 'blake2s256': self.BLAKE2S256_SAMPLE
- },
- 'target': '40e71b8614fcd89ccd17ca2b1d9e66c5b00a6d03',
- 'dir_id': self.DIRECTORY_ID,
- 'name': 'bob',
- 'type': 10,
- }]
-
- mock_storage.directory_ls.return_value = stub_dir_entries
- mock_storage.directory_missing.return_value = []
-
- # when
actual_directory_ls = list(service.lookup_directory(
- 'directory-sha1'))
+ directory))
- # then
- self.assertEqual(actual_directory_ls, expected_dir_entries)
+ expected_directory_ls = self.directory_ls(directory)
- mock_query.parse_hash_with_algorithms_or_throws.assert_called_with(
- 'directory-sha1', ['sha1'], 'Only sha1_git is supported.')
- mock_storage.directory_ls.assert_called_with(
- 'directory-sha1-bin')
+ self.assertEqual(actual_directory_ls, expected_directory_ls)
- @patch('swh.web.common.service.storage')
- def test_lookup_directory_empty(self, mock_storage):
- empty_dir_sha1 = '4b825dc642cb6eb9a060e54bf8d69288fbee4904'
- mock_storage.directory_ls.return_value = []
+ @given(empty_directory())
+ def test_lookup_directory_empty(self, empty_directory):
- # when
- actual_directory_ls = list(service.lookup_directory(empty_dir_sha1))
+ actual_directory_ls = list(service.lookup_directory(empty_directory))
- # then
self.assertEqual(actual_directory_ls, [])
- self.assertFalse(mock_storage.directory_ls.called)
-
- @patch('swh.web.common.service.lookup_snapshot')
- @patch('swh.web.common.service.get_origin_visit')
- def test_lookup_revision_by_nothing_found(self, mock_get_origin_visit,
- mock_lookup_snapshot):
- # given
- mock_get_origin_visit.return_value = {'snapshot': self.SHA1_SAMPLE}
- mock_lookup_snapshot.return_value = {'branches': {}}
+ @given(origin())
+ def test_lookup_revision_by_nothing_found(self, origin):
- # when
with self.assertRaises(NotFoundExc):
- service.lookup_revision_by(1)
-
- @patch('swh.web.common.service.lookup_revision')
- @patch('swh.web.common.service.lookup_snapshot')
- @patch('swh.web.common.service.get_origin_visit')
- def test_lookup_revision_by(self, mock_get_origin_visit,
- mock_lookup_snapshot, mock_lookup_revision):
- # given
- expected_rev = self.SAMPLE_REVISION
-
- mock_get_origin_visit.return_value = {'snapshot': self.SHA1_SAMPLE}
- mock_lookup_snapshot.return_value = \
- {
- 'branches': {
- 'master2': {
- 'target_type': 'revision',
- 'target': expected_rev['id']
- }
- }
- }
-
- mock_lookup_revision.return_value = expected_rev
-
- # when
- actual_revision = service.lookup_revision_by(10, 'master2', 'some-ts')
-
- # then
- self.assertEqual(actual_revision, expected_rev)
-
- @patch('swh.web.common.service.lookup_snapshot')
- @patch('swh.web.common.service.get_origin_visit')
- def test_lookup_revision_with_context_by_ko(self, mock_get_origin_visit,
- mock_lookup_snapshot):
- # given
- mock_get_origin_visit.return_value = {'snapshot': self.SHA1_SAMPLE}
- mock_lookup_snapshot.return_value = {'branches': {}}
-
- # when
- origin_id = 1
- branch_name = 'master3'
- ts = None
+ service.lookup_revision_by(origin['id'], 'invalid-branch-name')
+
+ @given(origin())
+ def test_lookup_revision_by(self, origin):
+
+ branches = self._get_origin_branches(origin)
+ branch_name = random.choice(list(branches.keys()))
+
+ actual_revision = \
+ service.lookup_revision_by(origin['id'], branch_name, None)
+
+ expected_revision = \
+ self.revision_get(branches[branch_name]['target'])
+
+ self.assertEqual(actual_revision, expected_revision)
+
+ @given(origin(), revision())
+ def test_lookup_revision_with_context_by_ko(self, origin, revision):
+
with self.assertRaises(NotFoundExc):
- service.lookup_revision_with_context_by(origin_id, branch_name, ts,
- 'sha1')
-
- @patch('swh.web.common.service.storage')
- @patch('swh.web.common.service.lookup_revision')
- @patch('swh.web.common.service.lookup_snapshot')
- @patch('swh.web.common.service.get_origin_visit')
- @patch('swh.web.common.service.lookup_revision_with_context')
- def test_lookup_revision_with_context_by(
- self, mock_lookup_revision_with_context, mock_get_origin_visit,
- mock_lookup_snapshot, mock_lookup_revision, mock_storage
- ):
- # given
- stub_root_rev = self.SAMPLE_REVISION
-
- mock_get_origin_visit.return_value = {'snapshot': self.SHA1_SAMPLE}
- mock_lookup_snapshot.return_value = \
- {
- 'branches': {
- 'master2': {
- 'target_type': 'revision',
- 'target': stub_root_rev['id']
- }
- }
- }
-
- mock_lookup_revision.return_value = stub_root_rev
- stub_rev = {'id': 'rev-found'}
- mock_lookup_revision_with_context.return_value = stub_rev
-
- mock_storage.revision_get.return_value = [self.SAMPLE_REVISION_RAW]
-
- # when
- origin_id = 1
- branch_name = 'master2'
- ts = None
- sha1_git = 'sha1'
+ service.lookup_revision_with_context_by(origin['id'],
+ 'invalid-branch-name',
+ None,
+ revision)
+
+ @given(origin())
+ def test_lookup_revision_with_context_by(self, origin):
+
+ branches = self._get_origin_branches(origin)
+ branch_name = random.choice(list(branches.keys()))
+
+ root_rev = branches[branch_name]['target']
+ root_rev_log = self.revision_log(root_rev)
+
+ children = defaultdict(list)
+
+ for rev in root_rev_log:
+ for rev_p in rev['parents']:
+ children[rev_p].append(rev['id'])
+
+ rev = root_rev_log[-1]['id']
+
actual_root_rev, actual_rev = service.lookup_revision_with_context_by(
- origin_id, branch_name, ts, sha1_git)
+ origin['id'], branch_name, None, rev)
- # then
- self.assertEqual(actual_root_rev, stub_root_rev)
- self.assertEqual(actual_rev, stub_rev)
+ expected_root_rev = self.revision_get(root_rev)
+ expected_rev = self.revision_get(rev)
+ expected_rev['children'] = children[rev]
- mock_lookup_revision_with_context.assert_called_once_with(
- self.SAMPLE_REVISION_RAW, sha1_git, 100)
+ self.assertEqual(actual_root_rev, expected_root_rev)
+ self.assertEqual(actual_rev, expected_rev)
def test_lookup_revision_through_ko_not_implemented(self):
- # then
+
with self.assertRaises(NotImplementedError):
service.lookup_revision_through({
'something-unknown': 10,
})
- @patch('swh.web.common.service.lookup_revision_with_context_by')
- def test_lookup_revision_through_with_context_by(self, mock_lookup):
- # given
- stub_rev = {'id': 'rev'}
- mock_lookup.return_value = stub_rev
-
- # when
- actual_revision = service.lookup_revision_through({
- 'origin_id': 1,
- 'branch_name': 'master',
- 'ts': None,
- 'sha1_git': 'sha1-git'
- }, limit=1000)
-
- # then
- self.assertEqual(actual_revision, stub_rev)
-
- mock_lookup.assert_called_once_with(
- 1, 'master', None, 'sha1-git', 1000)
-
- @patch('swh.web.common.service.lookup_revision_by')
- def test_lookup_revision_through_with_revision_by(self, mock_lookup):
- # given
- stub_rev = {'id': 'rev'}
- mock_lookup.return_value = stub_rev
-
- # when
- actual_revision = service.lookup_revision_through({
- 'origin_id': 2,
- 'branch_name': 'master2',
- 'ts': 'some-ts',
- }, limit=10)
-
- # then
- self.assertEqual(actual_revision, stub_rev)
-
- mock_lookup.assert_called_once_with(
- 2, 'master2', 'some-ts')
-
- @patch('swh.web.common.service.lookup_revision_with_context')
- def test_lookup_revision_through_with_context(self, mock_lookup):
- # given
- stub_rev = {'id': 'rev'}
- mock_lookup.return_value = stub_rev
-
- # when
- actual_revision = service.lookup_revision_through({
- 'sha1_git_root': 'some-sha1-root',
- 'sha1_git': 'some-sha1',
- })
-
- # then
- self.assertEqual(actual_revision, stub_rev)
-
- mock_lookup.assert_called_once_with(
- 'some-sha1-root', 'some-sha1', 100)
-
- @patch('swh.web.common.service.lookup_revision')
- def test_lookup_revision_through_with_revision(self, mock_lookup):
- # given
- stub_rev = {'id': 'rev'}
- mock_lookup.return_value = stub_rev
-
- # when
- actual_revision = service.lookup_revision_through({
- 'sha1_git': 'some-sha1',
- })
-
- # then
- self.assertEqual(actual_revision, stub_rev)
-
- mock_lookup.assert_called_once_with(
- 'some-sha1')
-
- @patch('swh.web.common.service.lookup_revision_through')
- def test_lookup_directory_through_revision_ko_not_found(
- self, mock_lookup_rev):
- # given
- mock_lookup_rev.return_value = None
-
- # when
+ @given(origin())
+ def test_lookup_revision_through_with_context_by(self, origin):
+
+ branches = self._get_origin_branches(origin)
+ branch_name = random.choice(list(branches.keys()))
+
+ root_rev = branches[branch_name]['target']
+ root_rev_log = self.revision_log(root_rev)
+ rev = root_rev_log[-1]['id']
+
+ self.assertEqual(service.lookup_revision_through({
+ 'origin_id': origin['id'],
+ 'branch_name': branch_name,
+ 'ts': None,
+ 'sha1_git': rev
+ }),
+ service.lookup_revision_with_context_by(
+ origin['id'], branch_name, None, rev)
+ )
+
+ @given(origin())
+ def test_lookup_revision_through_with_revision_by(self, origin):
+
+ branches = self._get_origin_branches(origin)
+ branch_name = random.choice(list(branches.keys()))
+
+ self.assertEqual(service.lookup_revision_through({
+ 'origin_id': origin['id'],
+ 'branch_name': branch_name,
+ 'ts': None,
+ }),
+ service.lookup_revision_by(
+ origin['id'], branch_name, None)
+ )
+
+ @given(ancestor_revisions())
+ def test_lookup_revision_through_with_context(self, ancestor_revisions):
+
+ sha1_git = ancestor_revisions['sha1_git']
+ sha1_git_root = ancestor_revisions['sha1_git_root']
+
+ self.assertEqual(service.lookup_revision_through({
+ 'sha1_git_root': sha1_git_root,
+ 'sha1_git': sha1_git,
+ }),
+ service.lookup_revision_with_context(
+ sha1_git_root, sha1_git)
+
+ )
+
+ @given(revision())
+ def test_lookup_revision_through_with_revision(self, revision):
+
+ self.assertEqual(service.lookup_revision_through({
+ 'sha1_git': revision
+ }),
+ service.lookup_revision(revision)
+ )
+
+ @given(revision())
+ def test_lookup_directory_through_revision_ko_not_found(self, revision):
+
with self.assertRaises(NotFoundExc):
service.lookup_directory_through_revision(
- {'id': 'rev'}, 'some/path', 100)
-
- mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 100)
-
- @patch('swh.web.common.service.lookup_revision_through')
- @patch('swh.web.common.service.lookup_directory_with_revision')
- def test_lookup_directory_through_revision_ok_with_data(
- self, mock_lookup_dir, mock_lookup_rev):
- # given
- mock_lookup_rev.return_value = {'id': 'rev-id'}
- mock_lookup_dir.return_value = {'type': 'dir',
- 'content': []}
-
- # when
- rev_id, dir_result = service.lookup_directory_through_revision(
- {'id': 'rev'}, 'some/path', 100)
- # then
- self.assertEqual(rev_id, 'rev-id')
- self.assertEqual(dir_result, {'type': 'dir',
- 'content': []})
-
- mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 100)
- mock_lookup_dir.assert_called_once_with('rev-id', 'some/path', False)
-
- @patch('swh.web.common.service.lookup_revision_through')
- @patch('swh.web.common.service.lookup_directory_with_revision')
- def test_lookup_directory_through_revision_ok_with_content(
- self, mock_lookup_dir, mock_lookup_rev):
- # given
- mock_lookup_rev.return_value = {'id': 'rev-id'}
- stub_result = {'type': 'file',
- 'revision': 'rev-id',
- 'content': {'data': b'blah',
- 'sha1': 'sha1'}}
- mock_lookup_dir.return_value = stub_result
-
- # when
- rev_id, dir_result = service.lookup_directory_through_revision(
- {'id': 'rev'}, 'some/path', 10, with_data=True)
- # then
- self.assertEqual(rev_id, 'rev-id')
- self.assertEqual(dir_result, stub_result)
-
- mock_lookup_rev.assert_called_once_with({'id': 'rev'}, 10)
- mock_lookup_dir.assert_called_once_with('rev-id', 'some/path', True)
+ {'sha1_git': revision}, 'some/invalid/path')
+
+ @given(revision())
+ def test_lookup_directory_through_revision_ok(self, revision):
+
+ revision_data = self.revision_get(revision)
+ dir_entries = [e for e in self.directory_ls(revision_data['directory'])
+ if e['type'] == 'file']
+ dir_entry = random.choice(dir_entries)
+
+ self.assertEqual(
+ service.lookup_directory_through_revision({'sha1_git': revision},
+ dir_entry['name']),
+ (revision,
+ service.lookup_directory_with_revision(
+ revision, dir_entry['name']))
+ )
+
+ @given(revision())
+ def test_lookup_directory_through_revision_ok_with_data(self, revision):
+
+ revision_data = self.revision_get(revision)
+ dir_entries = [e for e in self.directory_ls(revision_data['directory'])
+ if e['type'] == 'file']
+ dir_entry = random.choice(dir_entries)
+
+ self.assertEqual(
+ service.lookup_directory_through_revision({'sha1_git': revision},
+ dir_entry['name'],
+ with_data=True),
+ (revision,
+ service.lookup_directory_with_revision(
+ revision, dir_entry['name'], with_data=True))
+ )
diff --git a/swh/web/tests/data.py b/swh/web/tests/data.py
index 40f1e9971..2403402ef 100644
--- a/swh/web/tests/data.py
+++ b/swh/web/tests/data.py
@@ -1,237 +1,249 @@
# Copyright (C) 2018 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import os
from swh.indexer.language import LanguageIndexer
from swh.indexer.fossology_license import FossologyLicenseIndexer
from swh.indexer.mimetype import MimetypeIndexer
from swh.indexer.ctags import CtagsIndexer
from swh.indexer.storage import get_indexer_storage
-from swh.model.hashutil import hash_to_hex, DEFAULT_ALGORITHMS
+from swh.model.hashutil import hash_to_hex, hash_to_bytes, DEFAULT_ALGORITHMS
+from swh.model.identifiers import directory_identifier
from swh.loader.git.from_disk import GitLoaderFromArchive
from swh.storage.algos.dir_iterators import dir_iterator
# Module used to initialize data that will be provided as tests input
# Configuration for git loader
_TEST_LOADER_CONFIG = {
'storage': {
'cls': 'memory',
'args': {}
},
'send_contents': True,
'send_directories': True,
'send_revisions': True,
'send_releases': True,
'send_snapshot': True,
'content_size_limit': 100 * 1024 * 1024,
'content_packet_size': 10,
'content_packet_size_bytes': 100 * 1024 * 1024,
'directory_packet_size': 10,
'revision_packet_size': 10,
'release_packet_size': 10,
'save_data': False,
}
# Base content indexer configuration
_TEST_INDEXER_BASE_CONFIG = {
'storage': {
'cls': 'memory',
'args': {},
},
'objstorage': {
'cls': 'memory',
'args': {},
},
'indexer_storage': {
'cls': 'memory',
'args': {},
}
}
# MimetypeIndexer with custom configuration for tests
class _MimetypeIndexer(MimetypeIndexer):
def parse_config_file(self, *args, **kwargs):
return {
**_TEST_INDEXER_BASE_CONFIG,
'tools': {
'name': 'file',
'version': '1:5.30-1+deb9u1',
'configuration': {
"type": "library",
"debian-package": "python3-magic"
}
}
}
# LanguageIndexer with custom configuration for tests
class _LanguageIndexer(LanguageIndexer):
def parse_config_file(self, *args, **kwargs):
return {
**_TEST_INDEXER_BASE_CONFIG,
'tools': {
'name': 'pygments',
'version': '2.0.1+dfsg-1.1+deb8u1',
'configuration': {
'type': 'library',
'debian-package': 'python3-pygments',
'max_content_size': 10240,
}
}
}
# FossologyLicenseIndexer with custom configuration for tests
class _FossologyLicenseIndexer(FossologyLicenseIndexer):
def parse_config_file(self, *args, **kwargs):
return {
**_TEST_INDEXER_BASE_CONFIG,
'workdir': '/tmp/swh/indexer.fossology.license',
'tools': {
'name': 'nomos',
'version': '3.1.0rc2-31-ga2cbb8c',
'configuration': {
'command_line': 'nomossa <filepath>',
},
}
}
# CtagsIndexer with custom configuration for tests
class _CtagsIndexer(CtagsIndexer):
def parse_config_file(self, *args, **kwargs):
return {
**_TEST_INDEXER_BASE_CONFIG,
'workdir': '/tmp/swh/indexer.ctags',
'languages': {'c': 'c'},
'tools': {
'name': 'universal-ctags',
'version': '~git7859817b',
'configuration': {
'command_line': '''ctags --fields=+lnz --sort=no --links=no ''' # noqa
'''--output-format=json <filepath>'''
},
}
}
# Lightweight git repositories that will be loaded to generate
# input data for tests
_TEST_ORIGINS = [
{
'id': 1,
'type': 'git',
'url': 'https://github.com/wcoder/highlightjs-line-numbers.js',
'archive': 'highlightjs-line-numbers.js.zip'
},
{
'id': 2,
'type': 'git',
'url': 'https://github.com/memononen/libtess2',
'archive': 'libtess2.zip'
+ },
+ {
+ 'id': 3,
+ 'type': 'git',
+ 'url': 'repo_with_submodules',
+ 'archive': 'repo_with_submodules.tgz'
}
]
# Tests data initialization
def _init_tests_data():
# Load git repositories from archives
loader = GitLoaderFromArchive(config=_TEST_LOADER_CONFIG)
for origin in _TEST_ORIGINS:
origin_repo_archive = \
os.path.join(os.path.dirname(__file__),
'resources/repos/%s' % origin['archive'])
loader.load(origin['url'], origin_repo_archive, None)
# Get reference to the memory storage
storage = loader.storage
contents = set()
directories = set()
revisions = set()
releases = set()
snapshots = set()
# Get all objects loaded into the test archive
for origin in _TEST_ORIGINS:
snp = storage.snapshot_get_latest(origin['id'])
snapshots.add(hash_to_hex(snp['id']))
for branch_name, branch_data in snp['branches'].items():
if branch_data['target_type'] == 'revision':
revisions.add(branch_data['target'])
elif branch_data['target_type'] == 'release':
release = next(storage.release_get([branch_data['target']]))
revisions.add(release['target'])
releases.add(hash_to_hex(branch_data['target']))
for rev_log in storage.revision_shortlog(set(revisions)):
rev_id = rev_log[0]
revisions.add(rev_id)
for rev in storage.revision_get(revisions):
dir_id = rev['directory']
directories.add(hash_to_hex(dir_id))
for entry in dir_iterator(storage, dir_id):
if entry['type'] == 'file':
contents.add(entry['sha1'])
- else:
+ elif entry['type'] == 'dir':
directories.add(hash_to_hex(entry['target']))
# Get all checksums for each content
contents_metadata = storage.content_get_metadata(contents)
contents = []
for content_metadata in contents_metadata:
contents.append({
algo: hash_to_hex(content_metadata[algo])
for algo in DEFAULT_ALGORITHMS
})
# Create indexer storage instance that will be shared by indexers
idx_storage = get_indexer_storage('memory', {})
# Instantiate content indexers that will be used in tests
# and force them to use the memory storages
indexers = {}
for idx_name, idx_class in (('mimetype_indexer', _MimetypeIndexer),
('language_indexer', _LanguageIndexer),
('license_indexer', _FossologyLicenseIndexer),
('ctags_indexer', _CtagsIndexer)):
idx = idx_class()
idx.storage = storage
idx.objstorage = storage.objstorage
idx.idx_storage = idx_storage
idx.register_tools(idx.config['tools'])
indexers[idx_name] = idx
+ # Add the empty directory to the test archive
+ empty_dir_id = directory_identifier({'entries': []})
+ empty_dir_id_bin = hash_to_bytes(empty_dir_id)
+ storage.directory_add([{'id': empty_dir_id_bin, 'entries': []}])
+
# Return tests data
return {
'storage': storage,
'idx_storage': idx_storage,
**indexers,
'origins': _TEST_ORIGINS,
'contents': contents,
'directories': list(directories),
'releases': list(releases),
'revisions': list(map(hash_to_hex, revisions)),
'snapshots': list(snapshots)
}
_tests_data = None
def get_tests_data():
"""
Initialize tests data and return them in a dict.
"""
global _tests_data
if _tests_data is None:
_tests_data = _init_tests_data()
return _tests_data
diff --git a/swh/web/tests/resources/repos/repo_with_submodules.tgz b/swh/web/tests/resources/repos/repo_with_submodules.tgz
new file mode 100644
index 000000000..5c205ea0f
Binary files /dev/null and b/swh/web/tests/resources/repos/repo_with_submodules.tgz differ
diff --git a/swh/web/tests/strategies.py b/swh/web/tests/strategies.py
index 5bc8dd851..73c3c8f19 100644
--- a/swh/web/tests/strategies.py
+++ b/swh/web/tests/strategies.py
@@ -1,298 +1,341 @@
# Copyright (C) 2018 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import random
from collections import defaultdict
from datetime import datetime
from hypothesis import settings
from hypothesis.strategies import (
just, sampled_from, lists, composite, datetimes
)
from string import ascii_letters, hexdigits
-from swh.model.hashutil import hash_to_hex
+from swh.model.hashutil import hash_to_hex, hash_to_bytes
+from swh.model.identifiers import directory_identifier
from swh.storage.algos.revisions_walker import get_revisions_walker
-from swh.storage.tests.algos.test_snapshot import origins
+from swh.storage.tests.algos.test_snapshot import origins as new_origin # noqa
from swh.web.tests.data import get_tests_data
# Module dedicated to the generation of input data for tests through
# the use of hypothesis.
# Some of these data are sampled from a test archive created and populated
# in the swh.web.tests.data module.
# Set some hypothesis settings
settings.register_profile("swh-web", settings(deadline=None, max_examples=1))
settings.load_profile("swh-web")
# The following strategies exploit the hypothesis capabilities
def _known_swh_object(object_type):
tests_data = get_tests_data()
return sampled_from(tests_data[object_type])
def _unknown_swh_object(draw, object_type):
tests_data = get_tests_data()
+ storage = tests_data['storage']
while True:
sha1_git = draw(sha1())
- if sha1_git not in tests_data[object_type]:
+ # some tests will use the generated id to create a revision on the fly
+ if object_type == 'revisions':
+ obj = next(storage.revision_get([hash_to_bytes(sha1_git)]))
+ if obj is None:
+ return sha1_git
+ elif sha1_git not in tests_data[object_type]:
return sha1_git
def sha1():
"""
Hypothesis strategy returning a valid hexadecimal sha1 value.
"""
sha1 = ''.join(random.choice(hexdigits) for x in range(40))
return just(sha1.lower())
def invalid_sha1():
"""
Hypothesis strategy returning an invalid sha1 representation.
"""
invalid_sha1 = ''.join(random.choice(ascii_letters) for x in range(50))
return just(invalid_sha1.lower())
def sha256():
"""
Hypothesis strategy returning a valid hexadecimal sha256 value.
"""
sha256 = ''.join(random.choice(hexdigits) for x in range(64))
return just(sha256.lower())
def content():
"""
Hypothesis strategy returning a random content ingested
into the test archive.
"""
return _known_swh_object('contents')
def contents():
"""
Hypothesis strategy returning random contents ingested
into the test archive.
"""
return lists(content(), min_size=2, max_size=8)
@composite
def unknown_content(draw):
"""
Hypothesis strategy returning a random content not ingested
into the test archive.
"""
tests_data = get_tests_data()
while True:
unknown_content = {
'blake2s256': draw(sha256()),
'sha1': draw(sha1()),
'sha1_git': draw(sha1()),
'sha256': draw(sha256())
}
if unknown_content not in tests_data['contents']:
return unknown_content
def unknown_contents():
"""
Hypothesis strategy returning random contents not ingested
into the test archive.
"""
return lists(unknown_content(), min_size=2, max_size=8)
def directory():
"""
Hypothesis strategy returning a random directory ingested
into the test archive.
"""
return _known_swh_object('directories')
+def empty_directory():
+ """
+ Hypothesis strategy returning the empty directory ingested
+ into the test archive.
+ """
+ return just(directory_identifier({'entries': []}))
+
+
@composite
def unknown_directory(draw):
"""
Hypothesis strategy returning a random directory not ingested
into the test archive.
"""
return _unknown_swh_object(draw, 'directories')
def origin():
"""
Hypothesis strategy returning a random origin not ingested
into the test archive.
"""
- return origins()
+ return _known_swh_object('origins')
def visit_dates():
"""
Hypothesis strategy returning a list of visit dates.
"""
return lists(datetimes(min_value=datetime(2015, 1, 1, 0, 0),
max_value=datetime(2018, 12, 31, 0, 0)),
min_size=2, max_size=8, unique=True)
def release():
"""
Hypothesis strategy returning a random release ingested
into the test archive.
"""
return _known_swh_object('releases')
@composite
def unknown_release(draw):
"""
Hypothesis strategy returning a random revision not ingested
into the test archive.
"""
return _unknown_swh_object(draw, 'releases')
def revision():
"""
Hypothesis strategy returning a random revision ingested
into the test archive.
"""
return _known_swh_object('revisions')
@composite
def unknown_revision(draw):
"""
Hypothesis strategy returning a random revision not ingested
into the test archive.
"""
return _unknown_swh_object(draw, 'revisions')
+def revisions():
+ """
+ Hypothesis strategy returning random revisions ingested
+ into the test archive.
+ """
+ return lists(revision(), min_size=2, max_size=8)
+
+
+def unknown_revisions():
+ """
+ Hypothesis strategy returning random revisions not ingested
+ into the test archive.
+ """
+ return lists(unknown_revision(), min_size=2, max_size=8)
+
+
def snapshot():
"""
Hypothesis strategy returning a random snapshot ingested
into the test archive.
"""
return _known_swh_object('snapshots')
@composite
def unknown_snapshot(draw):
"""
Hypothesis strategy returning a random revision not ingested
into the test archive.
"""
return _unknown_swh_object(draw, 'snapshots')
def _get_origin_dfs_revisions_walker():
storage = get_tests_data()['storage']
- origin = random.choice(get_tests_data()['origins'])
+ origin = random.choice(get_tests_data()['origins'][:-1])
snapshot = storage.snapshot_get_latest(origin['id'])
head = snapshot['branches'][b'HEAD']['target']
return get_revisions_walker('dfs', storage, head)
def ancestor_revisions():
"""
Hypothesis strategy returning a pair of revisions ingested into the
test archive with an ancestor relation.
"""
# get a dfs revisions walker for one of the origins
# loaded into the test archive
revisions_walker = _get_origin_dfs_revisions_walker()
master_revisions = []
children = defaultdict(list)
init_rev_found = False
# get revisions only authored in the master branch
for rev in revisions_walker:
for rev_p in rev['parents']:
children[rev_p].append(rev['id'])
if not init_rev_found:
master_revisions.append(rev)
if not rev['parents']:
init_rev_found = True
# head revision
root_rev = master_revisions[0]
# pick a random revision, different from head, only authored
# in the master branch
ancestor_rev_idx = random.choice(list(range(1, len(master_revisions)-1)))
ancestor_rev = master_revisions[ancestor_rev_idx]
ancestor_child_revs = children[ancestor_rev['id']]
return just({
'sha1_git_root': hash_to_hex(root_rev['id']),
'sha1_git': hash_to_hex(ancestor_rev['id']),
'children': [hash_to_hex(r) for r in ancestor_child_revs]
})
def non_ancestor_revisions():
"""
Hypothesis strategy returning a pair of revisions ingested into the
test archive with no ancestor relation.
"""
# get a dfs revisions walker for one of the origins
# loaded into the test archive
revisions_walker = _get_origin_dfs_revisions_walker()
merge_revs = []
children = defaultdict(list)
# get all merge revisions
for rev in revisions_walker:
if len(rev['parents']) > 1:
merge_revs.append(rev)
for rev_p in rev['parents']:
children[rev_p].append(rev['id'])
# find a merge revisions whose parents have a unique child revision
random.shuffle(merge_revs)
selected_revs = None
for merge_rev in merge_revs:
if all(len(children[rev_p]) == 1
for rev_p in merge_rev['parents']):
selected_revs = merge_rev['parents']
return just({
'sha1_git_root': hash_to_hex(selected_revs[0]),
'sha1_git': hash_to_hex(selected_revs[1])
})
# The following strategies returns data specific to some tests
# that can not be generated and thus are hardcoded.
def contents_with_ctags():
"""
Hypothesis strategy returning contents ingested into the test
archive. Those contents are ctags compatible, that is running
ctags on those lay results.
"""
return just({
'sha1s': ['0ab37c02043ebff946c1937523f60aadd0844351',
'15554cf7608dde6bfefac7e3d525596343a85b6f',
'2ce837f1489bdfb8faf3ebcc7e72421b5bea83bd',
'30acd0b47fc25e159e27a980102ddb1c4bea0b95',
'4f81f05aaea3efb981f9d90144f746d6b682285b',
'5153aa4b6e4455a62525bc4de38ed0ff6e7dd682',
'59d08bafa6a749110dfb65ba43a61963d5a5bf9f',
'7568285b2d7f31ae483ae71617bd3db873deaa2c',
'7ed3ee8e94ac52ba983dd7690bdc9ab7618247b4',
'8ed7ef2e7ff9ed845e10259d08e4145f1b3b5b03',
'9b3557f1ab4111c8607a4f2ea3c1e53c6992916c',
'9c20da07ed14dc4fcd3ca2b055af99b2598d8bdd',
'c20ceebd6ec6f7a19b5c3aebc512a12fbdc9234b',
'e89e55a12def4cd54d5bff58378a3b5119878eb7',
'e8c0654fe2d75ecd7e0b01bee8a8fc60a130097e',
'eb6595e559a1d34a2b41e8d4835e0e4f98a5d2b5'],
'symbol_name': 'ABS'
})
+
+
+def revision_with_submodules():
+ """
+ Hypothesis strategy returning a revision that is known to
+ point to a directory with revision entries (aka git submodule)
+ """
+ return just({
+ 'rev_sha1_git': 'ffcb69001f3f6745dfd5b48f72ab6addb560e234',
+ 'rev_dir_sha1_git': 'd92a21446387fa28410e5a74379c934298f39ae2',
+ 'rev_dir_rev_path': 'libtess2'
+ })
diff --git a/swh/web/tests/testcase.py b/swh/web/tests/testcase.py
index ddc88e0e7..244d74ff0 100644
--- a/swh/web/tests/testcase.py
+++ b/swh/web/tests/testcase.py
@@ -1,145 +1,160 @@
# Copyright (C) 2015-2018 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import shutil
from subprocess import run, PIPE
from django.core.cache import cache
from hypothesis.extra.django import TestCase
from swh.model.hashutil import hash_to_bytes
from swh.web import config
from swh.web.common import converters, service
from swh.web.tests.data import get_tests_data
ctags_json_missing = \
shutil.which('ctags') is None or \
b'+json' not in run(['ctags', '--version'], stdout=PIPE).stdout
fossology_missing = shutil.which('nomossa') is None
class WebTestCase(TestCase):
"""Base TestCase class for swh-web.
It is initialized with references to in-memory storages containing
raw tests data.
It also defines class methods to retrieve those tests data in
a json serializable format in order to ease tests implementation.
"""
@classmethod
def setUpClass(cls):
super().setUpClass()
tests_data = get_tests_data()
cls.storage = tests_data['storage']
cls.idx_storage = tests_data['idx_storage']
cls.mimetype_indexer = tests_data['mimetype_indexer']
cls.language_indexer = tests_data['language_indexer']
cls.license_indexer = tests_data['license_indexer']
cls.ctags_indexer = tests_data['ctags_indexer']
# Update swh-web configuration to use the in-memory storage
# instantiated in the tests.data module
swh_config = config.get_config()
swh_config.update({'storage': cls.storage})
service.storage = cls.storage
# Update swh-web configuration to use the in-memory indexer storage
# instantiated in the tests.data modules
swh_config.update({'indexer_storage': cls.idx_storage})
service.idx_storage = cls.idx_storage
@classmethod
def content_add_mimetype(cls, cnt_id):
cls.mimetype_indexer.run([hash_to_bytes(cnt_id)],
'update-dups')
@classmethod
def content_get_mimetype(cls, cnt_id):
mimetype = next(cls.idx_storage.content_mimetype_get(
[hash_to_bytes(cnt_id)]))
return converters.from_filetype(mimetype)
@classmethod
def content_add_language(cls, cnt_id):
cls.language_indexer.run([hash_to_bytes(cnt_id)],
'update-dups')
@classmethod
def content_get_language(cls, cnt_id):
lang = next(cls.idx_storage.content_language_get(
[hash_to_bytes(cnt_id)]))
return converters.from_swh(lang, hashess={'id'})
@classmethod
def content_add_license(cls, cnt_id):
cls.license_indexer.run([hash_to_bytes(cnt_id)],
'update-dups')
@classmethod
def content_get_license(cls, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
lic = next(cls.idx_storage.content_fossology_license_get(
[cnt_id_bytes]))
return converters.from_swh({'id': cnt_id_bytes,
'facts': lic[cnt_id_bytes]},
hashess={'id'})
@classmethod
def content_add_ctags(cls, cnt_id):
cls.ctags_indexer.run([hash_to_bytes(cnt_id)],
'update-dups')
@classmethod
def content_get_ctags(cls, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
ctags = cls.idx_storage.content_ctags_get([cnt_id_bytes])
for ctag in ctags:
yield converters.from_swh(ctag, hashess={'id'})
@classmethod
def content_get_metadata(cls, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
metadata = next(cls.storage.content_get_metadata([cnt_id_bytes]))
return converters.from_swh(metadata,
hashess={'sha1', 'sha1_git', 'sha256',
'blake2s256'})
@classmethod
def content_get(cls, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
cnt = next(cls.storage.content_get([cnt_id_bytes]))
return converters.from_content(cnt)
@classmethod
def directory_ls(cls, dir_id):
cnt_id_bytes = hash_to_bytes(dir_id)
dir_content = map(converters.from_directory_entry,
cls.storage.directory_ls(cnt_id_bytes))
return list(dir_content)
@classmethod
def release_get(cls, rel_id):
rel_id_bytes = hash_to_bytes(rel_id)
rel_data = next(cls.storage.release_get([rel_id_bytes]))
return converters.from_release(rel_data)
@classmethod
def revision_get(cls, rev_id):
rev_id_bytes = hash_to_bytes(rev_id)
rev_data = next(cls.storage.revision_get([rev_id_bytes]))
return converters.from_revision(rev_data)
@classmethod
- def revision_log(cls, rev_id):
+ def revision_log(cls, rev_id, limit=None):
rev_id_bytes = hash_to_bytes(rev_id)
- return map(converters.from_revision,
- cls.storage.revision_log([rev_id_bytes]))
+ return list(map(converters.from_revision,
+ cls.storage.revision_log([rev_id_bytes], limit=limit)))
+
+ @classmethod
+ def snapshot_get_latest(cls, origin_id):
+ snp = cls.storage.snapshot_get_latest(origin_id)
+ return converters.from_snapshot(snp)
+
+ @classmethod
+ def origin_visit_get(cls, origin_id):
+ visits = cls.storage.origin_visit_get(origin_id)
+ return list(map(converters.from_origin_visit, visits))
+
+ @classmethod
+ def snapshot_get(cls, snapshot_id):
+ snp = cls.storage.snapshot_get(hash_to_bytes(snapshot_id))
+ return converters.from_snapshot(snp)
def setUp(self):
cache.clear()