KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run molecule automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
