planetmint/tests/commands/test_utils.py

# Copyright © 2020 Interplanetary Database Association e.V.,
# Planetmint and IPDB software contributors.
# SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
# Code is Apache-2.0 and docs are CC-BY-4.0

import argparse
from argparse import Namespace
import logging

import pytest
from planetmint.config import Config
from unittest.mock import patch


@pytest.fixture
def reset_planetmint_config(monkeypatch):
    monkeypatch.setattr("planetmint.config", Config().init_config("tarantool_db"))


def test_input_on_stderr():
    from planetmint.commands.utils import input_on_stderr, _convert

    with patch("builtins.input", return_value="I love cats"):
        assert input_on_stderr() == "I love cats"

    # input_on_stderr uses `_convert` internally, from now on we will
    # just use that function

    assert _convert("hack the planet") == "hack the planet"
    assert _convert("42") == "42"
    assert _convert("42", default=10) == 42
    assert _convert("", default=10) == 10
    assert _convert("42", convert=int) == 42
    assert _convert("True", convert=bool) is True
    assert _convert("False", convert=bool) is False
    assert _convert("t", convert=bool) is True
    assert _convert("3.14", default=1.0) == 3.14
    assert _convert("TrUe", default=False) is True

    with pytest.raises(ValueError):
        assert _convert("TRVE", default=False)

    with pytest.raises(ValueError):
        assert _convert("ಠ_ಠ", convert=int)


@pytest.mark.usefixtures("ignore_local_config_file", "reset_planetmint_config")
def test_configure_planetmint_configures_planetmint():
    from planetmint.commands.utils import configure_planetmint
    from planetmint.config_utils import is_configured

    assert not is_configured()

    @configure_planetmint
    def test_configure(args):
        assert is_configured()

    args = Namespace(config=None)
    test_configure(args)


@pytest.mark.usefixtures("ignore_local_config_file", "reset_planetmint_config", "reset_logging_config")
@pytest.mark.parametrize(
    "log_level",
    tuple(map(logging.getLevelName, (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL))),
)
def test_configure_planetmint_logging(log_level):
    # TODO: See following comment:
    # This is a dirty test. If a test *preceding* this test makes use of the logger, and then another test *after* this
    # test also makes use of the logger, somehow we get logger.disabled == True, and the later test fails. We need to
    # either engineer this somehow to leave the test env in the same state as it finds it, or make an assessment
    # whether or not we even need this test, and potentially just remove it.

    from planetmint.commands.utils import configure_planetmint

    @configure_planetmint
    def test_configure_logger(args):
        pass

    args = Namespace(config=None, log_level=log_level)
    test_configure_logger(args)
    assert Config().get()["log"]["level_console"] == log_level
    assert Config().get()["log"]["level_logfile"] == log_level


def test_start_raises_if_command_not_implemented():
    from planetmint.commands import utils
    from planetmint.commands.planetmint import create_parser

    parser = create_parser()

    with pytest.raises(NotImplementedError):
        # Will raise because `scope`, the third parameter,
        # doesn't contain the function `run_start`
        utils.start(parser, ["start"], {})


def test_start_raises_if_no_arguments_given():
    from planetmint.commands import utils
    from planetmint.commands.planetmint import create_parser

    parser = create_parser()

    with pytest.raises(SystemExit):
        utils.start(parser, [], {})


@patch("multiprocessing.cpu_count", return_value=42)
def test_start_sets_multiprocess_var_based_on_cli_args(mock_cpu_count):
    from planetmint.commands import utils

    def run_mp_arg_test(args):
        return args

    parser = argparse.ArgumentParser()
    subparser = parser.add_subparsers(title="Commands", dest="command")
    mp_arg_test_parser = subparser.add_parser("mp_arg_test")
    mp_arg_test_parser.add_argument("-m", "--multiprocess", nargs="?", type=int, default=False)

    scope = {"run_mp_arg_test": run_mp_arg_test}
    assert utils.start(parser, ["mp_arg_test"], scope).multiprocess == 1
    assert utils.start(parser, ["mp_arg_test", "--multiprocess"], scope).multiprocess == 42