diff --git a/swh/web/admin/urls.py b/swh/web/admin/urls.py
index 2140ea9d..3e3c8f10 100644
--- a/swh/web/admin/urls.py
+++ b/swh/web/admin/urls.py
@@ -1,25 +1,28 @@
# Copyright (C) 2018-2022 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.conf.urls import url
from django.contrib.auth.views import LoginView
from django.shortcuts import redirect
-import swh.web.admin.add_forge_now # noqa
from swh.web.admin.adminurls import AdminUrls
import swh.web.admin.deposit # noqa
import swh.web.admin.origin_save # noqa
+from swh.web.config import is_feature_enabled
+
+if is_feature_enabled("add_forge_now"):
+ import swh.web.admin.add_forge_now # noqa
def _admin_default_view(request):
return redirect("admin-origin-save")
urlpatterns = [
url(r"^$", _admin_default_view, name="admin"),
url(r"^login/$", LoginView.as_view(template_name="login.html"), name="login"),
]
urlpatterns += AdminUrls.get_url_patterns()
diff --git a/swh/web/common/utils.py b/swh/web/common/utils.py
index 8a0a73c5..1fe59d07 100644
--- a/swh/web/common/utils.py
+++ b/swh/web/common/utils.py
@@ -1,522 +1,523 @@
# Copyright (C) 2017-2022 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 datetime import datetime, timezone
import os
import re
from typing import Any, Dict, List, Optional
import urllib.parse
from xml.etree import ElementTree
from bs4 import BeautifulSoup
from docutils.core import publish_parts
import docutils.parsers.rst
import docutils.utils
from docutils.writers.html5_polyglot import HTMLTranslator, Writer
from iso8601 import ParseError, parse_date
from pkg_resources import get_distribution
from prometheus_client.registry import CollectorRegistry
import requests
from requests.auth import HTTPBasicAuth
from django.core.cache import cache
from django.http import HttpRequest, QueryDict
from django.shortcuts import redirect
from django.urls import resolve
from django.urls import reverse as django_reverse
from swh.web.auth.utils import (
ADD_FORGE_MODERATOR_PERMISSION,
ADMIN_LIST_DEPOSIT_PERMISSION,
)
from swh.web.common.exc import BadInputExc
from swh.web.common.typing import QueryParameters
from swh.web.config import SWH_WEB_SERVER_NAME, get_config, search
SWH_WEB_METRICS_REGISTRY = CollectorRegistry(auto_describe=True)
swh_object_icons = {
"alias": "mdi mdi-star",
"branch": "mdi mdi-source-branch",
"branches": "mdi mdi-source-branch",
"content": "mdi mdi-file-document",
"cnt": "mdi mdi-file-document",
"directory": "mdi mdi-folder",
"dir": "mdi mdi-folder",
"origin": "mdi mdi-source-repository",
"ori": "mdi mdi-source-repository",
"person": "mdi mdi-account",
"revisions history": "mdi mdi-history",
"release": "mdi mdi-tag",
"rel": "mdi mdi-tag",
"releases": "mdi mdi-tag",
"revision": "mdi mdi-rotate-90 mdi-source-commit",
"rev": "mdi mdi-rotate-90 mdi-source-commit",
"snapshot": "mdi mdi-camera",
"snp": "mdi mdi-camera",
"visits": "mdi mdi-calendar-month",
}
def reverse(
viewname: str,
url_args: Optional[Dict[str, Any]] = None,
query_params: Optional[QueryParameters] = None,
current_app: Optional[str] = None,
urlconf: Optional[str] = None,
request: Optional[HttpRequest] = None,
) -> str:
"""An override of django reverse function supporting query parameters.
Args:
viewname: the name of the django view from which to compute a url
url_args: dictionary of url arguments indexed by their names
query_params: dictionary of query parameters to append to the
reversed url
current_app: the name of the django app tighten to the view
urlconf: url configuration module
request: build an absolute URI if provided
Returns:
str: the url of the requested view with processed arguments and
query parameters
"""
if url_args:
url_args = {k: v for k, v in url_args.items() if v is not None}
url = django_reverse(
viewname, urlconf=urlconf, kwargs=url_args, current_app=current_app
)
if query_params:
query_params = {k: v for k, v in query_params.items() if v is not None}
if query_params and len(query_params) > 0:
query_dict = QueryDict("", mutable=True)
for k in sorted(query_params.keys()):
query_dict[k] = query_params[k]
url += "?" + query_dict.urlencode(safe="/;:")
if request is not None:
url = request.build_absolute_uri(url)
return url
def datetime_to_utc(date):
"""Returns datetime in UTC without timezone info
Args:
date (datetime.datetime): input datetime with timezone info
Returns:
datetime.datetime: datetime in UTC without timezone info
"""
if date.tzinfo and date.tzinfo != timezone.utc:
return date.astimezone(tz=timezone.utc)
else:
return date
def parse_iso8601_date_to_utc(iso_date: str) -> datetime:
"""Given an ISO 8601 datetime string, parse the result as UTC datetime.
Returns:
a timezone-aware datetime representing the parsed date
Raises:
swh.web.common.exc.BadInputExc: provided date does not respect ISO 8601 format
Samples:
- 2016-01-12
- 2016-01-12T09:19:12+0100
- 2007-01-14T20:34:22Z
"""
try:
date = parse_date(iso_date)
return datetime_to_utc(date)
except ParseError as e:
raise BadInputExc(e)
def shorten_path(path):
"""Shorten the given path: for each hash present, only return the first
8 characters followed by an ellipsis"""
sha256_re = r"([0-9a-f]{8})[0-9a-z]{56}"
sha1_re = r"([0-9a-f]{8})[0-9a-f]{32}"
ret = re.sub(sha256_re, r"\1...", path)
return re.sub(sha1_re, r"\1...", ret)
def format_utc_iso_date(iso_date, fmt="%d %B %Y, %H:%M UTC"):
"""Turns a string representation of an ISO 8601 datetime string
to UTC and format it into a more human readable one.
For instance, from the following input
string: '2017-05-04T13:27:13+02:00' the following one
is returned: '04 May 2017, 11:27 UTC'.
Custom format string may also be provided
as parameter
Args:
iso_date (str): a string representation of an ISO 8601 date
fmt (str): optional date formatting string
Returns:
str: a formatted string representation of the input iso date
"""
if not iso_date:
return iso_date
date = parse_iso8601_date_to_utc(iso_date)
return date.strftime(fmt)
def gen_path_info(path):
"""Function to generate path data navigation for use
with a breadcrumb in the swh web ui.
For instance, from a path /folder1/folder2/folder3,
it returns the following list::
[{'name': 'folder1', 'path': 'folder1'},
{'name': 'folder2', 'path': 'folder1/folder2'},
{'name': 'folder3', 'path': 'folder1/folder2/folder3'}]
Args:
path: a filesystem path
Returns:
list: a list of path data for navigation as illustrated above.
"""
path_info = []
if path:
sub_paths = path.strip("/").split("/")
path_from_root = ""
for p in sub_paths:
path_from_root += "/" + p
path_info.append({"name": p, "path": path_from_root.strip("/")})
return path_info
def parse_rst(text, report_level=2):
"""
Parse a reStructuredText string with docutils.
Args:
text (str): string with reStructuredText markups in it
report_level (int): level of docutils report messages to print
(1 info 2 warning 3 error 4 severe 5 none)
Returns:
docutils.nodes.document: a parsed docutils document
"""
parser = docutils.parsers.rst.Parser()
components = (docutils.parsers.rst.Parser,)
settings = docutils.frontend.OptionParser(
components=components
).get_default_values()
settings.report_level = report_level
document = docutils.utils.new_document("rst-doc", settings=settings)
parser.parse(text, document)
return document
def get_client_ip(request):
"""
Return the client IP address from an incoming HTTP request.
Args:
request (django.http.HttpRequest): the incoming HTTP request
Returns:
str: The client IP address
"""
x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
if x_forwarded_for:
ip = x_forwarded_for.split(",")[0]
else:
ip = request.META.get("REMOTE_ADDR")
return ip
def is_swh_web_development(request: HttpRequest) -> bool:
"""Indicate if we are running a development version of swh-web.
"""
site_base_url = request.build_absolute_uri("/")
return any(
host in site_base_url for host in ("localhost", "127.0.0.1", "testserver")
)
def is_swh_web_staging(request: HttpRequest) -> bool:
"""Indicate if we are running a staging version of swh-web.
"""
config = get_config()
site_base_url = request.build_absolute_uri("/")
return any(
server_name in site_base_url for server_name in config["staging_server_names"]
)
def is_swh_web_production(request: HttpRequest) -> bool:
"""Indicate if we are running the public production version of swh-web.
"""
return SWH_WEB_SERVER_NAME in request.build_absolute_uri("/")
browsers_supported_image_mimes = set(
[
"image/gif",
"image/png",
"image/jpeg",
"image/bmp",
"image/webp",
"image/svg",
"image/svg+xml",
]
)
def context_processor(request):
"""
Django context processor used to inject variables
in all swh-web templates.
"""
config = get_config()
if (
hasattr(request, "user")
and request.user.is_authenticated
and not hasattr(request.user, "backend")
):
# To avoid django.template.base.VariableDoesNotExist errors
# when rendering templates when standard Django user is logged in.
request.user.backend = "django.contrib.auth.backends.ModelBackend"
return {
"swh_object_icons": swh_object_icons,
"available_languages": None,
"swh_client_config": config["client_config"],
"oidc_enabled": bool(config["keycloak"]["server_url"]),
"browsers_supported_image_mimes": browsers_supported_image_mimes,
"keycloak": config["keycloak"],
"site_base_url": request.build_absolute_uri("/"),
"DJANGO_SETTINGS_MODULE": os.environ["DJANGO_SETTINGS_MODULE"],
"status": config["status"],
"swh_web_dev": is_swh_web_development(request),
"swh_web_staging": is_swh_web_staging(request),
"swh_web_version": get_distribution("swh.web").version,
"iframe_mode": False,
"ADMIN_LIST_DEPOSIT_PERMISSION": ADMIN_LIST_DEPOSIT_PERMISSION,
"ADD_FORGE_MODERATOR_PERMISSION": ADD_FORGE_MODERATOR_PERMISSION,
+ "FEATURES": get_config()["features"],
}
def resolve_branch_alias(
snapshot: Dict[str, Any], branch: Optional[Dict[str, Any]]
) -> Optional[Dict[str, Any]]:
"""
Resolve branch alias in snapshot content.
Args:
snapshot: a full snapshot content
branch: a branch alias contained in the snapshot
Returns:
The real snapshot branch that got aliased.
"""
while branch and branch["target_type"] == "alias":
if branch["target"] in snapshot["branches"]:
branch = snapshot["branches"][branch["target"]]
else:
from swh.web.common import archive
snp = archive.lookup_snapshot(
snapshot["id"], branches_from=branch["target"], branches_count=1
)
if snp and branch["target"] in snp["branches"]:
branch = snp["branches"][branch["target"]]
else:
branch = None
return branch
class _NoHeaderHTMLTranslator(HTMLTranslator):
"""
Docutils translator subclass to customize the generation of HTML
from reST-formatted docstrings
"""
def __init__(self, document):
super().__init__(document)
self.body_prefix = []
self.body_suffix = []
_HTML_WRITER = Writer()
_HTML_WRITER.translator_class = _NoHeaderHTMLTranslator
def rst_to_html(rst: str) -> str:
"""
Convert reStructuredText document into HTML.
Args:
rst: A string containing a reStructuredText document
Returns:
Body content of the produced HTML conversion.
"""
settings = {
"initial_header_level": 2,
"halt_level": 4,
"traceback": True,
"file_insertion_enabled": False,
"raw_enabled": False,
}
pp = publish_parts(rst, writer=_HTML_WRITER, settings_overrides=settings)
return f'
{pp["html_body"]}
'
def prettify_html(html: str) -> str:
"""
Prettify an HTML document.
Args:
html: Input HTML document
Returns:
The prettified HTML document
"""
return BeautifulSoup(html, "lxml").prettify()
def _deposits_list_url(
deposits_list_base_url: str, page_size: int, username: Optional[str]
) -> str:
params = {"page_size": str(page_size)}
if username is not None:
params["username"] = username
return f"{deposits_list_base_url}?{urllib.parse.urlencode(params)}"
def get_deposits_list(username: Optional[str] = None) -> List[Dict[str, Any]]:
"""Return the list of software deposits using swh-deposit API
"""
config = get_config()["deposit"]
deposits_list_base_url = config["private_api_url"] + "deposits"
deposits_list_auth = HTTPBasicAuth(
config["private_api_user"], config["private_api_password"]
)
deposits_list_url = _deposits_list_url(
deposits_list_base_url, page_size=1, username=username
)
nb_deposits = requests.get(
deposits_list_url, auth=deposits_list_auth, timeout=30
).json()["count"]
deposits_data = cache.get(f"swh-deposit-list-{username}")
if not deposits_data or deposits_data["count"] != nb_deposits:
deposits_list_url = _deposits_list_url(
deposits_list_base_url, page_size=nb_deposits, username=username
)
deposits_data = requests.get(
deposits_list_url, auth=deposits_list_auth, timeout=30,
).json()
cache.set(f"swh-deposit-list-{username}", deposits_data)
return deposits_data["results"]
def origin_visit_types() -> List[str]:
"""Return the exhaustive list of visit types for origins
ingested into the archive.
"""
try:
return sorted(search().visit_types_count().keys())
except Exception:
return []
def redirect_to_new_route(request, new_route, permanent=True):
"""Redirect a request to another route with url args and query parameters
eg: /origin//log?path=test can be redirected as
/log?url=&path=test. This can be used to deprecate routes
"""
request_path = resolve(request.path_info)
args = {**request_path.kwargs, **request.GET.dict()}
return redirect(reverse(new_route, query_params=args), permanent=permanent,)
NAMESPACES = {
"swh": "https://www.softwareheritage.org/schema/2018/deposit",
"schema": "http://schema.org/",
}
def parse_swh_metadata_provenance(raw_metadata: str) -> Optional[str]:
"""Parse swh metadata-provenance out of the raw metadata deposit. If found, returns the
value, None otherwise.
.. code-block:: xml
https://example.org/metadata/url
Args:
raw_metadata: raw metadata out of deposits received
Returns:
Either the metadata provenance url if any or None otherwise
"""
metadata = ElementTree.fromstring(raw_metadata)
url = metadata.findtext(
"swh:deposit/swh:metadata-provenance/schema:url", namespaces=NAMESPACES,
)
return url or None
def parse_swh_deposit_origin(raw_metadata: str) -> Optional[str]:
"""Parses and from metadata document,
if any. They are mutually exclusive and tested as such in the deposit.
.. code-block:: xml
.. code-block:: xml
Returns:
The one not null if any, None otherwise
"""
metadata = ElementTree.fromstring(raw_metadata)
for origin_tag in ["create_origin", "add_to_origin"]:
elt = metadata.find(
f"swh:deposit/swh:{origin_tag}/swh:origin[@url]", namespaces=NAMESPACES
)
if elt is not None:
return elt.attrib["url"]
return None
diff --git a/swh/web/config.py b/swh/web/config.py
index 432b034f..81ac7051 100644
--- a/swh/web/config.py
+++ b/swh/web/config.py
@@ -1,209 +1,218 @@
# Copyright (C) 2017-2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import os
from typing import Any, Dict
from swh.core import config
from swh.counters import get_counters
from swh.indexer.storage import get_indexer_storage
from swh.scheduler import get_scheduler
from swh.search import get_search
from swh.storage import get_storage
from swh.vault import get_vault
from swh.web import settings
SWH_WEB_SERVER_NAME = "archive.softwareheritage.org"
SWH_WEB_INTERNAL_SERVER_NAME = "archive.internal.softwareheritage.org"
SWH_WEB_STAGING_SERVER_NAMES = [
"webapp.staging.swh.network",
"webapp.internal.staging.swh.network",
]
SETTINGS_DIR = os.path.dirname(settings.__file__)
DEFAULT_CONFIG = {
"allowed_hosts": ("list", []),
"storage": (
"dict",
{"cls": "remote", "url": "http://127.0.0.1:5002/", "timeout": 10,},
),
"indexer_storage": (
"dict",
{"cls": "remote", "url": "http://127.0.0.1:5007/", "timeout": 1,},
),
"counters": (
"dict",
{"cls": "remote", "url": "http://127.0.0.1:5011/", "timeout": 1,},
),
"search": (
"dict",
{"cls": "remote", "url": "http://127.0.0.1:5010/", "timeout": 10,},
),
"search_config": (
"dict",
{"metadata_backend": "swh-indexer-storage",}, # or "swh-search"
),
"log_dir": ("string", "/tmp/swh/log"),
"debug": ("bool", False),
"serve_assets": ("bool", False),
"host": ("string", "127.0.0.1"),
"port": ("int", 5004),
"secret_key": ("string", "development key"),
# do not display code highlighting for content > 1MB
"content_display_max_size": ("int", 5 * 1024 * 1024),
"snapshot_content_max_size": ("int", 1000),
"throttling": (
"dict",
{
"cache_uri": None, # production: memcached as cache (127.0.0.1:11211)
# development: in-memory cache so None
"scopes": {
"swh_api": {
"limiter_rate": {"default": "120/h"},
"exempted_networks": ["127.0.0.0/8"],
},
"swh_api_origin_search": {
"limiter_rate": {"default": "10/m"},
"exempted_networks": ["127.0.0.0/8"],
},
"swh_vault_cooking": {
"limiter_rate": {"default": "120/h", "GET": "60/m"},
"exempted_networks": ["127.0.0.0/8"],
},
"swh_save_origin": {
"limiter_rate": {"default": "120/h", "POST": "10/h"},
"exempted_networks": ["127.0.0.0/8"],
},
"swh_api_origin_visit_latest": {
"limiter_rate": {"default": "700/m"},
"exempted_networks": ["127.0.0.0/8"],
},
},
},
),
"vault": ("dict", {"cls": "remote", "args": {"url": "http://127.0.0.1:5005/",}}),
"scheduler": ("dict", {"cls": "remote", "url": "http://127.0.0.1:5008/"}),
"development_db": ("string", os.path.join(SETTINGS_DIR, "db.sqlite3")),
"test_db": ("dict", {"name": "swh-web-test"}),
"production_db": ("dict", {"name": "swh-web"}),
"deposit": (
"dict",
{
"private_api_url": "https://deposit.softwareheritage.org/1/private/",
"private_api_user": "swhworker",
"private_api_password": "some-password",
},
),
"e2e_tests_mode": ("bool", False),
"es_workers_index_url": ("string", ""),
"history_counters_url": (
"string",
(
"http://counters1.internal.softwareheritage.org:5011"
"/counters_history/history.json"
),
),
"client_config": ("dict", {}),
"keycloak": ("dict", {"server_url": "", "realm_name": ""}),
"graph": (
"dict",
{
"server_url": "http://graph.internal.softwareheritage.org:5009/graph/",
"max_edges": {"staff": 0, "user": 100000, "anonymous": 1000},
},
),
"status": (
"dict",
{
"server_url": "https://status.softwareheritage.org/",
"json_path": "1.0/status/578e5eddcdc0cc7951000520",
},
),
"counters_backend": ("string", "swh-storage"), # or "swh-counters"
"staging_server_names": ("list", SWH_WEB_STAGING_SERVER_NAMES),
"instance_name": ("str", "archive-test.softwareheritage.org"),
"give": ("dict", {"public_key": "", "token": ""}),
+ "features": ("dict", {"add_forge_now": False}),
}
swhweb_config: Dict[str, Any] = {}
def get_config(config_file="web/web"):
"""Read the configuration file `config_file`.
If an environment variable SWH_CONFIG_FILENAME is defined, this
takes precedence over the config_file parameter.
In any case, update the app with parameters (secret_key, conf)
and return the parsed configuration as a dict.
If no configuration file is provided, return a default
configuration.
"""
if not swhweb_config:
config_filename = os.environ.get("SWH_CONFIG_FILENAME")
if config_filename:
config_file = config_filename
cfg = config.load_named_config(config_file, DEFAULT_CONFIG)
swhweb_config.update(cfg)
config.prepare_folders(swhweb_config, "log_dir")
if swhweb_config.get("search"):
swhweb_config["search"] = get_search(**swhweb_config["search"])
else:
swhweb_config["search"] = None
swhweb_config["storage"] = get_storage(**swhweb_config["storage"])
swhweb_config["vault"] = get_vault(**swhweb_config["vault"])
swhweb_config["indexer_storage"] = get_indexer_storage(
**swhweb_config["indexer_storage"]
)
swhweb_config["scheduler"] = get_scheduler(**swhweb_config["scheduler"])
swhweb_config["counters"] = get_counters(**swhweb_config["counters"])
return swhweb_config
def search():
"""Return the current application's search.
"""
return get_config()["search"]
def storage():
"""Return the current application's storage.
"""
return get_config()["storage"]
def vault():
"""Return the current application's vault.
"""
return get_config()["vault"]
def indexer_storage():
"""Return the current application's indexer storage.
"""
return get_config()["indexer_storage"]
def scheduler():
"""Return the current application's scheduler.
"""
return get_config()["scheduler"]
def counters():
"""Return the current application's counters.
"""
return get_config()["counters"]
+
+
+def is_feature_enabled(feature_name: str) -> bool:
+ """Determine whether a feature is enabled or not. If feature_name is not found at all,
+ it's considered disabled.
+
+ """
+ return get_config()["features"].get(feature_name, False)
diff --git a/swh/web/settings/tests.py b/swh/web/settings/tests.py
index 78eb67ff..2d7119fd 100644
--- a/swh/web/settings/tests.py
+++ b/swh/web/settings/tests.py
@@ -1,130 +1,131 @@
# Copyright (C) 2017-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
"""
Django tests settings for swh-web.
"""
import os
import sys
from swh.web.config import get_config
scope1_limiter_rate = 3
scope1_limiter_rate_post = 1
scope2_limiter_rate = 5
scope2_limiter_rate_post = 2
scope3_limiter_rate = 1
scope3_limiter_rate_post = 1
save_origin_rate_post = 10
swh_web_config = get_config()
_pytest = "pytest" in sys.argv[0] or "PYTEST_XDIST_WORKER" in os.environ
swh_web_config.update(
{
# enable django debug mode only when running pytest
"debug": _pytest,
"secret_key": "test",
"history_counters_url": "",
"throttling": {
"cache_uri": None,
"scopes": {
"swh_api": {
"limiter_rate": {"default": "60/min"},
"exempted_networks": ["127.0.0.0/8"],
},
"swh_api_origin_search": {
"limiter_rate": {"default": "100/min"},
"exempted_networks": ["127.0.0.0/8"],
},
"swh_api_origin_visit_latest": {
"limiter_rate": {"default": "6000/min"},
"exempted_networks": ["127.0.0.0/8"],
},
"swh_vault_cooking": {
"limiter_rate": {"default": "120/h", "GET": "60/m"},
"exempted_networks": ["127.0.0.0/8"],
},
"swh_save_origin": {
"limiter_rate": {
"default": "120/h",
"POST": "%s/h" % save_origin_rate_post,
}
},
"scope1": {
"limiter_rate": {
"default": "%s/min" % scope1_limiter_rate,
"POST": "%s/min" % scope1_limiter_rate_post,
}
},
"scope2": {
"limiter_rate": {
"default": "%s/min" % scope2_limiter_rate,
"POST": "%s/min" % scope2_limiter_rate_post,
}
},
"scope3": {
"limiter_rate": {
"default": "%s/min" % scope3_limiter_rate,
"POST": "%s/min" % scope3_limiter_rate_post,
},
"exempted_networks": ["127.0.0.0/8"],
},
},
},
"keycloak": {
# disable keycloak use when not running pytest
"server_url": "http://localhost:8080/auth/" if _pytest else "",
"realm_name": "SoftwareHeritage",
},
+ "features": {"add_forge_now": True,},
}
)
from .common import * # noqa
from .common import LOGGING # noqa, isort: skip
ALLOWED_HOSTS = ["*"]
DATABASES = {
"default": {
"ENGINE": "django.db.backends.postgresql",
"NAME": swh_web_config["test_db"]["name"],
}
}
# when running cypress tests, make the webapp fetch data from memory storages
if not _pytest:
swh_web_config.update(
{
"debug": True,
"e2e_tests_mode": True,
# ensure scheduler not available to avoid side effects in cypress tests
"scheduler": {"cls": "remote", "url": ""},
}
)
from django.conf import settings
from swh.web.tests.data import get_tests_data, override_storages
test_data = get_tests_data()
override_storages(
test_data["storage"],
test_data["idx_storage"],
test_data["search"],
test_data["counters"],
)
# using sqlite3 for frontend tests
settings.DATABASES["default"].update(
{"ENGINE": "django.db.backends.sqlite3", "NAME": "swh-web-test.sqlite3"}
)
else:
# Silent DEBUG output when running unit tests
LOGGING["handlers"]["console"]["level"] = "INFO" # type: ignore
diff --git a/swh/web/templates/layout.html b/swh/web/templates/layout.html
index 775bd1e0..af165417 100644
--- a/swh/web/templates/layout.html
+++ b/swh/web/templates/layout.html
@@ -1,303 +1,307 @@
{% comment %}
Copyright (C) 2015-2021 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
{% endcomment %}
{% load js_reverse %}
{% load static %}
{% load render_bundle from webpack_loader %}
{% load swh_templatetags %}
{% block title %}{% endblock %}
{% render_bundle 'vendors' %}
{% render_bundle 'webapp' %}
{% render_bundle 'guided_tour' %}
{{ request.user.is_authenticated|json_script:"swh_user_logged_in" }}
{% block header %}{% endblock %}
{% if not swh_web_dev and not swh_web_staging %}
{% endif %}
Operational
{% url 'logout' as logout_url %}
{% if user.is_authenticated %}
Logged in as
{% if 'OIDC' in user.backend %}
{{ user.username }},
logout
{% else %}
{{ user.username }},
logout
{% endif %}
{% elif oidc_enabled %}
{% if request.path != logout_url %}
login
{% else %}
login
{% endif %}
{% else %}
{% if request.path != logout_url %}
login
{% else %}
login
{% endif %}
{% endif %}
{% comment %} {% endcomment %}
{% if swh_web_staging %}
Staging v{{ swh_web_version }}
{% elif swh_web_dev %}
Development v{{ swh_web_version|split:"+"|first }}
{% endif %}
{% block content %}{% endblock %}
{% include "includes/global-modals.html" %}
diff --git a/swh/web/tests/test_config.py b/swh/web/tests/test_config.py
new file mode 100644
index 00000000..fed25f62
--- /dev/null
+++ b/swh/web/tests/test_config.py
@@ -0,0 +1,23 @@
+# Copyright (C) 2022 The Software Heritage developers
+# See the AUTHORS file at the top-level directory of this distribution
+# License: GNU General Public License version 3, or any later version
+# See top-level LICENSE file for more information
+
+import pytest
+
+from swh.web.config import get_config, is_feature_enabled
+
+
+@pytest.mark.parametrize(
+ "feature_name", ["inexistant-feature", "awesome-stuff"],
+)
+def test_is_feature_enabled(feature_name):
+ config = get_config()
+ # by default, feature non configured are considered disabled
+ assert is_feature_enabled(feature_name) is False
+
+ for enabled in [True, False]:
+ # Let's configure the feature
+ config["features"] = {feature_name: enabled}
+ # and check its configuration is properly read
+ assert is_feature_enabled(feature_name) is enabled
diff --git a/swh/web/urls.py b/swh/web/urls.py
index a4bf47ee..705223b5 100644
--- a/swh/web/urls.py
+++ b/swh/web/urls.py
@@ -1,84 +1,86 @@
-# Copyright (C) 2017-2021 The Software Heritage developers
+# Copyright (C) 2017-2022 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_js_reverse.views import urls_js
from django.conf import settings
from django.conf.urls import (
handler400,
handler403,
handler404,
handler500,
include,
url,
)
from django.contrib.auth.views import LogoutView
from django.contrib.staticfiles.views import serve
from django.shortcuts import render
from django.views.generic.base import RedirectView
from swh.web.browse.identifiers import swhid_browse
from swh.web.common.exc import (
swh_handle400,
swh_handle403,
swh_handle404,
swh_handle500,
)
from swh.web.common.utils import origin_visit_types
-from swh.web.config import get_config
+from swh.web.config import get_config, is_feature_enabled
swh_web_config = get_config()
favicon_view = RedirectView.as_view(
url="/static/img/icons/swh-logo-32x32.png", permanent=True
)
def _default_view(request):
return render(request, "homepage.html", {"visit_types": origin_visit_types()})
urlpatterns = [
url(r"^admin/", include("swh.web.admin.urls")),
url(r"^favicon\.ico$", favicon_view),
url(r"^api/", include("swh.web.api.urls")),
url(r"^browse/", include("swh.web.browse.urls")),
url(r"^$", _default_view, name="swh-web-homepage"),
url(r"^jsreverse/$", urls_js, name="js_reverse"),
# keep legacy SWHID resolving URL with trailing slash for backward compatibility
url(
r"^(?P(swh|SWH):[0-9]+:[A-Za-z]+:[0-9A-Fa-f]+.*)/$",
swhid_browse,
name="browse-swhid-legacy",
),
url(
r"^(?P(swh|SWH):[0-9]+:[A-Za-z]+:[0-9A-Fa-f]+.*)$",
swhid_browse,
name="browse-swhid",
),
url(r"^", include("swh.web.misc.urls")),
- url(r"^", include("swh.web.add_forge_now.views")),
url(r"^", include("swh.web.auth.views")),
url(r"^logout/$", LogoutView.as_view(template_name="logout.html"), name="logout"),
]
+if is_feature_enabled("add_forge_now"):
+ urlpatterns += (url(r"^", include("swh.web.add_forge_now.views")),)
+
# allow to serve assets through django staticfiles
# even if settings.DEBUG is False
def insecure_serve(request, path, **kwargs):
return serve(request, path, insecure=True, **kwargs)
# enable to serve compressed assets through django development server
if swh_web_config["serve_assets"]:
static_pattern = r"^%s(?P.*)$" % settings.STATIC_URL[1:]
urlpatterns.append(url(static_pattern, insecure_serve))
handler400 = swh_handle400 # noqa
handler403 = swh_handle403 # noqa
handler404 = swh_handle404 # noqa
handler500 = swh_handle500 # noqa