KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
