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

How to use get_container_name method in localstack

Best Python code snippet using localstack_python

guestutils.py

Source:guestutils.py

...21	return 'default'22def get_service_name():23	# returns the friendly name of the service the container is a member of.24	return os.environ['SERVICE_NAME']25def get_container_name():26	# returns the friendly name of the container itself.27	return os.environ['CONTAINER_NAME']28def get_container_host_address():29	# returns the IP address or hostname of the host of the container. Useful if your application needs to advertise itself to some service discovery system.30	return os.environ['CONTAINER_HOST_ADDRESS']31	service = get_service_name()32	all_service_instances = client.read('/' + service).value33	return all_service_instances[get_container_name()]34def get_container_internal_address():35	# returns the IP address assigned to the container itself by Docker (its private IP address).36	# host_ip = get_container_host_address()37	# print 'this is the modifIED VERSION'38	# dockerclient = docker.Client(base_url = "tcp://"+host_ip+":4243", 39	# 	version = '1.10', 40	# 	timeout=10)41	# details =  dockerclient.inspect_container(get_container_name())42	# return str(details['NetworkSettings']['IPAddress'])43	return socket.gethostbyname(socket.gethostname())44def get_port(name, default = 'default_port'):45	# will return the exposed (internal) port number of a given named port for the current container instance. This is useful to set configuration parameters for example.46	# 'port_mapping': {'rpc': {'external': ('0.0.0.0', '9160/tcp'), 'exposed': '9160/tcp'}, 'storage': {'external': ('0.0.0.0', '7000/tcp'), 'exposed': '7000/tcp'}47	if default != 'default_port':48		return get_specific_exposed_port(get_service_name(), get_container_name(), name, default)49	return get_specific_exposed_port(get_service_name(), get_container_name(), name)50def get_node_list(service, ports=[], minimum=1, labels = []):51	# It takes in a service name and an optional list of port names and returns the list of IP addresses/hostname of the containers of that service. For each port specified, in order, it will append :<port number> to each host with the external port number. For example, if you want to return the list of ZooKeeper endpoints with their client ports:52	# get_node_list('zookeeper', ports=['client']) -> ['c414.ore1.domain.com:2181', 'c415.ore1.domain.com:2181']53	nodes = []54	all_service_instances = client.read('/' + service).value55	all_service_instances = ast.literal_eval(all_service_instances)56	for key in all_service_instances.keys():57		instance = all_service_instances[key]58		#59		# only proceed if the labels are correct60		# @LABELS61		#62		fits_query=True63		for label in labels:64			if label not in instance['labels']:65				fits_query = False66		if fits_query:67			node = instance['instance_host']68			portlist = ""69			for port in ports:70				p = get_specific_port(instance['service_name'], instance['instance_name'], port)71				p = ":" + p72				portlist = portlist + p73			nodes.append(str(node + portlist))74	return nodes75def get_specific_host(service, container):76	# which can be used to return the hostname or IP address of a specific container from a given service, and77	all_service_instances = client.read('/' + service).value78	all_service_instances = ast.literal_eval(all_service_instances)79	try:80		return all_service_instances[container]['instance_host']81	except: 82		return None83def get_specific_port(service, container, port, default='default'):84	# to retrieve the external port number of a specific named port of a given container.85	all_service_instances = client.read('/'+service).value86	all_service_instances = ast.literal_eval(all_service_instances)87	my_instance = all_service_instances[container]88	port_mappings = my_instance['port_mapping']89	port_mapping = port_mappings.get(port)90	if port_mapping is None:91		return default92	return port_mapping['external'][1].replace('/tcp', '')93def get_specific_exposed_port(service, container, port, default='default'):94	# to retrieve the exposed (internal) port number of a specific named port of a given container.95	all_service_instances = client.read('/'+service).value96	all_service_instances = ast.literal_eval(all_service_instances)97	print all_service_instances.keys()98	my_instance = all_service_instances[container]99	port_mappings = my_instance['port_mapping']100	port_mapping = port_mappings.get(port)101	if port_mapping is None:102		return default103	return port_mapping['exposed'].replace('/tcp', '')104def test_all():105	try:106		print 'get_environment_name():'107		print get_environment_name()108	except Exception, e:109		print e.__doc__110		print e.message111		print 'failed'112	try:113		print 'get_service_name():'114		print get_service_name()115	except Exception, e:116		print e.__doc__117		print e.message118		print 'failed'119	try:120		print 'get_container_name():'121		print get_container_name()122	except Exception, e:123		print e.__doc__124		print e.message125		print 'failed'126	try:127		print 'get_container_host_address():'128		print get_container_host_address()129	except Exception, e:130		print e.__doc__131		print e.message132		print 'failed'133	try:134		print 'get_container_internal_address():'135		print get_container_internal_address()...

run.py

Source:run.py

