Page MenuHomeSoftware Heritage

testcase.py
No OneTemporary

testcase.py

# Copyright (C) 2015-2019 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.common import converters
from swh.web.tests.data import get_tests_data, override_storages
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.
"""
def _pre_setup(self):
cache.clear()
tests_data = get_tests_data(reset=True)
self.storage = tests_data['storage']
self.idx_storage = tests_data['idx_storage']
self.mimetype_indexer = tests_data['mimetype_indexer']
self.license_indexer = tests_data['license_indexer']
self.ctags_indexer = tests_data['ctags_indexer']
# Update swh-web configuration to use the in-memory storages
# instantiated in the tests.data module
override_storages(tests_data['storage'], tests_data['idx_storage'])
super()._pre_setup()
def content_add_mimetype(self, cnt_id):
self.mimetype_indexer.run([hash_to_bytes(cnt_id)],
'update-dups')
def content_get_mimetype(self, cnt_id):
mimetype = next(self.idx_storage.content_mimetype_get(
[hash_to_bytes(cnt_id)]))
return converters.from_filetype(mimetype)
def content_add_language(self, cnt_id):
raise NotImplementedError('Language indexer is disabled.')
self.language_indexer.run([hash_to_bytes(cnt_id)],
'update-dups')
def content_get_language(self, cnt_id):
lang = next(self.idx_storage.content_language_get(
[hash_to_bytes(cnt_id)]))
return converters.from_swh(lang, hashess={'id'})
def content_add_license(self, cnt_id):
self.license_indexer.run([hash_to_bytes(cnt_id)],
'update-dups')
def content_get_license(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
lic = next(self.idx_storage.content_fossology_license_get(
[cnt_id_bytes]))
return converters.from_swh({'id': cnt_id_bytes,
'facts': lic[cnt_id_bytes]},
hashess={'id'})
def content_add_ctags(self, cnt_id):
self.ctags_indexer.run([hash_to_bytes(cnt_id)],
'update-dups')
def content_get_ctags(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
ctags = self.idx_storage.content_ctags_get([cnt_id_bytes])
for ctag in ctags:
yield converters.from_swh(ctag, hashess={'id'})
def content_get_metadata(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
metadata = next(self.storage.content_get_metadata([cnt_id_bytes]))
return converters.from_swh(metadata,
hashess={'sha1', 'sha1_git', 'sha256',
'blake2s256'})
def content_get(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
cnt = next(self.storage.content_get([cnt_id_bytes]))
return converters.from_content(cnt)
def directory_ls(self, dir_id):
cnt_id_bytes = hash_to_bytes(dir_id)
dir_content = map(converters.from_directory_entry,
self.storage.directory_ls(cnt_id_bytes))
return list(dir_content)
def release_get(self, rel_id):
rel_id_bytes = hash_to_bytes(rel_id)
rel_data = next(self.storage.release_get([rel_id_bytes]))
return converters.from_release(rel_data)
def revision_get(self, rev_id):
rev_id_bytes = hash_to_bytes(rev_id)
rev_data = next(self.storage.revision_get([rev_id_bytes]))
return converters.from_revision(rev_data)
def revision_log(self, rev_id, limit=None):
rev_id_bytes = hash_to_bytes(rev_id)
return list(map(converters.from_revision,
self.storage.revision_log([rev_id_bytes], limit=limit)))
def snapshot_get_latest(self, origin_url):
snp = self.storage.snapshot_get_latest(origin_url)
return converters.from_snapshot(snp)
def origin_get(self, origin_info):
origin = self.storage.origin_get(origin_info)
return converters.from_origin(origin)
def origin_visit_get(self, origin_url):
visits = self.storage.origin_visit_get(origin_url)
return list(map(converters.from_origin_visit, visits))
def origin_visit_get_by(self, origin_url, visit_id):
visit = self.storage.origin_visit_get_by(origin_url, visit_id)
return converters.from_origin_visit(visit)
def snapshot_get(self, snapshot_id):
snp = self.storage.snapshot_get(hash_to_bytes(snapshot_id))
return converters.from_snapshot(snp)
def snapshot_get_branches(self, snapshot_id, branches_from='',
branches_count=1000, target_types=None):
snp = self.storage.snapshot_get_branches(
hash_to_bytes(snapshot_id), branches_from.encode(),
branches_count, target_types)
return converters.from_snapshot(snp)

File Metadata

Mime Type
text/x-python
Expires
Jun 4 2025, 7:24 PM (9 w, 5 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3354029

Event Timeline