Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/conftest.py
| # Copyright (C) 2018-2021 The Software Heritage developers | # Copyright (C) 2018-2021 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 | ||||||||
| from datetime import timedelta | |||||||||
| import json | import json | ||||||||
| import os | import os | ||||||||
| import shutil | import shutil | ||||||||
| from subprocess import PIPE, run | from subprocess import PIPE, run | ||||||||
| import sys | import sys | ||||||||
| from typing import Any, Dict, List, Optional | from typing import Any, Dict, List, Optional | ||||||||
| from _pytest.python import Function | from _pytest.python import Function | ||||||||
| from hypothesis import HealthCheck, settings | from hypothesis import HealthCheck, settings | ||||||||
| import pytest | import pytest | ||||||||
| from django.core.cache import cache | from django.core.cache import cache | ||||||||
| from rest_framework.test import APIClient, APIRequestFactory | from rest_framework.test import APIClient, APIRequestFactory | ||||||||
| from swh.model.hashutil import ALGORITHMS, hash_to_bytes | 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.origin import origin_get_latest_visit_status | ||||||||
| from swh.storage.algos.snapshot import snapshot_get_all_branches, snapshot_get_latest | 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.auth.utils import OIDC_SWH_WEB_CLIENT_ID | ||||||||
| from swh.web.common import converters | 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.common.typing import OriginVisitInfo | ||||||||
| from swh.web.config import get_config | from swh.web.config import get_config | ||||||||
| from swh.web.tests.data import get_tests_data, override_storages | from swh.web.tests.data import get_tests_data, override_storages | ||||||||
| # Used to skip some tests | # Used to skip some tests | ||||||||
| ctags_json_missing = ( | ctags_json_missing = ( | ||||||||
| shutil.which("ctags") is None | shutil.which("ctags") is None | ||||||||
| or b"+json" not in run(["ctags", "--version"], stdout=PIPE).stdout | or b"+json" not in run(["ctags", "--version"], stdout=PIPE).stdout | ||||||||
| ▲ Show 20 Lines • Show All 391 Lines • ▼ Show 20 Lines | def inner(func): | ||||||||
| ) | ) | ||||||||
| nextitem = parent_test # prevents pytest from tearing down module fixtures | nextitem = parent_test # prevents pytest from tearing down module fixtures | ||||||||
| item.ihook.pytest_runtest_setup(item=item) | item.ihook.pytest_runtest_setup(item=item) | ||||||||
| item.ihook.pytest_runtest_call(item=item) | item.ihook.pytest_runtest_call(item=item) | ||||||||
| item.ihook.pytest_runtest_teardown(item=item, nextitem=nextitem) | item.ihook.pytest_runtest_teardown(item=item, nextitem=nextitem) | ||||||||
| return inner | 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", | |||||||||
ardumont: not that it matters here but it's svn ;) | |||||||||
| "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), | |||||||||
| } | |||||||||
| ) | |||||||||
| # 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", | |||||||||
Not Done Inline Actions
same ;) ardumont: same ;) | |||||||||
| "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() | |||||||||
not that it matters here but it's svn ;)