def single_node_test(ctx, config):
    """
    - ceph-deploy.single_node_test: null

    #rhbuild testing
    - ceph-deploy.single_node_test:
        rhbuild: 1.2.3

    """
    log.info("Testing ceph-deploy on single node")
    if config is None:
        config = {}
    overrides = ctx.config.get('overrides', {})
    teuthology.deep_merge(config, overrides.get('ceph-deploy', {}))

    if config.get('rhbuild'):
        log.info("RH Build, Skip Download")
        with contextutil.nested(
            lambda: cli_test(ctx=ctx, config=config),
        ):
            yield
    else:
        with contextutil.nested(
            lambda: install_fn.ship_utilities(ctx=ctx, config=None),
            lambda: download_ceph_deploy(ctx=ctx, config=config),
            lambda: cli_test(ctx=ctx, config=config),
        ):
            yield

def task(ctx, config):
    """
    Creates a set of RSA keys, distributes the same key pair
    to all hosts listed in ctx.cluster, and adds all hosts
    to all others authorized_keys list. 

    During cleanup it will delete .ssh/id_rsa, .ssh/id_rsa.pub 
    and remove the entries in .ssh/authorized_keys while leaving
    pre-existing entries in place. 
    """

    if config is None:
        config = {}
    assert isinstance(config, dict), \
        "task hadoop only supports a dictionary for configuration"

    # this does not need to do cleanup and does not depend on 
    # ctx, so I'm keeping it outside of the nested calls
    public_key_string, private_key_string = generate_keys()

    with contextutil.nested(
        lambda: tweak_ssh_config(ctx, config),
        lambda: push_keys_to_host(ctx, config, public_key_string, private_key_string),
        #lambda: tweak_ssh_config(ctx, config),
        ):
        yield

def task(ctx, config):
    if config is None:
        config = {}
    if not isinstance(config, dict):
        raise TypeError("config must be a dict")
    with nested(lambda: install(ctx=ctx, config=config), lambda: smoke(ctx=ctx, config=config)):
        yield

def task(ctx, config):
    """
    handle iscsi admin login after a tgt connection has been established.

    Assume a default host client of client.0 and a sending client of
    client.0 if not specified otherwise.

    Sample tests could be:

    iscsi:

        This sets up a tgt link from client.0 to client.0

    iscsi: [client.1, client.2]

        This sets up a tgt link from client.1 to client.0 and a tgt link
        from client.2 to client.0

    iscsi:
        client.0: client.1
        client.1: client.0

        This sets up a tgt link from client.0 to client.1 and a tgt link
        from client.1 to client.0

    Note that the iscsi image name is iscsi-image, so this only works
    for one image being tested at any one time.
    """
    try:
        pairs = config.items()
    except AttributeError:
        pairs = [('client.0', 'client.0')]
    with contextutil.nested(
            lambda: start_iscsi_initiators(ctx=ctx, tgt_link=pairs),):
        yield

def task(ctx, config):
    """
    Set up and tear down a Ceph cluster.

    For example::

        tasks:
        - install:
             extras: yes
        - ssh_keys:
        - ceph-deploy:
             branch:
                stable: bobtail
             mon_initial_members: 1
             only_mon: true
             keep_running: true

        tasks:
        - install:
             extras: yes
        - ssh_keys:
        - ceph-deploy:
             branch:
                dev: master
             conf:
                mon:
                   debug mon = 20

        tasks:
        - install:
             extras: yes
        - ssh_keys:
        - ceph-deploy:
             branch:
                testing:
             dmcrypt: yes
             separate_journal_disk: yes

    """
    if config is None:
        config = {}

    assert isinstance(config, dict), \
        "task ceph-deploy only supports a dictionary for configuration"

    overrides = ctx.config.get('overrides', {})
    teuthology.deep_merge(config, overrides.get('ceph-deploy', {}))

    if config.get('branch') is not None:
        assert isinstance(
            config['branch'], dict), 'branch must be a dictionary'

    log.info('task ceph-deploy with config ' + str(config))

    with contextutil.nested(
        lambda: install_fn.ship_utilities(ctx=ctx, config=None),
        lambda: download_ceph_deploy(ctx=ctx, config=config),
        lambda: build_ceph_cluster(ctx=ctx, config=config),
    ):
        yield

