Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/conftest.py
| Show All 10 Lines | |||||
| from subprocess import run, PIPE | from subprocess import run, PIPE | ||||
| 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 | from rest_framework.test import APIClient | ||||
| from swh.model.hashutil import 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.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 or | shutil.which('ctags') is None or | ||||
| b'+json' not in run(['ctags', '--version'], stdout=PIPE).stdout | b'+json' not in run(['ctags', '--version'], stdout=PIPE).stdout | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 114 Lines • ▼ Show 20 Lines | def __init__(self, tests_data): | ||||
| # Forward calls to non overridden Storage methods to wrapped | # Forward calls to non overridden Storage methods to wrapped | ||||
| # storage instance | # storage instance | ||||
| for method_name, method in inspect.getmembers( | for method_name, method in inspect.getmembers( | ||||
| self.storage, predicate=inspect.ismethod): | self.storage, predicate=inspect.ismethod): | ||||
| if (not hasattr(self, method_name) and | if (not hasattr(self, method_name) and | ||||
| not method_name.startswith('_')): | not method_name.startswith('_')): | ||||
| setattr(self, method_name, _call_storage_method(method)) | setattr(self, method_name, _call_storage_method(method)) | ||||
| def content_find(self, content): | |||||
| 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]) 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 = next(self.storage.content_get_metadata([cnt_id_bytes])) | metadata = next(self.storage.content_get_metadata([cnt_id_bytes])) | ||||
| return converters.from_swh(metadata, | return converters.from_swh(metadata, | ||||
| hashess={'sha1', 'sha1_git', 'sha256', | hashess={'sha1', 'sha1_git', 'sha256', | ||||
| 'blake2s256'}) | 'blake2s256'}) | ||||
| def content_get(self, cnt_id): | def content_get(self, cnt_id): | ||||
| cnt_id_bytes = hash_to_bytes(cnt_id) | cnt_id_bytes = hash_to_bytes(cnt_id) | ||||
| cnt = next(self.storage.content_get([cnt_id_bytes])) | cnt = next(self.storage.content_get([cnt_id_bytes])) | ||||
| return converters.from_content(cnt) | return converters.from_content(cnt) | ||||
| def directory_get(self, dir_id): | |||||
| return { | |||||
| 'id': dir_id, | |||||
| 'content': self.directory_ls(dir_id) | |||||
| } | |||||
| def directory_ls(self, dir_id): | def directory_ls(self, dir_id): | ||||
| cnt_id_bytes = hash_to_bytes(dir_id) | cnt_id_bytes = hash_to_bytes(dir_id) | ||||
| dir_content = map(converters.from_directory_entry, | dir_content = map(converters.from_directory_entry, | ||||
| self.storage.directory_ls(cnt_id_bytes)) | self.storage.directory_ls(cnt_id_bytes)) | ||||
| return list(dir_content) | return list(dir_content) | ||||
| def release_get(self, rel_id): | def release_get(self, rel_id): | ||||
| rel_id_bytes = hash_to_bytes(rel_id) | rel_id_bytes = hash_to_bytes(rel_id) | ||||
| ▲ Show 20 Lines • Show All 108 Lines • Show Last 20 Lines | |||||