diff --git a/swh/loader/cli.py b/swh/loader/cli.py index 06e35e5..613042b 100644 --- a/swh/loader/cli.py +++ b/swh/loader/cli.py @@ -1,66 +1,68 @@ # Copyright (C) 2019 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 import logging import click import pkg_resources +from typing import Any + from swh.core.cli import CONTEXT_SETTINGS from swh.scheduler.cli.utils import parse_options logger = logging.getLogger(__name__) LOADERS = {entry_point.name.split('.', 1)[1]: entry_point for entry_point in pkg_resources.iter_entry_points('swh.workers') if entry_point.name.split('.', 1)[0] == 'loader'} SUPPORTED_LOADERS = list(LOADERS) -def get_loader(name: str, **kwargs): +def get_loader(name: str, **kwargs) -> Any: """Given a loader name, instantiate it. Args: - name (str): Lister's name - conf (dict): Configuration dict (lister db cnx, policy, priority...) + name: Loader's name + kwargs: Configuration dict (url...) Returns: An instantiated loader """ if name not in LOADERS: raise ValueError( - 'Invalid lister %s: only supported listers are %s' % + 'Invalid loader %s: only supported loaders are %s' % (name, SUPPORTED_LOADERS)) registry_entry = LOADERS[name].load()() logger.debug(f'registry: {registry_entry}') loader_cls = registry_entry['loader'] logger.debug(f'loader class: {loader_cls}') return loader_cls(**kwargs) @click.command(name='run', context_settings=CONTEXT_SETTINGS) @click.option('--type', '-t', help='Loader to run', type=click.Choice(SUPPORTED_LOADERS)) @click.option('--url', '-u', default=None, help="Origin url to load") @click.argument('options', nargs=-1) @click.pass_context def run(ctx, type, url, options): """Loader cli tools Load an origin from its url with loader """ (_, kw) = parse_options(options) logger.debug(f'kw: {kw}') loader = get_loader(type, url=url, **kw) result = loader.load() click.echo(result) diff --git a/swh/loader/tests/__init__.py b/swh/loader/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/swh/loader/tests/conftest.py b/swh/loader/tests/conftest.py new file mode 100644 index 0000000..650ef58 --- /dev/null +++ b/swh/loader/tests/conftest.py @@ -0,0 +1,24 @@ +# Copyright (C) 2019 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 + +import pytest + +from typing import Any, Dict + + +@pytest.fixture +def swh_loader_config() -> Dict[str, Any]: + return { + 'storage': { + 'cls': 'memory', + }, + 'deposit': { + 'url': 'https://deposit.softwareheritage.org/1/private', + 'auth': { + 'username': 'user', + 'password': 'pass', + } + }, + } diff --git a/swh/loader/tests/test_cli.py b/swh/loader/tests/test_cli.py new file mode 100644 index 0000000..c7920bc --- /dev/null +++ b/swh/loader/tests/test_cli.py @@ -0,0 +1,91 @@ +# Copyright (C) 2019 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 + +import pytest + + +from swh.loader.cli import run, get_loader, SUPPORTED_LOADERS +from swh.loader.package.loader import PackageLoader + +from click.testing import CliRunner +from unittest.mock import patch + + +def test_get_loader_wrong_input(swh_config): + """Unsupported loader should raise + + """ + loader_type = 'unknown' + assert loader_type not in SUPPORTED_LOADERS + with pytest.raises(ValueError, match='Invalid loader'): + get_loader(loader_type, url='db-url') + + +def test_get_loader(swh_config): + """Instantiating a supported loader should be ok + + """ + loader_input = { + 'archive': { + 'url': 'some-url', + 'artifacts': [], + }, + 'debian': { + 'url': 'some-url', + 'date': 'something', + 'packages': [], + }, + 'deposit': { + 'url': 'some-url', + 'deposit_id': 1, + }, + 'npm': { + 'url': 'https://www.npmjs.com/package/onepackage', + }, + 'pypi': { + 'url': 'some-url', + }, + } + for loader_type, kwargs in loader_input.items(): + loader = get_loader(loader_type, **kwargs) + assert isinstance(loader, PackageLoader) + + +help_msg = """Usage: run [OPTIONS] [OPTIONS]... + + Loader cli tools + + Load an origin from its url with loader + +Options: + -t, --type [archive|debian|deposit|npm|pypi] + Loader to run + -u, --url TEXT Origin url to load + -h, --help Show this message and exit. +""" + + +def test_run_help(swh_config): + """Help message should be ok + + """ + runner = CliRunner() + result = runner.invoke(run, ['-h']) + assert result.exit_code == 0 + assert result.output.startswith(help_msg) + + +@patch('swh.loader.package.pypi.loader.PyPILoader') +def test_run_pypi(mock_loader, swh_config): + """Triggering a load should be ok + + """ + runner = CliRunner() + result = runner.invoke(run, [ + '--type', 'pypi', + '--url', 'https://some-url' + ]) + assert result.exit_code == 0 + mock_loader.assert_called_once_with(url='https://some-url') # constructor