KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_container_id method in localstack

Best Python code snippet using localstack_python

test_models.py

Source:test_models.py

...62    recurse_callbacks(callbacks=callbacks)63    with capture_on_commit_callbacks(execute=True):64        s = SessionFactory(workstation_image=wsi)65    try:66        assert get_container_id(name=s.service.container_name)67        s.refresh_from_db()68        assert s.status == s.STARTED69        container = inspect_container(name=s.service.container_name)70        expected_labels = {71            "job": f"{s._meta.app_label}-{s._meta.model_name}-{s.pk}",72            "traefik.enable": "true",73            f"traefik.http.routers.{s.hostname}-http.entrypoints": "workstation-http",74            f"traefik.http.routers.{s.hostname}-http.rule": f"Host(`{s.hostname}`)",75            f"traefik.http.routers.{s.hostname}-http.service": f"{s.hostname}-http",76            f"traefik.http.routers.{s.hostname}-websocket.entrypoints": "workstation-websocket",77            f"traefik.http.routers.{s.hostname}-websocket.rule": f"Host(`{s.hostname}`)",78            f"traefik.http.routers.{s.hostname}-websocket.service": f"{s.hostname}-websocket",79            f"traefik.http.services.{s.hostname}-http.loadbalancer.server.port": "8080",80            f"traefik.http.services.{s.hostname}-websocket.loadbalancer.server.port": "4114",81        }82        for k, v in expected_labels.items():83            assert container["Config"]["Labels"][k] == v84        networks = container["NetworkSettings"]["Networks"]85        assert len(networks) == 186        assert settings.WORKSTATIONS_NETWORK_NAME in networks87        with capture_on_commit_callbacks(execute=True):88            s.user_finished = True89            s.save()90        with pytest.raises(ObjectDoesNotExist):91            # noinspection PyStatementEffect92            get_container_id(name=s.service.container_name)93    finally:94        stop_all_sessions()95@pytest.mark.django_db96def test_correct_session_stopped(http_image, settings):97    # Execute celery tasks in place98    settings.task_eager_propagates = (True,)99    settings.task_always_eager = (True,)100    with capture_on_commit_callbacks() as callbacks:101        wsi = WorkstationImageFactory(image__from_path=http_image)102    recurse_callbacks(callbacks=callbacks)103    try:104        with capture_on_commit_callbacks(execute=True):105            s1, s2 = (106                SessionFactory(workstation_image=wsi),107                SessionFactory(workstation_image=wsi),108            )109        assert get_container_id(name=s1.service.container_name)110        assert get_container_id(name=s2.service.container_name)111        s2.refresh_from_db()112        auth_token_pk = s2.auth_token.pk113        with capture_on_commit_callbacks(execute=True):114            s2.user_finished = True115            s2.save()116        assert get_container_id(name=s1.service.container_name)117        with pytest.raises(ObjectDoesNotExist):118            # noinspection PyStatementEffect119            get_container_id(name=s2.service.container_name)120        with pytest.raises(ObjectDoesNotExist):121            # auth token should be deleted when the service is stopped122            AuthToken.objects.get(pk=auth_token_pk)123    finally:124        stop_all_sessions()125@pytest.mark.django_db126def test_session_cleanup(http_image, settings):127    # Execute celery tasks in place128    settings.task_eager_propagates = (True,)129    settings.task_always_eager = (True,)130    with capture_on_commit_callbacks() as callbacks:131        wsi = WorkstationImageFactory(image__from_path=http_image)132    recurse_callbacks(callbacks=callbacks)133    default_region = "eu-nl-1"134    try:135        with capture_on_commit_callbacks(execute=True):136            s1, s2, s3 = (137                SessionFactory(workstation_image=wsi, region=default_region),138                SessionFactory(139                    workstation_image=wsi,140                    maximum_duration=timedelta(seconds=0),141                    region=default_region,142                ),143                # An expired service in a different region144                SessionFactory(145                    workstation_image=wsi,146                    maximum_duration=timedelta(seconds=0),147                    region="us-east-1",148                ),149            )150        assert get_container_id(name=s1.service.container_name)151        assert get_container_id(name=s2.service.container_name)152        assert get_container_id(name=s3.service.container_name)153        # Stop expired services in the default region154        stop_expired_services(155            app_label="workstations",156            model_name="session",157            region=default_region,158        )159        assert get_container_id(name=s1.service.container_name)160        with pytest.raises(ObjectDoesNotExist):161            # noinspection PyStatementEffect162            get_container_id(name=s2.service.container_name)163        assert get_container_id(name=s3.service.container_name)164    finally:165        stop_all_sessions()166@pytest.mark.django_db167def test_workstation_ready(http_image, settings):168    # Execute celery tasks in place169    settings.task_eager_propagates = (True,)170    settings.task_always_eager = (True,)171    # Do not execute the callbacks as the image should not be ready172    wsi = WorkstationImageFactory(image__from_path=http_image)173    assert wsi.ready is False174    with capture_on_commit_callbacks(execute=True):175        s = SessionFactory(workstation_image=wsi)176    s.refresh_from_db()177    assert s.status == s.FAILED...

