Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/tests/test_cli.py
# Copyright (C) 2019-2020 The Software Heritage developers | # Copyright (C) 2019-2021 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 datetime | import datetime | ||||
from click.formatting import HelpFormatter | from click.formatting import HelpFormatter | ||||
from click.testing import CliRunner | from click.testing import CliRunner | ||||
import pytest | import pytest | ||||
from swh.loader.cli import SUPPORTED_LOADERS, get_loader, list, run | from swh.loader.cli import SUPPORTED_LOADERS, get_loader | ||||
from swh.loader.cli import loader as loader_cli | |||||
from swh.loader.package.loader import PackageLoader | from swh.loader.package.loader import PackageLoader | ||||
def test_get_loader_wrong_input(swh_config): | def test_get_loader_wrong_input(swh_config): | ||||
"""Unsupported loader should raise | """Unsupported loader should raise | ||||
""" | """ | ||||
loader_type = "unknown" | loader_type = "unknown" | ||||
assert loader_type not in SUPPORTED_LOADERS | assert loader_type not in SUPPORTED_LOADERS | ||||
with pytest.raises(ValueError, match="Invalid loader"): | with pytest.raises(ValueError, match="Invalid loader"): | ||||
get_loader(loader_type, url="db-url") | get_loader(loader_type, url="db-url") | ||||
def test_get_loader(swh_config): | def test_get_loader(swh_loader_config): | ||||
"""Instantiating a supported loader should be ok | """Instantiating a supported loader should be ok | ||||
""" | """ | ||||
loader_input = { | loader_input = { | ||||
"archive": {"url": "some-url", "artifacts": [],}, | "archive": {"url": "some-url", "artifacts": []}, | ||||
"debian": {"url": "some-url", "date": "something", "packages": [],}, | "debian": {"url": "some-url", "date": "something", "packages": [],}, | ||||
"deposit": {"url": "some-url", "deposit_id": 1,}, | |||||
"npm": {"url": "https://www.npmjs.com/package/onepackage",}, | "npm": {"url": "https://www.npmjs.com/package/onepackage",}, | ||||
"pypi": {"url": "some-url",}, | "pypi": {"url": "some-url",}, | ||||
} | } | ||||
for loader_type, kwargs in loader_input.items(): | for loader_type, kwargs in loader_input.items(): | ||||
kwargs["storage"] = swh_loader_config["storage"] | |||||
loader = get_loader(loader_type, **kwargs) | loader = get_loader(loader_type, **kwargs) | ||||
assert isinstance(loader, PackageLoader) | assert isinstance(loader, PackageLoader) | ||||
def _write_usage(command, args, max_width=80): | def _write_usage(command, args, max_width=80): | ||||
hf = HelpFormatter(width=max_width) | hf = HelpFormatter(width=max_width) | ||||
hf.write_usage(command, args) | hf.write_usage(command, args) | ||||
return hf.getvalue()[:-1] | return hf.getvalue()[:-1] | ||||
def test_run_help(swh_config): | def test_run_help(swh_config): | ||||
"""Help message should be ok | """Help message should be ok | ||||
""" | """ | ||||
runner = CliRunner() | runner = CliRunner() | ||||
result = runner.invoke(run, ["-h"]) | |||||
result = runner.invoke(loader_cli, ["run", "-h"]) | |||||
assert result.exit_code == 0 | assert result.exit_code == 0 | ||||
usage_prefix = _write_usage( | usage_prefix = _write_usage( | ||||
"run", f"[OPTIONS] [{'|'.join(SUPPORTED_LOADERS)}] URL [OPTIONS]..." | "loader", f"run [OPTIONS] [{'|'.join(SUPPORTED_LOADERS)}]\n" | ||||
) | ) | ||||
expected_help_msg = f"""{usage_prefix} | assert result.output.startswith(usage_prefix) | ||||
Ingest with loader <type> the origin located at <url> | |||||
Options: | |||||
-h, --help Show this message and exit. | |||||
""" | |||||
assert result.output.startswith(expected_help_msg) | |||||
def test_run_pypi(mocker, swh_config): | def test_run_pypi(mocker, swh_config): | ||||
"""Triggering a load should be ok | """Triggering a load should be ok | ||||
""" | """ | ||||
mock_loader = mocker.patch("swh.loader.package.pypi.loader.PyPILoader") | mock_loader = mocker.patch("swh.loader.package.pypi.loader.PyPILoader.load") | ||||
runner = CliRunner() | runner = CliRunner() | ||||
result = runner.invoke(run, ["pypi", "https://some-url"]) | result = runner.invoke( | ||||
loader_cli, ["-C", swh_config, "run", "pypi", "url=https://some-url",] | |||||
) | |||||
assert result.exit_code == 0 | assert result.exit_code == 0 | ||||
mock_loader.assert_called_once_with(url="https://some-url") # constructor | mock_loader.assert_called_once_with() | ||||
def test_run_with_visit_date(mocker, swh_config): | def test_run_with_visit_date(mocker, swh_config): | ||||
"""iso visit_date parameter should be parsed as datetime | """iso visit_date parameter should be parsed as datetime | ||||
""" | """ | ||||
mock_loader = mocker.patch("swh.loader.cli.get_loader") | mock_loader = mocker.patch("swh.loader.cli.get_loader") | ||||
runner = CliRunner() | runner = CliRunner() | ||||
input_date = "2016-05-03 15:16:32+00" | input_date = "2016-05-03 15:16:32+00" | ||||
result = runner.invoke( | result = runner.invoke( | ||||
run, ["npm", "https://some-url", f"visit_date='{input_date}'"] | loader_cli, ["run", "npm", "https://some-url", f"visit_date='{input_date}'"] | ||||
) | ) | ||||
assert result.exit_code == 0 | assert result.exit_code == 0 | ||||
expected_parsed_date = datetime.datetime( | expected_parsed_date = datetime.datetime( | ||||
2016, 5, 3, 15, 16, 32, tzinfo=datetime.timezone.utc | 2016, 5, 3, 15, 16, 32, tzinfo=datetime.timezone.utc | ||||
) | ) | ||||
mock_loader.assert_called_once_with( | mock_loader.assert_called_once_with( | ||||
"npm", url="https://some-url", visit_date=expected_parsed_date | "npm", | ||||
storage={"cls": "memory"}, | |||||
url="https://some-url", | |||||
visit_date=expected_parsed_date, | |||||
) | ) | ||||
def test_list_help(mocker, swh_config): | def test_list_help(mocker, swh_config): | ||||
"""Triggering a load should be ok | """Triggering a load should be ok | ||||
""" | """ | ||||
runner = CliRunner() | runner = CliRunner() | ||||
result = runner.invoke(list, ["--help"]) | result = runner.invoke(loader_cli, ["list", "--help"]) | ||||
assert result.exit_code == 0 | assert result.exit_code == 0 | ||||
usage_prefix = _write_usage( | usage_prefix = _write_usage("loader", "list [OPTIONS]\n") | ||||
"list", f"[OPTIONS] [[{'|'.join(['all'] + SUPPORTED_LOADERS)}]]" | |||||
) | |||||
expected_help_msg = f"""{usage_prefix} | expected_help_msg = f"""{usage_prefix} | ||||
[[{'|'.join(['all'] + SUPPORTED_LOADERS)}]] | |||||
List supported loaders and optionally their arguments | List supported loaders and optionally their arguments | ||||
Options: | Options: | ||||
-h, --help Show this message and exit. | -h, --help Show this message and exit. | ||||
""" | """ | ||||
assert result.output.startswith(expected_help_msg) | assert result.output.startswith(expected_help_msg) | ||||
def test_list_help_npm(mocker, swh_config): | def test_list_help_npm(mocker, swh_config): | ||||
"""Triggering a load should be ok | """Triggering a load should be ok | ||||
""" | """ | ||||
runner = CliRunner() | runner = CliRunner() | ||||
result = runner.invoke(list, ["npm"]) | result = runner.invoke(loader_cli, ["list", "npm"]) | ||||
assert result.exit_code == 0 | assert result.exit_code == 0 | ||||
expected_help_msg = """ | expected_help_msg = """ | ||||
Loader: Load npm origin's artifact releases into swh archive. | Loader: Load npm origin's artifact releases into swh archive. | ||||
signature: (url: str) | |||||
""" | """ | ||||
assert result.output.startswith(expected_help_msg[1:]) | assert result.output.startswith(expected_help_msg[1:]) |