Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/data.py
# Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2019 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import os | import os | ||||
import random | import random | ||||
from copy import deepcopy | from copy import deepcopy | ||||
from datetime import datetime | |||||
from typing import Dict | from typing import Dict | ||||
from rest_framework.decorators import api_view | from rest_framework.decorators import api_view | ||||
from rest_framework.response import Response | from rest_framework.response import Response | ||||
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 import from_disk | ||||
from swh.model.from_disk import DentryPerms | |||||
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.model.model import ( | |||||
Directory, DirectoryEntry, Revision, Snapshot, TimestampWithTimezone, | |||||
Person, Timestamp, SnapshotBranch, TargetType | |||||
) | |||||
from swh.loader.git.from_disk import GitLoaderFromArchive | from swh.loader.git.from_disk import GitLoaderFromArchive | ||||
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 212 Lines • ▼ Show 20 Lines | def _init_tests_data(): | ||||
# Create indexer storage instance that will be shared by indexers | # Create indexer storage instance that will be shared by indexers | ||||
idx_storage = get_indexer_storage('memory', {}) | idx_storage = get_indexer_storage('memory', {}) | ||||
# 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': []}]) | ||||
# Generate an origin whose root directory contains a single | |||||
# directory named 'directory' | |||||
origin = {'url': 'https://git-server-domain/project-name'} | |||||
visit_date = datetime.now() | |||||
dir_entry = DirectoryEntry(name=b'directory', type='dir', | |||||
target=empty_dir_id_bin, | |||||
perms=DentryPerms.directory) | |||||
dir = Directory(entries=[dir_entry]) | |||||
person = Person(name=b'abcd', email=b'abcd@company.org', | |||||
fullname=b'abcd <abcd@company.org>') | |||||
timestamp = visit_date.timestamp() | |||||
date = TimestampWithTimezone(negative_utc=False, offset=0, | |||||
timestamp=Timestamp(seconds=timestamp, | |||||
microseconds=0)) | |||||
rev = Revision(author=person, committer=person, date=date, | |||||
committer_date=date, message=b'Initial commit', | |||||
metadata=[], parents=[], synthetic=False, type='git', | |||||
directory=dir.id) | |||||
snp = Snapshot(branches={ | |||||
b'master': SnapshotBranch(target_type=TargetType.REVISION, | |||||
target=rev.id), | |||||
b'HEAD': SnapshotBranch(target_type=TargetType.ALIAS, | |||||
target=b'master') | |||||
}) | |||||
storage.directory_add([dir.to_dict()]) | |||||
storage.revision_add([rev.to_dict()]) | |||||
storage.snapshot_add([snp.to_dict()]) | |||||
storage.origin_add_one(origin) | |||||
origin_visit = storage.origin_visit_add(origin['url'], visit_date, | |||||
type='git') | |||||
storage.origin_visit_update(origin['url'], origin_visit['visit'], | |||||
snapshot=snp.id) | |||||
_TEST_ORIGINS.append({ | |||||
'type': 'git', | |||||
'url': origin['url'], | |||||
'visit_date': [visit_date.isoformat()] | |||||
}) | |||||
vlorentz: this is way too long for such a simple thing to add :/ | |||||
anlambertAuthorUnsubmitted Done Inline ActionsSo imagine how long it was when using raw dicts instead models. anlambert: So imagine how long it was when using raw dicts instead models. | |||||
# Return tests data | # Return tests data | ||||
return { | return { | ||||
'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), | ||||
▲ Show 20 Lines • Show All 82 Lines • ▼ Show 20 Lines | Args: | ||||
data_path (str): path to a directory relative to the tests | data_path (str): path to a directory relative to the tests | ||||
folder of swh-web | folder of swh-web | ||||
data_dict (dict): the dict that will store files metadata | data_dict (dict): the dict that will store files metadata | ||||
ext_key (bool): whether to use file extensions or filenames | ext_key (bool): whether to use file extensions or filenames | ||||
as dict keys | as dict keys | ||||
""" | """ | ||||
test_contents_dir = os.path.join( | test_contents_dir = os.path.join( | ||||
os.path.dirname(__file__), data_path).encode('utf-8') | os.path.dirname(__file__), data_path).encode('utf-8') | ||||
directory = Directory.from_disk(path=test_contents_dir, data=True, | directory = from_disk.Directory.from_disk(path=test_contents_dir, | ||||
save_path=True) | data=True, save_path=True) | ||||
objects = directory.collect() | objects = directory.collect() | ||||
for c in objects['content'].values(): | for c in objects['content'].values(): | ||||
c['status'] = 'visible' | c['status'] = 'visible' | ||||
sha1 = hash_to_hex(c['sha1']) | sha1 = hash_to_hex(c['sha1']) | ||||
if ext_key: | if ext_key: | ||||
key = c['path'].decode('utf-8').split('.')[-1] | key = c['path'].decode('utf-8').split('.')[-1] | ||||
filename = 'test.' + key | filename = 'test.' + key | ||||
else: | else: | ||||
▲ Show 20 Lines • Show All 108 Lines • Show Last 20 Lines |
this is way too long for such a simple thing to add :/