Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.