def task(ctx, config):
    """
    Set up and tear down a Ceph cluster.

    For example::

        tasks:
        - install:
             extras: yes
        - ssh_keys:
        - ceph-deploy:
             branch:
                stable: bobtail
             mon_initial_members: 1

        tasks:
        - install:
             extras: yes
        - ssh_keys:
        - ceph-deploy:
             branch:
                dev: master
             conf:
                mon:
                   debug mon = 20

        tasks:
        - install:
             extras: yes
        - ssh_keys:
        - ceph-deploy:
             branch:
                testing:
    """
    if config is None:
        config = {}

    overrides = ctx.config.get('overrides', {})
    teuthology.deep_merge(config, overrides.get('ceph-deploy', {}))

    assert isinstance(config, dict), \
        "task ceph-deploy only supports a dictionary for configuration"

    overrides = ctx.config.get('overrides', {})
    teuthology.deep_merge(config, overrides.get('ceph-deploy', {}))

    if config.get('branch') is not None:
        assert isinstance(config['branch'], dict), 'branch must be a dictionary'

    with contextutil.nested(
         lambda: ceph_fn.ship_utilities(ctx=ctx, config=None),
         lambda: download_ceph_deploy(ctx=ctx, config=config),
         lambda: build_ceph_cluster(ctx=ctx, config=dict(
                 conf=config.get('conf', {}),
                 branch=config.get('branch',{}),
                 mon_initial_members=config.get('mon_initial_members', None),
                 test_mon_destroy=config.get('test_mon_destroy', None),
                 )),
        ):
        yield

def task(ctx, config):
    """
    Spin up apache configured to run a rados gateway.
    Only one should be run per machine, since it uses a hard-coded port for now.

    For example, to run rgw on all clients::

        tasks:
        - ceph:
        - rgw:

    To only run on certain clients::

        tasks:
        - ceph:
        - rgw: [client.0, client.3]

    or

        tasks:
        - ceph:
        - rgw:
            client.0:
            client.3:

    To run radosgw through valgrind:

        tasks:
        - ceph:
        - rgw:
            client.0:
              valgrind: [--tool=memcheck]
            client.3:
              valgrind: [--tool=memcheck]

    """
    if config is None:
        config = dict(('client.{id}'.format(id=id_), None)
                  for id_ in teuthology.all_roles_of_type(ctx.cluster, 'client'))
    elif isinstance(config, list):
        config = dict((name, None) for name in config)

    for _, roles_for_host in ctx.cluster.remotes.iteritems():
        running_rgw = False
        for role in roles_for_host:
            if role in config.iterkeys():
                assert not running_rgw, "Only one client per host can run rgw."
                running_rgw = True

    with contextutil.nested(
        lambda: create_dirs(ctx=ctx, config=config),
        lambda: ship_config(ctx=ctx, config=config),
        lambda: start_rgw(ctx=ctx, config=config),
        lambda: start_apache(ctx=ctx, config=config),
        ):
        yield

def task(ctx, config):
    """
    Configures dnsmasq to add cnames for teuthology remotes. The task expects a
    dictionary, where each key is a role. If all cnames for that role use the
    same address as that role, the cnames can be given as a list. For example,
    this entry configures dnsmasq on the remote associated with client.0, adding
    two cnames for the ip address associated with client.0:

        - dnsmasq:
            client.0:
            - client0.example.com
            - c0.example.com

    If the addresses do not all match the given role, a dictionary can be given
    to specify the ip address by its target role. For example:

        - dnsmasq:
            client.0:
              client.0.example.com: client.0
              client.1.example.com: client.1
    """
    # apply overrides
    overrides = config.get('overrides', {})
    misc.deep_merge(config, overrides.get('dnsmasq', {}))

    # multiple roles may map to the same remote, so collect names by remote
    remote_names = {}
    for role, cnames in config.iteritems():
        remote = get_remote_for_role(ctx, role)
        if remote is None:
            raise ConfigError('no remote for role %s' % role)

        names = remote_names.get(remote, {})

        if isinstance(cnames, list):
            # when given a list of cnames, point to local ip
            for cname in cnames:
                names[cname] = remote.ip_address
        elif isinstance(cnames, dict):
            # when given a dict, look up the remote ip for each
            for cname, client in cnames.iteritems():
                r = get_remote_for_role(ctx, client)
                if r is None:
                    raise ConfigError('no remote for role %s' % client)
                names[cname] = r.ip_address

        remote_names[remote] = names

    # run a subtask for each unique remote
    subtasks = []
    for remote, cnames in remote_names.iteritems():
        subtasks.extend([ lambda r=remote, cn=cnames: setup_dnsmasq(r, cn) ])

    with contextutil.nested(*subtasks):
        yield

