How to use ansible_args method in molecule

Best Python code snippet using molecule_python

tasks.py

Source:tasks.py

1import invoke2from invoke import task3from invoke import Collection4import click5import subprocess6import getpass7import os8import re9import json10try:  # py311    from shlex import quote12except ImportError:  # py213    from pipes import quote14CONTEXT_SETTINGS = dict(15    max_content_width=12016)17@click.group(context_settings=CONTEXT_SETTINGS)18def cli():19    pass20@task21def _virtualenv(ctx, virtualenv_path=None):22    """23    Initialize a virtualenv folder.24    """25    ctx.virtualenv_path = os.path.dirname(__file__) + '/' + ctx.virtualenv_path26    virtualenv_path = virtualenv_path or ctx.virtualenv_path27    if not check_virtualenv(ctx, virtualenv_path):28        if not os.path.exists(os.path.dirname(virtualenv_path)):29            os.makedirs(os.path.dirname(virtualenv_path))30        ctx.run(' '.join(['virtualenv', virtualenv_path]))31        if not check_virtualenv(ctx, virtualenv_path):32            raise Exception('python install fails')33@task(pre=[_virtualenv])34def galaxy(ctx, virtualenv_path=None):35    """36    Install needed ansible dependencies with ansible-galaxy.37    """38    virtualenv_path = ctx.virtualenv_path;39    ctx.run(_vcommand(virtualenv_path, 'ansible-galaxy',40                      'install',41                      '-r', os.path.dirname(__file__) + '/requirements.yml',42                      '--roles-path', os.path.dirname(__file__) +'/dependencies/'))43@task44def _ansible(ctx, virtualenv_path=None, skip=None, version=None):45    """46    Install ansible inside a virtualenv folder.47    """48    skip = skip if (skip is not None) \49        else ctx.ansible.skip50    dependencies = ctx.dependencies51    if not skip:52        virtualenv_path = virtualenv_path or ctx.virtualenv_path53        package = ctx.ansible.package_name54        version = version or ctx.ansible.version55        _pip_package(ctx, package, version)56        for dependency in dependencies:57            _pip_package(ctx, dependency['name'],58                         dependency.get('version', None))59    else:60        print('ansible not managed (ansible.skip: yes)')61@task62def _sphinx(ctx, virtualenv_path=None, skip=None, version=None):63    """64    Install sphinx inside a virtualenv folder.65    """66    skip = skip if (skip is not None) \67        else ctx.sphinx.skip68    dependencies = ctx.sphinx.dependencies69    if not skip:70        virtualenv_path = virtualenv_path or ctx.virtualenv_path71        package = ctx.sphinx.package_name72        version = version or ctx.sphinx.version73        _pip_package(ctx, package, version)74        for dependency in dependencies:75            _pip_package(ctx, dependency['name'],76                         dependency.get('version', None))77    else:78        print('sphinx not managed (sphinx.skip: yes)')79@task80def _recommonmark(ctx, virtualenv_path=None, skip=None, version=None):81    """82    Install recommonmark inside a virtualenv folder.83    """84    skip = skip if (skip is not None) \85        else ctx.recommonmark.skip86    dependencies = ctx.dependencies87    if not skip:88        virtualenv_path = virtualenv_path or ctx.virtualenv_path89        package = ctx.recommonmark.package_name90        version = version or ctx.recommonmark.version91        _pip_package(ctx, package, version)92        for dependency in dependencies:93            _pip_package(ctx, dependency['name'],94                         dependency.get('version', None))95    else:96        print('recommonmark not managed (recommonmark.skip: yes)')97def _pip_package(ctx, package, version=None, virtualenv_path=None):98    """99    Install a pypi package (with pip) inside a virtualenv folder.100    """101    virtualenv_path = virtualenv_path or ctx.virtualenv_path102    if not check_pip(ctx, virtualenv_path, package, version):103        pip_install(ctx, virtualenv_path, package, version)104        if not check_pip(ctx, virtualenv_path, package, version):105            raise Exception('{} install fails'.format(package))106@task(pre=[_virtualenv, _ansible, _sphinx, _recommonmark, galaxy])107def configure(ctx):108    """109    Trigger virtualenv, ansible initialization.110    All this tools are needed to handle ansible jobs and documentation111    generation.112    """113    pass114def _ansible_playbook(ctx, virtualenv_path=None, playbook=None,115                      extra_vars={}, ansible_args=''):116    """117    Run an ansible playbook118    """119    virtualenv_path = virtualenv_path or ctx.virtualenv_path120    _check_hosts(virtualenv_path)121    playbook = playbook or ctx.default_playbook122    playbook = os.path.dirname(__file__) + '/' + playbook123    os.environ['PATH'] = \124        os.path.join(virtualenv_path, 'bin') + ':' + os.environ['PATH']125    os.environ['ANSIBLE_CONFIG'] = os.path.dirname(__file__) + '/' + 'etc/ansible.cfg'126    os.environ['ANSIBLE_ROOT'] = os.path.dirname(__file__)127    args = [playbook]128    if ansible_args:129        args.extend(ansible_args.split(' '))130    if extra_vars:131        args.append('--extra-vars')132        args.append(quote(json.dumps(extra_vars)))133    ctx.run(_vcommand(virtualenv_path, 'ansible-playbook', *args), pty=True)134def _docs_makefile(target, ctx, virtualenv_path=None):135    """136    Trigger a sphinx Makefile target. Used to delegate all documentation jobs to137    original sphinx Makefile.138    """139    virtualenv_path = virtualenv_path or ctx.virtualenv_path140    os.environ['PATH'] = \141        ':'.join([142            os.path.abspath(os.path.join(virtualenv_path, 'bin')),143            os.environ['PATH']])144    args = ['make', '-C', '.', target]145    ctx.run(' '.join(args), pty=True)146@task(pre=[configure])147def docs(ctx, virtualenv_path=None):148    """149    Rebuild documentation.150    """151    _docs_makefile('html', ctx, virtualenv_path)152@task(name='docs-clean', pre=[configure])153def docs_clean(ctx, virtualenv_path=None):154    """155    Clean generated documentation.156    """157    _docs_makefile('clean', ctx, virtualenv_path)158@task(name='docs-live', pre=[docs, configure])159def docs_live(ctx, virtualenv_path=None):160    """161    Live build of documentation on each modification. Open a browser with a162    local server to serve documentation. Opened page is reloaded each time163    documentation is generated.164    """165    virtualenv_path = virtualenv_path or ctx.virtualenv_path166    os.environ['PATH'] = \167        ':'.join([168            os.path.abspath(os.path.join(virtualenv_path, 'bin')),169            os.environ['PATH']])170    command = ' '.join([171        'sphinx-autobuild',172        '-B',173        '--ignore', '"*.swp"',174        '--ignore', '"*.log"',175        '--ignore', '"*~"',176        '--ignore', '"*~"',177        '-b', 'html',178        os.path.dirname(__file__) + '/docs/source',179        os.path.dirname(__file__) + '/docs/build/html'180    ])181    ctx.run(command, pty=True)182def _check_hosts(virtualenv_path):183    """184    Check if user present in hosts file185    """186    inventory_path = os.path.dirname(__file__) + '/inventory/hosts'187    hosts = subprocess.check_output([os.path.join(virtualenv_path, 'bin', 'ansible'), 'all', '-i', inventory_path, '--list-hosts'])188    if getpass.getuser() not in hosts:189        print 'Error : You must add your username (' + getpass.getuser() + ') in inventory/hosts'190        quit()191@task(192    name='list-hosts',193    pre=[configure],194    help={195        'limit': 'limit selected hosts by a pattern or a group; for example `-l qualification`'196    }197)198def ansible_list_hosts(ctx, limit=None):199    """200    List hosts from inventory201    """202    inventory_path = os.path.dirname(__file__) + '/inventory/hosts'203    ansible_cmd = ['ansible', 'all', '-i', inventory_path, '--list-hosts']204    if limit is not None:205        ansible_cmd.extend(['-l', limit])206    hosts = subprocess.check_output(ansible_cmd)207    print hosts208@task(209      name='run',210      pre=[configure],211      help={212          'virtualenv-path': 'custom virtualenv folder',213          'playbook': 'playbook to run',214          'check': 'run the ansible playbook with check option',215          'diff': 'run the ansible playbook with diff option',216          'ansible-args': 'additional ansible args',217          'extra-vars': 'extra ansible vars (json format)'218      }219)220def ansible_run(ctx, virtualenv_path=None, playbook=None, check=False, diff=False, extra_vars='', ansible_args=''):221    """222    Run an arbitrary ansible playbook.223    """224    extra_vars_parsed = {}225    ansible_args = ansible_args + ' ' + ctx.ansible_default_args226    if check:227        ansible_args = ansible_args + ' --check '228    if diff:229        ansible_args = ansible_args + ' --diff '230    if extra_vars:231        extra_vars_parsed = json.loads(extra_vars)232    _ansible_playbook(ctx, virtualenv_path=virtualenv_path, playbook=playbook,233                      ansible_args=ansible_args, extra_vars=extra_vars_parsed)234@task(name='up', pre=[configure])235def vagrant_up(ctx, environment='localhost', ansible_args=''):236    """237    Deploy a complete vagrant environment238    """239    _ansible_playbook(ctx, playbook='playbooks/vagrant-up.yml',240                      ansible_args=ansible_args + '--ask-become-pass',241                      extra_vars={'target_host': environment})242@task(name='destroy', pre=[configure])243def vagrant_destroy(ctx, environment='localhost', ansible_args=''):244    """245    Destroy a complete vagrant environment (erase vms)246    """247    _ansible_playbook(ctx, playbook='playbooks/vagrant-destroy.yml',248                      ansible_args=ansible_args,249                      extra_vars={'target_host': environment})250@task(name='halt', pre=[configure])251def vagrant_halt(ctx, ansible_args=''):252    """253    Halt a complete vagrant environment (shutdown vms, vms NOT erased)254    """255    _ansible_playbook(ctx, playbook='playbooks/vagrant-halt.yml',256                      ansible_args=ansible_args,257                      extra_vars={'target_host': environment})258@task(name='ssh', pre=[configure],259      help={'host': 'host to ssh to'})260def vagrant_ssh(ctx, host):261    """262    ssh in `host` vagrant vm263    """264    command_ssh = ' '.join([265        'ssh',266        '-F',267        os.path.join(os.path.dirname(__file__) + '/vagrant', host, 'ssh_config'),268        host269    ])270    ctx.run(command_ssh, pty=True)271def check_virtualenv(ctx, virtualenv_path):272    """273    Check if virtualenv is initialized in virtualenv folder (based on274    bin/python file).275    """276    r = ctx.run(' '.join([277        os.path.join(virtualenv_path, 'bin/python'),278        '--version'279    ]), warn=True, hide='both')280    return r.ok281def check_pip(ctx, virtualenv_path, package, version):282    """283    Check if a pypi package is installed in virtualenv folder.284    """285    r = ctx.run(' '.join([286        os.path.join(virtualenv_path, 'bin/pip'),287        'show',288        package289    ]), hide='both', warn=True)290    if not r.ok:291        # pip show package error - package is not here292        return False293    if version is None:294        # no version check needed295        return True296    # package here, check version297    m = re.search(r'^Version: (.*)$', r.stdout, re.MULTILINE)298    result = m is not None and m.group(1).strip() == version299    return result300def pip_install(ctx, virtualenv_path, package, version):301    """302    Install a pypi package in a virtualenv folder with pip.303    """304    pkgspec = None305    if version is None:306        pkgspec = package307    else:308        pkgspec = '{}=={}'.format(package, version)309    ctx.run(' '.join([310        os.path.join(virtualenv_path, 'bin/pip'),311        'install',312        pkgspec313    ]))314def _vcommand(virtualenv_path, command, *args):315    """316    Run a command from virtualenv folder.317    """318    cl = []319    cl.append(os.path.join(virtualenv_path, 'bin', command))320    cl.extend(args)321    return ' '.join(cl)322def _command(command, *args):323    """324    Run a command.325    """326    cl = []327    cl.append(os.path.join(command))328    cl.extend(args)329    return ' '.join(cl)330vagrant_ns = Collection('vagrant', vagrant_up, vagrant_destroy, vagrant_halt,331                                   vagrant_ssh)332ansible_ns = Collection('ansible', ansible_list_hosts, ansible_run)333ns = Collection(configure, galaxy, vagrant_ns, ansible_ns,334                docs, docs_live, docs_clean)335ns.configure({336    'ansible': {337        'package_name': 'ansible'338    },339    'sphinx': {340        'package_name': 'sphinx',341        'dependencies': [342            { 'name': 'sphinx-bootstrap-theme' },343            { 'name': 'sphinx-autobuild' }344        ]345    },346    'recommonmark': {347        'package_name': 'recommonmark'348    },349    'dependencies': []350})351if __name__ == '__main__':...

