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

How to use container_exists method in autotest

Best Python code snippet using autotest_python

podman_api.py

Source:podman_api.py

...150            logger.warning(f"Could not create container {name}. {result.message.get('cause')}")151        return ''152    def container_delete(self, name: str) -> None:153        logger.info(f'Delete container {name}')154        container_exists = self.container_exists(name)155        if container_exists:156            container_status = 'unkown'157            container_details = self.container_inspect(name)158            container_states = container_details.get('state')159            if container_states:160                container_status = container_states.get('Status')161            if container_status == 'exited' or container_status == 'configured':162                url = f'/{self.api_version}/libpod/containers/{name}'163                resp = self.podman_socket.delete(url=url)164                result = PodmanApiResponse(resp)165            else:166                logger.warning(f"Can not delete container with status {container_status}")167                return168        else:169            logger.warning(f"container {name} does not exist")170            return171        if result.successfully:172            logger.info(f"Deleted container {name}")173        else:174            logger.warning(f"Could not delete Container {name}")175            if result.message:176                logger.warning(f"{result.message.get('cause')}")177    def container_start(self, name: str) -> None:178        logger.info(f'Start container {name}')179        container_exists = self.container_exists(name)180        if container_exists:181            url = f'/{self.api_version}/libpod/containers/{name}/start'182            resp = self.podman_socket.post(url=url)183            result = PodmanApiResponse(resp)184        else:185            logger.warning(f"Could not start container {name}. Container does not exists")186            return187        if result.successfully:188            logger.info(f"Started container {name}")189        else:190            logger.warning(f"Could not start container {name}. {result.message.get('cause')}")191    def container_stop(self, name: str) -> None:192        logger.info(f'Stop container {name}')193        if self.container_exists(name):194            url = f'/{self.api_version}/libpod/containers/{name}/stop'195            resp = self.podman_socket.post(url=url, timeout=60)196            result = PodmanApiResponse(resp)197        else:198            logger.warning(f"Could not stop container {name}. Container does not exists")199            return200        if result.successfully:201            logger.info(f"Stopped container {name}")202        else:203            logger.warning(f"Could not stop container {name}. {result.message.get('cause')}")204    def container_inspect(self, name: str) -> Dict[str, Any]:205        logger.debug(f'Inspect container {name}')206        if self.container_exists(name):207            url = f'/{self.api_version}/libpod/containers/{name}/json'208            resp = self.podman_socket.get(url)209            result = PodmanApiResponse(resp)210        else:211            logger.warning(f"container {name} does not exist")212            return {}213        if result.successfully and isinstance(result.message, dict):214            return result.message215        else:216            return {}217    def container_exists(self, name: str) -> bool:218        if name:219            url = f'/{self.api_version}/libpod/containers/{name}/exists'220            resp = self.podman_socket.get(url)221            result = PodmanApiResponse(resp)222        else:223            logger.warning("No container name was given")224            return False225        if result.successfully:226            return True227        else:228            return False229    def container_pause(self, name: str) -> None:230        logger.info(f'Pause container {name}')231        container_exists = self.container_exists(name)232        if container_exists:233            url = f'/{self.api_version}/libpod/containers/{name}/pause'234            resp = self.podman_socket.post(url=url)235            result = PodmanApiResponse(resp)236        else:237            logger.warning(f"container {name} does not exist")238            return239        if result.successfully:240            logger.info(f"Paused container {name}")241        else:242            logger.warning(f"Could not pause Container {name}")243            if result.message:244                logger.warning(f"{result.message.get('cause')}")245    def container_unpause(self, name: str) -> None:246        logger.info(f'Unpause container {name}')247        container_exists = self.container_exists(name)248        if container_exists:249            url = f'/{self.api_version}/libpod/containers/{name}/unpause'250            resp = self.podman_socket.post(url=url)251            result = PodmanApiResponse(resp)252        else:253            logger.warning(f"container {name} does not exist")254            return255        if result.successfully:256            logger.info(f"Unpaused container {name}")257        else:258            logger.warning(f"Could not unpause container {name}")259            if result.message:260                logger.warning(f"{result.message.get('cause')}")261    def container_wait(262        self,263        name: str,264        condition: str = "exited",265        request_interval: str = "250ms"266    ) -> None:267        logger.info(f'Wait for container {name}')268        container_exists = self.container_exists(name)269        if container_exists:270            url = f'/{self.api_version}/libpod/containers/{name}/wait'271            resp = self.podman_socket.post(272                url=url,273                query_params={274                    'condition': condition,275                    'interval': request_interval276                },277                timeout=1000278            )279            result = PodmanApiResponse(resp)280        else:281            logger.warning(f"container {name} does not exist")282            return283        if result.successfully:284            logger.info(f"Contidion {condition} of container {name} reached")285        else:286            logger.warning(f"Could not wait for container {name}")287            if result.message:288                logger.warning(f"{result.message.get('cause')}")289    def container_exec(self, name: str, cmd: List) -> None:290        logger.info(f'Execute {cmd} in container {name}')291        body = {'Cmd': cmd}292        url = f'/{self.api_version}/libpod/containers/{name}/exec'293        resp = self.podman_socket.post(294            url=url,295            body=body,296        )297        result = PodmanApiResponse(resp)298        if result.successfully:299            logger.info(f"Successfully executed {cmd} in container {name}")300        else:301            logger.warning(f"Could not execute {cmd} in container {name}. {result.message.get('cause')}")302    def container_logs(303        self,304        name: str,305        follow: bool = False,306        since: str = None,307        until: str = None,308        stderr: bool = True,309        stdout: bool = True,310        timestamp: bool = False311    ) -> str:312        logger.info(f'Get logs from container {name}')313        container_exists = self.container_exists(name)314        if container_exists:315            url = f'/{self.api_version}/libpod/containers/{name}/logs'316            resp = self.podman_socket.get(317                url=url,318                query_params={319                    "follow": follow,320                    "since": since,321                    "until": until,322                    "stderr": stderr,323                    "stdout": stdout,324                    "timestamp": timestamp325                }326            )327            decoded_content = resp.content.decode('utf-8')...

