scullery/tests/test_cli.py

import pytest
import os
import vagrant
import dotenv
import libvirt


@pytest.fixture
def script():
    is_packaged = False
    script = 'scullery'
    if not is_packaged:
        script = f'{script}.py'
    return script


@pytest.fixture
def testbase():
    return os.path.abspath(os.path.dirname(os.getenv('PYTEST_CURRENT_TEST')))


@pytest.fixture
def config(testbase, request):
    return '{}/configs/{}.ini'.format(testbase, request.param)


@pytest.fixture
def virt():
    return libvirt.openReadOnly('qemu:///system')

def test_no_arguments(script_runner, script):
    result = script_runner.run(script)
    assert result.success is False
    assert 'the following arguments are required: --suite' in result.stderr


def test_no_config(script_runner, script):
    result = script_runner.run(script, '--suite', 'foo', print_result=True) 
    assert result.success is False
    assert 'Unable to locate configuration file' in result.stderr


@pytest.mark.parametrize('section,message', [
        ('box', 'No "box" section found in the configuration file'),
        ('suites', 'No suites configured'),
        ('boxes', 'No boxes configured'),
        ('suite_box', 'Specified suite does not reference a box'),
        ('box_name', 'Box configuration is incomplete'),
        ('box_image', 'Box configuration is incomplete'),
    ])
def test_config_incomplete(script_runner, script, testbase, section, message):
    configfile = '{}/configs/missing_{}.ini'.format(testbase, section)
    result = script_runner.run(script, '--config', configfile, '--suite', 'one_minion')
    assert result.success is False
    assert message in result.stderr


@pytest.mark.parametrize('section,message', [
        ('suite_box', 'Suite references an undefined box')
    ])
def test_config_undefined(script_runner, script, testbase, section, message):
    configfile = '{}/configs/undefined_{}.ini'.format(testbase, section)
    result = script_runner.run(script, '--config', configfile, '--suite', 'one_minion')
    assert result.success is False
    assert message in result.stderr


@pytest.mark.parametrize('config', ['complete'], indirect=True)
@pytest.mark.parametrize('suite,report', [
        ('one_minion', "[Status(name='scullery-minion0', state='not_created', provider='libvirt')]"),
        ('two_minions', "[Status(name='scullery-minion0', state='not_created', provider='libvirt'), Status(name='scullery-minion1', state='not_created', provider='libvirt')]"),
        ('one_master', "[Status(name='scullery-master0', state='not_created', provider='libvirt')]"),
        ('one_minion_one_master', "[Status(name='scullery-master0', state='not_created', provider='libvirt'), Status(name='scullery-minion0', state='not_created', provider='libvirt')]"),
        ('two_minions_one_master', "[Status(name='scullery-master0', state='not_created', provider='libvirt'), Status(name='scullery-minion0', state='not_created', provider='libvirt'), Status(name='scullery-minion1', state='not_created', provider='libvirt')]")
    ])
def test_status_report_not_running(script_runner, script, config, suite, report):
    result = script_runner.run(script, '--config', config, '--suite', suite, '--status') 
    assert result.success
    assert result.stderr.endswith("INFO - main_interactive: Status report: {}\n".format(report))


@pytest.mark.parametrize('config', ['complete'], indirect=True)
@pytest.mark.parametrize('suite', [
        'one_minion',
        'two_minions',
        'one_master',
        'one_minion_one_master',
        'two_minions_one_master'
    ])
def test_launch_stop(script_runner, script, virt, config, suite):
    cmd = (script, '--config', config, '--suite', suite)
    result = script_runner.run(*cmd)
    assert result.success
    assert 'Launching {} ...\n'.format(suite) in result.stderr
    domains = []
    print(virt.getURI())
    print(virt.listDomainsID())
    for domain in virt.listDomainsID():
        print(domain)
        domains.append(virt.lookupByID(domain).name())
    print(str(domains))
    # consider refining this
    assert any('scullery' in domain for domain in domains)
    result = script_runner.run(*cmd, '--stop')
    assert result.success
    assert 'Destroying machines ...\n' in result.stderr
    domains = []
    for domain in virt.listDomainsID():
        domains.append(virt.lookupByID(domain).name())
    assert not any('scullery' in domain for domain in domains)


@pytest.mark.parametrize('config', ['complete'], indirect=True)
@pytest.mark.parametrize('suite,masters,minions', [
        ('one_minion', None, 'scullery-minion0'),
        ('two_minions', None, 'scullery-minion0,scullery-minion1'),
        ('one_master', 'scullery-master0', None),
        ('one_minion_one_master', 'scullery-master0', 'scullery-minion0'),
        ('two_minions_one_master', 'scullery-master0', 'scullery-minion0,scullery-minion1')
    ])
def test_envfile(script_runner, script, config, suite, masters, minions):
    cmd = (script, '--config', config, '--suite', suite)
    envfile = '.scullery_env'
    result = script_runner.run(*cmd, '--env')
    assert result.success
    assert os.path.isfile(envfile)
    envmap = dotenv.dotenv_values(envfile)
    if masters is not None:
        assert envmap['SCULLERY_MASTERS'] == masters 
    if minions is not None:
        assert envmap['SCULLERY_MINIONS'] == minions
    script_runner.run(*cmd, '--stop')
    assert os.path.isfile(envfile) is False


@pytest.mark.parametrize('config', ['complete'], indirect=True)
@pytest.mark.parametrize('suite', [
        'one_minion',
        'two_minions',
        'one_master',
        'one_minion_one_master',
        'two_minions_one_master'
    ])
def test_stop(script_runner, script, virt, config, suite):
    cmd = (script, '--config', config, '--suite', suite, '--stop')
    result = script_runner.run(*cmd)
    assert result.success
    domains = []
    for domain in virt.listDomainsID():
        domains.append(virt.lookupByID(domain).name())
    assert not any('scullery' in domain for domain in domains)


@pytest.mark.parametrize('config', ['complete'], indirect=True)
@pytest.mark.parametrize('suite', [
        'one_minion',
        'two_minions',
        'one_master',
        'one_minion_one_master',
        'two_minions_one_master'
    ])
def test_already_stopped(script_runner, script, virt, config, suite):
    cmd = (script, '--config', config, '--suite', suite, '--stop', '--debug')
    result = script_runner.run(*cmd)
    assert result.success
    assert result.stderr.endswith('DEBUG - main_interactive: Deployment is not running\n')