diff --git a/swh/loader/pypi/tests/common.py b/swh/loader/pypi/tests/common.py index 2159d13..b80ec57 100644 --- a/swh/loader/pypi/tests/common.py +++ b/swh/loader/pypi/tests/common.py @@ -1,116 +1,114 @@ # Copyright (C) 2018 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from swh.loader.pypi.client import PyPiClient RESOURCES_PATH = './swh/loader/pypi/tests/resources' class PyPiClientWithCache(PyPiClient): """Force the use of the cache to bypass pypi calls """ def __init__(self, temp_directory, cache_dir): super().__init__(temp_directory=temp_directory, cache=True, cache_dir=cache_dir) class LoaderNoStorage: - """Mixin class to inhibit the persistence and keep in memory the data - sent for storage. - - cf. LoaderNoStorage + """Mixin class to inhibit the persistence (storage calls) and keep in + memory the data sent. """ CONFIG_BASE_FILENAME = '' # do not provide a real path ADDITIONAL_CONFIG = { 'storage': ('dict', { 'cls': 'remote', 'args': { 'url': 'http://nowhere:5002/', # do not provide a real storage } }), # do not send any data to the storage 'send_contents': ('bool', False), 'send_directories': ('bool', False), 'send_revisions': ('bool', False), 'send_releases': ('bool', False), 'send_snapshot': ('bool', False), 'debug': ('bool', False), } def __init__(self, client=None): super().__init__(client=client) self.all_contents = [] self.all_directories = [] self.all_revisions = [] self.all_releases = [] self.all_snapshots = [] # typed data self.objects = { 'content': self.all_contents, 'directory': self.all_directories, 'revision': self.all_revisions, 'release': self.all_releases, 'snapshot': self.all_snapshots } def _add(self, type, l): """Add without duplicates and keeping the insertion order. Args: type (str): Type of objects concerned by the action l ([object]): List of 'type' object """ col = self.objects[type] for o in l: if o in col: continue col.extend([o]) def maybe_load_contents(self, all_contents): self._add('content', all_contents) def maybe_load_directories(self, all_directories): self._add('directory', all_directories) def maybe_load_revisions(self, all_revisions): self._add('revision', all_revisions) def maybe_load_releases(self, releases): raise ValueError('If called, the test must break.') def maybe_load_snapshot(self, snapshot): self.objects['snapshot'].append(snapshot) def _store_origin_visit(self): pass def open_fetch_history(self): pass def close_fetch_history_success(self, fetch_history_id): pass def close_fetch_history_failure(self, fetch_history_id): pass def update_origin_visit(self, origin_id, visit, status): pass # Override to do nothing at the end def close_failure(self): pass def close_success(self): pass def pre_cleanup(self): pass