docker_sdk_wrappers.py

Source:docker_sdk_wrappers.py

1from functools import wraps2import luigi3from luigi.contrib.docker_runner import DockerTask4def complete(task_obj):5    task_obj.complete()6    return task_obj7def container_status(container_inspect_info, status="running"):8    prev_state = container_inspect_info.get('State', None)9    prev_status = ''10    if prev_state:11        prev_status = prev_state.get("Status", "")12    if prev_status == status:13        return True14    else:15        return False16def create_container(self_obj, *args0, **kwargs0):17    client_method = getattr(self_obj._client, "exec_start")18    @wraps(client_method)19    def wrapped_client_method(*args, **kwargs):20        return client_method(*args0, **kwargs0)21    return wrapped_client_method22def build_image_from_dockerfile(self_obj, *args0, **kwargs0):23    """24    path (str) â Path to the directory containing the Dockerfile25    fileobj â A file object to use as the Dockerfile. (Or a file-like object)26    tag (str) â A tag to add to the final image27    quiet (bool) â Whether to return the status28    nocache (bool) â Donât use the cache when set to True29    rm (bool) â Remove intermediate containers. The docker build command now defaults to --rm=true, but we have kept30    the old default of False to preserve backward compatibility31    timeout (int) â HTTP timeout32    custom_context (bool) â Optional if using fileobj33    encoding (str) â The encoding for a stream. Set to gzip for compressing34    pull (bool) â Downloads any updates to the FROM image in Dockerfiles35    forcerm (bool) â Always remove intermediate containers, even after unsuccessful builds36    dockerfile (str) â path within the build context to the Dockerfile37    buildargs (dict) â A dictionary of build arguments38    container_limits (dict) â39    A dictionary of limits applied to each container created by the build process. Valid keys:40    memory (int): set memory limit for build41    memswap (int): Total memory (memory + swap), -1 to disable42    swap43    cpushares (int): CPU shares (relative weight)44    cpusetcpus (str): CPUs in which to allow execution, e.g., "0-3", "0,1"45    decode (bool) â If set to True, the returned stream will be decoded into dicts on the fly. Default False46    shmsize (int) â Size of /dev/shm in bytes. The size must be greater than 0. If omitted the system uses 64MB47    labels (dict) â A dictionary of labels to set on the image48    cache_from (list) â A list of images used for build cache resolution49    target (str) â Name of the build-stage to build in a multi-stage Dockerfile50    network_mode (str) â networking mode for the run commands during build51    squash (bool) â Squash the resulting images layers into a single layer.52    extra_hosts (dict) â Extra hosts to add to /etc/hosts in building containers, as a mapping of hostname to IP address.53    platform (str) â Platform in the format os[/arch[/variant]]54    isolation (str) â Isolation technology used during build. Default: None.55    use_config_proxy (bool) â If True, and if the docker client configuration file (~/.docker/config.json by default)56    contains a proxy configuration, the corresponding environment variables will be set in the container being built.57    :param self_obj:58    :param args0:59    :param kwargs0:60    :return:61    """62    client_method = getattr(self_obj._client, "build")63    @wraps(client_method)64    def wrapped_client_method(*args, **kwargs):65        return client_method(*args0, **kwargs0)66    return wrapped_client_method67class PruneContainers(luigi.ExternalTask, DockerTask):68    filters = luigi.DictParameter(default=None)69    is_complete = luigi.BoolParameter(default=False)70    def complete(self):71        if not self.is_complete:72            if self.filters:73                filters = self.filters74            else:75                filters = None76            dict_to_return = self._client.prune_containers(filters)77            self.is_complete = True78        return self.is_complete79    def run(self):80        pass81class PruneNetworks(luigi.ExternalTask, DockerTask):82    filters = luigi.DictParameter(default=None)83    is_complete = luigi.BoolParameter(False)84    def complete(self):85        if not self.is_complete:86            if self.filters:87                filters = self.filters88            else:89                filters = None90            dict_to_return = self._client.prune_networks(filters)91            self.is_complete = True92        return self.is_complete93    def run(self):94        pass95class ContainerExists(luigi.ExternalTask, DockerTask):96    name = luigi.Parameter(default="")97    container_exists = luigi.BoolParameter(False)98    check_complete_on_run = luigi.BoolParameter(default=True)99    def complete(self):100        pre_existing = self._client.containers(all=True, filters={"name": self.name})101        if len(pre_existing) >= 1:102            self.container_exists = True103            return True104        else:105            self.container_exists = False106            return True107    def run(self):108        self.complete()109class StopContainers(luigi.ExternalTask, DockerTask):110    is_complete = luigi.BoolParameter(default=False)111    check_complete_on_run = luigi.BoolParameter(default=True)112    all_containers_stopped = luigi.BoolParameter(default=False)113    def complete(self):114        all_containers = self._client.containers(all=True)115        if all_containers:116            all_stopped = all([c['State'] == "exited" for c in all_containers])117            if all_stopped:118                self.all_containers_stopped = True119                return True120            else:121                return False122        else:123            return True124    def run(self):125        if self.all_containers_stopped:126            pass127        else:128            if not self.is_complete:129                all_containers = self._client.containers(all=True)130                if all_containers:131                    all_stopped = all([c['State'] == "exited" for c in all_containers])132                    while not all_stopped:133                        for c in all_containers:134                            if c['State'] != "exited":135                                try:136                                    self._client.kill(c['Id'])137                                except Exception as e:138                                    print(e)139                        all_containers = self._client.containers(all=True)140                        all_stopped = all([c['State'] == "exited" for c in all_containers])141                    if all_stopped:142                        self.all_containers_stopped = True143                        self.is_complete = True144class ContainerIsRunning(luigi.ExternalTask, DockerTask):145    name = luigi.Parameter(default="")146    container_is_running = luigi.BoolParameter(False)147    is_complete = luigi.BoolParameter(False)148    check_complete_on_run = luigi.BoolParameter(default=True)149    def complete(self):150        container_exists = complete(ContainerExists(name=self.name)).container_exists151        if container_exists:152            prev_container_info = self._client.inspect_container(self.name)153            is_running = container_status(prev_container_info, status="running")154            if is_running:155                self.is_complete = True156                self.container_is_running = True157            else:158                self.is_complete = True159                self.container_is_running = False160        else:161            self.is_complete = True162            self.container_is_running = False163        return self.is_complete164    def run(self):165        self.complete()166class StartContainer(luigi.ExternalTask, DockerTask):167    name = luigi.Parameter(default="")168    container_is_running = luigi.BoolParameter(False)169    is_complete = luigi.BoolParameter(False)170    check_complete_on_run = luigi.BoolParameter(default=True)171    def complete(self):172        container_exists = complete(ContainerExists(name=self.name)).container_exists173        if container_exists:174            is_running = complete(ContainerIsRunning(name=self.name)).container_is_running175            if is_running:176                self.is_complete = True177                self.container_is_running = True178            else:179                self.is_complete = False180                self.container_is_running = False181        else:182            self.is_complete = False183            self.container_is_running = False184        return self.is_complete185    def run(self):186        if not self.is_complete:187            container_exists = complete(ContainerExists(name=self.name)).container_exists188            if container_exists:189                prev_container_info = self._client.inspect_container(self.name)190                is_exited = container_status(prev_container_info, status="exited")191                if is_exited:192                    self._client.start(prev_container_info['Id'])193                    while not self.container_is_running:194                        self.complete()195class CreateNetwork(luigi.ExternalTask, DockerTask):196    network_name = luigi.Parameter("grid")197    is_complete = luigi.BoolParameter(False)198    def complete(self):199        if not self.is_complete:200            network_info = self._client.create_network(self.network_name, check_duplicate=True)201            self.is_complete = True202            return self.is_complete203        else:204            return self.is_complete205    def run(self):...

