Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/tests/common.py
| # Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 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 General Public License version 3, or any later version | # License: GNU 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 os | from os import path | ||||
| import os.path | from urllib.parse import urlparse | ||||
| RESOURCES_PATH = os.path.join(os.path.dirname(__file__), 'resources') | from swh.model.hashutil import hash_to_bytes, hash_to_hex | ||||
| package = '8sync' | |||||
| package_url = 'https://ftp.gnu.org/gnu/8sync/' | DATADIR = path.join(path.abspath(path.dirname(__file__)), 'resources') | ||||
| tarball = [{'date': '944729610', | |||||
| 'archive': 'https://ftp.gnu.org/gnu/8sync/8sync-0.1.0.tar.gz'}] | |||||
| def get_response_cb(request, context): | |||||
| """Mount point callback to fetch on disk the content of a request | |||||
| def init_test_data(mock_tarball_request): | Args: | ||||
| """Initialize the loader with the mock of the tarballs | request (requests.Request): Object requests | ||||
| context (requests.Context): Object holding requests metadata | |||||
| information (headers, etc...) | |||||
| Returns: | |||||
| File descriptor on the on disk file to read from the test context | |||||
| """ | |||||
| url = urlparse(request.url) | |||||
| dirname = url.hostname # pypi.org | files.pythonhosted.org | |||||
| # url.path: pypi/<project>/json -> local file: pypi_<project>_json | |||||
| filename = url.path[1:].replace('/', '_') | |||||
| filepath = path.join(DATADIR, dirname, filename) | |||||
| fd = open(filepath, 'rb') | |||||
| context.headers['content-length'] = str(path.getsize(filepath)) | |||||
| return fd | |||||
| def decode_target(target): | |||||
| if not target: | |||||
| return target | |||||
| target_type = target['target_type'] | |||||
| if target_type == 'alias': | |||||
| decoded_target = target['target'].decode('utf-8') | |||||
| else: | |||||
| decoded_target = hash_to_hex(target['target']) | |||||
| return { | |||||
| 'target': decoded_target, | |||||
| 'target_type': target_type | |||||
| } | |||||
| def check_snapshot(expected_snapshot, expected_branches, storage): | |||||
| """Check for snapshot match. | |||||
| Provide the hashes as hexadecimal, the conversion is done | |||||
| within the method. | |||||
| Args: | |||||
| expected_snapshot (Union[str, dict]): Either the snapshot | |||||
| identifier or the full | |||||
| snapshot | |||||
| expected_branches ([dict]): expected branches or nothing is | |||||
| the full snapshot is provided | |||||
| """ | """ | ||||
| for version in tarball: | if isinstance(expected_snapshot, dict) and not expected_branches: | ||||
| tarball_url = version['archive'] | expected_snapshot_id = expected_snapshot['id'] | ||||
| tarball_filename = tarball_url.split('/')[-1] | expected_branches = expected_snapshot['branches'] | ||||
| tarball_filepath = os.path.join(RESOURCES_PATH, 'tarballs', | else: | ||||
| tarball_filename) | expected_snapshot_id = expected_snapshot | ||||
| with open(tarball_filepath, mode='rb') as tarball_file: | |||||
| tarball_content = tarball_file.read() | snap = storage.snapshot_get(hash_to_bytes(expected_snapshot_id)) | ||||
| mock_tarball_request.get( | assert snap is not None | ||||
| tarball_url, content=tarball_content, | |||||
| headers={'content-length': str(len(tarball_content))}) | branches = { | ||||
| branch.decode('utf-8'): decode_target(target) | |||||
| for branch, target in snap['branches'].items() | |||||
| } | |||||
| assert expected_branches == branches | |||||