Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/browse/test_snapshot_context.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import random | import random | ||||
from hypothesis import given | from hypothesis import given | ||||
from swh.web.browse.snapshot_context import ( | from swh.web.browse.snapshot_context import ( | ||||
get_origin_visit_snapshot, | get_origin_visit_snapshot, | ||||
get_snapshot_content, | get_snapshot_content, | ||||
get_snapshot_context, | get_snapshot_context, | ||||
_get_release, | _get_release, | ||||
) | ) | ||||
from swh.web.browse.utils import gen_revision_url | |||||
from swh.web.common.identifiers import get_swh_persistent_id | from swh.web.common.identifiers import get_swh_persistent_id | ||||
from swh.web.common.origin_visits import get_origin_visit, get_origin_visits | from swh.web.common.origin_visits import get_origin_visit, get_origin_visits | ||||
from swh.web.common.typing import ( | from swh.web.common.typing import ( | ||||
SnapshotBranchInfo, | SnapshotBranchInfo, | ||||
SnapshotReleaseInfo, | SnapshotReleaseInfo, | ||||
SnapshotContext, | SnapshotContext, | ||||
) | ) | ||||
from swh.web.common.utils import format_utc_iso_date, reverse | from swh.web.common.utils import format_utc_iso_date, reverse | ||||
▲ Show 20 Lines • Show All 111 Lines • ▼ Show 20 Lines | ): | ||||
origin_info=None, | origin_info=None, | ||||
origin_visits_url=None, | origin_visits_url=None, | ||||
release=None, | release=None, | ||||
release_id=None, | release_id=None, | ||||
query_params=query_params, | query_params=query_params, | ||||
releases=releases, | releases=releases, | ||||
releases_url=releases_url, | releases_url=releases_url, | ||||
revision_id=revision_id, | revision_id=revision_id, | ||||
revision_info=_get_revision_info(archive_data, revision_id), | |||||
root_directory=root_directory, | root_directory=root_directory, | ||||
snapshot_id=snapshot, | snapshot_id=snapshot, | ||||
snapshot_sizes=snapshot_sizes, | snapshot_sizes=snapshot_sizes, | ||||
snapshot_swhid=snapshot_swhid, | snapshot_swhid=snapshot_swhid, | ||||
url_args=url_args, | url_args=url_args, | ||||
visit_info=None, | visit_info=None, | ||||
) | ) | ||||
if revision_id: | |||||
expected["revision_info"]["revision_url"] = gen_revision_url( | |||||
revision_id, snapshot_context | |||||
) | |||||
assert snapshot_context == expected | assert snapshot_context == expected | ||||
_check_branch_release_revision_parameters( | _check_branch_release_revision_parameters( | ||||
archive_data, expected, browse_context, kwargs, branches, releases | archive_data, expected, browse_context, kwargs, branches, releases | ||||
) | ) | ||||
@given(origin_with_multiple_visits()) | @given(origin_with_multiple_visits()) | ||||
▲ Show 20 Lines • Show All 69 Lines • ▼ Show 20 Lines | ): | ||||
origin_info={"url": origin["url"]}, | origin_info={"url": origin["url"]}, | ||||
origin_visits_url=origin_visits_url, | origin_visits_url=origin_visits_url, | ||||
release=None, | release=None, | ||||
release_id=None, | release_id=None, | ||||
query_params=query_params, | query_params=query_params, | ||||
releases=releases, | releases=releases, | ||||
releases_url=releases_url, | releases_url=releases_url, | ||||
revision_id=revision_id, | revision_id=revision_id, | ||||
revision_info=_get_revision_info(archive_data, revision_id), | |||||
root_directory=root_directory, | root_directory=root_directory, | ||||
snapshot_id=snapshot, | snapshot_id=snapshot, | ||||
snapshot_sizes=snapshot_sizes, | snapshot_sizes=snapshot_sizes, | ||||
snapshot_swhid=snapshot_swhid, | snapshot_swhid=snapshot_swhid, | ||||
url_args={}, | url_args={}, | ||||
visit_info=visit_info, | visit_info=visit_info, | ||||
) | ) | ||||
if revision_id: | |||||
expected["revision_info"]["revision_url"] = gen_revision_url( | |||||
revision_id, snapshot_context | |||||
) | |||||
assert snapshot_context == expected | assert snapshot_context == expected | ||||
_check_branch_release_revision_parameters( | _check_branch_release_revision_parameters( | ||||
archive_data, expected, browse_context, kwargs, branches, releases | archive_data, expected, browse_context, kwargs, branches, releases | ||||
) | ) | ||||
def _check_branch_release_revision_parameters( | def _check_branch_release_revision_parameters( | ||||
Show All 12 Lines | ): | ||||
url_args.pop("origin_url", None) | url_args.pop("origin_url", None) | ||||
query_params = dict(kwargs) | query_params = dict(kwargs) | ||||
query_params.pop("snapshot_id", None) | query_params.pop("snapshot_id", None) | ||||
expected_branch = dict(base_expected_context) | expected_branch = dict(base_expected_context) | ||||
expected_branch["branch"] = branch["name"] | expected_branch["branch"] = branch["name"] | ||||
expected_branch["revision_id"] = branch["revision"] | expected_branch["revision_id"] = branch["revision"] | ||||
expected_branch["revision_info"] = _get_revision_info( | |||||
archive_data, branch["revision"] | |||||
) | |||||
expected_branch["root_directory"] = branch["directory"] | expected_branch["root_directory"] = branch["directory"] | ||||
expected_branch["query_params"] = {"branch": branch["name"], **query_params} | expected_branch["query_params"] = {"branch": branch["name"], **query_params} | ||||
expected_branch["revision_info"]["revision_url"] = gen_revision_url( | |||||
branch["revision"], expected_branch | |||||
) | |||||
assert snapshot_context == expected_branch | assert snapshot_context == expected_branch | ||||
if releases: | if releases: | ||||
release = random.choice(releases) | release = random.choice(releases) | ||||
snapshot_context = get_snapshot_context( | snapshot_context = get_snapshot_context( | ||||
**kwargs, browse_context=browse_context, release_name=release["name"] | **kwargs, browse_context=browse_context, release_name=release["name"] | ||||
) | ) | ||||
expected_release = dict(base_expected_context) | expected_release = dict(base_expected_context) | ||||
expected_release["branch"] = None | expected_release["branch"] = None | ||||
expected_release["release"] = release["name"] | expected_release["release"] = release["name"] | ||||
expected_release["release_id"] = release["id"] | expected_release["release_id"] = release["id"] | ||||
if release["target_type"] == "revision": | if release["target_type"] == "revision": | ||||
expected_release["revision_id"] = release["target"] | expected_release["revision_id"] = release["target"] | ||||
expected_release["revision_info"] = _get_revision_info( | |||||
archive_data, release["target"] | |||||
) | |||||
expected_release["root_directory"] = release["directory"] | expected_release["root_directory"] = release["directory"] | ||||
expected_release["query_params"] = {"release": release["name"], **query_params} | expected_release["query_params"] = {"release": release["name"], **query_params} | ||||
expected_release["revision_info"]["revision_url"] = gen_revision_url( | |||||
release["target"], expected_release | |||||
) | |||||
assert snapshot_context == expected_release | assert snapshot_context == expected_release | ||||
revision_log = archive_data.revision_log(branch["revision"]) | revision_log = archive_data.revision_log(branch["revision"]) | ||||
revision = revision_log[-1] | revision = revision_log[-1] | ||||
snapshot_context = get_snapshot_context( | snapshot_context = get_snapshot_context( | ||||
**kwargs, browse_context=browse_context, revision_id=revision["id"] | **kwargs, browse_context=browse_context, revision_id=revision["id"] | ||||
Show All 22 Lines | branches.append( | ||||
url=revision_browse_url, | url=revision_browse_url, | ||||
) | ) | ||||
) | ) | ||||
expected_revision = dict(base_expected_context) | expected_revision = dict(base_expected_context) | ||||
expected_revision["branch"] = revision["id"] | expected_revision["branch"] = revision["id"] | ||||
expected_revision["branches"] = branches | expected_revision["branches"] = branches | ||||
expected_revision["revision_id"] = revision["id"] | expected_revision["revision_id"] = revision["id"] | ||||
expected_revision["revision_info"] = _get_revision_info( | |||||
archive_data, revision["id"] | |||||
) | |||||
expected_revision["root_directory"] = revision["directory"] | expected_revision["root_directory"] = revision["directory"] | ||||
expected_revision["query_params"] = {"revision": revision["id"], **query_params} | expected_revision["query_params"] = {"revision": revision["id"], **query_params} | ||||
expected_revision["revision_info"]["revision_url"] = gen_revision_url( | |||||
revision["id"], expected_revision | |||||
) | |||||
assert snapshot_context == expected_revision | assert snapshot_context == expected_revision | ||||
@given(origin_with_releases()) | @given(origin_with_releases()) | ||||
def test_get_release_large_snapshot(archive_data, origin): | def test_get_release_large_snapshot(archive_data, origin): | ||||
snapshot = archive_data.snapshot_get_latest(origin["url"]) | snapshot = archive_data.snapshot_get_latest(origin["url"]) | ||||
release_id = random.choice( | release_id = random.choice( | ||||
[ | [ | ||||
v["target"] | v["target"] | ||||
for v in snapshot["branches"].values() | for v in snapshot["branches"].values() | ||||
if v["target_type"] == "release" | if v["target_type"] == "release" | ||||
] | ] | ||||
) | ) | ||||
release_data = archive_data.release_get(release_id) | release_data = archive_data.release_get(release_id) | ||||
# simulate large snapshot processing by providing releases parameter | # simulate large snapshot processing by providing releases parameter | ||||
# as an empty list | # as an empty list | ||||
release = _get_release( | release = _get_release( | ||||
releases=[], release_name=release_data["name"], snapshot_id=snapshot["id"] | releases=[], release_name=release_data["name"], snapshot_id=snapshot["id"] | ||||
) | ) | ||||
assert release_data["name"] == release["name"] | assert release_data["name"] == release["name"] | ||||
assert release_data["id"] == release["id"] | assert release_data["id"] == release["id"] | ||||
def _get_revision_info(archive_data, revision_id): | |||||
revision_info = None | |||||
if revision_id: | |||||
revision_info = archive_data.revision_get(revision_id) | |||||
revision_info["message_header"] = revision_info["message"].split("\n")[0] | |||||
revision_info["date"] = format_utc_iso_date(revision_info["date"]) | |||||
revision_info["committer_date"] = format_utc_iso_date( | |||||
revision_info["committer_date"] | |||||
) | |||||
return revision_info |