test_container.py

Source:test_container.py

...14    assert c.status.paused is False15    ip_addr = c.status.ip_addr16    assert not c.is_port_live(22)17    assert not scan_port(ip_addr, 22)18    assert container_exists(c.name)19    c.pause()20    assert c.status.exists is True21    assert c.status.running is True22    assert c.status.paused is True23    assert_raises(RuntimeError, c.poll, 22)24    c.unpause()25    assert c.status.exists is True26    assert c.status.running is True27    assert c.status.paused is False28    c.stop()29    assert c.status.exists is True30    assert c.status.running is False31    assert c.status.paused is False32    assert not c.is_port_live(22)33    assert_raises(RuntimeError, c.poll, 22)34    c.start()35    assert c.status.exists is True36    assert c.status.running is True37    assert c.status.paused is False38    c.remove()39    assert c.status.exists is False40    assert c.status.running is False41    assert c.status.paused is False42    assert not container_exists(c.name)43    assert not c.is_port_live(22)44    assert_raises(socket.error, scan_port, ip_addr, 22)45#-------------------------------------------------------------------------------46# Container context manager47def test_container():48    name = 'foobarbaz1'49    assert not container_exists(name)50    class Foo(Exception): pass51    raised = False52    try:53        with container('mbodenhamer/alpine-data', name=name) as c:54            assert c.name == name55            assert container_exists(name)56            raise Foo57    except Foo:58        raised = True59    # Test that container() deletes the container even if there is an error60    assert raised61    assert not container_exists(name)62#-------------------------------------------------------------------------------63# Polling64def test_polling():65    with container('mbodenhamer/echoserver') as c:66        c.poll(5000, timeout=5)67        assert c.is_port_live(5000)68        assert not c.is_port_live(5001)69        assert_raises(RuntimeError, c.poll, 5001, timeout=1)70#-------------------------------------------------------------------------------71if __name__ == '__main__': # pragma: no cover72    from syn.base_utils import run_all_tests...

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.