def task(ctx, config):
    """
    Install packages for a given project.

    tasks:
    - install:
        project: ceph
        branch: bar
    - install:
        project: samba
        branch: foo
        extra_packages: ['samba']

    Overrides are project specific:

    overrides:
      install:
        ceph:
          sha1: ...

    :param ctx: the argparse.Namespace object
    :param config: the config dict
    """
    if config is None:
        config = {}
    assert isinstance(config, dict), \
        "task install only supports a dictionary for configuration"

    project, = config.get('project', 'ceph'),
    log.debug('project %s' % project)
    overrides = ctx.config.get('overrides')
    if overrides:
        install_overrides = overrides.get('install', {})
        teuthology.deep_merge(config, install_overrides.get(project, {}))
    log.debug('config %s' % config)

    flavor = get_flavor(config)
    log.info("Using flavor: %s", flavor)

    ctx.summary['flavor'] = flavor

    with contextutil.nested(
        lambda: install(ctx=ctx, config=dict(
            branch=config.get('branch'),
            tag=config.get('tag'),
            sha1=config.get('sha1'),
            flavor=flavor,
            extra_packages=config.get('extra_packages', []),
            extras=config.get('extras', None),
            wait_for_package=ctx.config.get('wait_for_package', False),
            project=project,
        )),
        lambda: ship_utilities(ctx=ctx, config=None),
    ):
        yield

def task(ctx, config):
    """
    Start up tgt.

    To start on on all clients::

        tasks:
        - ceph:
        - tgt:

    To start on certain clients::

        tasks:
        - ceph:
        - tgt: [client.0, client.3]

    or

        tasks:
        - ceph:
        - tgt:
            client.0:
            client.3:

    An image blocksize size can also be specified::
        
        tasks:
        - ceph:
        - tgt:
            image_size = 20480

    The general flow of things here is:
        1. Find clients on which tgt is supposed to run (start_tgtd)
        2. Remotely start up tgt daemon
    On cleanup:
        3. Stop tgt daemon

    The iscsi administration is handled by the iscsi task.
    """
    if config:
        config = {key : val for key, val in config.items()
                if key.startswith('client')}
    # config at this point should only contain keys starting with 'client'
    start_tgtd = []
    remotes = ctx.cluster.only(teuthology.is_type('client')).remotes
    log.info(remotes)
    if not config:
        start_tgtd = ['client.{id}'.format(id=id_)
            for id_ in teuthology.all_roles_of_type(ctx.cluster, 'client')]
    else:
        start_tgtd = config
    log.info(start_tgtd)
    with contextutil.nested(
            lambda: start_tgt_remotes(ctx=ctx, start_tgtd=start_tgtd),):
        yield

def start_iscsi_initiators(ctx, tgt_link):
    """
    This is the sub-task that assigns an rbd to an iscsiadm control and
    performs a login (thereby creating a /dev/sd device).  It performs
    a logout when finished.
    """
    remotes = ctx.cluster.only(teuthology.is_type('client')).remotes
    tgtd_list = []
    for role, host in tgt_link:
        rem = _get_remote(remotes, role)
        rem_name = _get_remote_name(remotes, host)
        rem.run(
            args=[
                'sudo',
                'iscsiadm',
                '-m',
                'discovery',
                '-t',
                'st',
                '-p',
                rem_name,
        ])
        proc = rem.run(
            args=[
                'sudo',
                'iscsiadm',
                '-m',
                'node',
                '--login',
        ])
        if proc.exitstatus == 0:
            tgtd_list.append((rem, rem_name))
        general_io_test(ctx, rem, host)
    try:
        with contextutil.nested(
            lambda: generic_mkfs(ctx=ctx, config={host: {'fs_type': 'xfs'}},
                    devname_rtn=tgt_devname_rtn),
            lambda: generic_mount(ctx=ctx, config={host: None},
                    devname_rtn=tgt_devname_rtn),
            ):
            yield
    finally:
        for rem_info in tgtd_list:
            rem = rem_info[0]
            rem_name = rem_info[1]
            rem.run(
                args=[
                    'sudo',
                    'iscsiadm',
                    '-m',
                    'node',
                    '--logout',
            ])

def task(ctx, config):
    """
    Run some s3-tests suite against rgw, verify opslog socket returns data

    Must restrict testing to a particular client::

        tasks:
        - ceph:
        - rgw: [client.0]
        - s3tests: [client.0]

    To pass extra arguments to nose (e.g. to run a certain test)::

        tasks:
        - ceph:
        - rgw: [client.0]
        - s3tests:
            client.0:
              extra_args: ['test_s3:test_object_acl_grand_public_read']
            client.1:
              extra_args: ['--exclude', 'test_100_continue']
    """
    assert (
        config is None or isinstance(config, list) or isinstance(config, dict)
    ), "task s3tests only supports a list or dictionary for configuration"
    all_clients = ["client.{id}".format(id=id_) for id_ in teuthology.all_roles_of_type(ctx.cluster, "client")]
    if config is None:
        config = all_clients
    if isinstance(config, list):
        config = dict.fromkeys(config)
    clients = config.keys()

    overrides = ctx.config.get("overrides", {})
    # merge each client section, not the top level.
    for (client, cconf) in config.iteritems():
        teuthology.deep_merge(cconf, overrides.get("rgw-logsocket", {}))

    log.debug("config is %s", config)

    s3tests_conf = {}
    for client in clients:
        s3tests_conf[client] = ConfigObj(
            indent_type="",
            infile={"DEFAULT": {"port": 7280, "is_secure": "no"}, "fixtures": {}, "s3 main": {}, "s3 alt": {}},
        )

    with contextutil.nested(
        lambda: download(ctx=ctx, config=config),
        lambda: create_users(ctx=ctx, config=dict(clients=clients, s3tests_conf=s3tests_conf)),
        lambda: configure(ctx=ctx, config=dict(clients=config, s3tests_conf=s3tests_conf)),
        lambda: run_tests(ctx=ctx, config=config),
    ):
        yield

