spoc/tests/test_cli.py

import pytest
import requests
from argparse import Namespace
from unittest.mock import patch

import spoc
import spoc_cli

class MockResponse:
    def __init__(self, status_code, reason):
        self.status_code = status_code
        self.reason = reason

@pytest.mark.parametrize('exception,expected',[
    (spoc.AppAlreadyInstalledError('someapp'),
     'Application someapp is already installed\n'),
    (spoc.AppNotInstalledError('someapp'),
     'Application someapp is not installed\n'),
    (spoc.AppNotInRepoError('someapp'),
     'Application someapp does not exist in the repository\n'),
    (spoc.AppNotUpdateableError('someapp'),
     'Application someapp does not have a newer version to update\n'),
])
def test_handle_repo_error(exception, expected, capsys):
    spoc_cli.handle_app_error(exception)

    captured = capsys.readouterr()
    assert captured.err == expected

@pytest.mark.parametrize('exception,expected',[
    (requests.HTTPError(response=MockResponse(401, 'Unauthorized')),
     'Invalid username/password'),
    (requests.HTTPError(response=MockResponse(404, 'Not Found')),
     '404 Not Found'),
    (requests.ConnectTimeout(),
     'requests.exceptions.ConnectTimeout'),
])
def test_handle_app_error(exception, expected, capsys):
    spoc_cli.handle_repo_error(exception)

    captured = capsys.readouterr()
    expected = f'Repository "{spoc.config.REGISTRY_HOST}" cannot be reached due to: {expected}\n'
    assert captured.err == expected

@patch('spoc.list_installed', return_value={'anotherapp': '0.1', 'someapp': '0.1'})
@patch('spoc.list_online')
@patch('spoc.list_updates')
def test_listing_installed(list_updates, list_online, list_installed, capsys):
    spoc_cli.listing('installed')

    list_installed.assert_called_once()
    list_online.assert_not_called()
    list_updates.assert_not_called()

    captured = capsys.readouterr()
    assert captured.out == 'anotherapp 0.1\nsomeapp 0.1\n'

@patch('spoc.list_installed')
@patch('spoc.list_online', return_value={'anotherapp': '0.2', 'someapp': '0.2'})
@patch('spoc.list_updates')
def test_listing_online(list_updates, list_online, list_installed, capsys):
    spoc_cli.listing('online')

    list_installed.assert_not_called()
    list_online.assert_called_once()
    list_updates.assert_not_called()

    captured = capsys.readouterr()
    assert captured.out == 'anotherapp 0.2\nsomeapp 0.2\n'

@patch('spoc.list_installed')
@patch('spoc.list_online')
@patch('spoc.list_updates', return_value={'anotherapp': '0.1 -> 0.2', 'someapp': '0.1 -> 0.2'})
def test_listing_updates(list_updates, list_online, list_installed, capsys):
    spoc_cli.listing('updates')

    list_installed.assert_not_called()
    list_online.assert_not_called()
    list_updates.assert_called_once()

    captured = capsys.readouterr()
    assert captured.out == 'anotherapp 0.1 -> 0.2\nsomeapp 0.1 -> 0.2\n'

@patch('spoc.list_installed')
@patch('spoc.list_online')
@patch('spoc.list_updates')
def test_listing_invalid(list_updates, list_online, list_installed, capsys):
    spoc_cli.listing('invalid')

    list_installed.assert_not_called()
    list_online.assert_not_called()
    list_updates.assert_not_called()

    captured = capsys.readouterr()
    assert not captured.out

@patch('spoc.install')
def test_install(install):
    spoc_cli.install('someapp')

    install.assert_called_once_with('someapp')

@patch('spoc.update')
def test_update(update):
    spoc_cli.update('someapp')

    update.assert_called_once_with('someapp')

@patch('spoc.uninstall')
def test_uninstall(uninstall):
    spoc_cli.uninstall('someapp')

    uninstall.assert_called_once_with('someapp')

@patch('spoc.start')
def test_start(start):
    spoc_cli.start('someapp')

    start.assert_called_once_with('someapp')

@patch('spoc.stop')
def test_stop(stop):
    spoc_cli.stop('someapp')

    stop.assert_called_once_with('someapp')

@patch('spoc.status', return_value='STATUS')
def test_status(status, capsys):
    spoc_cli.status('someapp')

    status.assert_called_once_with('someapp')
    captured = capsys.readouterr()
    assert captured.out == 'STATUS\n'

@pytest.mark.parametrize('value,expected',[
    ('1', True),
    ('on', True),
    ('Enable', True),
    ('TRUE', True),
    ('0', False),
    ('off', False),
    ('Disable', False),
    ('FALSE', False),
    ('whatever', False),
])
@patch('spoc.set_autostart')
def test_set_autostart(set_autostart, value, expected):
    spoc_cli.set_autostart('someapp', value)

    set_autostart.assert_called_once_with('someapp', expected)

@patch('spoc.start_autostarted')
def test_start_autostarted(start_autostarted):
    spoc_cli.start_autostarted()

    start_autostarted.assert_called_once()

@patch('spoc.stop_all')
def test_stop_all(stop_all):
    spoc_cli.stop_all()

    stop_all.assert_called_once()

@patch('builtins.input', return_value='someuser')
@patch('getpass.getpass', return_value='somepass')
@patch('spoc.login')
def test_login(login, getpass, nput, capsys):
    spoc_cli.login('somehost')

    nput.assert_called_once_with('Username: ')
    getpass.assert_called_once()
    login.assert_called_once_with('somehost', 'someuser', 'somepass')
    captured = capsys.readouterr()
    assert captured.out == 'Login OK\n'

