KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use my_container_name method in autotest
Best Python code snippet using autotest_python
subprocess_docker_wrapper.py
Source:subprocess_docker_wrapper.py 
1import os, sys2import subprocess3import re4#print os.environ['http_proxy']5#print os.environ['HTTP_PROXY']6os.environ['http_proxy']=''7os.environ['HTTP_PROXY']=''8# Enable tcp service to allow non-root communication with docker daemon9#sudo service docker stop10#sudo docker --daemon=true -H tcp://127.0.0.1:237611#sudo docker --daemon=true -H tcp://127.0.0.1:2376 -d --bip=172.17.42.1/1612DOCKER = ['docker', '-H', '127.0.0.1:2376']13#DOCKER = ['docker']14 15def docker_run_rm(container_name, image_name, *exec_command):16 docker_command = [] 17 docker_command.extend(DOCKER) 18 docker_command.extend(['run', '--rm', '--name', container_name, image_name])19 docker_command.extend(exec_command)20 p = subprocess.Popen(docker_command,21 stdout=subprocess.PIPE, stderr=subprocess.PIPE)22 out = p.stdout.read()23 print out24 err = p.stderr.read()25 print err26 return out27def docker_run(container_name, image_name, *exec_command):28 # TODO: 29 #if keep_live:30 # docker_command.extend(['-t'])31 docker_command = []32 docker_command.extend(DOCKER)33 docker_command.extend(['run', '--name', container_name, image_name])34 docker_command.extend(exec_command)35 p = subprocess.Popen(docker_command,36 stdout=subprocess.PIPE, stderr=subprocess.PIPE)37 out = p.stdout.read()38 print out39 err = p.stderr.read()40 print err 41def docker_rm(container_name, force=False):42 docker_command = []43 docker_command.extend(DOCKER)44 docker_command.extend(['rm'])45 docker_command.extend(['--force=' + str(force)])46 docker_command.extend([container_name])47 try:48 out = subprocess.check_output(docker_command)49 except subprocess.CalledProcessError as e:50 print "Status: FAIL", e.returncode, e.output51 else:52 return out53def docker_ps_a():54 docker_command = []55 docker_command.extend(DOCKER)56 docker_command.extend(['ps', '-a'])57 p = subprocess.Popen(docker_command,58 stdout=subprocess.PIPE, stderr=subprocess.PIPE)59 out = p.stdout.read()60 print out61 err = p.stderr.read()62 print err 63def docker_images():64 docker_command = []65 docker_command.extend(DOCKER)66 docker_command.extend(['images'])67 out = subprocess.check_output(docker_command)68 #out = p.stdout.read()69 #print out70 #err = p.stderr.read()71 #print err 72 image_names = re.findall(r'^(.*?)\s', out, re.MULTILINE)73 image_names.pop(0) # remove the 'REPOSITORY' column header from the list74 print image_names75 return image_names76def docker_create(container_name, image_name, keep_live=False, shared_directory=None):77 docker_command = []78 docker_command.extend(DOCKER)79 docker_command.extend(['create'])80 if keep_live:81 docker_command.extend(['-t'])82 if shared_directory:83 docker_command.extend(['--volume=' + shared_directory + ':/shared_with_docker_host'])84 docker_command.extend(['--name', container_name, image_name])85 try:86 print docker_command87 out = subprocess.check_output(docker_command)88 except subprocess.CalledProcessError as e:89 print "Status: FAIL", e.returncode, e.output90 sys.exit(1)91 else:92 print out93 return out94def docker_start(container_name):95 docker_command = []96 docker_command.extend(DOCKER)97 docker_command.extend(['start', '--attach=false', container_name])98 try:99 out = subprocess.check_output(docker_command)100 except subprocess.CalledProcessError as e:101 print "Status: FAIL", e.returncode, e.output102 sys.exit(1)103 else:104 print out105def docker_stop(container_name):106 docker_command = []107 docker_command.extend(DOCKER)108 docker_command.extend(['stop', container_name])109 try:110 out = subprocess.check_output(docker_command)111 except subprocess.CalledProcessError as e:112 print "Status: FAIL", e.returncode, e.output113 else:114 print out 115def docker_exec(container_name, exec_command, *exec_command_arguments):116 docker_command = []117 docker_command.extend(DOCKER)118 docker_command.extend(['exec', container_name, exec_command])119 docker_command.extend(exec_command_arguments)120 print docker_command121 try:122 out = subprocess.check_output(docker_command)123 except subprocess.CalledProcessError as e:124 print "Status: FAIL", e.returncode, e.output125 sys.exit(1)126 else:127 return out128if __name__ == '__main__':129 docker_images()130 sys.exit()131 my_container_name = 'delme'132 #my_image_name = 'my_gcc_4_8_2_image:latest'133 my_image_name = 'ubuntu:14.04'134 135 #docker_ps_a()136 #docker_run_rm(my_container_name)137 #docker_ps_a()138 #docker_run(my_container_name)139 #docker_ps_a()140 docker_rm(my_container_name)141 #docker_ps_a()142 docker_create(my_container_name, my_image_name, keep_live=True)143 docker_start(my_container_name)144 #docker_ps_a()145 #docker_rm(my_container_name)146 #docker_ps_a()147 #docker_run_rm(my_container_name, my_image_name, ['ls', '-a'])148 #docker_run(my_container_name, my_image_name, ['ls', '-a'])149 #docker_ps_a()150 #docker_rm(my_container_name)151 #docker_ps_a()152 #image_names = docker_images()153 #out = docker_exec(my_container_name, 'gcc', '--version')154 out = docker_exec(my_container_name, 'ls')155 print out156 docker_stop(my_container_name)157 docker_rm(my_container_name)...
docker.py
Source:docker.py
1from metux.util import log2from metux.util.docker import Docker3from os.path import dirname4from metux.util.specobject import SpecObject5class ContainerDriverDocker(SpecObject):6 spec_mandatory = ['rootfs-image', 'command']7 def __init__(self, param):8 SpecObject.__init__(self, param)9 self.my_params = []10 self.my_docker = Docker()11 self.my_env = {}12 self.my_jail = None13 if 'name' in param:14 self.my_container_name = param['name']15 else:16 self.my_container_name = None17 self.add_params(param)18 def add_params(self, p):19 self.add_mounts(p.get('mounts'))20 self.add_env(p.get('env'))21 self.add_tempdirs(p.get('tempdirs'))22 self.add_opts(p.get('opts'))23 self.add_devices(p.get('devices'))24 def add_mount(self, m):25 if m is not None:26 if m['type'] == 'bind':27 self.add_bind_mount(m['source'], m['target'])28 else:29 raise FT_ConfigError("unsupported mount type: "+m['type'])30 def add_bind_mount(self, source, target):31 self.my_params.append('-v')32 self.my_params.append(source+':'+target)33 def add_mounts(self, mounts):34 if mounts is not None:35 for m in mounts:36 self.add_mount(m)37 def add_tempdir(self, t):38 if t is not None:39 self.my_params.append('--mount')40 self.my_params.append('type=tmpfs,destination=%s' % t)41 def add_tempdirs(self, dirs):42 if dirs is not None:43 for t in dirs:44 self.add_tempdir(t)45 def add_env(self, env):46 if env is not None:47 self.my_env.update(env)48 def add_opts(self, opts):49 if opts is not None:50 for o,v in opts.iteritems():51 if o == 'ipc':52 self.my_params.append('--ipc')53 self.my_params.append(v)54 elif o == 'user':55 self.my_params.append('--user')56 self.my_params.append(v)57 elif o == 'name' and v is not None:58 self.my_container_name = v59 elif o == 'readonly':60 if v:61 self.my_params.append('--read-only')62 else:63 raise FT_ConfigError("container opt %s (%s) not understood" % (o,v))64 def add_devices(self, devs):65 if devs is not None:66 for d in devs:67 self.my_params.append('--device')68 self.my_params.append(d)69 def add_network(self, net):70 self.my_params.append('--network')71 self.my_params.append(net)72 def add_ip(self, ipaddr):73 self.my_params.append('--ip=%s' % ipaddr)74 def run_foreground(self):75 log.info("docker jail running foreground")76 self.check_mandatory(self.spec_mandatory)77 self.my_docker.container_qrun(image = self['rootfs-image'],78 cmdline = self.get_cf_list('command'),79 extra_args = self.my_params,80 env = self.my_env,81 name = self.my_container_name)82 # FIXME: check for already running83 def run_detached(self):84 log.info("docker jail running detached")85 self.check_mandatory(self.spec_mandatory)86 id = self.my_docker.container_run_detached(87 image = self['rootfs-image'],88 cmdline = self.get_cf_list('command'),89 extra_args = self.my_params,90 env = self.my_env,91 name = self.my_container_name)92 log.info("new container ID: "+id)93 return id94 def create(self):95 log.info("creating container: %s" % self.my_container_name)96 self.my_jail = self.my_docker.container_create(97 image = self['rootfs-image'],98 cmdline = self.get_cf_list('command'),99 extra_args = self.my_params,100 env = self.my_env,101 name = self.my_container_name,102 auto_destroy = False)103 log.info("new container ID: "+str(self.my_jail))104 def start(self):105 log.info("starting container: %s" % self.my_container_name)106 return self.my_jail.start()107 def join_network(self, netname):108 self.my_jail.join_network(netname)109 def destroy(self):110 self.my_docker.container_get(self.my_container_name).destroy()111 """execute inside running container"""112 def execute(self, args):113 return self.my_docker.container_get(self.my_container_name).execute(args)114 def check_running(self):115 return len(self.my_docker.container_running_id(self.my_container_name)) > 0116 def check_rootfs(self):117 return self.my_docker.image_check(self['rootfs-image'])118 def network_create(self, name, label=None, driver=None, internal=None, subnet=None, ip_range=None):119 return self.my_docker.network_create(120 name = name,121 label = label,122 driver = driver,123 internal = internal,124 subnet = subnet,125 ip_range = ip_range126 )127 def signal(self, sig):...
libcloud_storage.py
Source:libcloud_storage.py
1"""2Apache Libcloud Storage State3=============================4Manage cloud storage using libcloud5:codeauthor: ``Anthony Shaw <anthonyshaw@apache.org>``6Apache Libcloud Storage (object/blob) management for a full list7of supported clouds, see http://libcloud.readthedocs.io/en/latest/storage/supported_providers.html8Clouds include Amazon S3, Google Storage, Aliyun, Azure Blobs, Ceph, OpenStack swift9.. versionadded:: 2018.3.010:configuration:11 This module uses a configuration profile for one or multiple Storage providers12 .. code-block:: yaml13 libcloud_storage:14 profile_test1:15 driver: google_storage16 key: GOOG0123456789ABCXYZ17 secret: mysecret18 profile_test2:19 driver: s320 key: 1234521 secret: mysecret22Examples23--------24Creating a container and uploading a file25~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~26.. code-block:: yaml27 web_things:28 libcloud_storage.container_present:29 name: my_container_name30 profile: profile131 libcloud_storage.object_present:32 name: my_file.jpg33 container: my_container_name34 path: /path/to/local/file.jpg35 profile: profile136Downloading a file37~~~~~~~~~~~~~~~~~~38This example will download the file from the remote cloud and keep it locally39.. code-block:: yaml40 web_things:41 libcloud_storage.file_present:42 name: my_file.jpg43 container: my_container_name44 path: /path/to/local/file.jpg45 profile: profile146:depends: apache-libcloud47"""48import logging49log = logging.getLogger(__name__)50def __virtual__():51 if "libcloud_storage.list_containers" in __salt__:52 return True53 return (False, "libcloud_storage module could not be loaded")54def state_result(result, message, name, changes):55 return {"result": result, "comment": message, "name": name, "changes": changes}56def container_present(name, profile):57 """58 Ensures a container is present.59 :param name: Container name60 :type name: ``str``61 :param profile: The profile key62 :type profile: ``str``63 """64 containers = __salt__["libcloud_storage.list_containers"](profile)65 match = [z for z in containers if z["name"] == name]66 if len(match) > 0:67 return state_result(True, "Container already exists", name, {})68 else:69 result = __salt__["libcloud_storage.create_container"](name, profile)70 return state_result(True, "Created new container", name, result)71def container_absent(name, profile):72 """73 Ensures a container is absent.74 :param name: Container name75 :type name: ``str``76 :param profile: The profile key77 :type profile: ``str``78 """79 containers = __salt__["libcloud_storage.list_containers"](profile)80 match = [z for z in containers if z["name"] == name]81 if len(match) == 0:82 return state_result(True, "Container already absent", name, {})83 else:84 result = __salt__["libcloud_storage.delete_container"](name, profile)85 return state_result(result, "Deleted container", name, {})86def object_present(container, name, path, profile):87 """88 Ensures a object is presnt.89 :param container: Container name90 :type container: ``str``91 :param name: Object name in cloud92 :type name: ``str``93 :param path: Local path to file94 :type path: ``str``95 :param profile: The profile key96 :type profile: ``str``97 """98 existing_object = __salt__["libcloud_storage.get_container_object"](99 container, name, profile100 )101 if existing_object is not None:102 return state_result(True, "Object already present", name, {})103 else:104 result = __salt__["libcloud_storage.upload_object"](105 path, container, name, profile106 )107 return state_result(result, "Uploaded object", name, {})108def object_absent(container, name, profile):109 """110 Ensures a object is absent.111 :param container: Container name112 :type container: ``str``113 :param name: Object name in cloud114 :type name: ``str``115 :param profile: The profile key116 :type profile: ``str``117 """118 existing_object = __salt__["libcloud_storage.get_container_object"](119 container, name, profile120 )121 if existing_object is None:122 return state_result(True, "Object already absent", name, {})123 else:124 result = __salt__["libcloud_storage.delete_object"](container, name, profile)125 return state_result(result, "Deleted object", name, {})126def file_present(container, name, path, profile, overwrite_existing=False):127 """128 Ensures a object is downloaded locally.129 :param container: Container name130 :type container: ``str``131 :param name: Object name in cloud132 :type name: ``str``133 :param path: Local path to file134 :type path: ``str``135 :param profile: The profile key136 :type profile: ``str``137 :param overwrite_existing: Replace if already exists138 :type overwrite_existing: ``bool``139 """140 result = __salt__["libcloud_storage.download_object"](141 path, container, name, profile, overwrite_existing142 )...
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 autotest 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?
