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

How to use get_container_status method in localstack

Best Python code snippet using localstack_python

main.py

Source:main.py

...80                    if dockers[challenge]["type"] == "compose":81                        segment += "compose\n"82                        segment += " " * 4 + "Containers:\n"83                        for container in list_compose_containers(dockers[challenge]["directory"]):84                            status, exitcode = get_container_status(container)85                            ports = get_container_ports(container)86                            segment += " " * 8 + f"`{container}`: {status}" + \87                                       (f" with code {exitcode}" if status == "exited" else "")88                            segment += (", ports " + ",".join(ports)) if len(ports) != 0 else ""89                            segment += "\n"90                    elif dockers[challenge]["type"] == "container":91                        segment += "single container\n"92                        status, exitcode = get_container_status(dockers[challenge]["container-name"])93                        ports = get_container_ports(dockers[challenge]["container-name"])94                        segment += " " * 4 + f"Status: {status}" + \95                                   (f" with code {exitcode}" if status == "exited" else "")96                        segment += (", ports " + ",".join(ports)) if len(ports) != 0 else ""97                        segment += "\n"98                    else:99                        await message.channel.send("Unknown type on " + challenge)100                        return101                    segment += "\n"102                    # Compensate for Discord's 2k character limit103                    if len(segment) + len(output) > 2000:104                        await message.channel.send(output)105                        output = segment106                    else:107                        output += segment108            await message.channel.send(output)109        elif command == "restart":110            if len(input_message) < 2: 111                await message.channel.send("Not enough arguments")112                return113            name = input_message[1]114            output = ""115            async with message.channel.typing():116                if name in dockers:117                    # Challenge shorthand name recognised118                    if dockers[name]["type"] == "compose":119                        restart_compose(dockers[name]["directory"])120                        output += f"Compose cluster `{name}` restarted.\n\n"121                        for container in list_compose_containers(dockers[name]["directory"]):122                            status, exitcode = get_container_status(container)123                            output += " " * 4 + f"`{container}`: {status}" + \124                                      (f" with code {exitcode}" if status == "exited" else "") + "\n"125                    elif dockers[name]["type"] == "container":126                        res = restart_container(dockers[name]["container-name"])127                        if not res[0]:128                            await message.channel.send("Failed: " + res[1])129                            return130                        output += "Single container `" + dockers[name]["container-name"] + "` restarted. \n\n"131                        status, exitcode = get_container_status(dockers[name]["container-name"])132                        output += " " * 4 + "`" + dockers[name]["container-name"] + f"`: {status}" + \133                                  (f" with code {exitcode}" if status == "exited" else "") + "\n"134                    else:135                        await message.channel.send("Unknown type on " + name)136                        return137                else:138                    # Challenge shorthand not recognised, assuming unknown container name139                    res = restart_container(name)140                    if not res[0]:141                        output = "Failed: " + res[1]142                    else:143                        output += f"Single container `{name}` restarted.\n\n"144                        status, exitcode = get_container_status(name)145                        output += " " * 4 + f"`{name}`: {status}" + \146                                  (f" with code {exitcode}" if status == "exited" else "") + "\n"147            await message.channel.send(output)148        elif command == "stop":149            if len(input_message) < 2: 150                await message.channel.send("Not enough arguments")151                return152            name = input_message[1]153            output = ""154            async with message.channel.typing():155                if name == "*":156                    m = await message.channel.send("Are you sure you want to stop all containers? React with :white_check_mark: to proceed.")157                    await m.add_reaction("â")158                    waiting_messages[m.id] = {"type":"stop", "time":time.time()}159                    return160                elif name in dockers:161                    # Challenge shorthand name recognised162                    if dockers[name]["type"] == "compose":163                        stop_compose(dockers[name]["directory"])164                        output += f"Compose cluster `{name}` stopped.\n\n"165                        for container in list_compose_containers(dockers[name]["directory"]):166                            status, exitcode = get_container_status(container)167                            output += " " * 4 + f"`{container}`: {status}" + \168                                      (f" with code {exitcode}" if status == "exited" else "") + "\n"169                    elif dockers[name]["type"] == "container":170                        res = stop_container(dockers[name]["container-name"])171                        if not res[0]:172                            await message.channel.send("Failed: " + res[1])173                            return174                        output += "Single container `" + dockers[name]["container-name"] + "` stopped. \n\n"175                        status, exitcode = get_container_status(dockers[name]["container-name"])176                        output += " " * 4 + "`" + dockers[name]["container-name"] + f"`: {status}" + \177                                  (f" with code {exitcode}" if status == "exited" else "") + "\n"178                    else:179                        await message.channel.send("Unknown type on " + name)180                        return181                else:182                    # Challenge shorthand not recognised, assuming unknown container name183                    res = stop_container(name)184                    if not res[0]:185                        output = "Failed: " + res[1]186                    else:187                        output += f"Single container `{name}` stopped.\n\n"188                        status, exitcode = get_container_status(name)189                        output += " " * 4 + f"`{name}`: {status}" + \190                                  (f" with code {exitcode}" if status == "exited" else "") + "\n"191            await message.channel.send(output)192        elif command == "start":193            if len(input_message) < 2: 194                await message.channel.send("Not enough arguments")195                return196            name = input_message[1]197            output = ""198            async with message.channel.typing():199                if name == "*":200                    m = await message.channel.send("Are you sure you want to start all containers? React with :white_check_mark: to proceed.")201                    await m.add_reaction("â")202                    waiting_messages[m.id] = {"type":"start", "time":time.time()}203                    return204                elif name in dockers:205                    # Challenge shorthand name recognised206                    if dockers[name]["type"] == "compose":207                        start_compose(dockers[name]["directory"])208                        output += f"Compose cluster `{name}` started.\n\n"209                        for container in list_compose_containers(dockers[name]["directory"]):210                            status, exitcode = get_container_status(container)211                            output += " " * 4 + f"`{container}`: {status}" + \212                                      (f" with code {exitcode}" if status == "exited" else "") + "\n"213                    elif dockers[name]["type"] == "container":214                        res = start_container(dockers[name]["container-name"])215                        if not res[0]:216                            await message.channel.send("Failed: " + res[1])217                            return218                        output += "Single container `" + dockers[name]["container-name"] + "` started. \n\n"219                        status, exitcode = get_container_status(dockers[name]["container-name"])220                        output += " " * 4 + "`" + dockers[name]["container-name"] + f"`: {status}" + \221                                  (f" with code {exitcode}" if status == "exited" else "") + "\n"222                    else:223                        await message.channel.send("Unknown type on " + name)224                        return225                else:226                    # Challenge shorthand not recognised, assuming unknown container name227                    res = start_container(name)228                    if not res[0]:229                        output = "Failed: " + res[1]230                    else:231                        output += f"Single container `{name}` started.\n\n"232                        status, exitcode = get_container_status(name)233                        output += " " * 4 + f"`{name}`: {status}" + \234                                  (f" with code {exitcode}" if status == "exited" else "") + "\n"235            await message.channel.send(output)236        elif command == "init":237            async with message.channel.typing():238                m = await message.channel.send("Are you sure you want to initialise all challenges? Likelihood is that this has always been done and you do not need to do it. This will cause outages and realistically you probably do not need to run it. Unless you're Tom, in which case go ahead. If you're not, you can do this or you can walk away and forget this ever happened. Did I mention you probably don't need to do this?\nReact with :white_check_mark: to proceed.")239                await m.add_reaction("â")240                waiting_messages[m.id] = {"type":"init", "time":time.time()}241                return242        elif command == "logs":243            name = input_message[1]244            if name in dockers:245                if dockers[name]["type"] == "compose":246                    await message.channel.send("This function only works with containers directly.")247                    return248                else:249                    logs, _ = container_logs(name)250                    if len(logs) > 1930:251                        logs = logs[len(logs)-1930:]252                    logs = escape_ansi(logs)253                    await message.channel.send(f"Logs for `{name}`\n```\n{logs}\n```")254			255        else:256            await message.channel.send("Unknown command")257@client.event258async def on_reaction_add(reaction, user):259    if user == client.user:260        return261    # holy wall of text262    if reaction.message.id in waiting_messages:263        if (time.time() - waiting_messages[reaction.message.id]["time"]) < 30:264            r = waiting_messages.pop(reaction.message.id)265            if r["type"] == "start":266                await reaction.message.channel.send("Starting all challenges")267                for chall in dockers:268                    print("Starting", chall)269                    m = await reaction.message.channel.send(f"Starting `{chall}`...")270                    if dockers[chall]["type"] == "compose":271                        start_compose(dockers[chall]["directory"])272                    elif dockers[chall]["type"] == "container":273                        c = dockers[chall]274                        start_container(c["container-name"])275                    await m.edit(content=f"`{chall}` started.")276                await reaction.message.channel.send(":tada: All containers started.")277            elif r["type"] == "stop":278                await reaction.message.channel.send("Stopping all challenges")279                for chall in dockers:280                    print("Stopping", chall)281                    m = await reaction.message.channel.send(f"Stopping `{chall}`...")282                    if dockers[chall]["type"] == "compose":283                        stop_compose(dockers[chall]["directory"])284                    elif dockers[chall]["type"] == "container":285                        c = dockers[chall]286                        stop_container(c["container-name"])287                    await m.edit(content=f"`{chall}` stopped.")288                await reaction.message.channel.send(":tada: All containers stopped.")289            elif r["type"] == "init":290                await reaction.message.channel.send("Initialising all challenges. This might take a moment or two.")291                for container in get_all_containers():292                    m = await reaction.message.channel.send(f"Removing `{container}`...")293                    remove_container(container)294                    await m.edit(content=f"`{container}` removed.")295                for chall in dockers:296                    m = await reaction.message.channel.send(f"Creating `{chall}`...")297                    if dockers[chall]["type"] == "compose":298                        create_compose(dockers[chall]["directory"])299                    elif dockers[chall]["type"] == "container":300                        c = dockers[chall]301                        if "cpu" in c:302                            cpus = c["cpu"]303                        else:304                            cpus = default_cpu305                        if "ram" in c:306                            ram = c["ram"]307                        else:308                            ram = default_ram309                        create_container(c["container-name"], c["create-args"], c["image"], ram, cpus)310                    await m.edit(content=f"`{chall}` created.")311                await reaction.message.channel.send(":tada: All challenges initialised.")312# Helper functions313def escape_ansi(line):314    ansi_escape = re.compile(r'(\x9B|\x1B\[)[0-?]*[ -\/]*[@-~]')315    return ansi_escape.sub('', line)316def run_command(command: list) -> tuple:317    """318    Runs a command319    :param command: elements of command to be run320    :return: text output of command, boolean indicating if stderr was used321    """322    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)323    if result.stdout == b"":324        result = result.stderr325        error = True326    else:327        result = result.stdout328        error = False329    return result.decode().strip("\n"), error330def validate_container_command(output: str, name: str) -> tuple:331    """332    Validates output from one of the *_container functions, eg start_container.333    :param output: output from the command line334    :param name: container name or ID335    :return: success boolean, message336    """337    if "no such container" in output:338        return False, "no such container"339    elif output != name.lower():340        return False, output341    else:342        return True, "ok"343# Miscellaneous functions344def load_dockers():345    """346    Loads Docker container information from the file specified as docker-info-file in the settings JSON347    :return: None348    """349    global dockers350    dockers = json.load(open(docker_info_file))351def list_compose_containers(directory: str) -> list:352    """353    Lists containers that form a compose cluster354    :param directory: The directory of the Docker compose cluster355    :return: list of container names356    """357    os.chdir(directory)358    result, _ = run_command(["sudo", "docker-compose", "ps"])359    result = result.split("\n")[2:]360    os.chdir(bot_location)361    return [line.split(" ")[0] for line in result]362def get_container_status(name: str) -> tuple:363    """364    Gets the status of the specified container (eg created, running, stopped, etc)365    :param name: container name od ID366    :return: status as string, exit code as int (zero if not exited)367    """368    result, _ = run_command(["sudo", "docker", "inspect", name])369    if "no such container" in result:370        return "no such container", 0371    try:372        j = json.loads(result)[0]373    except json.decoder.JSONDecodeError:374        return result, 0375    if "State" not in j:376        return "created", 0...

