acceptancetests/assess_cloud.py

#!/usr/bin/env python
from argparse import ArgumentParser
import logging
from textwrap import dedent
import yaml

from deploy_stack import (
    BootstrapManager,
    )
from jujupy import (
    get_juju_home,
    ModelClient,
    FakeBackend,
    FakeControllerState,
    )
from jujupy.wait_condition import (
    ConditionList,
    )
from utility import (
    add_basic_testing_arguments,
    configure_logging,
    )


def client_from_args(args):
    """Return client from args, as generated by parse_args.

    If the path given is FAKE, fake_juju_client() is used.  Otherwise, the
    client is determined based on the path and version.
    """
    client_class = ModelClient
    if args.juju_bin == 'FAKE':
        controller_state = FakeControllerState()
        version = '2.0.0'
        backend = FakeBackend(controller_state, full_path=args.juju_bin,
                              version=version)
    else:
        version = ModelClient.get_version(args.juju_bin)
        backend = None
    juju_home = get_juju_home()
    with open(args.clouds_file) as f:
        clouds = yaml.safe_load(f)
    if args.config is None:
        config = {}
    else:
        with open(args.config) as f:
            config = yaml.safe_load(f)
    juju_data = client_class.config_class.from_cloud_region(
        args.cloud, args.region, config, clouds, juju_home)
    return client_class(juju_data, version, args.juju_bin, debug=args.debug,
                        soft_deadline=args.deadline, _backend=backend)


def assess_cloud_combined(bs_manager):
    """Assess several operations on a cloud.

    This tests bootstrap, deploy, remove-unit and destroy-controller.
    """
    client = bs_manager.client
    with bs_manager.booted_context(upload_tools=False):
        old_status = client.get_status()
        client.juju('deploy', ('ubuntu'))
        new_status = client.wait_for_started()
        new_machines = [k for k, v in new_status.iter_new_machines(old_status)]
        client.juju('remove-unit', 'ubuntu/0')
        new_status = client.wait_for(
            ConditionList([client.make_remove_machine_condition(n)
                           for n in new_machines]))


def assess_cloud_provisioning(bs_manager, series=None):
    """Assess provisioning operations on a cloud.

    This was created for testing Azure streams.  It tests bootstrap,
    add-machine, remove-machine and destroy-controller.  It does not test
    charms.

    It tests "xenial" and "win2012r2" as representative series on Azure.  It
    tests "xenial" rather than "bionic" because "bionic" will be used for
    bootstrap by default.
    """
    if series is None:
        series = ['win2012r2', 'xenial', 'centos7']
    logging.info(
        'Testing provisioning for series: {}'.format(', '.join(series)))
    client = bs_manager.client
    with bs_manager.booted_context(upload_tools=False):
        old_status = client.get_status()
        for current_series in series:
            client.juju('add-machine', ('--series', current_series))
        new_status = client.wait_for_started()
        new_machines = [k for k, v in new_status.iter_new_machines(old_status)]
        conditions = []
        for machine in new_machines:
            conditions.append(client.remove_machine(machine))
        new_status = client.wait_for(ConditionList(conditions))


def assess_cloud_kill_controller(bs_manager):
    client = bs_manager.client
    with bs_manager.booted_context(upload_tools=False):
        controller_client = client.get_controller_client()
        # We expect juju to die with a connection error (because we just
        # stopped its jujud).  This would normally be seen as a bug, so we
        # suppress it.
        controller_client.juju('run', (
            '--machine', '0', 'sudo service jujud-machine-0 stop'),
            check=False, suppress_err=True)
        bs_manager.has_controller = False


def parse_args(args):
    parser = ArgumentParser(description=dedent("""\
        Test a specified cloud.

        This tests basic provider operations and charm store support.

        The cloud.yaml file must be provided, followed by the name of the
        cloud to test.
        """))
    subparsers = parser.add_subparsers(dest='test')
    for test in ['combined', 'kill-controller', 'provisioning']:
        subparser = subparsers.add_parser(test)
        subparser.add_argument('clouds_file',
                               help='A clouds.yaml file to use for testing.')
        subparser.add_argument('cloud', help='Specific cloud to test.')
        add_basic_testing_arguments(subparser, env=False, existing=False)
        subparser.add_argument('--config')
        if test == 'provisioning':
            subparser.add_argument('--machine-series', action='append',
                                   help='A machine series to add.')
    return parser.parse_args(args)


def main(argv=None):
    args = parse_args(argv)
    configure_logging(args.verbose)
    client = client_from_args(args)
    bs_manager = BootstrapManager.from_client(args, client)
    if args.test == 'combined':
        assess_cloud_combined(bs_manager)
    elif args.test == 'provisioning':
        assess_cloud_provisioning(bs_manager, args.machine_series)
    else:
        assess_cloud_kill_controller(bs_manager)


if __name__ == '__main__':
    main()