Page MenuHomeSoftware Heritage
Files F9336947

No OneTemporary
Actions

View Options

diff --git a/swh/web/browse/utils.py b/swh/web/browse/utils.py
index 5d72f120..09d649e0 100644
--- a/swh/web/browse/utils.py
+++ b/swh/web/browse/utils.py
@@ -1,1104 +1,1106 @@
-# Copyright (C) 2017-2019 The Software Heritage developers
+# Copyright (C) 2017-2020 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 base64
import magic
import pypandoc
import stat
import textwrap
from collections import defaultdict
from threading import Lock
from django.core.cache import cache
from django.utils.safestring import mark_safe
from django.utils.html import escape
import sentry_sdk
from swh.model.identifiers import persistent_identifier
from swh.web.common import highlightjs, service
from swh.web.common.exc import NotFoundExc, http_status_code_message
from swh.web.common.origin_visits import get_origin_visit
from swh.web.common.utils import (
reverse, format_utc_iso_date, get_swh_persistent_id,
swh_object_icons
)
from swh.web.config import get_config
def get_directory_entries(sha1_git):
"""Function that retrieves the content of a directory
from the archive.
The directories entries are first sorted in lexicographical order.
Sub-directories and regular files are then extracted.
Args:
sha1_git: sha1_git identifier of the directory
Returns:
A tuple whose first member corresponds to the sub-directories list
and second member the regular files list
Raises:
NotFoundExc if the directory is not found
"""
cache_entry_id = 'directory_entries_%s' % sha1_git
cache_entry = cache.get(cache_entry_id)
if cache_entry:
return cache_entry
entries = list(service.lookup_directory(sha1_git))
for e in entries:
e['perms'] = stat.filemode(e['perms'])
if e['type'] == 'rev':
# modify dir entry name to explicitly show it points
# to a revision
e['name'] = '%s @ %s' % (e['name'], e['target'][:7])
dirs = [e for e in entries if e['type'] in ('dir', 'rev')]
files = [e for e in entries if e['type'] == 'file']
dirs = sorted(dirs, key=lambda d: d['name'])
files = sorted(files, key=lambda f: f['name'])
cache.set(cache_entry_id, (dirs, files))
return dirs, files
_lock = Lock()
def get_mimetype_and_encoding_for_content(content):
"""Function that returns the mime type and the encoding associated to
a content buffer using the magic module under the hood.
Args:
content (bytes): a content buffer
Returns:
A tuple (mimetype, encoding), for instance ('text/plain', 'us-ascii'),
associated to the provided content.
"""
# https://pypi.org/project/python-magic/
# packaged as python3-magic in debian buster
if hasattr(magic, 'from_buffer'):
m = magic.Magic(mime=True, mime_encoding=True)
mime_encoding = m.from_buffer(content)
mime_type, encoding = mime_encoding.split(';')
encoding = encoding.replace(' charset=', '')
# https://pypi.org/project/file-magic/
# packaged as python3-magic in debian stretch
else:
# TODO: Remove that code when production environment is upgraded
# to debian buster
# calls to the file-magic API are not thread-safe so they must
# be protected with a Lock to guarantee they will succeed
_lock.acquire()
magic_result = magic.detect_from_content(content)
_lock.release()
mime_type = magic_result.mime_type
encoding = magic_result.encoding
return mime_type, encoding
# maximum authorized content size in bytes for HTML display
# with code highlighting
content_display_max_size = get_config()['content_display_max_size']
snapshot_content_max_size = get_config()['snapshot_content_max_size']
def _re_encode_content(mimetype, encoding, content_data):
# encode textual content to utf-8 if needed
if mimetype.startswith('text/'):
# probably a malformed UTF-8 content, re-encode it
# by replacing invalid chars with a substitution one
if encoding == 'unknown-8bit':
content_data = content_data.decode('utf-8', 'replace')\
.encode('utf-8')
elif encoding not in ['utf-8', 'binary']:
content_data = content_data.decode(encoding, 'replace')\
.encode('utf-8')
elif mimetype.startswith('application/octet-stream'):
# file may detect a text content as binary
# so try to decode it for display
encodings = ['us-ascii', 'utf-8']
encodings += ['iso-8859-%s' % i for i in range(1, 17)]
for enc in encodings:
try:
content_data = content_data.decode(enc).encode('utf-8')
except Exception as exc:
sentry_sdk.capture_exception(exc)
else:
# ensure display in content view
encoding = enc
mimetype = 'text/plain'
break
return mimetype, encoding, content_data
def request_content(query_string, max_size=content_display_max_size,
raise_if_unavailable=True, re_encode=True):
"""Function that retrieves a content from the archive.
Raw bytes content is first retrieved, then the content mime type.
If the mime type is not stored in the archive, it will be computed
using Python magic module.
Args:
query_string: a string of the form "[ALGO_HASH:]HASH" where
optional ALGO_HASH can be either ``sha1``, ``sha1_git``,
``sha256``, or ``blake2s256`` (default to ``sha1``) and HASH
the hexadecimal representation of the hash value
max_size: the maximum size for a content to retrieve (default to 1MB,
no size limit if None)
Returns:
A tuple whose first member corresponds to the content raw bytes
and second member the content mime type
Raises:
NotFoundExc if the content is not found
"""
content_data = service.lookup_content(query_string)
filetype = None
language = None
license = None
# requests to the indexer db may fail so properly handle
# those cases in order to avoid content display errors
try:
filetype = service.lookup_content_filetype(query_string)
language = service.lookup_content_language(query_string)
license = service.lookup_content_license(query_string)
except Exception as exc:
sentry_sdk.capture_exception(exc)
mimetype = 'unknown'
encoding = 'unknown'
if filetype:
mimetype = filetype['mimetype']
encoding = filetype['encoding']
# workaround when encountering corrupted data due to implicit
# conversion from bytea to text in the indexer db (see T818)
# TODO: Remove that code when all data have been correctly converted
if mimetype.startswith('\\'):
filetype = None
content_data['error_code'] = 200
content_data['error_message'] = ''
content_data['error_description'] = ''
if not max_size or content_data['length'] < max_size:
try:
content_raw = service.lookup_content_raw(query_string)
except Exception as exc:
if raise_if_unavailable:
raise exc
else:
sentry_sdk.capture_exception(exc)
content_data['raw_data'] = None
content_data['error_code'] = 404
content_data['error_description'] = \
'The bytes of the content are currently not available in the archive.' # noqa
content_data['error_message'] = \
http_status_code_message[content_data['error_code']]
else:
content_data['raw_data'] = content_raw['data']
if not filetype:
mimetype, encoding = \
get_mimetype_and_encoding_for_content(content_data['raw_data']) # noqa
if re_encode:
mimetype, encoding, raw_data = _re_encode_content(
mimetype, encoding, content_data['raw_data'])
content_data['raw_data'] = raw_data
else:
content_data['raw_data'] = None
content_data['mimetype'] = mimetype
content_data['encoding'] = encoding
if language:
content_data['language'] = language['lang']
else:
content_data['language'] = 'not detected'
if license:
content_data['licenses'] = ', '.join(license['facts'][0]['licenses'])
else:
content_data['licenses'] = 'not detected'
return content_data
_browsers_supported_image_mimes = set(['image/gif', 'image/png',
'image/jpeg', 'image/bmp',
'image/webp', 'image/svg',
'image/svg+xml'])
def prepare_content_for_display(content_data, mime_type, path):
"""Function that prepares a content for HTML display.
The function tries to associate a programming language to a
content in order to perform syntax highlighting client-side
using highlightjs. The language is determined using either
the content filename or its mime type.
If the mime type corresponds to an image format supported
by web browsers, the content will be encoded in base64
for displaying the image.
Args:
content_data (bytes): raw bytes of the content
mime_type (string): mime type of the content
path (string): path of the content including filename
Returns:
A dict containing the content bytes (possibly different from the one
provided as parameter if it is an image) under the key 'content_data
and the corresponding highlightjs language class under the
key 'language'.
"""
language = highlightjs.get_hljs_language_from_filename(path)
if not language:
language = highlightjs.get_hljs_language_from_mime_type(mime_type)
if not language:
language = 'nohighlight'
elif mime_type.startswith('application/'):
mime_type = mime_type.replace('application/', 'text/')
if mime_type.startswith('image/'):
if mime_type in _browsers_supported_image_mimes:
- content_data = base64.b64encode(content_data)
- content_data = content_data.decode('utf-8')
+ content_data = base64.b64encode(content_data).decode('ascii')
else:
content_data = None
if mime_type.startswith('image/svg'):
mime_type = 'image/svg+xml'
+ if mime_type.startswith('text/'):
+ content_data = content_data.decode('utf-8', errors='replace')
+
return {'content_data': content_data,
'language': language,
'mimetype': mime_type}
def process_snapshot_branches(snapshot):
"""
Process a dictionary describing snapshot branches: extract those
targeting revisions and releases, put them in two different lists,
then sort those lists in lexicographical order of the branches' names.
Args:
snapshot_branches (dict): A dict describing the branches of a snapshot
as returned for instance by
:func:`swh.web.common.service.lookup_snapshot`
Returns:
tuple: A tuple whose first member is the sorted list of branches
targeting revisions and second member the sorted list of branches
targeting releases
"""
snapshot_branches = snapshot['branches']
branches = {}
branch_aliases = {}
releases = {}
revision_to_branch = defaultdict(set)
revision_to_release = defaultdict(set)
release_to_branch = defaultdict(set)
for branch_name, target in snapshot_branches.items():
if not target:
# FIXME: display branches with an unknown target anyway
continue
target_id = target['target']
target_type = target['target_type']
if target_type == 'revision':
branches[branch_name] = {
'name': branch_name,
'revision': target_id,
}
revision_to_branch[target_id].add(branch_name)
elif target_type == 'release':
release_to_branch[target_id].add(branch_name)
elif target_type == 'alias':
branch_aliases[branch_name] = target_id
# FIXME: handle pointers to other object types
def _enrich_release_branch(branch, release):
releases[branch] = {
'name': release['name'],
'branch_name': branch,
'date': format_utc_iso_date(release['date']),
'id': release['id'],
'message': release['message'],
'target_type': release['target_type'],
'target': release['target'],
}
def _enrich_revision_branch(branch, revision):
branches[branch].update({
'revision': revision['id'],
'directory': revision['directory'],
'date': format_utc_iso_date(revision['date']),
'message': revision['message']
})
releases_info = service.lookup_release_multiple(
release_to_branch.keys()
)
for release in releases_info:
branches_to_update = release_to_branch[release['id']]
for branch in branches_to_update:
_enrich_release_branch(branch, release)
if release['target_type'] == 'revision':
revision_to_release[release['target']].update(
branches_to_update
)
revisions = service.lookup_revision_multiple(
set(revision_to_branch.keys()) | set(revision_to_release.keys())
)
for revision in revisions:
if not revision:
continue
for branch in revision_to_branch[revision['id']]:
_enrich_revision_branch(branch, revision)
for release in revision_to_release[revision['id']]:
releases[release]['directory'] = revision['directory']
for branch_alias, branch_target in branch_aliases.items():
if branch_target in branches:
branches[branch_alias] = dict(branches[branch_target])
else:
snp = service.lookup_snapshot(snapshot['id'],
branches_from=branch_target,
branches_count=1)
if snp and branch_target in snp['branches']:
if snp['branches'][branch_target] is None:
continue
target_type = snp['branches'][branch_target]['target_type']
target = snp['branches'][branch_target]['target']
if target_type == 'revision':
branches[branch_alias] = snp['branches'][branch_target]
revision = service.lookup_revision(target)
_enrich_revision_branch(branch_alias, revision)
elif target_type == 'release':
release = service.lookup_release(target)
_enrich_release_branch(branch_alias, release)
if branch_alias in branches:
branches[branch_alias]['name'] = branch_alias
ret_branches = list(sorted(branches.values(), key=lambda b: b['name']))
ret_releases = list(sorted(releases.values(), key=lambda b: b['name']))
return ret_branches, ret_releases
def get_snapshot_content(snapshot_id):
"""Returns the lists of branches and releases
associated to a swh snapshot.
That list is put in cache in order to speedup the navigation
in the swh-web/browse ui.
.. warning:: At most 1000 branches contained in the snapshot
will be returned for performance reasons.
Args:
snapshot_id (str): hexadecimal representation of the snapshot
identifier
Returns:
A tuple with two members. The first one is a list of dict describing
the snapshot branches. The second one is a list of dict describing the
snapshot releases.
Raises:
NotFoundExc if the snapshot does not exist
"""
cache_entry_id = 'swh_snapshot_%s' % snapshot_id
cache_entry = cache.get(cache_entry_id)
if cache_entry:
return cache_entry['branches'], cache_entry['releases']
branches = []
releases = []
if snapshot_id:
snapshot = service.lookup_snapshot(
snapshot_id, branches_count=snapshot_content_max_size)
branches, releases = process_snapshot_branches(snapshot)
cache.set(cache_entry_id, {
'branches': branches,
'releases': releases,
})
return branches, releases
def get_origin_visit_snapshot(origin_info, visit_ts=None, visit_id=None,
snapshot_id=None):
"""Returns the lists of branches and releases
associated to a swh origin for a given visit.
The visit is expressed by a timestamp. In the latter case,
the closest visit from the provided timestamp will be used.
If no visit parameter is provided, it returns the list of branches
found for the latest visit.
That list is put in cache in order to speedup the navigation
in the swh-web/browse ui.
.. warning:: At most 1000 branches contained in the snapshot
will be returned for performance reasons.
Args:
origin_info (dict): a dict filled with origin information
(id, url, type)
visit_ts (int or str): an ISO date string or Unix timestamp to parse
visit_id (int): optional visit id for disambiguation in case
several visits have the same timestamp
Returns:
A tuple with two members. The first one is a list of dict describing
the origin branches for the given visit.
The second one is a list of dict describing the origin releases
for the given visit.
Raises:
NotFoundExc if the origin or its visit are not found
"""
visit_info = get_origin_visit(origin_info, visit_ts, visit_id, snapshot_id)
return get_snapshot_content(visit_info['snapshot'])
def gen_link(url, link_text=None, link_attrs=None):
"""
Utility function for generating an HTML link to insert
in Django templates.
Args:
url (str): an url
link_text (str): optional text for the produced link,
if not provided the url will be used
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="url">link_text</a>'
"""
attrs = ' '
if link_attrs:
for k, v in link_attrs.items():
attrs += '%s="%s" ' % (k, v)
if not link_text:
link_text = url
link = '<a%shref="%s">%s</a>' \
% (attrs, escape(url), escape(link_text))
return mark_safe(link)
def _snapshot_context_query_params(snapshot_context):
query_params = None
if snapshot_context and snapshot_context['origin_info']:
origin_info = snapshot_context['origin_info']
query_params = {'origin': origin_info['url']}
if 'timestamp' in snapshot_context['url_args']:
query_params['timestamp'] = \
snapshot_context['url_args']['timestamp']
if 'visit_id' in snapshot_context['query_params']:
query_params['visit_id'] = \
snapshot_context['query_params']['visit_id']
elif snapshot_context:
query_params = {'snapshot_id': snapshot_context['snapshot_id']}
return query_params
def gen_revision_url(revision_id, snapshot_context=None):
"""
Utility function for generating an url to a revision.
Args:
revision_id (str): a revision id
snapshot_context (dict): if provided, generate snapshot-dependent
browsing url
Returns:
str: The url to browse the revision
"""
query_params = _snapshot_context_query_params(snapshot_context)
return reverse('browse-revision',
url_args={'sha1_git': revision_id},
query_params=query_params)
def gen_revision_link(revision_id, shorten_id=False, snapshot_context=None,
link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a revision HTML view
to insert in Django templates.
Args:
revision_id (str): a revision id
shorten_id (boolean): whether to shorten the revision id to 7
characters for the link text
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
link_text (str): optional text for the generated link
(the revision id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
str: An HTML link in the form '<a href="revision_url">revision_id</a>'
"""
if not revision_id:
return None
revision_url = gen_revision_url(revision_id, snapshot_context)
if shorten_id:
return gen_link(revision_url, revision_id[:7], link_attrs)
else:
if not link_text:
link_text = revision_id
return gen_link(revision_url, link_text, link_attrs)
def gen_directory_link(sha1_git, snapshot_context=None, link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a directory HTML view
to insert in Django templates.
Args:
sha1_git (str): directory identifier
link_text (str): optional text for the generated link
(the directory id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="directory_view_url">link_text</a>'
"""
if not sha1_git:
return None
query_params = _snapshot_context_query_params(snapshot_context)
directory_url = reverse('browse-directory',
url_args={'sha1_git': sha1_git},
query_params=query_params)
if not link_text:
link_text = sha1_git
return gen_link(directory_url, link_text, link_attrs)
def gen_snapshot_link(snapshot_id, snapshot_context=None, link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a snapshot HTML view
to insert in Django templates.
Args:
snapshot_id (str): snapshot identifier
link_text (str): optional text for the generated link
(the snapshot id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="snapshot_view_url">link_text</a>'
"""
query_params = _snapshot_context_query_params(snapshot_context)
snapshot_url = reverse('browse-snapshot',
url_args={'snapshot_id': snapshot_id},
query_params=query_params)
if not link_text:
link_text = snapshot_id
return gen_link(snapshot_url, link_text, link_attrs)
def gen_content_link(sha1_git, snapshot_context=None, link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a content HTML view
to insert in Django templates.
Args:
sha1_git (str): content identifier
link_text (str): optional text for the generated link
(the content sha1_git will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="content_view_url">link_text</a>'
"""
if not sha1_git:
return None
query_params = _snapshot_context_query_params(snapshot_context)
content_url = reverse('browse-content',
url_args={'query_string': 'sha1_git:' + sha1_git},
query_params=query_params)
if not link_text:
link_text = sha1_git
return gen_link(content_url, link_text, link_attrs)
def get_revision_log_url(revision_id, snapshot_context=None):
"""
Utility function for getting the URL for a revision log HTML view
(possibly in the context of an origin).
Args:
revision_id (str): revision identifier the history heads to
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
Returns:
The revision log view URL
"""
query_params = {'revision': revision_id}
if snapshot_context and snapshot_context['origin_info']:
origin_info = snapshot_context['origin_info']
url_args = {'origin_url': origin_info['url']}
if 'timestamp' in snapshot_context['url_args']:
url_args['timestamp'] = \
snapshot_context['url_args']['timestamp']
if 'visit_id' in snapshot_context['query_params']:
query_params['visit_id'] = \
snapshot_context['query_params']['visit_id']
revision_log_url = reverse('browse-origin-log',
url_args=url_args,
query_params=query_params)
elif snapshot_context:
url_args = {'snapshot_id': snapshot_context['snapshot_id']}
revision_log_url = reverse('browse-snapshot-log',
url_args=url_args,
query_params=query_params)
else:
revision_log_url = reverse('browse-revision-log',
url_args={'sha1_git': revision_id})
return revision_log_url
def gen_revision_log_link(revision_id, snapshot_context=None,
link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a revision log HTML view
(possibly in the context of an origin) to insert in Django templates.
Args:
revision_id (str): revision identifier the history heads to
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
link_text (str): optional text to use for the generated link
(the revision id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form
'<a href="revision_log_view_url">link_text</a>'
"""
if not revision_id:
return None
revision_log_url = get_revision_log_url(revision_id, snapshot_context)
if not link_text:
link_text = revision_id
return gen_link(revision_log_url, link_text, link_attrs)
def gen_person_mail_link(person, link_text=None):
"""
Utility function for generating a mail link to a person to insert
in Django templates.
Args:
person (dict): dictionary containing person data
(*name*, *email*, *fullname*)
link_text (str): optional text to use for the generated mail link
(the person name will be used by default)
Returns:
str: A mail link to the person or the person name if no email is
present in person data
"""
person_name = person['name'] or person['fullname'] or 'None'
if link_text is None:
link_text = person_name
person_email = person['email'] if person['email'] else None
if person_email is None and '@' in person_name and ' ' not in person_name:
person_email = person_name
if person_email:
return gen_link(url='mailto:%s' % person_email,
link_text=link_text)
else:
return person_name
def gen_release_link(sha1_git, snapshot_context=None, link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a release HTML view
to insert in Django templates.
Args:
sha1_git (str): release identifier
link_text (str): optional text for the generated link
(the release id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="release_view_url">link_text</a>'
"""
query_params = _snapshot_context_query_params(snapshot_context)
release_url = reverse('browse-release',
url_args={'sha1_git': sha1_git},
query_params=query_params)
if not link_text:
link_text = sha1_git
return gen_link(release_url, link_text, link_attrs)
def format_log_entries(revision_log, per_page, snapshot_context=None):
"""
Utility functions that process raw revision log data for HTML display.
Its purpose is to:
* add links to relevant browse views
* format date in human readable format
* truncate the message log
Args:
revision_log (list): raw revision log as returned by the swh-web api
per_page (int): number of log entries per page
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
"""
revision_log_data = []
for i, rev in enumerate(revision_log):
if i == per_page:
break
author_name = 'None'
author_fullname = 'None'
committer_fullname = 'None'
if rev['author']:
author_name = gen_person_mail_link(rev['author'])
author_fullname = rev['author']['fullname']
if rev['committer']:
committer_fullname = rev['committer']['fullname']
author_date = format_utc_iso_date(rev['date'])
committer_date = format_utc_iso_date(rev['committer_date'])
tooltip = 'revision %s\n' % rev['id']
tooltip += 'author: %s\n' % author_fullname
tooltip += 'author date: %s\n' % author_date
tooltip += 'committer: %s\n' % committer_fullname
tooltip += 'committer date: %s\n\n' % committer_date
if rev['message']:
tooltip += textwrap.indent(rev['message'], ' '*4)
revision_log_data.append({
'author': author_name,
'id': rev['id'][:7],
'message': rev['message'],
'date': author_date,
'commit_date': committer_date,
'url': gen_revision_url(rev['id'], snapshot_context),
'tooltip': tooltip
})
return revision_log_data
def get_snapshot_context(snapshot_id=None, origin_url=None,
timestamp=None, visit_id=None):
"""
Utility function to compute relevant information when navigating
the archive in a snapshot context. The snapshot is either
referenced by its id or it will be retrieved from an origin visit.
Args:
snapshot_id (str): hexadecimal representation of a snapshot identifier,
all other parameters will be ignored if it is provided
origin_url (str): the origin_url
(e.g. https://github.com/(user)/(repo)/)
timestamp (str): a datetime string for retrieving the closest
visit of the origin
visit_id (int): optional visit id for disambiguation in case
of several visits with the same timestamp
Returns:
A dict with the following entries:
* origin_info: dict containing origin information
* visit_info: dict containing visit information
* branches: the list of branches for the origin found
during the visit
* releases: the list of releases for the origin found
during the visit
* origin_browse_url: the url to browse the origin
* origin_branches_url: the url to browse the origin branches
* origin_releases_url': the url to browse the origin releases
* origin_visit_url: the url to browse the snapshot of the origin
found during the visit
* url_args: dict containing url arguments to use when browsing in
the context of the origin and its visit
Raises:
NotFoundExc: if no snapshot is found for the visit of an origin.
"""
origin_info = None
visit_info = None
url_args = None
query_params = {}
branches = []
releases = []
browse_url = None
visit_url = None
branches_url = None
releases_url = None
swh_type = 'snapshot'
if origin_url:
swh_type = 'origin'
origin_info = service.lookup_origin({'url': origin_url})
visit_info = get_origin_visit(origin_info, timestamp, visit_id,
snapshot_id)
fmt_date = format_utc_iso_date(visit_info['date'])
visit_info['fmt_date'] = fmt_date
snapshot_id = visit_info['snapshot']
if not snapshot_id:
raise NotFoundExc('No snapshot associated to the visit of origin '
'%s on %s' % (escape(origin_url), fmt_date))
# provided timestamp is not necessarily equals to the one
# of the retrieved visit, so get the exact one in order
# use it in the urls generated below
if timestamp:
timestamp = visit_info['date']
branches, releases = \
get_origin_visit_snapshot(origin_info, timestamp, visit_id,
snapshot_id)
url_args = {'origin_url': origin_info['url']}
query_params = {'visit_id': visit_id}
browse_url = reverse('browse-origin-visits',
url_args=url_args)
if timestamp:
url_args['timestamp'] = format_utc_iso_date(timestamp,
'%Y-%m-%dT%H:%M:%S')
visit_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
visit_info['url'] = visit_url
branches_url = reverse('browse-origin-branches',
url_args=url_args,
query_params=query_params)
releases_url = reverse('browse-origin-releases',
url_args=url_args,
query_params=query_params)
elif snapshot_id:
branches, releases = get_snapshot_content(snapshot_id)
url_args = {'snapshot_id': snapshot_id}
browse_url = reverse('browse-snapshot',
url_args=url_args)
branches_url = reverse('browse-snapshot-branches',
url_args=url_args)
releases_url = reverse('browse-snapshot-releases',
url_args=url_args)
releases = list(reversed(releases))
snapshot_sizes = service.lookup_snapshot_sizes(snapshot_id)
is_empty = sum(snapshot_sizes.values()) == 0
swh_snp_id = persistent_identifier('snapshot', snapshot_id)
return {
'swh_type': swh_type,
'swh_object_id': swh_snp_id,
'snapshot_id': snapshot_id,
'snapshot_sizes': snapshot_sizes,
'is_empty': is_empty,
'origin_info': origin_info,
'visit_info': visit_info,
'branches': branches,
'releases': releases,
'branch': None,
'release': None,
'browse_url': browse_url,
'branches_url': branches_url,
'releases_url': releases_url,
'url_args': url_args,
'query_params': query_params
}
# list of common readme names ordered by preference
# (lower indices have higher priority)
_common_readme_names = [
"readme.markdown",
"readme.md",
"readme.rst",
"readme.txt",
"readme"
]
def get_readme_to_display(readmes):
"""
Process a list of readme files found in a directory
in order to find the adequate one to display.
Args:
readmes: a list of dict where keys are readme file names and values
are readme sha1s
Returns:
A tuple (readme_name, readme_sha1)
"""
readme_name = None
readme_url = None
readme_sha1 = None
readme_html = None
lc_readmes = {k.lower(): {'orig_name': k, 'sha1': v}
for k, v in readmes.items()}
# look for readme names according to the preference order
# defined by the _common_readme_names list
for common_readme_name in _common_readme_names:
if common_readme_name in lc_readmes:
readme_name = lc_readmes[common_readme_name]['orig_name']
readme_sha1 = lc_readmes[common_readme_name]['sha1']
readme_url = reverse('browse-content-raw',
url_args={'query_string': readme_sha1},
query_params={'re_encode': 'true'})
break
# otherwise pick the first readme like file if any
if not readme_name and len(readmes.items()) > 0:
readme_name = next(iter(readmes))
readme_sha1 = readmes[readme_name]
readme_url = reverse('browse-content-raw',
url_args={'query_string': readme_sha1},
query_params={'re_encode': 'true'})
# convert rst README to html server side as there is
# no viable solution to perform that task client side
if readme_name and readme_name.endswith('.rst'):
cache_entry_id = 'readme_%s' % readme_sha1
cache_entry = cache.get(cache_entry_id)
if cache_entry:
readme_html = cache_entry
else:
try:
rst_doc = request_content(readme_sha1)
readme_html = pypandoc.convert_text(rst_doc['raw_data'],
'html', format='rst')
cache.set(cache_entry_id, readme_html)
except Exception as exc:
sentry_sdk.capture_exception(exc)
readme_html = 'Readme bytes are not available'
return readme_name, readme_url, readme_html
def get_swh_persistent_ids(swh_objects, snapshot_context=None):
"""
Returns a list of dict containing info related to persistent
identifiers of swh objects.
Args:
swh_objects (list): a list of dict with the following keys:
* type: swh object type
(content/directory/release/revision/snapshot)
* id: swh object id
snapshot_context (dict): optional parameter describing the snapshot in
which the object has been found
Returns:
list: a list of dict with the following keys:
* object_type: the swh object type
(content/directory/release/revision/snapshot)
* object_icon: the swh object icon to use in HTML views
* swh_id: the computed swh object persistent identifier
* swh_id_url: the url resolving the persistent identifier
* show_options: boolean indicating if the persistent id options
must be displayed in persistent ids HTML view
"""
swh_ids = []
for swh_object in swh_objects:
if not swh_object['id']:
continue
swh_id = get_swh_persistent_id(swh_object['type'], swh_object['id'])
show_options = swh_object['type'] == 'content' or \
(snapshot_context and snapshot_context['origin_info'] is not None)
object_icon = swh_object_icons[swh_object['type']]
swh_ids.append({
'object_type': swh_object['type'],
'object_id': swh_object['id'],
'object_icon': object_icon,
'swh_id': swh_id,
'swh_id_url': reverse('browse-swh-id',
url_args={'swh_id': swh_id}),
'show_options': show_options
})
return swh_ids
diff --git a/swh/web/tests/browse/views/test_content.py b/swh/web/tests/browse/views/test_content.py
index 1904a2c6..42b9c5ce 100644
--- a/swh/web/tests/browse/views/test_content.py
+++ b/swh/web/tests/browse/views/test_content.py
@@ -1,382 +1,386 @@
-# Copyright (C) 2017-2019 The Software Heritage developers
+# Copyright (C) 2017-2020 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
from django.utils.html import escape
from hypothesis import given
from swh.web.browse.utils import (
get_mimetype_and_encoding_for_content, prepare_content_for_display,
_re_encode_content
)
from swh.web.common.exc import NotFoundExc
from swh.web.common.utils import reverse, get_swh_persistent_id
from swh.web.common.utils import gen_path_info
from swh.web.tests.django_asserts import (
assert_contains, assert_not_contains, assert_template_used
)
from swh.web.tests.strategies import (
content, content_text_non_utf8, content_text_no_highlight,
content_image_type, content_text, invalid_sha1, unknown_content,
content_utf8_detected_as_binary
)
@given(content_text())
def test_content_view_text(client, archive_data, content):
sha1_git = content['sha1_git']
url = reverse('browse-content',
url_args={'query_string': content['sha1']},
query_params={'path': content['path']})
url_raw = reverse('browse-content-raw',
url_args={'query_string': content['sha1']})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
mimetype = content_display['mimetype']
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
if mimetype.startswith('text/'):
assert_contains(resp, '<code class="%s">' %
content_display['language'])
assert_contains(resp, escape(content_display['content_data']))
assert_contains(resp, url_raw)
swh_cnt_id = get_swh_persistent_id('content', sha1_git)
swh_cnt_id_url = reverse('browse-swh-id',
url_args={'swh_id': swh_cnt_id})
assert_contains(resp, swh_cnt_id)
assert_contains(resp, swh_cnt_id_url)
@given(content_text_no_highlight())
def test_content_view_text_no_highlight(client, archive_data, content):
sha1_git = content['sha1_git']
url = reverse('browse-content',
url_args={'query_string': content['sha1']})
url_raw = reverse('browse-content-raw',
url_args={'query_string': content['sha1']})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
assert_contains(resp, '<code class="nohighlight">')
assert_contains(resp, escape(content_display['content_data']))
assert_contains(resp, url_raw)
swh_cnt_id = get_swh_persistent_id('content', sha1_git)
swh_cnt_id_url = reverse('browse-swh-id',
url_args={'swh_id': swh_cnt_id})
assert_contains(resp, swh_cnt_id)
assert_contains(resp, swh_cnt_id_url)
@given(content_text_non_utf8())
def test_content_view_no_utf8_text(client, archive_data, content):
sha1_git = content['sha1_git']
url = reverse('browse-content',
url_args={'query_string': content['sha1']})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
swh_cnt_id = get_swh_persistent_id('content', sha1_git)
swh_cnt_id_url = reverse('browse-swh-id',
url_args={'swh_id': swh_cnt_id})
assert_contains(resp, swh_cnt_id_url)
assert_contains(resp, escape(content_display['content_data']))
@given(content_image_type())
def test_content_view_image(client, archive_data, content):
url = reverse('browse-content',
url_args={'query_string': content['sha1']})
url_raw = reverse('browse-content-raw',
url_args={'query_string': content['sha1']})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
mimetype = content_display['mimetype']
content_data = content_display['content_data']
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
assert_contains(resp, '<img src="data:%s;base64,%s"/>'
% (mimetype, content_data))
assert_contains(resp, url_raw)
@given(content_text())
def test_content_view_text_with_path(client, archive_data, content):
path = content['path']
url = reverse('browse-content',
url_args={'query_string': content['sha1']},
query_params={'path': path})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
assert_contains(resp, '<nav class="bread-crumbs')
content_display = _process_content_for_display(archive_data, content)
mimetype = content_display['mimetype']
if mimetype.startswith('text/'):
hljs_language = content['hljs_language']
assert_contains(resp, '<code class="%s">' % hljs_language)
assert_contains(resp, escape(content_display['content_data']))
split_path = path.split('/')
root_dir_sha1 = split_path[0]
filename = split_path[-1]
path = path.replace(root_dir_sha1 + '/', '').replace(filename, '')
path_info = gen_path_info(path)
root_dir_url = reverse('browse-directory',
url_args={'sha1_git': root_dir_sha1})
assert_contains(resp, '<li class="swh-path">',
count=len(path_info)+1)
assert_contains(resp, '<a href="' + root_dir_url + '">' +
root_dir_sha1[:7] + '</a>')
for p in path_info:
dir_url = reverse('browse-directory',
url_args={'sha1_git': root_dir_sha1,
'path': p['path']})
assert_contains(resp, '<a href="' + dir_url + '">' +
p['name'] + '</a>')
assert_contains(resp, '<li>' + filename + '</li>')
url_raw = reverse('browse-content-raw',
url_args={'query_string': content['sha1']},
query_params={'filename': filename})
assert_contains(resp, url_raw)
url = reverse('browse-content',
url_args={'query_string': content['sha1']},
query_params={'path': filename})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
assert_not_contains(resp, '<nav class="bread-crumbs')
invalid_path = '%s/foo/bar/baz' % root_dir_sha1
url = reverse('browse-content',
url_args={'query_string': content['sha1']},
query_params={'path': invalid_path})
resp = client.get(url)
assert resp.status_code == 404
@given(content_text())
def test_content_raw_text(client, archive_data, content):
url = reverse('browse-content-raw',
url_args={'query_string': content['sha1']})
resp = client.get(url)
content_data = archive_data.content_get(content['sha1'])['data']
assert resp.status_code == 200
assert resp['Content-Type'] == 'text/plain'
assert resp['Content-disposition'] == ('filename=%s_%s' %
('sha1', content['sha1']))
assert resp.content == content_data
filename = content['path'].split('/')[-1]
url = reverse('browse-content-raw',
url_args={'query_string': content['sha1']},
query_params={'filename': filename})
resp = client.get(url)
assert resp.status_code == 200
assert resp['Content-Type'] == 'text/plain'
assert resp['Content-disposition'] == 'filename=%s' % filename
assert resp.content == content_data
@given(content_text_non_utf8())
def test_content_raw_no_utf8_text(client, content):
url = reverse('browse-content-raw',
url_args={'query_string': content['sha1']})
resp = client.get(url)
assert resp.status_code == 200
_, encoding = get_mimetype_and_encoding_for_content(resp.content)
assert encoding == content['encoding']
@given(content_image_type())
def test_content_raw_bin(client, archive_data, content):
url = reverse('browse-content-raw',
url_args={'query_string': content['sha1']})
resp = client.get(url)
filename = content['path'].split('/')[-1]
content_data = archive_data.content_get(content['sha1'])['data']
assert resp.status_code == 200
assert resp['Content-Type'] == 'application/octet-stream'
assert resp['Content-disposition'] == \
'attachment; filename=%s_%s' % \
('sha1', content['sha1'])
assert resp.content == content_data
url = reverse('browse-content-raw',
url_args={'query_string': content['sha1']},
query_params={'filename': filename})
resp = client.get(url)
assert resp.status_code == 200
assert resp['Content-Type'] == 'application/octet-stream'
assert resp['Content-disposition'] == \
'attachment; filename=%s' % filename
assert resp.content == content_data
@given(invalid_sha1(), unknown_content())
def test_content_request_errors(client, invalid_sha1, unknown_content):
url = reverse('browse-content',
url_args={'query_string': invalid_sha1})
resp = client.get(url)
assert resp.status_code == 400
assert_template_used(resp, 'error.html')
url = reverse('browse-content',
url_args={'query_string': unknown_content['sha1']})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
@given(content())
def test_content_bytes_missing(client, archive_data, mocker, content):
mock_service = mocker.patch('swh.web.browse.utils.service')
content_data = archive_data.content_get_metadata(content['sha1'])
content_data['data'] = None
mock_service.lookup_content.return_value = content_data
mock_service.lookup_content_filetype.side_effect = Exception()
mock_service.lookup_content_raw.side_effect = NotFoundExc(
'Content bytes not available!')
url = reverse('browse-content',
url_args={'query_string': content['sha1']})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'browse/content.html')
def test_content_too_large(client, mocker):
mock_request_content = mocker.patch(
'swh.web.browse.views.content.request_content')
stub_content_too_large_data = {
'checksums': {
'sha1': '8624bcdae55baeef00cd11d5dfcfa60f68710a02',
'sha1_git': '94a9ed024d3859793618152ea559a168bbcbb5e2',
'sha256': ('8ceb4b9ee5adedde47b31e975c1d90c73ad27b6b16'
'5a1dcd80c7c545eb65b903'),
'blake2s256': ('38702b7168c7785bfe748b51b45d9856070ba90'
'f9dc6d90f2ea75d4356411ffe')
},
'length': 30000000,
'raw_data': None,
'mimetype': 'text/plain',
'encoding': 'us-ascii',
'language': 'not detected',
'licenses': 'GPL',
'error_code': 200,
'error_message': '',
'error_description': ''
}
content_sha1 = stub_content_too_large_data['checksums']['sha1']
mock_request_content.return_value = stub_content_too_large_data
url = reverse('browse-content',
url_args={'query_string': content_sha1})
url_raw = reverse('browse-content-raw',
url_args={'query_string': content_sha1})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
assert_contains(resp, 'Content is too large to be displayed')
assert_contains(resp, url_raw)
@given(content())
def test_content_uppercase(client, content):
url = reverse('browse-content-uppercase-checksum',
url_args={'query_string': content['sha1'].upper()})
resp = client.get(url)
assert resp.status_code == 302
redirect_url = reverse('browse-content',
url_args={'query_string': content['sha1']})
assert resp['location'] == redirect_url
@given(content_utf8_detected_as_binary())
def test_content_utf8_detected_as_binary_display(client, archive_data,
content):
url = reverse('browse-content',
url_args={'query_string': content['sha1']})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
assert_contains(resp, escape(content_display['content_data']))
def _process_content_for_display(archive_data, content):
content_data = archive_data.content_get(content['sha1'])
mime_type, encoding = get_mimetype_and_encoding_for_content(
content_data['data'])
mime_type, encoding, content_data = _re_encode_content(
mime_type, encoding, content_data['data'])
- return prepare_content_for_display(content_data, mime_type,
- content['path'])
+ content_display = prepare_content_for_display(content_data, mime_type,
+ content['path'])
+
+ assert type(content_display['content_data']) == str
+
+ return content_display
diff --git a/swh/web/tests/browse/views/test_origin.py b/swh/web/tests/browse/views/test_origin.py
index 26bed5cd..8518430c 100644
--- a/swh/web/tests/browse/views/test_origin.py
+++ b/swh/web/tests/browse/views/test_origin.py
@@ -1,902 +1,904 @@
-# Copyright (C) 2017-2019 The Software Heritage developers
+# Copyright (C) 2017-2020 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
import re
import swh.web.browse.utils
from django.utils.html import escape
from hypothesis import given
from swh.model.hashutil import hash_to_bytes
from swh.web.browse.utils import process_snapshot_branches
from swh.web.common.exc import NotFoundExc
from swh.web.common.utils import (
reverse, gen_path_info, format_utc_iso_date,
parse_timestamp, get_swh_persistent_id
)
from swh.web.tests.data import get_content
from swh.web.tests.django_asserts import assert_contains, assert_template_used
from swh.web.tests.strategies import (
origin, origin_with_multiple_visits, new_origin,
new_snapshot, visit_dates, revisions, origin_with_releases
)
@given(origin_with_multiple_visits())
def test_origin_visits_browse(client, archive_data, origin):
url = reverse('browse-origin-visits',
url_args={'origin_url': origin['url']})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/origin-visits.html')
url = reverse('browse-origin-visits',
url_args={'origin_url': origin['url']})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/origin-visits.html')
visits = archive_data.origin_visit_get(origin['url'])
for v in visits:
vdate = format_utc_iso_date(v['date'], '%Y-%m-%dT%H:%M:%SZ')
browse_dir_url = reverse('browse-origin-directory',
url_args={'origin_url': origin['url'],
'timestamp': vdate})
assert_contains(resp, browse_dir_url)
@given(origin_with_multiple_visits())
def test_origin_content_view(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin['url'])
def _get_archive_data(visit_idx):
snapshot = archive_data.snapshot_get(
origin_visits[visit_idx]['snapshot'])
head_rev_id = archive_data.snapshot_get_head(snapshot)
head_rev = archive_data.revision_get(head_rev_id)
dir_content = archive_data.directory_ls(head_rev['directory'])
dir_files = [e for e in dir_content if e['type'] == 'file']
dir_file = random.choice(dir_files)
branches, releases = process_snapshot_branches(snapshot)
return {
'branches': branches,
'releases': releases,
'root_dir_sha1': head_rev['directory'],
'content': get_content(dir_file['checksums']['sha1']),
'visit': origin_visits[visit_idx]
}
tdata = _get_archive_data(-1)
_origin_content_view_test_helper(client, origin, origin_visits,
tdata['branches'],
tdata['releases'],
tdata['root_dir_sha1'],
tdata['content'])
_origin_content_view_test_helper(client, origin, origin_visits,
tdata['branches'],
tdata['releases'],
tdata['root_dir_sha1'],
tdata['content'],
timestamp=tdata['visit']['date'])
visit_unix_ts = parse_timestamp(tdata['visit']['date']).timestamp()
visit_unix_ts = int(visit_unix_ts)
_origin_content_view_test_helper(client, origin, origin_visits,
tdata['branches'],
tdata['releases'],
tdata['root_dir_sha1'],
tdata['content'],
timestamp=visit_unix_ts)
tdata = _get_archive_data(0)
_origin_content_view_test_helper(client, origin, origin_visits,
tdata['branches'],
tdata['releases'],
tdata['root_dir_sha1'],
tdata['content'],
visit_id=tdata['visit']['visit'])
@given(origin())
def test_origin_root_directory_view(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin['url'])
visit = origin_visits[-1]
snapshot = archive_data.snapshot_get(visit['snapshot'])
head_rev_id = archive_data.snapshot_get_head(snapshot)
head_rev = archive_data.revision_get(head_rev_id)
root_dir_sha1 = head_rev['directory']
dir_content = archive_data.directory_ls(root_dir_sha1)
branches, releases = process_snapshot_branches(snapshot)
visit_unix_ts = parse_timestamp(visit['date']).timestamp()
visit_unix_ts = int(visit_unix_ts)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, dir_content)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, dir_content,
visit_id=visit['visit'])
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, dir_content,
timestamp=visit_unix_ts)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, dir_content,
timestamp=visit['date'])
origin = dict(origin)
del origin['type']
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, dir_content)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, dir_content,
visit_id=visit['visit'])
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, dir_content,
timestamp=visit_unix_ts)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, dir_content,
timestamp=visit['date'])
@given(origin())
def test_origin_sub_directory_view(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin['url'])
visit = origin_visits[-1]
snapshot = archive_data.snapshot_get(visit['snapshot'])
head_rev_id = archive_data.snapshot_get_head(snapshot)
head_rev = archive_data.revision_get(head_rev_id)
root_dir_sha1 = head_rev['directory']
subdirs = [e for e in archive_data.directory_ls(root_dir_sha1)
if e['type'] == 'dir']
branches, releases = process_snapshot_branches(snapshot)
visit_unix_ts = parse_timestamp(visit['date']).timestamp()
visit_unix_ts = int(visit_unix_ts)
if len(subdirs) == 0:
return
subdir = random.choice(subdirs)
subdir_content = archive_data.directory_ls(subdir['target'])
subdir_path = subdir['name']
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, subdir_content,
path=subdir_path)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, subdir_content,
path=subdir_path,
visit_id=visit['visit'])
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, subdir_content,
path=subdir_path,
timestamp=visit_unix_ts)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, subdir_content,
path=subdir_path,
timestamp=visit['date'])
origin = dict(origin)
del origin['type']
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, subdir_content,
path=subdir_path)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, subdir_content,
path=subdir_path,
visit_id=visit['visit'])
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, subdir_content,
path=subdir_path,
timestamp=visit_unix_ts)
_origin_directory_view_test_helper(client, origin, origin_visits, branches,
releases, root_dir_sha1, subdir_content,
path=subdir_path,
timestamp=visit['date'])
@given(origin())
def test_origin_branches(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin['url'])
visit = origin_visits[-1]
snapshot = archive_data.snapshot_get(visit['snapshot'])
snapshot_content = process_snapshot_branches(snapshot)
_origin_branches_test_helper(client, origin, snapshot_content)
origin = dict(origin)
origin['type'] = None
_origin_branches_test_helper(client, origin, snapshot_content)
@given(origin())
def test_origin_releases(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin['url'])
visit = origin_visits[-1]
snapshot = archive_data.snapshot_get(visit['snapshot'])
snapshot_content = process_snapshot_branches(snapshot)
_origin_releases_test_helper(client, origin, snapshot_content)
origin = dict(origin)
origin['type'] = None
_origin_releases_test_helper(client, origin, snapshot_content)
@given(new_origin(), new_snapshot(min_size=4, max_size=4), visit_dates(),
revisions(min_size=3, max_size=3))
def test_origin_snapshot_null_branch(client, archive_data, new_origin,
new_snapshot, visit_dates, revisions):
snp_dict = new_snapshot.to_dict()
new_origin = archive_data.origin_add([new_origin])[0]
for i, branch in enumerate(snp_dict['branches'].keys()):
if i == 0:
snp_dict['branches'][branch] = None
else:
snp_dict['branches'][branch] = {
'target_type': 'revision',
'target': hash_to_bytes(revisions[i-1]),
}
archive_data.snapshot_add([snp_dict])
visit = archive_data.origin_visit_add(
new_origin['url'], visit_dates[0], type='git')
archive_data.origin_visit_update(new_origin['url'], visit['visit'],
status='partial',
snapshot=snp_dict['id'])
url = reverse('browse-origin-directory',
url_args={'origin_url': new_origin['url']})
rv = client.get(url)
assert rv.status_code == 200
@given(new_origin(), new_snapshot(min_size=4, max_size=4), visit_dates(),
revisions(min_size=4, max_size=4))
def test_origin_snapshot_invalid_branch(client, archive_data, new_origin,
new_snapshot, visit_dates, revisions):
snp_dict = new_snapshot.to_dict()
new_origin = archive_data.origin_add([new_origin])[0]
for i, branch in enumerate(snp_dict['branches'].keys()):
snp_dict['branches'][branch] = {
'target_type': 'revision',
'target': hash_to_bytes(revisions[i]),
}
archive_data.snapshot_add([snp_dict])
visit = archive_data.origin_visit_add(
new_origin['url'], visit_dates[0], type='git')
archive_data.origin_visit_update(new_origin['url'], visit['visit'],
status='full',
snapshot=snp_dict['id'])
url = reverse('browse-origin-directory',
url_args={'origin_url': new_origin['url']},
query_params={'branch': 'invalid_branch'})
rv = client.get(url)
assert rv.status_code == 404
def test_origin_request_errors(client, archive_data, mocker):
mock_snapshot_service = mocker.patch(
'swh.web.browse.views.utils.snapshot_context.service')
mock_origin_service = mocker.patch('swh.web.browse.views.origin.service')
mock_utils_service = mocker.patch('swh.web.browse.utils.service')
mock_get_origin_visit_snapshot = mocker.patch(
'swh.web.browse.utils.get_origin_visit_snapshot')
mock_get_origin_visits = mocker.patch(
'swh.web.common.origin_visits.get_origin_visits')
mock_request_content = mocker.patch(
'swh.web.browse.views.utils.snapshot_context.request_content')
mock_origin_service.lookup_origin.side_effect = NotFoundExc(
'origin not found')
url = reverse('browse-origin-visits',
url_args={'origin_url': 'bar'})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
assert_contains(resp, 'origin not found', status_code=404)
mock_origin_service.lookup_origin.side_effect = None
mock_origin_service.lookup_origin.return_value = {'type': 'foo',
'url': 'bar',
'id': 457}
mock_get_origin_visits.return_value = []
url = reverse('browse-origin-directory',
url_args={'origin_url': 'bar'})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
assert_contains(resp, "No visit", status_code=404)
mock_get_origin_visits.return_value = [{'visit': 1}]
mock_get_origin_visit_snapshot.side_effect = NotFoundExc('visit not found')
url = reverse('browse-origin-directory',
url_args={'origin_url': 'bar'},
query_params={'visit_id': 2})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
assert re.search('Visit.*not found', resp.content.decode('utf-8'))
mock_get_origin_visits.return_value = [{
'date': '2015-09-26T09:30:52.373449+00:00',
'metadata': {},
'origin': 457,
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65',
'status': 'full',
'visit': 1
}]
mock_get_origin_visit_snapshot.side_effect = None
mock_get_origin_visit_snapshot.return_value = (
[{'directory': 'ae59ceecf46367e8e4ad800e231fc76adc3afffb',
'name': 'HEAD',
'revision': '7bc08e1aa0b08cb23e18715a32aa38517ad34672',
'date': '04 May 2017, 13:27 UTC',
'message': ''}],
[]
)
mock_utils_service.lookup_snapshot_sizes.return_value = {
'revision': 1,
'release': 0
}
mock_lookup_directory = mock_utils_service.lookup_directory
mock_lookup_directory.side_effect = NotFoundExc('Directory not found')
url = reverse('browse-origin-directory',
url_args={'origin_url': 'bar'})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
assert_contains(resp, 'Directory not found', status_code=404)
mock_origin_service.lookup_origin.side_effect = None
mock_origin_service.lookup_origin.return_value = {'type': 'foo',
'url': 'bar',
'id': 457}
mock_get_origin_visits.return_value = []
url = reverse('browse-origin-content',
url_args={'origin_url': 'bar',
'path': 'foo'})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
assert_contains(resp, "No visit", status_code=404)
mock_get_origin_visits.return_value = [{'visit': 1}]
mock_get_origin_visit_snapshot.side_effect = NotFoundExc('visit not found')
url = reverse('browse-origin-content',
url_args={'origin_url': 'bar',
'path': 'foo'},
query_params={'visit_id': 2})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
assert re.search('Visit.*not found', resp.content.decode('utf-8'))
mock_get_origin_visits.return_value = [{
'date': '2015-09-26T09:30:52.373449+00:00',
'metadata': {},
'origin': 457,
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65',
'status': 'full',
'type': 'git',
'visit': 1
}]
mock_get_origin_visit_snapshot.side_effect = None
mock_get_origin_visit_snapshot.return_value = ([], [])
mock_utils_service.lookup_snapshot_sizes.return_value = {
'revision': 0,
'release': 0
}
mock_utils_service.lookup_origin.return_value = {'type': 'foo',
'url': 'bar',
'id': 457}
url = reverse('browse-origin-content',
url_args={'origin_url': 'bar',
'path': 'baz'})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
assert re.search('snapshot.*is empty', resp.content.decode('utf-8'))
mock_get_origin_visit_snapshot.return_value = (
[{'directory': 'ae59ceecf46367e8e4ad800e231fc76adc3afffb',
'name': 'HEAD',
'revision': '7bc08e1aa0b08cb23e18715a32aa38517ad34672',
'date': '04 May 2017, 13:27 UTC',
'message': ''}],
[]
)
mock_utils_service.lookup_snapshot_sizes.return_value = {
'revision': 1,
'release': 0
}
mock_snapshot_service.lookup_directory_with_path.return_value = {
'target': '5ecd9f37b7a2d2e9980d201acd6286116f2ba1f1'
}
mock_request_content.side_effect = NotFoundExc('Content not found')
url = reverse('browse-origin-content',
url_args={'origin_url': 'bar',
'path': 'baz'})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
assert_contains(resp, 'Content not found', status_code=404)
mock_get_snapshot_context = mocker.patch(
'swh.web.browse.views.utils.snapshot_context.get_snapshot_context')
mock_get_snapshot_context.side_effect = NotFoundExc('Snapshot not found')
url = reverse('browse-origin-directory',
url_args={'origin_url': 'bar'})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, 'error.html')
assert_contains(resp, 'Snapshot not found', status_code=404)
def test_origin_empty_snapshot(client, mocker):
mock_utils_service = mocker.patch('swh.web.browse.utils.service')
mock_get_origin_visit_snapshot = mocker.patch(
'swh.web.browse.utils.get_origin_visit_snapshot')
mock_get_origin_visits = mocker.patch(
'swh.web.common.origin_visits.get_origin_visits')
mock_get_origin_visits.return_value = [{
'date': '2015-09-26T09:30:52.373449+00:00',
'metadata': {},
'origin': 457,
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65',
'status': 'full',
'type': 'git',
'visit': 1
}]
mock_get_origin_visit_snapshot.return_value = ([], [])
mock_utils_service.lookup_snapshot_sizes.return_value = {
'revision': 0,
'release': 0
}
mock_utils_service.lookup_origin.return_value = {
'id': 457,
'url': 'https://github.com/foo/bar'
}
url = reverse('browse-origin-directory',
url_args={'origin_url': 'bar'})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/directory.html')
resp_content = resp.content.decode('utf-8')
assert re.search('snapshot.*is empty', resp_content)
assert not re.search('swh-tr-link', resp_content)
@given(origin_with_releases())
def test_origin_release_browse(client, archive_data, origin):
# for swh.web.browse.utils.get_snapshot_content to only return one branch
snapshot_max_size = swh.web.browse.utils.snapshot_content_max_size
swh.web.browse.utils.snapshot_content_max_size = 1
try:
snapshot = archive_data.snapshot_get_latest(origin['url'])
release = [b for b in snapshot['branches'].values()
if b['target_type'] == 'release'][-1]
release_data = archive_data.release_get(release['target'])
url = reverse('browse-origin-directory',
url_args={'origin_url': origin['url']},
query_params={'release': release_data['name']})
resp = client.get(url)
assert resp.status_code == 200
assert_contains(resp, release_data['name'])
assert_contains(resp, release['target'])
finally:
swh.web.browse.utils.snapshot_content_max_size = snapshot_max_size
@given(origin_with_releases())
def test_origin_release_browse_not_found(client, archive_data, origin):
invalid_release_name = 'swh-foo-bar'
url = reverse('browse-origin-directory',
url_args={'origin_url': origin['url']},
query_params={'release': invalid_release_name})
resp = client.get(url)
assert resp.status_code == 404
assert re.search(f'Release {invalid_release_name}.*not found',
resp.content.decode('utf-8'))
def _origin_content_view_test_helper(client, origin_info, origin_visits,
origin_branches, origin_releases,
root_dir_sha1, content,
visit_id=None, timestamp=None):
content_path = '/'.join(content['path'].split('/')[1:])
url_args = {'origin_url': origin_info['url'],
'path': content_path}
if not visit_id:
visit_id = origin_visits[-1]['visit']
query_params = {}
if timestamp:
url_args['timestamp'] = timestamp
if visit_id:
query_params['visit_id'] = visit_id
url = reverse('browse-origin-content',
url_args=url_args,
query_params=query_params)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
+ assert type(content['data']) == str
+
assert_contains(resp, '<code class="%s">' %
content['hljs_language'])
assert_contains(resp, escape(content['data']))
split_path = content_path.split('/')
filename = split_path[-1]
path = content_path.replace(filename, '')[:-1]
path_info = gen_path_info(path)
del url_args['path']
if timestamp:
url_args['timestamp'] = format_utc_iso_date(
parse_timestamp(timestamp).isoformat(), '%Y-%m-%dT%H:%M:%S')
root_dir_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<li class="swh-path">',
count=len(path_info)+1)
assert_contains(resp, '<a href="%s">%s</a>' %
(root_dir_url, root_dir_sha1[:7]))
for p in path_info:
url_args['path'] = p['path']
dir_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<a href="%s">%s</a>' %
(dir_url, p['name']))
assert_contains(resp, '<li>%s</li>' % filename)
query_string = 'sha1_git:' + content['sha1_git']
url_raw = reverse('browse-content-raw',
url_args={'query_string': query_string},
query_params={'filename': filename})
assert_contains(resp, url_raw)
if 'args' in url_args:
del url_args['path']
origin_branches_url = reverse('browse-origin-branches',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<a href="%s">Branches (%s)</a>' %
(origin_branches_url, len(origin_branches)))
origin_releases_url = reverse('browse-origin-releases',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<a href="%s">Releases (%s)</a>' %
(origin_releases_url, len(origin_releases)))
assert_contains(resp, '<li class="swh-branch">',
count=len(origin_branches))
url_args['path'] = content_path
for branch in origin_branches:
query_params['branch'] = branch['name']
root_dir_branch_url = reverse('browse-origin-content',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<a href="%s">' % root_dir_branch_url)
assert_contains(resp, '<li class="swh-release">',
count=len(origin_releases))
query_params['branch'] = None
for release in origin_releases:
query_params['release'] = release['name']
root_dir_release_url = reverse('browse-origin-content',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<a href="%s">' % root_dir_release_url)
url = reverse('browse-origin-content',
url_args=url_args,
query_params=query_params)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/content.html')
swh_cnt_id = get_swh_persistent_id('content', content['sha1_git'])
swh_cnt_id_url = reverse('browse-swh-id',
url_args={'swh_id': swh_cnt_id})
assert_contains(resp, swh_cnt_id)
assert_contains(resp, swh_cnt_id_url)
assert_contains(resp, 'swh-take-new-snapshot')
def _origin_directory_view_test_helper(client, origin_info, origin_visits,
origin_branches, origin_releases,
root_directory_sha1, directory_entries,
visit_id=None, timestamp=None,
path=None):
dirs = [e for e in directory_entries
if e['type'] in ('dir', 'rev')]
files = [e for e in directory_entries
if e['type'] == 'file']
if not visit_id:
visit_id = origin_visits[-1]['visit']
url_args = {'origin_url': origin_info['url']}
query_params = {}
if timestamp:
url_args['timestamp'] = timestamp
else:
query_params['visit_id'] = visit_id
if path:
url_args['path'] = path
url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/directory.html')
assert resp.status_code == 200
assert_template_used(resp, 'browse/directory.html')
assert_contains(resp, '<td class="swh-directory">',
count=len(dirs))
assert_contains(resp, '<td class="swh-content">',
count=len(files))
if timestamp:
url_args['timestamp'] = format_utc_iso_date(
parse_timestamp(timestamp).isoformat(), '%Y-%m-%dT%H:%M:%S')
for d in dirs:
if d['type'] == 'rev':
dir_url = reverse('browse-revision',
url_args={'sha1_git': d['target']})
else:
dir_path = d['name']
if path:
dir_path = "%s/%s" % (path, d['name'])
dir_url_args = dict(url_args)
dir_url_args['path'] = dir_path
dir_url = reverse('browse-origin-directory',
url_args=dir_url_args,
query_params=query_params)
assert_contains(resp, dir_url)
for f in files:
file_path = f['name']
if path:
file_path = "%s/%s" % (path, f['name'])
file_url_args = dict(url_args)
file_url_args['path'] = file_path
file_url = reverse('browse-origin-content',
url_args=file_url_args,
query_params=query_params)
assert_contains(resp, file_url)
if 'path' in url_args:
del url_args['path']
root_dir_branch_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
nb_bc_paths = 1
if path:
nb_bc_paths = len(path.split('/')) + 1
assert_contains(resp, '<li class="swh-path">', count=nb_bc_paths)
assert_contains(resp, '<a href="%s">%s</a>' %
(root_dir_branch_url,
root_directory_sha1[:7]))
origin_branches_url = reverse('browse-origin-branches',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<a href="%s">Branches (%s)</a>' %
(origin_branches_url, len(origin_branches)))
origin_releases_url = reverse('browse-origin-releases',
url_args=url_args,
query_params=query_params)
nb_releases = len(origin_releases)
if nb_releases > 0:
assert_contains(resp, '<a href="%s">Releases (%s)</a>' %
(origin_releases_url, nb_releases))
if path:
url_args['path'] = path
assert_contains(resp, '<li class="swh-branch">',
count=len(origin_branches))
for branch in origin_branches:
query_params['branch'] = branch['name']
root_dir_branch_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<a href="%s">' % root_dir_branch_url)
assert_contains(resp, '<li class="swh-release">',
count=len(origin_releases))
query_params['branch'] = None
for release in origin_releases:
query_params['release'] = release['name']
root_dir_release_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
assert_contains(resp, '<a href="%s">' % root_dir_release_url)
assert_contains(resp, 'vault-cook-directory')
assert_contains(resp, 'vault-cook-revision')
swh_dir_id = get_swh_persistent_id('directory', directory_entries[0]['dir_id']) # noqa
swh_dir_id_url = reverse('browse-swh-id',
url_args={'swh_id': swh_dir_id})
assert_contains(resp, swh_dir_id)
assert_contains(resp, swh_dir_id_url)
assert_contains(resp, 'swh-take-new-snapshot')
def _origin_branches_test_helper(client, origin_info, origin_snapshot):
url_args = {'origin_url': origin_info['url']}
url = reverse('browse-origin-branches',
url_args=url_args)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/branches.html')
origin_branches = origin_snapshot[0]
origin_releases = origin_snapshot[1]
origin_branches_url = reverse('browse-origin-branches',
url_args=url_args)
assert_contains(resp, '<a href="%s">Branches (%s)</a>' %
(origin_branches_url, len(origin_branches)))
origin_releases_url = reverse('browse-origin-releases',
url_args=url_args)
nb_releases = len(origin_releases)
if nb_releases > 0:
assert_contains(resp, '<a href="%s">Releases (%s)</a>' %
(origin_releases_url, nb_releases))
assert_contains(resp, '<tr class="swh-branch-entry',
count=len(origin_branches))
for branch in origin_branches:
browse_branch_url = reverse(
'browse-origin-directory',
url_args={'origin_url': origin_info['url']},
query_params={'branch': branch['name']})
assert_contains(resp, '<a href="%s">' %
escape(browse_branch_url))
browse_revision_url = reverse(
'browse-revision',
url_args={'sha1_git': branch['revision']},
query_params={'origin': origin_info['url']})
assert_contains(resp, '<a href="%s">' %
escape(browse_revision_url))
def _origin_releases_test_helper(client, origin_info, origin_snapshot):
url_args = {'origin_url': origin_info['url']}
url = reverse('browse-origin-releases',
url_args=url_args)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, 'browse/releases.html')
origin_branches = origin_snapshot[0]
origin_releases = origin_snapshot[1]
origin_branches_url = reverse('browse-origin-branches',
url_args=url_args)
assert_contains(resp, '<a href="%s">Branches (%s)</a>' %
(origin_branches_url, len(origin_branches)))
origin_releases_url = reverse('browse-origin-releases',
url_args=url_args)
nb_releases = len(origin_releases)
if nb_releases > 0:
assert_contains(resp, '<a href="%s">Releases (%s)</a>' %
(origin_releases_url, nb_releases))
assert_contains(resp, '<tr class="swh-release-entry',
count=nb_releases)
for release in origin_releases:
browse_release_url = reverse(
'browse-release',
url_args={'sha1_git': release['id']},
query_params={'origin': origin_info['url']})
browse_revision_url = reverse(
'browse-revision',
url_args={'sha1_git': release['target']},
query_params={'origin': origin_info['url']})
assert_contains(resp, '<a href="%s">' %
escape(browse_release_url))
assert_contains(resp, '<a href="%s">' %
escape(browse_revision_url))
diff --git a/swh/web/tests/data.py b/swh/web/tests/data.py
index 9370b1e7..d69ee8d9 100644
--- a/swh/web/tests/data.py
+++ b/swh/web/tests/data.py
@@ -1,481 +1,484 @@
-# Copyright (C) 2018-2019 The Software Heritage developers
+# Copyright (C) 2018-2020 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
import random
from copy import deepcopy
from typing import Dict
from rest_framework.decorators import api_view
from rest_framework.response import Response
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.from_disk import Directory
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.search import get_search
from swh.storage.algos.dir_iterators import dir_iterator
from swh.web import config
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,
+ _re_encode_content
)
from swh.web.common import service
from swh.web.common.highlightjs import get_hljs_language_from_filename
# Module used to initialize data that will be provided as tests input
# Configuration for git loader
_TEST_LOADER_CONFIG = {
'storage': {
'cls': 'memory'
},
'save_data': False,
'max_content_size': 100 * 1024 * 1024,
}
# Base content indexer configuration
_TEST_INDEXER_BASE_CONFIG = {
'storage': {
'cls': 'memory'
},
'objstorage': {
'cls': 'memory',
'args': {},
},
'indexer_storage': {
'cls': 'memory',
'args': {},
}
}
def random_sha1():
return hash_to_hex(bytes(random.randint(0, 255) for _ in range(20)))
def random_sha256():
return hash_to_hex(bytes(random.randint(0, 255) for _ in range(32)))
def random_blake2s256():
return hash_to_hex(bytes(random.randint(0, 255) for _ in range(32)))
def random_content():
return {
'sha1': random_sha1(),
'sha1_git': random_sha1(),
'sha256': random_sha256(),
'blake2s256': random_blake2s256(),
}
# 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"
}
}
}
# 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 = [
{
'type': 'git',
'url': 'https://github.com/wcoder/highlightjs-line-numbers.js',
'archives': ['highlightjs-line-numbers.js.zip',
'highlightjs-line-numbers.js_visit2.zip'],
'visit_date': ['Dec 1 2018, 01:00 UTC',
'Jan 20 2019, 15:00 UTC']
},
{
'type': 'git',
'url': 'https://github.com/memononen/libtess2',
'archives': ['libtess2.zip'],
'visit_date': ['May 25 2018, 01:00 UTC']
},
{
'type': 'git',
'url': 'repo_with_submodules',
'archives': ['repo_with_submodules.tgz'],
'visit_date': ['Jan 1 2019, 01:00 UTC']
}
]
_contents = {}
# Tests data initialization
def _init_tests_data():
# To hold reference to the memory storage
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
for origin in _TEST_ORIGINS:
for i, archive in enumerate(origin['archives']):
origin_repo_archive = \
os.path.join(os.path.dirname(__file__),
'resources/repos/%s' % archive)
loader = GitLoaderFromArchive(origin['url'],
archive_path=origin_repo_archive,
config=_TEST_LOADER_CONFIG,
visit_date=origin['visit_date'][i])
if storage is None:
storage = loader.storage
else:
loader.storage = storage
loader.load()
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):
url = 'https://many.origins/%d' % (i+1)
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')
storage.origin_visit_update(
url, visit['visit'],
snapshot='1a8893e6a86f444e8be8e7bda6cb34fb1735a00e')
contents = set()
directories = set()
revisions = set()
releases = set()
snapshots = set()
content_path = {}
# Get all objects loaded into the test archive
for origin in _TEST_ORIGINS:
snp = storage.snapshot_get_latest(origin['url'])
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):
content_path[entry['sha1']] = '/'.join(
[hash_to_hex(dir_id), entry['path'].decode('utf-8')])
if entry['type'] == 'file':
contents.add(entry['sha1'])
elif entry['type'] == 'dir':
directories.add(hash_to_hex(entry['target']))
# Get all checksums for each content
result = storage.content_get_metadata(contents)
contents = []
for sha1, contents_metadata in result.items():
for content_metadata in contents_metadata:
contents.append({
algo: hash_to_hex(content_metadata[algo])
for algo in DEFAULT_ALGORITHMS
})
path = content_path[sha1]
cnt = next(storage.content_get([sha1]))
mimetype, encoding = get_mimetype_and_encoding_for_content(
cnt['data'])
+ _, _, cnt['data'] = _re_encode_content(
+ mimetype, encoding, cnt['data'])
content_display_data = prepare_content_for_display(
cnt['data'], mimetype, path)
contents[-1]['path'] = path
contents[-1]['mimetype'] = mimetype
contents[-1]['encoding'] = encoding
contents[-1]['hljs_language'] = content_display_data['language']
contents[-1]['data'] = content_display_data['content_data']
_contents[contents[-1]['sha1']] = contents[-1]
# Create indexer storage instance that will be shared by indexers
idx_storage = get_indexer_storage('memory', {})
# 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 {
'search': search,
'storage': storage,
'idx_storage': idx_storage,
'origins': _TEST_ORIGINS,
'contents': contents,
'directories': list(directories),
'releases': list(releases),
'revisions': list(map(hash_to_hex, revisions)),
'snapshots': list(snapshots),
'generated_checksums': set(),
}
def _init_indexers(tests_data):
# 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),
('license_indexer', _FossologyLicenseIndexer),
('ctags_indexer', _CtagsIndexer)):
idx = idx_class()
idx.storage = tests_data['storage']
idx.objstorage = tests_data['storage'].objstorage
idx.idx_storage = tests_data['idx_storage']
idx.register_tools(idx.config['tools'])
indexers[idx_name] = idx
return indexers
def get_content(content_sha1):
return _contents.get(content_sha1)
_tests_data = None
_current_tests_data = None
_indexer_loggers = {}
def get_tests_data(reset=False):
"""
Initialize tests data and return them in a dict.
"""
global _tests_data, _current_tests_data
if _tests_data is None:
_tests_data = _init_tests_data()
indexers = _init_indexers(_tests_data)
for (name, idx) in indexers.items():
# pytest makes the loggers use a temporary file; and deepcopy
# requires serializability. So we remove them, and add them
# back after the copy.
_indexer_loggers[name] = idx.log
del idx.log
_tests_data.update(indexers)
if reset or _current_tests_data is None:
_current_tests_data = deepcopy(_tests_data)
for (name, logger) in _indexer_loggers.items():
_current_tests_data[name].log = logger
return _current_tests_data
def override_storages(storage, idx_storage, search):
"""
Helper function to replace the storages from which archive data
are fetched.
"""
swh_config = config.get_config()
swh_config.update({
'storage': storage,
'indexer_storage': idx_storage,
'search': search,
})
service.storage = storage
service.idx_storage = idx_storage
service.search = search
# Implement some special endpoints used to provide input tests data
# when executing end to end tests with cypress
_content_code_data_exts = {} # 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]]
def _init_content_tests_data(data_path, data_dict, ext_key):
"""
Helper function to read the content of a directory, store it
into a test archive and add some files metadata (sha1 and/or
expected programming language) in a dict.
Args:
data_path (str): path to a directory relative to the tests
folder of swh-web
data_dict (dict): the dict that will store files metadata
ext_key (bool): whether to use file extensions or filenames
as dict keys
"""
test_contents_dir = os.path.join(
os.path.dirname(__file__), data_path).encode('utf-8')
directory = Directory.from_disk(path=test_contents_dir, data=True,
save_path=True)
objects = directory.collect()
for c in objects['content'].values():
c['status'] = 'visible'
sha1 = hash_to_hex(c['sha1'])
if ext_key:
key = c['path'].decode('utf-8').split('.')[-1]
filename = 'test.' + key
else:
filename = c['path'].decode('utf-8').split('/')[-1]
key = filename
language = get_hljs_language_from_filename(filename)
data_dict[key] = {'sha1': sha1,
'language': language}
del c['path']
del c['perms']
storage = get_tests_data()['storage']
storage.content_add(objects['content'].values())
def _init_content_code_data_exts():
"""
Fill a global dictionary which maps source file extension to
a code content example.
"""
global _content_code_data_exts
_init_content_tests_data('resources/contents/code/extensions',
_content_code_data_exts, True)
def _init_content_other_data_exts():
"""
Fill a global dictionary which maps a file extension to
a content example.
"""
global _content_other_data_exts
_init_content_tests_data('resources/contents/other/extensions',
_content_other_data_exts, True)
def _init_content_code_data_filenames():
"""
Fill a global dictionary which maps a filename to
a content example.
"""
global _content_code_data_filenames
_init_content_tests_data('resources/contents/code/filenames',
_content_code_data_filenames, False)
if config.get_config()['e2e_tests_mode']:
_init_content_code_data_exts()
_init_content_other_data_exts()
_init_content_code_data_filenames()
@api_view(['GET'])
def get_content_code_data_all_exts(request):
"""
Endpoint implementation returning a list of all source file
extensions to test for highlighting using cypress.
"""
return Response(sorted(_content_code_data_exts.keys()),
status=200, content_type='application/json')
@api_view(['GET'])
def get_content_code_data_by_ext(request, ext):
"""
Endpoint implementation returning metadata of a code content example
based on the source file extension.
"""
data = None
status = 404
if ext in _content_code_data_exts:
data = _content_code_data_exts[ext]
status = 200
return Response(data, status=status, content_type='application/json')
@api_view(['GET'])
def get_content_other_data_by_ext(request, ext):
"""
Endpoint implementation returning metadata of a content example
based on the file extension.
"""
_init_content_other_data_exts()
data = None
status = 404
if ext in _content_other_data_exts:
data = _content_other_data_exts[ext]
status = 200
return Response(data, status=status, content_type='application/json')
@api_view(['GET'])
def get_content_code_data_all_filenames(request):
"""
Endpoint implementation returning a list of all source filenames
to test for highlighting using cypress.
"""
return Response(sorted(_content_code_data_filenames.keys()),
status=200, content_type='application/json')
@api_view(['GET'])
def get_content_code_data_by_filename(request, filename):
"""
Endpoint implementation returning metadata of a code content example
based on the source filename.
"""
data = None
status = 404
if filename in _content_code_data_filenames:
data = _content_code_data_filenames[filename]
status = 200
return Response(data, status=status, content_type='application/json')

File Metadata

Mime Type
text/x-diff
Expires
Jul 4 2025, 7:49 AM (10 w, 3 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3273751

Event Timeline

About · Developer information · Legal