Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/data.py
Show All 14 Lines | |||||
from swh.indexer.fossology_license import FossologyLicenseIndexer | from swh.indexer.fossology_license import FossologyLicenseIndexer | ||||
from swh.indexer.mimetype import MimetypeIndexer | from swh.indexer.mimetype import MimetypeIndexer | ||||
from swh.indexer.ctags import CtagsIndexer | from swh.indexer.ctags import CtagsIndexer | ||||
from swh.indexer.storage import get_indexer_storage | from swh.indexer.storage import get_indexer_storage | ||||
from swh.model.from_disk import Directory | from swh.model.from_disk import Directory | ||||
from swh.model.hashutil import hash_to_hex, hash_to_bytes, DEFAULT_ALGORITHMS | from swh.model.hashutil import hash_to_hex, hash_to_bytes, DEFAULT_ALGORITHMS | ||||
from swh.model.identifiers import directory_identifier | from swh.model.identifiers import directory_identifier | ||||
from swh.loader.git.from_disk import GitLoaderFromArchive | from swh.loader.git.from_disk import GitLoaderFromArchive | ||||
from swh.search import get_search | |||||
from swh.storage.algos.dir_iterators import dir_iterator | from swh.storage.algos.dir_iterators import dir_iterator | ||||
from swh.web import config | from swh.web import config | ||||
from swh.web.browse.utils import ( | from swh.web.browse.utils import ( | ||||
get_mimetype_and_encoding_for_content, prepare_content_for_display | get_mimetype_and_encoding_for_content, prepare_content_for_display | ||||
) | ) | ||||
from swh.web.common import service | from swh.web.common import service | ||||
from swh.web.common.highlightjs import get_hljs_language_from_filename | from swh.web.common.highlightjs import get_hljs_language_from_filename | ||||
▲ Show 20 Lines • Show All 123 Lines • ▼ Show 20 Lines | |||||
_contents = {} | _contents = {} | ||||
# Tests data initialization | # Tests data initialization | ||||
def _init_tests_data(): | def _init_tests_data(): | ||||
# To hold reference to the memory storage | # To hold reference to the memory storage | ||||
storage = None | storage = None | ||||
# Create search instance | |||||
search = get_search('memory', {}) | |||||
search.initialize() | |||||
search.origin_update({'url': origin['url']} for origin in _TEST_ORIGINS) | |||||
# Load git repositories from archives | # Load git repositories from archives | ||||
for origin in _TEST_ORIGINS: | for origin in _TEST_ORIGINS: | ||||
for i, archive in enumerate(origin['archives']): | for i, archive in enumerate(origin['archives']): | ||||
origin_repo_archive = \ | origin_repo_archive = \ | ||||
os.path.join(os.path.dirname(__file__), | os.path.join(os.path.dirname(__file__), | ||||
'resources/repos/%s' % archive) | 'resources/repos/%s' % archive) | ||||
loader = GitLoaderFromArchive(origin['url'], | loader = GitLoaderFromArchive(origin['url'], | ||||
archive_path=origin_repo_archive, | archive_path=origin_repo_archive, | ||||
config=_TEST_LOADER_CONFIG, | config=_TEST_LOADER_CONFIG, | ||||
visit_date=origin['visit_date'][i]) | visit_date=origin['visit_date'][i]) | ||||
if storage is None: | if storage is None: | ||||
storage = loader.storage | storage = loader.storage | ||||
else: | else: | ||||
loader.storage = storage | loader.storage = storage | ||||
loader.load() | loader.load() | ||||
origin.update(storage.origin_get(origin)) # add an 'id' key if enabled | origin.update(storage.origin_get(origin)) # add an 'id' key if enabled | ||||
search.origin_update([{'url': origin['url'], 'has_visits': True}]) | |||||
for i in range(250): | for i in range(250): | ||||
url = 'https://many.origins/%d' % (i+1) | url = 'https://many.origins/%d' % (i+1) | ||||
storage.origin_add([{'url': url}]) | storage.origin_add([{'url': url}]) | ||||
search.origin_update([{'url': url, 'has_visits': True}]) | |||||
visit = storage.origin_visit_add(url, '2019-12-03 13:55:05', 'tar') | visit = storage.origin_visit_add(url, '2019-12-03 13:55:05', 'tar') | ||||
storage.origin_visit_update( | storage.origin_visit_update( | ||||
url, visit['visit'], | url, visit['visit'], | ||||
snapshot='1a8893e6a86f444e8be8e7bda6cb34fb1735a00e') | snapshot='1a8893e6a86f444e8be8e7bda6cb34fb1735a00e') | ||||
contents = set() | contents = set() | ||||
directories = set() | directories = set() | ||||
revisions = set() | revisions = set() | ||||
▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 Lines | def _init_tests_data(): | ||||
# Add the empty directory to the test archive | # Add the empty directory to the test archive | ||||
empty_dir_id = directory_identifier({'entries': []}) | empty_dir_id = directory_identifier({'entries': []}) | ||||
empty_dir_id_bin = hash_to_bytes(empty_dir_id) | empty_dir_id_bin = hash_to_bytes(empty_dir_id) | ||||
storage.directory_add([{'id': empty_dir_id_bin, 'entries': []}]) | storage.directory_add([{'id': empty_dir_id_bin, 'entries': []}]) | ||||
# Return tests data | # Return tests data | ||||
return { | return { | ||||
'search': search, | |||||
'storage': storage, | 'storage': storage, | ||||
'idx_storage': idx_storage, | 'idx_storage': idx_storage, | ||||
'origins': _TEST_ORIGINS, | 'origins': _TEST_ORIGINS, | ||||
'contents': contents, | 'contents': contents, | ||||
'directories': list(directories), | 'directories': list(directories), | ||||
'releases': list(releases), | 'releases': list(releases), | ||||
'revisions': list(map(hash_to_hex, revisions)), | 'revisions': list(map(hash_to_hex, revisions)), | ||||
'snapshots': list(snapshots), | 'snapshots': list(snapshots), | ||||
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines | if _tests_data is None: | ||||
_tests_data.update(indexers) | _tests_data.update(indexers) | ||||
if reset or _current_tests_data is None: | if reset or _current_tests_data is None: | ||||
_current_tests_data = deepcopy(_tests_data) | _current_tests_data = deepcopy(_tests_data) | ||||
for (name, logger) in _indexer_loggers.items(): | for (name, logger) in _indexer_loggers.items(): | ||||
_current_tests_data[name].log = logger | _current_tests_data[name].log = logger | ||||
return _current_tests_data | return _current_tests_data | ||||
def override_storages(storage, idx_storage): | def override_storages(storage, idx_storage, search): | ||||
""" | """ | ||||
Helper function to replace the storages from which archive data | Helper function to replace the storages from which archive data | ||||
are fetched. | are fetched. | ||||
""" | """ | ||||
swh_config = config.get_config() | swh_config = config.get_config() | ||||
swh_config.update({'storage': storage}) | swh_config.update({ | ||||
service.storage = storage | 'storage': storage, | ||||
'indexer_storage': idx_storage, | |||||
'search': search, | |||||
}) | |||||
swh_config.update({'indexer_storage': idx_storage}) | service.storage = storage | ||||
service.idx_storage = idx_storage | service.idx_storage = idx_storage | ||||
service.search = search | |||||
# Implement some special endpoints used to provide input tests data | # Implement some special endpoints used to provide input tests data | ||||
# when executing end to end tests with cypress | # when executing end to end tests with cypress | ||||
_content_code_data_exts = {} # type: Dict[str, Dict[str, str]] | _content_code_data_exts = {} # type: Dict[str, Dict[str, str]] | ||||
_content_code_data_filenames = {} # type: Dict[str, Dict[str, str]] | _content_code_data_filenames = {} # type: Dict[str, Dict[str, str]] | ||||
_content_other_data_exts = {} # type: Dict[str, Dict[str, str]] | _content_other_data_exts = {} # type: Dict[str, Dict[str, str]] | ||||
▲ Show 20 Lines • Show All 135 Lines • Show Last 20 Lines |