def task(ctx, config):
    if config is None:
        config = dict(('client.{id}'.format(id=id_), None)
                  for id_ in teuthology.all_roles_of_type(ctx.cluster, 'client'))
    elif isinstance(config, list):
        config = dict.fromkeys(config)

    with contextutil.nested(
        lambda: setup(ctx=ctx, config=config),
        lambda: execute(ctx=ctx, config=config),
        ):
        yield

def task(ctx, config):
    """
    Do the setup of a calamari server.

    - calamari_setup:
        version: 'v80.1'
        ice_tool_dir: <directory>
        iceball_location: <directory>

    Options are:

    version -- ceph version we are testing against (defaults to 80.1)
    ice_tool_dir -- optional local directory where ice-tool exists or will
                    be loaded (defaults to src in home directory)
    ice_version  -- version of ICE we're testing (with default)
    iceball_location -- Can be an HTTP URL, in which case fetch from this
                        location, using 'ice_version' and distro information
                        to select the right tarball.  Can also be a local
                        path.  If local path is '.', and iceball is
                        not already present, then we try to build
                        an iceball using the ice_tool_dir commands.
    ice_git_location -- location of ice tool on git
    start_browser -- If True, start a browser.  To be used by runs that will
                     bring up a browser quickly for human use.  Set to False
                     for overnight suites that are testing for problems in
                     the installation itself (defaults to False).
    email -- email address for the user (defaults to [email protected])
    no_epel -- indicates if we should remove epel files prior to yum
               installations.  Defaults to True.
    calamari_user -- user name to log into gui (defaults to admin)
    calamari_password -- calamari user password (defaults to admin)
    """
    cal_svr = None
    start_browser = config.get('start_browser', False)
    no_epel = config.get('no_epel', True)
    for remote_, roles in ctx.cluster.remotes.items():
        if 'client.0' in roles:
            cal_svr = remote_
            break
    if not cal_svr:
        raise RuntimeError('client.0 not found in roles')
    with contextutil.nested(
        lambda: adjust_yum_repos(ctx, cal_svr, no_epel),
        lambda: calamari_install(config, cal_svr),
        lambda: ceph_install(ctx, cal_svr),
        lambda: calamari_connect(ctx, cal_svr),
        lambda: browser(start_browser, cal_svr.hostname),
    ):
        yield

def single_node_test(ctx, config):
    """
    - ceph-deploy.single_node_test: null
    
    #rhbuild testing
    - ceph-deploy.single_node_test: 
        rhbuild: 1.2.3
        
    """
    log.info("Testing ceph-deploy on single node")
    if config is None:
        config = {}

    if config.get("rhbuild"):
        log.info("RH Build, Skip Download")
        with contextutil.nested(lambda: cli_test(ctx=ctx, config=config)):
            yield
    else:
        with contextutil.nested(
            lambda: install_fn.ship_utilities(ctx=ctx, config=None),
            lambda: download_ceph_deploy(ctx=ctx, config=config),
            lambda: cli_test(ctx=ctx, config=config),
        ):
            yield

def task(ctx, config):
    if config is None:
        config = {}
    assert isinstance(config, dict), "task hadoop config must be dictionary"

    overrides = ctx.config.get('overrides', {})
    teuthology.deep_merge(config, overrides.get('hadoop', {}))

    tasks = [
        lambda: install_hadoop(ctx=ctx, config=config),
        lambda: start_hadoop(ctx=ctx, config=config),
    ]

    with contextutil.nested(*tasks):
        yield

def task(ctx, config):
    """This is the main body of the task that gets run."""

    """Take car of some yaml parsing here"""
    if config is not None and not isinstance(config, list) and not isinstance(config, dict):
        assert False, "task parallel_example only supports a list or dictionary for configuration"
    if config is None:
        config = ["client.{id}".format(id=id_) for id_ in teuthology.all_roles_of_type(ctx.cluster, "client")]
    if isinstance(config, list):
        config = dict.fromkeys(config)
    clients = config.keys()

    """Run Multiple contextmanagers sequentially by nesting them."""
    with contextutil.nested(
        lambda: parallel_test(ctx=ctx, config=clients), lambda: sequential_test(ctx=ctx, config=clients)
    ):
        yield