Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/conftest.py
# Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2019 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 inspect | |||||
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 | ||||
import pytest | import pytest | ||||
▲ Show 20 Lines • Show All 130 Lines • ▼ Show 20 Lines | class _ArchiveData: | ||||
It is basically a proxy to Storage interface but it overrides some methods | It is basically a proxy to Storage interface but it overrides some methods | ||||
to retrieve those tests data in a json serializable format in order to ease | to retrieve those tests data in a json serializable format in order to ease | ||||
tests implementation. | tests implementation. | ||||
""" | """ | ||||
def __init__(self, tests_data): | def __init__(self, tests_data): | ||||
self.storage = tests_data['storage'] | self.storage = tests_data['storage'] | ||||
def _call_storage_method(method): | def __getattr__(self, key): | ||||
def call_storage_method(*args, **kwargs): | if key == 'storage': | ||||
return method(*args, **kwargs) | raise AttributeError(key) | ||||
return call_storage_method | |||||
# 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( | return getattr(self.storage, key) | ||||
self.storage, predicate=inspect.ismethod): | |||||
if (not hasattr(self, method_name) and | |||||
not method_name.startswith('_')): | |||||
setattr(self, method_name, _call_storage_method(method)) | |||||
def content_find(self, content): | def content_find(self, content): | ||||
cnt_ids_bytes = {algo_hash: hash_to_bytes(content[algo_hash]) | cnt_ids_bytes = {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]) if cnt else cnt | ||||
▲ Show 20 Lines • Show All 136 Lines • Show Last 20 Lines |