Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/conftest.py
| Show All 13 Lines | |||||
| import pytest | import pytest | ||||
| from django.core.cache import cache | from django.core.cache import cache | ||||
| from hypothesis import settings, HealthCheck | from hypothesis import settings, HealthCheck | ||||
| 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.web.common import converters | from swh.web.common import converters | ||||
| from swh.web.common.typing import OriginVisitInfo | |||||
| from swh.web.tests.data import get_tests_data, override_storages | from swh.web.tests.data import get_tests_data, override_storages | ||||
| 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_latest | from swh.storage.algos.snapshot import snapshot_get_latest | ||||
| # 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 202 Lines • ▼ Show 20 Lines | def origin_visit_get(self, origin_url): | ||||
| visit_status = self.storage.origin_visit_status_get_latest( | visit_status = self.storage.origin_visit_status_get_latest( | ||||
| origin_url, visits[i]["visit"] | origin_url, visits[i]["visit"] | ||||
| ) | ) | ||||
| visits[i] = converters.from_origin_visit( | visits[i] = converters.from_origin_visit( | ||||
| {**visits[i], **visit_status.to_dict()} | {**visits[i], **visit_status.to_dict()} | ||||
| ) | ) | ||||
| return visits | return visits | ||||
| def origin_visit_get_by(self, origin_url, visit_id): | def origin_visit_get_by(self, origin_url: str, visit_id: int) -> OriginVisitInfo: | ||||
| visit = self.storage.origin_visit_get_by(origin_url, visit_id) | 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) | visit_status = self.storage.origin_visit_status_get_latest(origin_url, visit_id) | ||||
| return converters.from_origin_visit({**visit, **visit_status.to_dict()}) | assert visit_status is not None | ||||
| return converters.from_origin_visit( | |||||
| {**visit_status.to_dict(), "type": visit.type} | |||||
| ) | |||||
| def origin_visit_status_get_latest( | def origin_visit_status_get_latest( | ||||
| self, | self, | ||||
| origin_url, | origin_url, | ||||
| type: Optional[str] = None, | type: Optional[str] = None, | ||||
| allowed_statuses: Optional[Iterable[str]] = None, | allowed_statuses: Optional[Iterable[str]] = None, | ||||
| require_snapshot: bool = False, | require_snapshot: bool = False, | ||||
| ): | ): | ||||
| ▲ Show 20 Lines • Show All 90 Lines • Show Last 20 Lines | |||||