Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/conftest.py
| # Copyright (C) 2018-2020 The Software Heritage developers | # Copyright (C) 2018-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 json | import json | ||||
| import os | import os | ||||
| import shutil | import shutil | ||||
| import sys | import sys | ||||
| from subprocess import run, PIPE | from subprocess import run, PIPE | ||||
| from typing import List, Optional | from typing import Any, Dict, List, Optional | ||||
| 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 | ||||
| ▲ Show 20 Lines • Show All 141 Lines • ▼ Show 20 Lines | class _ArchiveData: | ||||
| def __getattr__(self, key): | def __getattr__(self, key): | ||||
| if key == "storage": | if key == "storage": | ||||
| raise AttributeError(key) | raise AttributeError(key) | ||||
| # Forward calls to non overridden Storage methods to wrapped | # Forward calls to non overridden Storage methods to wrapped | ||||
| # storage instance | # storage instance | ||||
| return getattr(self.storage, key) | return getattr(self.storage, key) | ||||
| def content_find(self, content): | def content_find(self, content: Dict[str, Any]) -> Dict[str, Any]: | ||||
| cnt_ids_bytes = { | cnt_ids_bytes = { | ||||
| algo_hash: hash_to_bytes(content[algo_hash]) | algo_hash: hash_to_bytes(content[algo_hash]) | ||||
| for algo_hash in ALGORITHMS | for algo_hash in ALGORITHMS | ||||
| if content.get(algo_hash) | if content.get(algo_hash) | ||||
| } | } | ||||
| cnt = self.storage.content_find(cnt_ids_bytes) | cnt = self.storage.content_find(cnt_ids_bytes) | ||||
| return converters.from_content(cnt[0]) if cnt else cnt | return converters.from_content(cnt[0].to_dict()) if cnt else cnt | ||||
| def content_get_metadata(self, cnt_id): | def content_get_metadata(self, cnt_id): | ||||
| cnt_id_bytes = hash_to_bytes(cnt_id) | cnt_id_bytes = hash_to_bytes(cnt_id) | ||||
| metadata = self.storage.content_get_metadata([cnt_id_bytes]) | metadata = self.storage.content_get_metadata([cnt_id_bytes]) | ||||
| contents = metadata[cnt_id_bytes] | contents = metadata[cnt_id_bytes] | ||||
| content = None if not contents else contents[0] | content = None if not contents else contents[0] | ||||
| return converters.from_swh( | return converters.from_swh( | ||||
| content, hashess={"sha1", "sha1_git", "sha256", "blake2s256"} | content, hashess={"sha1", "sha1_git", "sha256", "blake2s256"} | ||||
| ▲ Show 20 Lines • Show All 172 Lines • Show Last 20 Lines | |||||