docker_client.py

Source:docker_client.py

...55    )56    return json.loads(result.stdout)57def stop_container(*, name):58    try:59        container_id = get_container_id(name=name)60        return _run_docker_command("stop", container_id)61    except ObjectDoesNotExist:62        return63def remove_container(*, name):64    try:65        container_id = get_container_id(name=name)66        try:67            _run_docker_command("rm", container_id)68        except CalledProcessError as error:69            if "Error: No such container" in error.stderr:70                raise ObjectDoesNotExist from error71            else:72                raise73    except ObjectDoesNotExist:74        return75def get_container_id(*, name):76    result = _run_docker_command(77        "ps", "--all", "--quiet", "--filter", f"name={name}"78    )79    return get([line for line in result.stdout.splitlines()])80def inspect_container(*, name):81    container_id = get_container_id(name=name)82    result = _run_docker_command(83        "inspect", "--format", "{{json .}}", container_id84    )85    return json.loads(result.stdout)86def get_logs(*, name, tail=None):87    container_id = get_container_id(name=name)88    args = ["logs", "--timestamps"]89    if tail is not None:90        args.extend(["--tail", str(tail)])91    result = _run_docker_command(*args, container_id)92    return result.stdout.splitlines() + result.stderr.splitlines()93def run_container(94    *,95    repo_tag,96    name,97    labels,98    environment,99    network,100    mem_limit,101    ports=None,...

sshlxc.py

Source:sshlxc.py

...25        # self.host == containerhost26        # this way SSHConnection parent class uses the containerhost as the SSH remote host27        self.connector = None28        # logging.warning(self._play_context.connection)29    def get_container_id(self):30        return self.containerspec31    def get_container_connector(self):32        return 'lxc'33    def _strip_sudo(self, executable, cmd):34        # Get the command without sudo35        sudoless = cmd.rsplit(executable + ' -c ', 1)[1]36        # Get the quotes37        quotes = sudoless.partition('echo')[0]38        # Get the string between the quotes39        cmd = sudoless[len(quotes):-len(quotes+'?')]40        # Drop the first command becasue we don't need it41        #cmd = cmd.split('; ', 1)[1]42        return cmd43    def host_command(self, cmd, do_become=False):44        if self._play_context.become and do_become:45            cmd = self._play_context.make_become_cmd(cmd)46        return super(Connection, self).exec_command(cmd, in_data=None, sudoable=True)47    def exec_command(self, cmd, in_data=None, executable='/bin/sh', sudoable=True):48        ''' run a command in the container '''49        cmd = '%s exec %s -- %s' % (self.get_container_connector(), self.get_container_id(), cmd)50        if self._play_context.become:51            # display.debug("_low_level_execute_command(): using become for this command")52            cmd = self._play_context.make_become_cmd(cmd)53        # display.vvv("CONTAINER (%s) %s" % (local_cmd), host=self.host)54        return super(Connection, self).exec_command(cmd, in_data, True)55    def container_path(self, path):56        return self.get_container_id() + path57    @contextmanager58    def tempfile(self):59        code, stdout, stderr = self.host_command('mktemp')60        if code != 0:61            raise AnsibleError("failed to make temp file:\n%s\n%s" % (stdout, stderr))62        tmp = stdout.strip().split('\n')[-1]63        yield tmp64        code, stdout, stderr = self.host_command(' '.join(['rm', tmp]))65        if code != 0:66            raise AnsibleError("failed to remove temp file %s:\n%s\n%s" % (tmp, stdout, stderr))67    def put_file(self, in_path, out_path):68        ''' transfer a file from local to remote container '''69        with self.tempfile() as tmp:70            super(Connection, self).put_file(in_path, tmp)71            self.host_command(' '.join(['lxc', 'exec', self.get_container_id(), '--', 'mkdir', '-p', os.path.dirname(out_path)]), do_become=True)72            self.host_command(' '.join(['lxc', 'file', 'push', '--debug', tmp, self.container_path(out_path)]), do_become=True)73    def fetch_file(self, in_path, out_path):74        ''' fetch a file from remote to local '''75        with self.tempfile() as tmp:76            self.host_command(' '.join(['lxc', 'file', 'pull', self.container_path(in_path), tmp]), do_become=True)77            super(Connection, self).fetch_file(tmp, out_path)78    def close(self):79        ''' Close the connection, nothing to do for us '''...

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.