container_client.py

Source:container_client.py

...101		return self.client.containers(all=False)102	#==============================================================================103	# docker get container status104	#==============================================================================105	def get_container_status(self,container):106		for c in self.client.containers(all=True):107			names_list = c['Names']108			name = "/"+container109			exist = name in names_list110			if exist:111				status = c['Status']112				if status[:2]=='Up':113					return CS_Running114				else:115					return CS_Stopped116				#elif status[:10]=='Exited (0)':117					#return CS_Stopped118				#else:119				#	return CS_Crashed120		return CS_NonExist121	#==============================================================================122	# docker get container id123	#==============================================================================124	def get_container_id(self,container):125		for c in self.client.containers(all=True):126			names_list = c['Names']127			name = "/"+container128			exist = name in names_list129			if exist:130				return c['Id']131		return None132	def inspect_container(self,container):133		return self.client.inspect_container(container)134	def diff_container(self,container):135		return self.client.diff(container)136	def copy_from_container(self,container,filepath):137		return self.client.copy(container=container,resource=filepath)138	def port_container(self,container,private_port):139		return self.client.port(container=container,private_port=private_port)140	def top_container(self,container):141		return self.client.top(container)142	def stats_controller(self,container):143		return self.client.stats(container)144class BaseContainer(ContainerClient):145	"""146	base container147	"""148	def __init__(self,docker_base_url,container_param):149		ContainerClient.__init__(self,docker_base_url)150		self.container_param = container_param151      		""" @type: L{ContainerParam}"""152	#==============================================================================153	# basic getter154	#==============================================================================155	def get_docker_base_url(self):156		return self.docker_base_url157	def get_container_param(self):158		return self.container_param159	#==============================================================================160	# useful functions161	#==============================================================================162	def get_status(self):163		return ContainerClient.get_container_status(self,self.container_param.name)164	def get_id(self):165		return ContainerClient.get_container_id(self,self.container_param.name)166	#==============================================================================167	# start/stop/restart168	#==============================================================================169	def start(self):170		# if not exist,then create it and start it171		status = self.get_status()172		if status == CS_NonExist:173			print "[BaseContainer] container {0} does't exist, so create it first...".format(self.name)174			ContainerClient.create_container(self,self.container_param)175			ContainerClient.start_container(self,self.container_param)176		elif status == CS_Running:177			print "[BaseContainer] container {0} has already been started!".format(self.name)178		elif status == CS_Stopped:179			ContainerClient.start_container(self,self.container_param)180	def stop(self):181		status = self.get_status()182		if status == CS_Running:183			ContainerClient.stop_container(self,self.container_param.name)184		elif status == CS_Stopped:185			print "[BaseContainer] container {0} has already been stopped!".format(self.name)186	def restart(self):187		status = self.get_status()188		if status == CS_Running:189			ContainerClient.stop_container(self,self.container_param.name)190			ContainerClient.start_container(self,self.container_param)191		elif status == CS_Stopped:192			print "[BaseContainer] container {0} has already been stopped! Start it first!".format(self.name)193		194class ApacheContainer(BaseContainer):195	def __init__(self,docker_base_url):196		container_param = self.create_container_param()197		BaseContainer.__init__(self,docker_base_url,container_param)198	def create_container_param(self):199		pass200class MysqlContainer(BaseContainer):201	def __init__(self,docker_base_url):202		container_param = self.create_container_param()203		BaseContainer.__init__(self,docker_base_url,container_param)204	def create_container_param(self):205		pass206class VgeoContainer(BaseContainer):207	def __init__(self,docker_base_url,container_param):208		BaseContainer.__init__(self,docker_base_url,container_param)209class RobustContainer(VgeoContainer):210	def __init__(self,docker_base_url):211		container_param = self.create_container_param()212		VgeoContainer.__init__(self,docker_base_url,container_param)213	def create_container_param(self):214		pass215class OpensimContainer(VgeoContainer):216	def __init__(self,docker_base_url):217		container_param = self.create_container_param()218		VgeoContainer.__init__(self,docker_base_url,container_param)219	def create_container_param(self):220		pass221#class ContainerTesting(ApacheContainer,MysqlContainer,RobustContainer,OpensimContainer):222#	pass223def test_docker_client():224	client = ContainerClient(DOCKER_SERVER_URL)225	print client.get_container_status('apache')226	print client.get_container_status('mysql')227	print client.get_container_status('robust')228	print client.get_container_status('huyu')229	print client.get_container_status('xwd')230	print client.get_container_status('mycontainer')231	print client.get_container_status('xxx')232def format_percentage(a,b):233	return "{0:.2f}%".format(float(a)/b * 100)234def test_stats():235	client = DockerClient(DOCKER_SERVER_URL)236	stats_obj = client.stats_controller("docker-registry")237	counter = 0238	for stat in stats_obj:239		d = json.loads(stat)240		total_usage = d['cpu_stats']['cpu_usage']['total_usage']241		usage_in_usermode = d['cpu_stats']['cpu_usage']['usage_in_usermode']242		usage_in_kernelmode = d['cpu_stats']['cpu_usage']['usage_in_kernelmode']243		system_cpu_usage = d['cpu_stats']['system_cpu_usage']244		#cpu_percentage = format_percentage(total_usage,system_cpu_usage)245		cpu_percentage = format_percentage(u,system_cpu_usage)...

monitor.py

Source:monitor.py

...8def get_container_info(container):9    msg = commands.getoutput('docker inspect '+container)10    data = json.loads(msg)11    return data[0]12def get_container_status(data):13    return data['State']['Running']14def stop_container_ui(container_ui_name):15    return commands.getoutput('docker stop '+container_ui_name)16def start_container_ui(container_ui_name):17    return commands.getoutput('docker start '+container_ui_name)18if get_container_status(get_container_info(registry_ui)):19    for i in container:20        if not get_container_status(get_container_info(i)):...

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.