@patch('builtins.input', return_value='someuser')
@patch('getpass.getpass', return_value='somepass')
@patch('spoc.login', side_effect=requests.ConnectTimeout())
def test_login_bad(login, getpass, nput, capsys):
    with pytest.raises(requests.ConnectTimeout):
        spoc_cli.login('somehost')

    nput.assert_called_once_with('Username: ')
    getpass.assert_called_once()
    login.assert_called_once_with('somehost', 'someuser', 'somepass')
    captured = capsys.readouterr()
    assert captured.out == ''

@patch('spoc.prune')
def test_prune(prune):
    spoc_cli.prune()

    prune.assert_called_once()

@patch('sys.argv', ['foo', 'list'])
@patch('spoc_cli.listing')
def test_main_listing(listing):
    spoc_cli.main()

    listing.assert_called_once_with('installed')

@patch('sys.argv', ['foo', 'list', 'online'])
@patch('spoc_cli.listing')
def test_main_listing_online(listing):
    spoc_cli.main()

    listing.assert_called_once_with('online')

@patch('sys.argv', ['foo', 'install', 'someapp'])
@patch('spoc_cli.install')
def test_main_install(install):
    spoc_cli.main()

    install.assert_called_once_with('someapp')

@patch('sys.argv', ['foo', 'update', 'someapp'])
@patch('spoc_cli.update')
def test_main_update(update):
    spoc_cli.main()

    update.assert_called_once_with('someapp')

@patch('sys.argv', ['foo', 'uninstall', 'someapp'])
@patch('spoc_cli.uninstall')
def test_main_uninstall(uninstall):
    spoc_cli.main()

    uninstall.assert_called_once_with('someapp')

@patch('sys.argv', ['foo', 'start', 'someapp'])
@patch('spoc_cli.start')
def test_main_start(start):
    spoc_cli.main()

    start.assert_called_once_with('someapp')

@patch('sys.argv', ['foo', 'stop', 'someapp'])
@patch('spoc_cli.stop')
def test_main_stop(stop):
    spoc_cli.main()

    stop.assert_called_once_with('someapp')

@patch('sys.argv', ['foo', 'status', 'someapp'])
@patch('spoc_cli.status')
def test_main_status(status):
    spoc_cli.main()

    status.assert_called_once_with('someapp')

@patch('sys.argv', ['foo', 'status'])
@patch('spoc_cli.status')
def test_main_status_all(status):
    spoc_cli.main()

    status.assert_called_once_with(None)

@patch('sys.argv', ['foo', 'autostart', 'someapp', 'on'])
@patch('spoc_cli.set_autostart')
def test_main_autostart(autostart):
    spoc_cli.main()

    autostart.assert_called_once_with('someapp', 'on')

@patch('sys.argv', ['foo', 'start-autostarted'])
@patch('spoc_cli.start_autostarted')
def test_main_start_autostarted(start_autostarted):
    spoc_cli.main()

    start_autostarted.assert_called_once()

@patch('sys.argv', ['foo', 'stop-all'])
@patch('spoc_cli.stop_all')
def test_main_stop_all(stop_all):
    spoc_cli.main()

    stop_all.assert_called_once()


@patch('sys.argv', ['foo', 'login', 'example.com'])
@patch('spoc_cli.login')
def test_main_login(login):
    spoc_cli.main()

    login.assert_called_once_with('example.com')

@patch('sys.argv', ['foo', 'prune'])
@patch('spoc_cli.prune')
def test_main_prune(prune):
    spoc_cli.main()

    prune.assert_called_once()

@patch('spoc_cli.parse_args', return_value=Namespace(action=None))
@patch('spoc_cli.listing')
@patch('spoc_cli.install')
@patch('spoc_cli.update')
@patch('spoc_cli.uninstall')
@patch('spoc_cli.start')
@patch('spoc_cli.stop')
@patch('spoc_cli.status')
@patch('spoc_cli.start_autostarted')
@patch('spoc_cli.stop_all')
@patch('spoc_cli.login')
@patch('spoc_cli.prune')
def test_main_invalid(prune, login, stop_all, start_autostarted, status, stop, start,
                      uninstall, update, install, listing, parse_args):
    spoc_cli.main()

    parse_args.assert_called_once()
    listing.assert_not_called()
    install.assert_not_called()
    update.assert_not_called()
    uninstall.assert_not_called()
    start.assert_not_called()
    stop.assert_not_called()
    status.assert_not_called()
    start_autostarted.assert_not_called()
    stop_all.assert_not_called()
    login.assert_not_called()
    prune.assert_not_called()

@pytest.mark.parametrize('argv', [
    ['list', 'invalid'],
    ['install'],
    ['update'],
    ['uninstall'],
    ['start'],
    ['stop'],
    ['autostart'],
    ['autostart', 'someapp'],
    ['login'],
    ['invalid'],
    [],
])
def test_main_systemexit(argv):
    argv.insert(0, 'foo')
    with patch('sys.argv', argv):
        with pytest.raises(SystemExit):
            spoc_cli.main()

@patch('sys.argv', ['foo', 'start', 'someapp'])
@patch('spoc_cli.start', side_effect=spoc.AppNotInstalledError('someapp'))
@patch('spoc_cli.handle_app_error')
def test_main_apperror(handle_app_error, start):
    spoc_cli.main()

    start.assert_called_once()
    handle_app_error.assert_called_once()

@patch('sys.argv', ['foo', 'login', 'somehost'])
@patch('spoc_cli.login', side_effect=requests.HTTPError(response=MockResponse(401, 'Unauthorized')))
@patch('spoc_cli.handle_repo_error')
def test_main_repoerror(handle_repo_error, login):
    spoc_cli.main()

    login.assert_called_once()
    handle_repo_error.assert_called_once()