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 getattr(self.storage, key) | |||||
| return call_storage_method | |||||
| # Forward calls to non overridden Storage methods to wrapped | |||||
| # storage instance | |||||
| for method_name, method in inspect.getmembers( | |||||
| 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 | |||||