You can contribute to extend the content of the Software Heritage archive by submitting an origin
save request. To do so, fill the required info in the form below:
Processing "save code now" request ...
A "Save code now" request takes the following parameters:
- Origin type: the type of version control system the software origin is using.
Currently, the supported types are:
git, for origins using Githg, for origins using Mercurialsvn, for origins using Subversioncvs, for origins using CVS
- Origin url: the url of the remote repository for the software origin.
In order to avoid saving errors from Software Heritage, you should provide the clone/checkout url
as given by the provider hosting the software origin.
It can easily be found in the
web interface used to browse the software origin.
For instance, if you want to save a git
origin into the archive, you should check that the command $ git clone <origin_url>
does not return an error before submitting a request.
Once submitted, your save request can either be:
- accepted: a visit to the provided origin will then be scheduled by Software Heritage in order to
load its content into the archive as soon as possible
- rejected: the provided origin url is blacklisted and no visit will be scheduled
- put in pending state: a manual review will then be performed in order to determine if the
origin can be safely loaded or not into the archive
Once a save request has been accepted, you can follow its current status in the
submitted save requests list.
If you submitted requests while authenticated, you will be able
to only display your own requests.
| Date |
Type |
Url |
Request |
Status |
Info |
|
{% endblock %}
diff --git a/swh/web/tests/conftest.py b/swh/web/tests/conftest.py
index fead2e30..5b0ef04f 100644
--- a/swh/web/tests/conftest.py
+++ b/swh/web/tests/conftest.py
@@ -1,482 +1,497 @@
# Copyright (C) 2018-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
from datetime import timedelta
import json
import os
import shutil
from subprocess import PIPE, run
import sys
from typing import Any, Dict, List, Optional
from _pytest.python import Function
from hypothesis import HealthCheck, settings
import pytest
from django.core.cache import cache
from rest_framework.test import APIClient, APIRequestFactory
from swh.model.hashutil import ALGORITHMS, hash_to_bytes
from swh.scheduler.tests.common import TASK_TYPES
from swh.storage.algos.origin import origin_get_latest_visit_status
from swh.storage.algos.snapshot import snapshot_get_all_branches, snapshot_get_latest
from swh.web.auth.utils import OIDC_SWH_WEB_CLIENT_ID
from swh.web.common import converters
from swh.web.common.origin_save import get_scheduler_load_task_types
from swh.web.common.typing import OriginVisitInfo
from swh.web.config import get_config
from swh.web.tests.data import get_tests_data, override_storages
# Used to skip some tests
ctags_json_missing = (
shutil.which("ctags") is None
or b"+json" not in run(["ctags", "--version"], stdout=PIPE).stdout
)
fossology_missing = shutil.which("nomossa") is None
# Register some hypothesis profiles
settings.register_profile("default", settings())
# we use getattr here to keep mypy happy regardless hypothesis version
function_scoped_fixture_check = (
[getattr(HealthCheck, "function_scoped_fixture")]
if hasattr(HealthCheck, "function_scoped_fixture")
else []
)
suppress_health_check = [
HealthCheck.too_slow,
HealthCheck.filter_too_much,
] + function_scoped_fixture_check
settings.register_profile(
"swh-web", settings(deadline=None, suppress_health_check=suppress_health_check,),
)
settings.register_profile(
"swh-web-fast",
settings(
deadline=None, max_examples=5, suppress_health_check=suppress_health_check,
),
)
def pytest_configure(config):
# Use fast hypothesis profile by default if none has been
# explicitly specified in pytest option
if config.getoption("--hypothesis-profile") is None:
settings.load_profile("swh-web-fast")
# Small hack in order to be able to run the unit tests
# without static assets generated by webpack.
# Those assets are not really needed for the Python tests
# but the django templates will fail to load due to missing
# generated file webpack-stats.json describing the js and css
# files to include.
# So generate a dummy webpack-stats.json file to overcome
# that issue.
test_dir = os.path.dirname(__file__)
# location of the static folder when running tests through tox
data_dir = os.path.join(sys.prefix, "share/swh/web")
static_dir = os.path.join(data_dir, "static")
if not os.path.exists(static_dir):
# location of the static folder when running tests locally with pytest
static_dir = os.path.join(test_dir, "../../../static")
webpack_stats = os.path.join(static_dir, "webpack-stats.json")
if os.path.exists(webpack_stats):
return
bundles_dir = os.path.join(test_dir, "../../../assets/src/bundles")
if not os.path.exists(bundles_dir):
# location of the bundles folder when running tests with tox
bundles_dir = os.path.join(data_dir, "assets/src/bundles")
_, bundles, _ = next(os.walk(bundles_dir))
mock_webpack_stats = {
"status": "done",
"publicPath": "/static",
"chunks": {},
"assets": {},
}
for bundle in bundles:
asset = f"js/{bundle}.js"
mock_webpack_stats["chunks"][bundle] = [asset]
mock_webpack_stats["assets"][asset] = {
"name": asset,
"publicPath": f"/static/{asset}",
}
with open(webpack_stats, "w") as outfile:
json.dump(mock_webpack_stats, outfile)
# Clear Django cache before each test
@pytest.fixture(autouse=True)
def django_cache_cleared():
cache.clear()
# Alias rf fixture from pytest-django
@pytest.fixture
def request_factory(rf):
return rf
# Fixture to get test client from Django REST Framework
@pytest.fixture
def api_client():
return APIClient()
# Fixture to get API request factory from Django REST Framework
@pytest.fixture
def api_request_factory():
return APIRequestFactory()
# Initialize tests data
@pytest.fixture(scope="function", autouse=True)
def tests_data():
data = get_tests_data(reset=True)
# Update swh-web configuration to use the in-memory storages
# instantiated in the tests.data module
override_storages(
data["storage"], data["idx_storage"], data["search"], data["counters"]
)
return data
# Fixture to manipulate data from a sample archive used in the tests
@pytest.fixture(scope="function")
def archive_data(tests_data):
return _ArchiveData(tests_data)
# Fixture to manipulate indexer data from a sample archive used in the tests
@pytest.fixture(scope="function")
def indexer_data(tests_data):
return _IndexerData(tests_data)
# Custom data directory for requests_mock
@pytest.fixture
def datadir():
return os.path.join(os.path.abspath(os.path.dirname(__file__)), "resources")
class _ArchiveData:
"""
Helper class to manage data from a sample test archive.
It is initialized with a reference to an in-memory storage
containing raw tests data.
It is basically a proxy to Storage interface but it overrides some methods
to retrieve those tests data in a json serializable format in order to ease
tests implementation.
"""
def __init__(self, tests_data):
self.storage = tests_data["storage"]
def __getattr__(self, key):
if key == "storage":
raise AttributeError(key)
# Forward calls to non overridden Storage methods to wrapped
# storage instance
return getattr(self.storage, key)
def content_find(self, content: Dict[str, Any]) -> Dict[str, Any]:
cnt_ids_bytes = {
algo_hash: hash_to_bytes(content[algo_hash])
for algo_hash in ALGORITHMS
if content.get(algo_hash)
}
cnt = self.storage.content_find(cnt_ids_bytes)
return converters.from_content(cnt[0].to_dict()) if cnt else cnt
def content_get(self, cnt_id: str) -> Dict[str, Any]:
cnt_id_bytes = hash_to_bytes(cnt_id)
content = self.storage.content_get([cnt_id_bytes])[0]
if content:
content_d = content.to_dict()
content_d.pop("ctime", None)
else:
content_d = None
return converters.from_swh(
content_d, hashess={"sha1", "sha1_git", "sha256", "blake2s256"}
)
def content_get_data(self, cnt_id: str) -> Optional[Dict[str, Any]]:
cnt_id_bytes = hash_to_bytes(cnt_id)
cnt_data = self.storage.content_get_data(cnt_id_bytes)
if cnt_data is None:
return None
return converters.from_content({"data": cnt_data, "sha1": cnt_id_bytes})
def directory_get(self, dir_id):
return {"id": dir_id, "content": self.directory_ls(dir_id)}
def directory_ls(self, dir_id):
cnt_id_bytes = hash_to_bytes(dir_id)
dir_content = map(
converters.from_directory_entry, self.storage.directory_ls(cnt_id_bytes)
)
return list(dir_content)
def release_get(self, rel_id: str) -> Optional[Dict[str, Any]]:
rel_id_bytes = hash_to_bytes(rel_id)
rel_data = self.storage.release_get([rel_id_bytes])[0]
return converters.from_release(rel_data) if rel_data else None
def revision_get(self, rev_id: str) -> Optional[Dict[str, Any]]:
rev_id_bytes = hash_to_bytes(rev_id)
rev_data = self.storage.revision_get([rev_id_bytes])[0]
return converters.from_revision(rev_data) if rev_data else None
def revision_log(self, rev_id, limit=None):
rev_id_bytes = hash_to_bytes(rev_id)
return list(
map(
converters.from_revision,
self.storage.revision_log([rev_id_bytes], limit=limit),
)
)
def snapshot_get_latest(self, origin_url):
snp = snapshot_get_latest(self.storage, origin_url)
return converters.from_snapshot(snp.to_dict())
def origin_get(self, origin_urls):
origins = self.storage.origin_get(origin_urls)
return [converters.from_origin(o.to_dict()) for o in origins]
def origin_visit_get(self, origin_url):
next_page_token = None
visits = []
while True:
visit_page = self.storage.origin_visit_get(
origin_url, page_token=next_page_token
)
next_page_token = visit_page.next_page_token
for visit in visit_page.results:
visit_status = self.storage.origin_visit_status_get_latest(
origin_url, visit.visit
)
visits.append(
converters.from_origin_visit(
{**visit_status.to_dict(), "type": visit.type}
)
)
if not next_page_token:
break
return visits
def origin_visit_get_by(self, origin_url: str, visit_id: int) -> OriginVisitInfo:
visit = self.storage.origin_visit_get_by(origin_url, visit_id)
assert visit is not None
visit_status = self.storage.origin_visit_status_get_latest(origin_url, visit_id)
assert visit_status is not None
return converters.from_origin_visit(
{**visit_status.to_dict(), "type": visit.type}
)
def origin_visit_status_get_latest(
self,
origin_url,
type: Optional[str] = None,
allowed_statuses: Optional[List[str]] = None,
require_snapshot: bool = False,
):
visit_status = origin_get_latest_visit_status(
self.storage,
origin_url,
type=type,
allowed_statuses=allowed_statuses,
require_snapshot=require_snapshot,
)
return (
converters.from_origin_visit(visit_status.to_dict())
if visit_status
else None
)
def snapshot_get(self, snapshot_id):
snp = snapshot_get_all_branches(self.storage, hash_to_bytes(snapshot_id))
return converters.from_snapshot(snp.to_dict())
def snapshot_get_branches(
self, snapshot_id, branches_from="", branches_count=1000, target_types=None
):
partial_branches = self.storage.snapshot_get_branches(
hash_to_bytes(snapshot_id),
branches_from.encode(),
branches_count,
target_types,
)
return converters.from_partial_branches(partial_branches)
def snapshot_get_head(self, snapshot):
if snapshot["branches"]["HEAD"]["target_type"] == "alias":
target = snapshot["branches"]["HEAD"]["target"]
head = snapshot["branches"][target]["target"]
else:
head = snapshot["branches"]["HEAD"]["target"]
return head
def snapshot_count_branches(self, snapshot_id):
counts = dict.fromkeys(("alias", "release", "revision"), 0)
counts.update(self.storage.snapshot_count_branches(hash_to_bytes(snapshot_id)))
counts.pop(None, None)
return counts
class _IndexerData:
"""
Helper class to manage indexer tests data
It is initialized with a reference to an in-memory indexer storage
containing raw tests data.
It also defines class methods to retrieve those tests data in
a json serializable format in order to ease tests implementation.
"""
def __init__(self, tests_data):
self.idx_storage = tests_data["idx_storage"]
self.mimetype_indexer = tests_data["mimetype_indexer"]
self.license_indexer = tests_data["license_indexer"]
self.ctags_indexer = tests_data["ctags_indexer"]
def content_add_mimetype(self, cnt_id):
self.mimetype_indexer.run([hash_to_bytes(cnt_id)])
def content_get_mimetype(self, cnt_id):
mimetype = self.idx_storage.content_mimetype_get([hash_to_bytes(cnt_id)])[
0
].to_dict()
return converters.from_filetype(mimetype)
def content_add_license(self, cnt_id):
self.license_indexer.run([hash_to_bytes(cnt_id)])
def content_get_license(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
licenses = self.idx_storage.content_fossology_license_get([cnt_id_bytes])
for license in licenses:
yield converters.from_swh(license.to_dict(), hashess={"id"})
def content_add_ctags(self, cnt_id):
self.ctags_indexer.run([hash_to_bytes(cnt_id)])
def content_get_ctags(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
ctags = self.idx_storage.content_ctags_get([cnt_id_bytes])
for ctag in ctags:
yield converters.from_swh(ctag, hashess={"id"})
@pytest.fixture
def keycloak_oidc(keycloak_oidc, mocker):
keycloak_config = get_config()["keycloak"]
keycloak_oidc.server_url = keycloak_config["server_url"]
keycloak_oidc.realm_name = keycloak_config["realm_name"]
keycloak_oidc.client_id = OIDC_SWH_WEB_CLIENT_ID
keycloak_oidc_client = mocker.patch("swh.web.auth.views.keycloak_oidc_client")
keycloak_oidc_client.return_value = keycloak_oidc
return keycloak_oidc
@pytest.fixture
def subtest(request):
"""A hack to explicitly set up and tear down fixtures.
This fixture allows you to set up and tear down fixtures within the test
function itself. This is useful (necessary!) for using Hypothesis inside
pytest, as hypothesis will call the test function multiple times, without
setting up or tearing down fixture state as it is normally the case.
Copied from the pytest-subtesthack project, public domain license
(https://github.com/untitaker/pytest-subtesthack).
"""
parent_test = request.node
def inner(func):
if hasattr(Function, "from_parent"):
item = Function.from_parent(
parent_test,
name=request.function.__name__ + "[]",
originalname=request.function.__name__,
callobj=func,
)
else:
item = Function(
name=request.function.__name__ + "[]", parent=parent_test, callobj=func
)
nextitem = parent_test # prevents pytest from tearing down module fixtures
item.ihook.pytest_runtest_setup(item=item)
item.ihook.pytest_runtest_call(item=item)
item.ihook.pytest_runtest_teardown(item=item, nextitem=nextitem)
return inner
@pytest.fixture
def swh_scheduler(swh_scheduler):
config = get_config()
scheduler = config["scheduler"]
config["scheduler"] = swh_scheduler
# create load-git and load-hg task types
for task_type in TASK_TYPES.values():
swh_scheduler.create_task_type(task_type)
# create load-svn task type
swh_scheduler.create_task_type(
{
"type": "load-svn",
- "description": "Update a mercurial repository",
+ "description": "Update a Subversion repository",
"backend_name": "swh.loader.svn.tasks.DumpMountAndLoadSvnRepository",
"default_interval": timedelta(days=64),
"min_interval": timedelta(hours=12),
"max_interval": timedelta(days=64),
"backoff_factor": 2,
"max_queue_length": None,
"num_retries": 7,
"retry_delay": timedelta(hours=2),
}
)
+ # create load-cvs task type
+ swh_scheduler.create_task_type(
+ {
+ "type": "load-cvs",
+ "description": "Update a CVS repository",
+ "backend_name": "swh.loader.cvs.tasks.DumpMountAndLoadSvnRepository",
+ "default_interval": timedelta(days=64),
+ "min_interval": timedelta(hours=12),
+ "max_interval": timedelta(days=64),
+ "backoff_factor": 2,
+ "max_queue_length": None,
+ "num_retries": 7,
+ "retry_delay": timedelta(hours=2),
+ }
+ )
# add method to add load-archive-files task type during tests
def add_load_archive_task_type():
swh_scheduler.create_task_type(
{
"type": "load-archive-files",
"description": "Load tarballs",
"backend_name": "swh.loader.package.archive.tasks.LoadArchive",
"default_interval": timedelta(days=64),
"min_interval": timedelta(hours=12),
"max_interval": timedelta(days=64),
"backoff_factor": 2,
"max_queue_length": None,
"num_retries": 7,
"retry_delay": timedelta(hours=2),
}
)
swh_scheduler.add_load_archive_task_type = add_load_archive_task_type
yield swh_scheduler
config["scheduler"] = scheduler
get_scheduler_load_task_types.cache_clear()
diff --git a/swh/web/tests/misc/test_origin_save.py b/swh/web/tests/misc/test_origin_save.py
index b357a138..bd9561d2 100644
--- a/swh/web/tests/misc/test_origin_save.py
+++ b/swh/web/tests/misc/test_origin_save.py
@@ -1,153 +1,153 @@
# Copyright (C) 2019-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
from datetime import datetime, timedelta, timezone
import json
import pytest
from swh.auth.django.utils import oidc_user_from_profile
from swh.web.common.models import SaveOriginRequest
from swh.web.common.origin_save import SAVE_REQUEST_ACCEPTED, SAVE_TASK_SUCCEEDED
from swh.web.common.utils import reverse
from swh.web.tests.utils import check_http_get_response
-VISIT_TYPES = ("git", "svn", "hg")
+VISIT_TYPES = ("git", "svn", "hg", "cvs")
PRIVILEGED_VISIT_TYPES = tuple(list(VISIT_TYPES) + ["archives"])
def test_old_save_url_redirection(client):
url = reverse("browse-origin-save")
redirect_url = reverse("origin-save")
resp = check_http_get_response(client, url, status_code=302)
assert resp["location"] == redirect_url
@pytest.mark.django_db
def test_save_origin_requests_list(client, mocker):
nb_origins_per_type = 10
for visit_type in VISIT_TYPES:
for i in range(nb_origins_per_type):
SaveOriginRequest.objects.create(
request_date=datetime.now(tz=timezone.utc),
visit_type=visit_type,
origin_url=f"https://{visit_type}.example.org/project{i}",
status=SAVE_REQUEST_ACCEPTED,
visit_date=datetime.now(tz=timezone.utc) + timedelta(hours=1),
loading_task_id=i,
loading_task_status=SAVE_TASK_SUCCEEDED,
)
mock_scheduler = mocker.patch("swh.web.common.origin_save.scheduler")
mock_scheduler.get_tasks.return_value = []
mock_scheduler.get_task_runs.return_value = []
# retrieve all save requests in 3 pages, sorted in descending order
# of request creation
for i, visit_type in enumerate(reversed(VISIT_TYPES)):
url = reverse(
"origin-save-requests-list",
url_args={"status": "all"},
query_params={
"draw": i + 1,
"search[value]": "",
"order[0][column]": "0",
"columns[0][name]": "request_date",
"order[0][dir]": "desc",
"length": nb_origins_per_type,
"start": i * nb_origins_per_type,
},
)
resp = check_http_get_response(
client, url, status_code=200, content_type="application/json"
)
sors = json.loads(resp.content.decode("utf-8"))
assert sors["draw"] == i + 1
assert sors["recordsFiltered"] == len(VISIT_TYPES) * nb_origins_per_type
assert sors["recordsTotal"] == len(VISIT_TYPES) * nb_origins_per_type
assert len(sors["data"]) == nb_origins_per_type
assert all(d["visit_type"] == visit_type for d in sors["data"])
# retrieve save requests filtered by visit type in a single page
for i, visit_type in enumerate(reversed(VISIT_TYPES)):
url = reverse(
"origin-save-requests-list",
url_args={"status": "all"},
query_params={
"draw": i + 1,
"search[value]": visit_type,
"order[0][column]": "0",
"columns[0][name]": "request_date",
"order[0][dir]": "desc",
"length": nb_origins_per_type,
"start": 0,
},
)
resp = check_http_get_response(
client, url, status_code=200, content_type="application/json"
)
sors = json.loads(resp.content.decode("utf-8"))
assert sors["draw"] == i + 1
assert sors["recordsFiltered"] == nb_origins_per_type
assert sors["recordsTotal"] == len(VISIT_TYPES) * nb_origins_per_type
assert len(sors["data"]) == nb_origins_per_type
assert all(d["visit_type"] == visit_type for d in sors["data"])
@pytest.mark.django_db
def test_save_origin_requests_list_user_filter(client, mocker, keycloak_oidc):
# anonymous user created a save request
sor = SaveOriginRequest.objects.create(
request_date=datetime.now(tz=timezone.utc),
visit_type="svn",
origin_url="https://svn.example.org/user/project",
status=SAVE_REQUEST_ACCEPTED,
visit_date=datetime.now(tz=timezone.utc) + timedelta(hours=1),
loading_task_id=1,
loading_task_status=SAVE_TASK_SUCCEEDED,
)
# authenticated user created a save request
user = oidc_user_from_profile(keycloak_oidc, keycloak_oidc.login())
client.login(code="", code_verifier="", redirect_uri="")
sor = SaveOriginRequest.objects.create(
request_date=datetime.now(tz=timezone.utc),
visit_type="git",
origin_url="https://git.example.org/user/project",
status=SAVE_REQUEST_ACCEPTED,
visit_date=datetime.now(tz=timezone.utc) + timedelta(hours=1),
loading_task_id=2,
loading_task_status=SAVE_TASK_SUCCEEDED,
user_ids=f'"{user.id}"',
)
# filter save requests according to user id
url = reverse(
"origin-save-requests-list",
url_args={"status": "all"},
query_params={
"draw": 1,
"search[value]": "",
"order[0][column]": "0",
"columns[0][name]": "request_date",
"order[0][dir]": "desc",
"length": 10,
"start": "0",
"user_requests_only": "1",
},
)
resp = check_http_get_response(
client, url, status_code=200, content_type="application/json"
)
sors = json.loads(resp.content.decode("utf-8"))
assert sors["recordsFiltered"] == 1
assert sors["recordsTotal"] == 2
assert sors["data"][0] == sor.to_dict()