How to use get_container_ip method in localstack

Best Python code snippet using localstack_python

failover.py

Source:failover.py

...84    for i in range(len(containers['sent'])):85        sentinels[i].reload()86        sentinel_status = sentinels[i].attrs['State']['Status']87        if sentinel_status == "running":88            res += f'\n Sentinel {i} [{get_container_ip(sentinels[i])}] status: {get_status_sentinel(sentinels[i])} master_addr:{get_master_addr(i)}'89        else:90            res += f'\n Sentinel {i} [{get_container_ip(sentinels[i])}] status: {sentinel_status}'91    for i in range(len(replicas)):92        replicas[i].reload()93        replica_status = replicas[i].attrs['State']['Status']94        if replica_status == "running":95            res += f'\n Replica  {i} [{get_container_ip(replicas[i])}] status: mapped-port: {get_container_mapped_port(replicas[i])} '96        else:97            res += f'\n Replica  {i} [{get_container_ip(replicas[i])}] status: {replica_status}'98    return res99def get_container_mapped_port(container):100    container_ports = container.attrs['NetworkSettings']['Ports']101    if container_ports and container_ports['6379/tcp']:102        return container_ports['6379/tcp'][0]['HostPort']103    return "None"104def get_container_aliases(container):105    return next(iter(container.attrs['NetworkSettings']['Networks'].values()))['Aliases']106def get_container_ip(container):107    return next(iter(container.attrs['NetworkSettings']['Networks'].values()))['IPAddress']108def get_master_addr(sentinel_to_query=-1):109    global containers110    global sentinel_stopped111    if sentinel_to_query == -1:112        sentinel_to_query = (1 + sentinel_stopped) % len(containers['sent'])113    res = containers['sent'][sentinel_to_query].exec_run("redis-cli -p 26379 SENTINEL get-master-addr-by-name mymaster")114    if res[0] != 0:115        raise116    return res[1].decode().split('\n')117def get_master_index():118    global containers119    res = []120    master_addr = get_master_addr()121    index = 0122    for c in containers['repl']:123        if args.mapped_port:124            if master_addr[1] == get_container_mapped_port(c):125                res.append(index)126        else:127            if master_addr[0] == get_container_ip(c):128                res.append(index)129            elif master_addr[0] in get_container_aliases(c):130                res.append(index)131        index = index + 1132    if len(res) == 1:133        return res[0]134    raise135def verify_num_sentinels_status_ok(expect_num_sentinels_ok):136    logging.info(f'----> Done wait {args.wait}sec. Verify {expect_num_sentinels_ok} sentinels status ok')137    containers_status = get_cluster_status()138    if containers_status.count("status=ok") != expect_num_sentinels_ok:139        assert False, f'Invalid containers_status: {containers_status}'140    logging.info(containers_status)141def run_failover():...

runtime.py

Source:runtime.py

...32                                       network=self.network)33            containers = self.client.containers.list(all=False, filters={"name": name})34            container = containers[0]35            return new_operator_instance(container.short_id, container.name,36                                         container.status, self.get_container_ip(container),37                                         container.ports)38        except APIError as e:39            raise DockerRuntimeError(e.explanation, e)40    def start_instance(self, name):41        try:42            container = self.client.containers.get(name)43            container.start()44            return new_operator_instance(container.short_id, container.name,45                                         container.status, self.get_container_ip(container),46                                         container.ports)47        except APIError as e:48            raise DockerRuntimeError(e.explanation, e)49    def stop_instance(self, name):50        try:51            container = self.client.containers.get(name)52            container.stop()53            return new_operator_instance(container.short_id, container.name,54                                         container.status, self.get_container_ip(container),55                                         container.ports)56        except APIError as e:57            raise DockerRuntimeError(e.explanation, e)58    def delete_instance(self, name):59        try:60            container = self.client.containers.get(name)61            container.remove(force=True)62            return new_operator_instance(container.short_id, container.name,63                                         "deleted", self.get_container_ip(container),64                                         container.ports)65        except APIError as e:66            raise DockerRuntimeError(e.explanation, e)67    def restart_instance(self, name):68        try:69            container = self.client.containers.get(name)70            container.restart()71            return new_operator_instance(container.short_id, container.name,72                                         container.status, self.get_container_ip(container),73                                         container.ports)74        except APIError as e:75            raise DockerRuntimeError(e.explanation, e)76    def list_instances(self, name):77        try:78            res = []79            containers = self.client.containers.list(80                all=True, filters={"name": f"phantoscope_{name}", "label": self.labels})81            for container in containers:82                res.append(new_operator_instance(container.short_id,83                                                 container.name,84                                                 container.status,85                                                 self.get_container_ip(container),86                                                 container.ports))87            return res88        except APIError as e:89            raise DockerRuntimeError(e.explanation, e)90    def inspect_instance(self, name):91        try:92            container = self.client.containers.get(name)93            return new_operator_instance(container.short_id, container.name,94                                         container.status, self.get_container_ip(container),95                                         container.ports)96        except APIError as e:97            raise DockerRuntimeError(e.explanation, e)98    def get_container_ip(self, container):99        return container.attrs["NetworkSettings"]["Networks"][self.network]["IPAddress"]100def runtime_client_getter(name):101    if name == "docker":...

adm.py

Source:adm.py

...40            print(line)41    print("å®¹å¨åç§°ï¼%s" % container_name)42    print("å¯ç ï¼%s" % pwd)43    return container_name, pwd44def get_container_ip(container_name):45    command = "(docker inspect --format '{{ .NetworkSettings.IPAddress }}' %s)"46    output = os.popen(command % container_name)47    ip = output.readlines()[0]48    print("å®¹å¨ ipæ¯: %s" % ip)49    return ip.strip()50def add_haproxy_config(container_name, ip, port):51    template = """52frontend ss-in-{container_name}53    bind *:{port}54    default_backend ss-out-{container_name}55backend ss-out-{container_name}56    server server1 {ip}:1090 maxconn 2048057    """58    param = dict(59        container_name=container_name,60        ip=ip,61        port=port,62    )63    return template.format(**param)64def output_serve_info():65    template = """66    æå¡å¨å°åï¼67    æå¡å¨ç«¯å£ï¼68    å¯ç ï¼69    """70if __name__ == '__main__':71    import sys72    input_ = sys.argv[1]73    if input_ == "add":74        name = sys.argv[2]75        run_docker_serve(name)76        get_container_ip(name)77    elif input_ == "haproxy":78        f = sys.argv[2]79        name = sys.argv[3]80        port = sys.argv[4]81        ip = get_container_ip(name)82        os.system("echo '%s' >> %s" % (add_haproxy_config(name, ip, port), f))83    else:84        print("""85        python {0} add container_name86        python {0} haproxy haproxy_cfg_path container_name...

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.