...47    servers = os.environ['ZOOKEEPER_SERVER_IDS'].split(',')48    for server in servers:49        node, id = server.split(':')50        conf['server.{}'.format(id)] = build_node_repr(node)51        if node == get_container_name():52            ZOOKEEPER_NODE_ID = id53# Verify that the number of declared nodes matches the size of the cluster.54ZOOKEEPER_NODE_COUNT = len(get_node_list(get_service_name()))55ZOOKEEPER_CLUSTER_SIZE = len(56    [i for i in conf.keys() if i.startswith('server.')])57# If no ZOOKEEPER_SERVER_IDS is defined, we expect to be in single-node mode so58# no more than one node can be declared in the cluster.59if ZOOKEEPER_CLUSTER_SIZE == 0 and ZOOKEEPER_NODE_COUNT != 1:60    sys.stderr.write(('Missing ZOOKEEPER_SERVER_IDS declaration for ' +61                      '{}-node ZooKeeper cluster!\n')62                     .format(ZOOKEEPER_NODE_COUNT))63    sys.exit(1)64# If we got nodes from ZOOKEEPER_SERVER_IDS, we expect exactly the same number65# of nodes declared in the cluster.66if ZOOKEEPER_CLUSTER_SIZE > 0 and \67        ZOOKEEPER_CLUSTER_SIZE != ZOOKEEPER_NODE_COUNT:68    sys.stderr.write(('Mismatched number of nodes between ' +69                      'ZOOKEEPER_SERVER_IDS ({}) and the declared ' +70                      'cluster ({})!\n')71                     .format(ZOOKEEPER_CLUSTER_SIZE), ZOOKEEPER_NODE_COUNT)72    sys.exit(1)73# Write out the ZooKeeper configuration file.74with open(ZOOKEEPER_CONFIG_FILE, 'w+') as f:75    for entry in conf.iteritems():76        f.write("%s=%s\n" % entry)77# Setup the logging configuration.78with open(ZOOKEEPER_LOG_CONFIG_FILE, 'w+') as f:79    f.write("""# Log4j configuration, logs to rotating file80log4j.rootLogger=INFO,R81log4j.appender.R=org.apache.log4j.RollingFileAppender82log4j.appender.R.File=/var/log/%s/%s.log83log4j.appender.R.MaxFileSize=100MB84log4j.appender.R.MaxBackupIndex=1085log4j.appender.R.layout=org.apache.log4j.PatternLayout86log4j.appender.R.layout.ConversionPattern=%s87""" % (get_service_name(), get_container_name(), LOG_PATTERN))88# Write out the 'myid' file in the data directory if in cluster mode.89if ZOOKEEPER_NODE_ID:90    if not os.path.exists(ZOOKEEPER_DATA_DIR):91        os.makedirs(ZOOKEEPER_DATA_DIR, mode=0750)92    with open(os.path.join(ZOOKEEPER_DATA_DIR, 'myid'), 'w+') as f:93        f.write('%s\n' % ZOOKEEPER_NODE_ID)94    sys.stderr.write(95        'Starting {}, node id#{} of a {}-node ZooKeeper cluster...\n'96        .format(get_container_name(), ZOOKEEPER_NODE_ID,97                ZOOKEEPER_CLUSTER_SIZE))98else:99    sys.stderr.write('Starting {} as a single-node ZooKeeper cluster...\n'100                     .format(get_container_name()))101jvmflags = [102    '-server',103    '-showversion',104    '-Dvisualvm.display.name="{}/{}"'.format(105        get_environment_name(), get_container_name()),106]107jmx_port = get_port('jmx', -1)108if jmx_port != -1:109    jvmflags += [110        '-Djava.rmi.server.hostname={}'.format(get_container_host_address()),111        '-Dcom.sun.management.jmxremote.port={}'.format(jmx_port),112        '-Dcom.sun.management.jmxremote.rmi.port={}'.format(jmx_port),113        '-Dcom.sun.management.jmxremote.authenticate=false',114        '-Dcom.sun.management.jmxremote.local.only=false',115        '-Dcom.sun.management.jmxremote.ssl=false',116    ]117os.environ['JVMFLAGS'] = ' '.join(jvmflags) + ' ' + os.environ.get('JVM_OPTS', '')118# Start ZooKeeper119os.execl('bin/zkServer.sh', 'zookeeper', 'start-foreground')

admin.py

Source:admin.py

...14            return obj.relation.product.name15        else:16            return None17    get_product_name.short_description = "Product"18    def get_container_name(self, obj):19        if(obj.relation is not None):20            return obj.relation.container.name21        else:22            return None23    get_container_name.short_description = "Container"24    def get_form(self, request, obj=None, **kwargs):25        form = super(ProductTagAdmin, self).get_form(request, obj, **kwargs)26        form.base_fields['relation'].label_from_instance = lambda obj: "({}) {} : {}".format(obj.id, obj.product.name, obj.container.name)27        return form28class ProductSpotAdmin(admin.ModelAdmin):29    list_display = ('get_shelf_name', 'position', 'get_tag_tag')30    def get_tag_tag(self, obj):31        if(obj.last_tag is not None):32            return obj.last_tag.tag33        else:34            return None35    get_tag_tag.short_description = "Last tag"36    def get_shelf_name(self, obj):37        return obj.shelf.name38    get_shelf_name.short_description = "Shelf"39    def get_form(self, request, obj=None, **kwargs):40        form = super(ProductSpotAdmin, self).get_form(request, obj, **kwargs)41        form.base_fields['last_tag'].label_from_instance = lambda obj: "({}) {}".format(obj.id, obj.tag)42        form.base_fields['shelf'].label_from_instance = lambda obj: "({}) {}".format(obj.id, obj.name)            43        return form44    45class ProductContainerRelationAdmin(admin.ModelAdmin):46    list_display = ('get_container_name', 'get_product_name', 'max_measurement')47    def get_container_name(self, obj):48        return obj.container.name49    get_container_name.short_description = "Container"50    def get_product_name(self, obj):51        return obj.product.name52    get_product_name.short_description = "Product"53    pass54    def get_form(self, request, obj=None, **kwargs):55            form = super(ProductContainerRelationAdmin, self).get_form(request, obj, **kwargs)56            form.base_fields['container'].label_from_instance = lambda obj: "({}) {}".format(obj.id, obj.name)57            form.base_fields['product'].label_from_instance = lambda obj: "({}) {}".format(obj.id, obj.name)            58            return form59    60class ProductSpotStateAdmin(admin.ModelAdmin):61    pass...

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.