test_cisco_file_transfer.py

Source:test_cisco_file_transfer.py

1from DEVICE_CREDS import my_device2def test_setup_initial_state(ansible_module):3    '''4    Transfer initial file to remote device5    '''6    ansible_args = dict( 7        source_file="/home/kbyers/scp_sidecar/tests/cisco_logging.txt",8        dest_file="cisco_logging.txt",9        enable_scp="true",       10    )11    ansible_args.update(my_device)12    module_out = ansible_module.cisco_file_transfer(**ansible_args)13def test_file_already_exists(ansible_module):14    '''15    Make sure file already exists and not 'changed'16    '''17    ansible_args = dict( 18        source_file="/home/kbyers/scp_sidecar/tests/cisco_logging.txt",19        dest_file="cisco_logging.txt",20    )21    ansible_args.update(my_device)22    module_out = ansible_module.cisco_file_transfer(**ansible_args)23    for host, result in module_out.items():24        assert result['changed'] is False25        assert result['msg'] == 'File exists and has correct MD5'26def test_xfer_file(ansible_module):27    '''28    Transfer a new file to the remote device29    '''30    # Will disable scp after test31    ansible_args = dict( 32        source_file="/home/kbyers/scp_sidecar/tests/cisco_logging1.txt",33        dest_file="cisco_logging.txt",34        enable_scp="true",       35    )36    ansible_args.update(my_device)37    module_out = ansible_module.cisco_file_transfer(**ansible_args)38    for host, result in module_out.items():39        assert result['changed'] is True40        assert result['msg'] == 'File successfully transferred to remote device'41def test_verify_file(ansible_module):42    '''43    Verify the new file on the remote device44    '''45    ansible_args = dict( 46        source_file="/home/kbyers/scp_sidecar/tests/cisco_logging1.txt",47        dest_file="cisco_logging.txt",48    )49    ansible_args.update(my_device)50    module_out = ansible_module.cisco_file_transfer(**ansible_args)51    for host, result in module_out.items():52        assert result['changed'] is False53        assert result['msg'] == 'File exists and has correct MD5'54def test_xfer_and_scp_enable(ansible_module):55    '''56    Transfer a new file to the remote device57    Ansible module must enable scp for this to work58    '''59    ansible_args = dict( 60        source_file="/home/kbyers/scp_sidecar/tests/cisco_logging.txt",61        dest_file="cisco_logging.txt",62        enable_scp="true",       63    )64    ansible_args.update(my_device)65    module_out = ansible_module.cisco_file_transfer(**ansible_args)66    for host, result in module_out.items():67        assert result['changed'] is True68        assert result['msg'] == 'File successfully transferred to remote device'69def test_overwrite(ansible_module):70    '''71    Verify overwrite when file already exists results in an error72    '''73    ansible_args = dict( 74        source_file="/home/kbyers/scp_sidecar/tests/cisco_logging1.txt",75        dest_file="cisco_logging.txt",76        enable_scp="true",       77        overwrite="false",78    )79    ansible_args.update(my_device)80    module_out = ansible_module.cisco_file_transfer(**ansible_args)81    for host, result in module_out.items():82        assert result['changed'] is False...

test_molecule.py

Source:test_molecule.py

1# Licensed under the Apache License, Version 2.0 (the "License");2# you may not use this file except in compliance with the License.3# You may obtain a copy of the License at4#5#    http://www.apache.org/licenses/LICENSE-2.06#7# Unless required by applicable law or agreed to in writing, software8# distributed under the License is distributed on an "AS IS" BASIS,9# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or10# implied.11# See the License for the specific language governing permissions and12# limitations under the License.13import subprocess14import pytest15def test_molecule(pytestconfig):16    cmd = ['python', '-m', 'molecule']17    scenario = pytestconfig.getoption("scenario")18    ansible_args = pytestconfig.getoption("ansible_args")19    if ansible_args:20        cmd.append('converge')21        if scenario:22            cmd.extend(['--scenario-name', scenario])23        cmd.append('--')24        cmd.extend(ansible_args.split())25    else:26        cmd.append('test')27        if scenario:28            cmd.extend(['--scenario-name', scenario])29        else:30            cmd.append('--all')31    try:32        assert subprocess.call(cmd) == 033    finally:34        if ansible_args:35            cmd = ['python', '-m', 'molecule', 'destroy']36            if scenario:37                cmd.extend(['--scenario-name', scenario])...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.