From 876f0c05dca23b36826185079bbec07fa6892527 Mon Sep 17 00:00:00 2001
From: Falmarri <463948+Falmarri@users.noreply.github.com>
Date: Sat, 9 Dec 2023 22:51:54 -0800
Subject: [PATCH 1/4] Use asyncio for subprocess calls

Removes the threads from compose_up and manages it using async. Also
uses async processing to format the log messages instead of piping
through sed. This should work on windows without having sed installed

Adds --parallel to support pull and build in parallel, same as docker
compose

Signed-off-by: Falmarri <463948+Falmarri@users.noreply.github.com>
---
 podman_compose.py            | 354 +++++++++++++++++++----------------
 tests/test_podman_compose.py |   8 +-
 2 files changed, 196 insertions(+), 166 deletions(-)

diff --git a/podman_compose.py b/podman_compose.py
index 20011455..ee661298 100755
--- a/podman_compose.py
+++ b/podman_compose.py
@@ -1,29 +1,24 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-
 # https://docs.docker.com/compose/compose-file/#service-configuration-reference
 # https://docs.docker.com/samples/
 # https://docs.docker.com/compose/gettingstarted/
 # https://docs.docker.com/compose/django/
 # https://docs.docker.com/compose/wordpress/
-
 # TODO: podman pod logs --color -n -f pod_testlogs
-
-
 import sys
 import os
 import getpass
 import argparse
 import itertools
 import subprocess
-import time
 import re
 import hashlib
 import random
 import json
 import glob
-
-from threading import Thread
+import asyncio.subprocess
+import signal
 
 import shlex
 
@@ -371,7 +366,7 @@ def transform(args, project_name, given_containers):
     return pods, containers
 
 
-def assert_volume(compose, mount_dict):
+async def assert_volume(compose, mount_dict):
     """
     inspect volume to get directory
     create volume if needed
@@ -398,7 +393,7 @@ def assert_volume(compose, mount_dict):
     # TODO: might move to using "volume list"
     # podman volume list --format '{{.Name}}\t{{.MountPoint}}' -f 'label=io.podman.compose.project=HERE'
     try:
-        _ = compose.podman.output([], "volume", ["inspect", vol_name]).decode("utf-8")
+        _ = (await compose.podman.output([], "volume", ["inspect", vol_name])).decode("utf-8")
     except subprocess.CalledProcessError as e:
         if is_ext:
             raise RuntimeError(f"External volume [{vol_name}] does not exists") from e
@@ -419,8 +414,8 @@ def assert_volume(compose, mount_dict):
         for opt, value in driver_opts.items():
             args.extend(["--opt", f"{opt}={value}"])
         args.append(vol_name)
-        compose.podman.output([], "volume", args)
-        _ = compose.podman.output([], "volume", ["inspect", vol_name]).decode("utf-8")
+        await compose.podman.output([], "volume", args)
+        _ = (await compose.podman.output([], "volume", ["inspect", vol_name])).decode("utf-8")
 
 
 def mount_desc_to_mount_args(
@@ -522,12 +517,12 @@ def get_mnt_dict(compose, cnt, volume):
     return fix_mount_dict(compose, volume, proj_name, srv_name)
 
 
-def get_mount_args(compose, cnt, volume):
+async def get_mount_args(compose, cnt, volume):
     volume = get_mnt_dict(compose, cnt, volume)
     # proj_name = compose.project_name
     srv_name = cnt["_service"]
     mount_type = volume["type"]
-    assert_volume(compose, volume)
+    await assert_volume(compose, volume)
     if compose.prefer_volume_over_mount:
         if mount_type == "tmpfs":
             # TODO: --tmpfs /tmp:rw,size=787448k,mode=1777
@@ -710,7 +705,7 @@ def norm_ports(ports_in):
     return ports_out
 
 
-def assert_cnt_nets(compose, cnt):
+async def assert_cnt_nets(compose, cnt):
     """
     create missing networks
     """
@@ -733,7 +728,7 @@ def assert_cnt_nets(compose, cnt):
             ext_desc.get("name", None) or net_desc.get("name", None) or default_net_name
         )
         try:
-            compose.podman.output([], "network", ["exists", net_name])
+            await compose.podman.output([], "network", ["exists", net_name])
         except subprocess.CalledProcessError as e:
             if is_ext:
                 raise RuntimeError(
@@ -776,8 +771,8 @@ def assert_cnt_nets(compose, cnt):
                 if gateway:
                     args.extend(("--gateway", gateway))
             args.append(net_name)
-            compose.podman.output([], "network", args)
-            compose.podman.output([], "network", ["exists", net_name])
+            await compose.podman.output([], "network", args)
+            await compose.podman.output([], "network", ["exists", net_name])
 
 
 def get_net_args(compose, cnt):
@@ -898,7 +893,7 @@ def get_net_args(compose, cnt):
     return net_args
 
 
-def container_to_args(compose, cnt, detached=True):
+async def container_to_args(compose, cnt, detached=True):
     # TODO: double check -e , --add-host, -v, --read-only
     dirname = compose.dirname
     pod = cnt.get("pod", None) or ""
@@ -955,9 +950,9 @@ def container_to_args(compose, cnt, detached=True):
     for i in tmpfs_ls:
         podman_args.extend(["--tmpfs", i])
     for volume in cnt.get("volumes", []):
-        podman_args.extend(get_mount_args(compose, cnt, volume))
+        podman_args.extend(await get_mount_args(compose, cnt, volume))
 
-    assert_cnt_nets(compose, cnt)
+    await assert_cnt_nets(compose, cnt)
     podman_args.extend(get_net_args(compose, cnt))
 
     logging = cnt.get("logging", None)
@@ -1161,12 +1156,22 @@ def __init__(self, compose, podman_path="podman", dry_run=False):
         self.podman_path = podman_path
         self.dry_run = dry_run
 
-    def output(self, podman_args, cmd="", cmd_args=None):
+    async def output(self, podman_args, cmd="", cmd_args=None):
         cmd_args = cmd_args or []
         xargs = self.compose.get_podman_args(cmd) if cmd else []
         cmd_ls = [self.podman_path, *podman_args, cmd] + xargs + cmd_args
         log(cmd_ls)
-        return subprocess.check_output(cmd_ls)
+        p = await asyncio.subprocess.create_subprocess_exec(
+            *cmd_ls,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE
+        )
+
+        stdout_data, stderr_data = await p.communicate()
+        if p.returncode == 0:
+            return stdout_data
+        else:
+            raise subprocess.CalledProcessError(p.returncode, " ".join(cmd_ls), stderr_data)
 
     def exec(
         self,
@@ -1180,55 +1185,71 @@ def exec(
         log(" ".join([str(i) for i in cmd_ls]))
         os.execlp(self.podman_path, *cmd_ls)
 
-    def run(
+    async def run(
         self,
         podman_args,
         cmd="",
         cmd_args=None,
         wait=True,
         sleep=1,
-        obj=None,
         log_formatter=None,
+        *,
+        # Intentionally mutable default argument to hold references to tasks
+        task_reference=set()
     ):
-        if obj is not None:
-            obj.exit_code = None
         cmd_args = list(map(str, cmd_args or []))
         xargs = self.compose.get_podman_args(cmd) if cmd else []
         cmd_ls = [self.podman_path, *podman_args, cmd] + xargs + cmd_args
         log(" ".join([str(i) for i in cmd_ls]))
         if self.dry_run:
             return None
-        # subprocess.Popen(
-        #   args, bufsize = 0, executable = None, stdin = None, stdout = None, stderr = None, preexec_fn = None,
-        #   close_fds = False, shell = False, cwd = None, env = None, universal_newlines = False, startupinfo = None,
-        #   creationflags = 0
-        # )
         if log_formatter is not None:
+
+            async def format_out(stdout):
+                while True:
+                    l = await stdout.readline()
+                    if l:
+                        print(log_formatter, l.decode('utf-8'), end='')
+                    if stdout.at_eof():
+                        break
+
+            # read, write = os.pipe()
             # Pipe podman process output through log_formatter (which can add colored prefix)
-            p = subprocess.Popen(
-                cmd_ls, stdout=subprocess.PIPE
-            )  # pylint: disable=consider-using-with
-            _ = subprocess.Popen(
-                log_formatter, stdin=p.stdout
+            p = await asyncio.subprocess.create_subprocess_exec(
+                *cmd_ls, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
             )  # pylint: disable=consider-using-with
-            p.stdout.close()  # Allow p_process to receive a SIGPIPE if logging process exits.
+
+            # This is hacky to make the tasks not get garbage collected
+            # https://github.com/python/cpython/issues/91887
+            out_t = asyncio.create_task(format_out(p.stdout))
+            task_reference.add(out_t)
+            out_t.add_done_callback(task_reference.discard)
+
+            err_t = asyncio.create_task(format_out(p.stderr))
+            task_reference.add(err_t)
+            err_t.add_done_callback(task_reference.discard)
         else:
-            p = subprocess.Popen(cmd_ls)  # pylint: disable=consider-using-with
+            p = await asyncio.subprocess.create_subprocess_exec(*cmd_ls)  # pylint: disable=consider-using-with
 
         if wait:
-            exit_code = p.wait()
+            try:
+                exit_code = await p.wait()
+            except asyncio.CancelledError:
+                p.terminate()
+                exit_code = await p.wait()
+
             log("exit code:", exit_code)
-            if obj is not None:
-                obj.exit_code = exit_code
+            return exit_code
 
         if sleep:
-            time.sleep(sleep)
+            log(f"Sleep {sleep}")
+            await asyncio.sleep(sleep)
         return p
 
-    def volume_ls(self, proj=None):
+    async def volume_ls(self, proj=None):
         if not proj:
             proj = self.compose.project_name
-        output = self.output(
+        output = (await self.output(
             [],
             "volume",
             [
@@ -1239,7 +1260,7 @@ def volume_ls(self, proj=None):
                 "--format",
                 "{{.Name}}",
             ],
-        ).decode("utf-8")
+        )).decode("utf-8")
         volumes = output.splitlines()
         return volumes
 
@@ -1487,7 +1508,7 @@ def get_podman_args(self, cmd):
             xargs.extend(shlex.split(args))
         return xargs
 
-    def run(self):
+    async def run(self):
         log("podman-compose version: " + __version__)
         args = self._parse_args()
         podman_path = args.podman_path
@@ -1504,8 +1525,8 @@ def run(self):
             # just to make sure podman is running
             try:
                 self.podman_version = (
-                    self.podman.output(["--version"], "", []).decode("utf-8").strip()
-                    or ""
+                        (await self.podman.output(["--version"], "", [])).decode("utf-8").strip()
+                        or ""
                 )
                 self.podman_version = (self.podman_version.split() or [""])[-1]
             except subprocess.CalledProcessError:
@@ -1521,7 +1542,7 @@ def run(self):
         if compose_required:
             self._parse_compose_file()
         cmd = self.commands[cmd_name]
-        retcode = cmd(self, args)
+        retcode = await cmd(self, args)
         if isinstance(retcode, int):
             sys.exit(retcode)
 
@@ -1919,6 +1940,8 @@ def __call__(self, func):
         def wrapped(*args, **kw):
             return func(*args, **kw)
 
+        if not asyncio.iscoroutinefunction(func):
+            raise Exception("Command must be async")
         wrapped._compose = self.compose
         # Trim extra indentation at start of multiline docstrings.
         wrapped.desc = self.cmd_desc or re.sub(r"^\s+", "", func.__doc__)
@@ -1947,7 +1970,7 @@ def wrapped(*args, **kw):
 
 
 @cmd_run(podman_compose, "version", "show version")
-def compose_version(compose, args):
+async def compose_version(compose, args):
     if getattr(args, "short", False):
         print(__version__)
         return
@@ -1956,7 +1979,7 @@ def compose_version(compose, args):
         print(json.dumps(res))
         return
     print("podman-compose version", __version__)
-    compose.podman.run(["--version"], "", [], sleep=0)
+    await compose.podman.run(["--version"], "", [], sleep=0)
 
 
 def is_local(container: dict) -> bool:
@@ -1972,15 +1995,15 @@ def is_local(container: dict) -> bool:
 
 
 @cmd_run(podman_compose, "wait", "wait running containers to stop")
-def compose_wait(compose, args):  # pylint: disable=unused-argument
+async def compose_wait(compose, args):  # pylint: disable=unused-argument
     containers = [cnt["name"] for cnt in compose.containers]
     cmd_args = ["--"]
     cmd_args.extend(containers)
-    compose.podman.exec([], "wait", cmd_args)
+    await compose.podman.exec([], "wait", cmd_args)
 
 
 @cmd_run(podman_compose, "systemd")
-def compose_systemd(compose, args):
+async def compose_systemd(compose, args):
     """
     create systemd unit file and register its compose stacks
 
@@ -2063,7 +2086,7 @@ def compose_systemd(compose, args):
 
 
 @cmd_run(podman_compose, "pull", "pull stack images")
-def compose_pull(compose, args):
+async def compose_pull(compose, args):
     img_containers = [cnt for cnt in compose.containers if "image" in cnt]
     if args.services:
         services = set(args.services)
@@ -2072,27 +2095,33 @@ def compose_pull(compose, args):
     if not args.force_local:
         local_images = {cnt["image"] for cnt in img_containers if is_local(cnt)}
         images -= local_images
-    for image in images:
-        compose.podman.run([], "pull", [image], sleep=0)
+
+    sem = asyncio.Semaphore(args.parallel)
+
+    async def _pull(image: str):
+        async with sem:
+            return await compose.podman.run([], "pull", [image], sleep=0)
+
+    await asyncio.gather(*[_pull(image) for image in images])
 
 
 @cmd_run(podman_compose, "push", "push stack images")
-def compose_push(compose, args):
+async def compose_push(compose, args):
     services = set(args.services)
     for cnt in compose.containers:
         if "build" not in cnt:
             continue
         if services and cnt["_service"] not in services:
             continue
-        compose.podman.run([], "push", [cnt["image"]], sleep=0)
+        await compose.podman.run([], "push", [cnt["image"]], sleep=0)
 
 
-def build_one(compose, args, cnt):
+async def build_one(compose, args, cnt):
     if "build" not in cnt:
         return None
     if getattr(args, "if_not_exists", None):
         try:
-            img_id = compose.podman.output(
+            img_id = await compose.podman.output(
                 [], "inspect", ["-t", "image", "-f", "{{.Id}}", cnt["image"]]
             )
         except subprocess.CalledProcessError:
@@ -2148,40 +2177,40 @@ def build_one(compose, args, cnt):
             )
         )
     build_args.append(ctx)
-    status = compose.podman.run([], "build", build_args, sleep=0)
+    status = await compose.podman.run([], "build", build_args, sleep=0)
     return status
 
 
 @cmd_run(podman_compose, "build", "build stack images")
-def compose_build(compose, args):
-    # keeps the status of the last service/container built
-    status = 0
+async def compose_build(compose, args):
+    tasks = []
+    sem = asyncio.Semaphore(args.parallel)
 
-    def parse_return_code(obj, current_status):
-        if obj and obj.returncode != 0:
-            return obj.returncode
-        return current_status
+    async def safe_build_task(cnt):
+        async with sem:
+            return await build_one(compose, args, cnt)
 
     if args.services:
         container_names_by_service = compose.container_names_by_service
         compose.assert_services(args.services)
         for service in args.services:
             cnt = compose.container_by_name[container_names_by_service[service][0]]
-            p = build_one(compose, args, cnt)
-            status = parse_return_code(p, status)
-            if status != 0:
-                return status
+            tasks.append(asyncio.create_task(safe_build_task(cnt)))
+
     else:
         for cnt in compose.containers:
-            p = build_one(compose, args, cnt)
-            status = parse_return_code(p, status)
-            if status != 0:
-                return status
+            tasks.append(asyncio.create_task(safe_build_task(cnt)))
+
+    status = 0
+    for t in asyncio.as_completed(tasks):
+        s = await t
+        if s is not None:
+            status = s
 
     return status
 
 
-def create_pods(compose, args):  # pylint: disable=unused-argument
+async def create_pods(compose, args):  # pylint: disable=unused-argument
     for pod in compose.pods:
         podman_args = [
             "create",
@@ -2196,7 +2225,7 @@ def create_pods(compose, args):  # pylint: disable=unused-argument
             ports = [ports]
         for i in ports:
             podman_args.extend(["-p", str(i)])
-        compose.podman.run([], "pod", podman_args)
+        await compose.podman.run([], "pod", podman_args)
 
 
 def get_excluded(compose, args):
@@ -2213,16 +2242,16 @@ def get_excluded(compose, args):
 @cmd_run(
     podman_compose, "up", "Create and start the entire stack or some of its services"
 )
-def compose_up(compose, args):
+async def compose_up(compose: PodmanCompose, args):
     proj_name = compose.project_name
     excluded = get_excluded(compose, args)
     if not args.no_build:
         # `podman build` does not cache, so don't always build
         build_args = argparse.Namespace(if_not_exists=(not args.build), **args.__dict__)
-        compose.commands["build"](compose, build_args)
+        ret = await compose.commands["build"](compose, build_args)
 
     hashes = (
-        compose.podman.output(
+        (await compose.podman.output(
             [],
             "ps",
             [
@@ -2232,7 +2261,7 @@ def compose_up(compose, args):
                 "--format",
                 '{{ index .Labels "io.podman.compose.config-hash"}}',
             ],
-        )
+        ))
         .decode("utf-8")
         .splitlines()
     )
@@ -2240,21 +2269,21 @@ def compose_up(compose, args):
     if args.force_recreate or len(diff_hashes):
         log("recreating: ...")
         down_args = argparse.Namespace(**dict(args.__dict__, volumes=False))
-        compose.commands["down"](compose, down_args)
+        await compose.commands["down"](compose, down_args)
         log("recreating: done\n\n")
     # args.no_recreate disables check for changes (which is not implemented)
 
     podman_command = "run" if args.detach and not args.no_start else "create"
 
-    create_pods(compose, args)
+    await create_pods(compose, args)
     for cnt in compose.containers:
         if cnt["_service"] in excluded:
             log("** skipping: ", cnt["name"])
             continue
-        podman_args = container_to_args(compose, cnt, detached=args.detach)
-        subproc = compose.podman.run([], podman_command, podman_args)
-        if podman_command == "run" and subproc and subproc.returncode:
-            compose.podman.run([], "start", [cnt["name"]])
+        podman_args = await container_to_args(compose, cnt, detached=args.detach)
+        subproc = await compose.podman.run([], podman_command, podman_args)
+        if podman_command == "run" and subproc:
+            await compose.podman.run([], "start", [cnt["name"]])
     if args.no_start or args.detach or args.dry_run:
         return
     # TODO: handle already existing
@@ -2264,54 +2293,41 @@ def compose_up(compose, args):
     if exit_code_from:
         args.abort_on_container_exit = True
 
-    threads = []
-
     max_service_length = 0
     for cnt in compose.containers:
         curr_length = len(cnt["_service"])
         max_service_length = (
             curr_length if curr_length > max_service_length else max_service_length
         )
-    has_sed = os.path.isfile("/bin/sed")
+
+    tasks = set()
+
+    loop = asyncio.get_event_loop()
+    loop.add_signal_handler(signal.SIGINT, lambda: [t.cancel("User exit") for t in tasks])
+
     for i, cnt in enumerate(compose.containers):
         # Add colored service prefix to output by piping output through sed
         color_idx = i % len(compose.console_colors)
         color = compose.console_colors[color_idx]
         space_suffix = " " * (max_service_length - len(cnt["_service"]) + 1)
-        log_formatter = "s/^/{}[{}]{}|\x1B[0m\\ /;".format(
+        log_formatter = "{}[{}]{}|\x1B[0m".format(
             color, cnt["_service"], space_suffix
         )
-        log_formatter = ["sed", "-e", log_formatter] if has_sed else None
         if cnt["_service"] in excluded:
             log("** skipping: ", cnt["name"])
             continue
-        # TODO: remove sleep from podman.run
-        obj = compose if exit_code_from == cnt["_service"] else None
-        thread = Thread(
-            target=compose.podman.run,
-            args=[[], "start", ["-a", cnt["name"]]],
-            kwargs={"obj": obj, "log_formatter": log_formatter},
-            daemon=True,
-            name=cnt["name"],
-        )
-        thread.start()
-        threads.append(thread)
-        time.sleep(1)
-
-    while threads:
-        to_remove = []
-        for thread in threads:
-            thread.join(timeout=1.0)
-            if not thread.is_alive():
-                to_remove.append(thread)
-                if args.abort_on_container_exit:
-                    time.sleep(1)
-                    exit_code = (
-                        compose.exit_code if compose.exit_code is not None else -1
-                    )
-                    sys.exit(exit_code)
-        for thread in to_remove:
-            threads.remove(thread)
+
+        tasks.add(asyncio.create_task(compose.podman.run([], "start", ["-a", cnt["name"]], wait=True, sleep=None, log_formatter=log_formatter), name=cnt["_service"]))
+
+    exit_code = 0
+    for task in asyncio.as_completed(tasks):
+        t = await task
+        if args.abort_on_container_exit:
+            [_.cancel() for _ in tasks if not _.cancelling() and not _.cancelled()]
+            if task.get_name() == exit_code_from:
+                exit_code = t
+
+    sys.exit(exit_code)
 
 
 def get_volume_names(compose, cnt):
@@ -2332,7 +2348,7 @@ def get_volume_names(compose, cnt):
 
 
 @cmd_run(podman_compose, "down", "tear down entire stack")
-def compose_down(compose, args):
+async def compose_down(compose, args):
     excluded = get_excluded(compose, args)
     podman_args = []
     timeout_global = getattr(args, "timeout", None)
@@ -2348,14 +2364,14 @@ def compose_down(compose, args):
             timeout = str_to_seconds(timeout_str)
         if timeout is not None:
             podman_stop_args.extend(["-t", str(timeout)])
-        compose.podman.run([], "stop", [*podman_stop_args, cnt["name"]], sleep=0)
+        await compose.podman.run([], "stop", [*podman_stop_args, cnt["name"]], sleep=0)
     for cnt in containers:
         if cnt["_service"] in excluded:
             continue
-        compose.podman.run([], "rm", [cnt["name"]], sleep=0)
+        await compose.podman.run([], "rm", [cnt["name"]], sleep=0)
     if args.remove_orphans:
         names = (
-            compose.podman.output(
+            (await compose.podman.output(
                 [],
                 "ps",
                 [
@@ -2365,14 +2381,14 @@ def compose_down(compose, args):
                     "--format",
                     "{{ .Names }}",
                 ],
-            )
+            ))
             .decode("utf-8")
             .splitlines()
         )
         for name in names:
-            compose.podman.run([], "stop", [*podman_args, name], sleep=0)
+            await compose.podman.run([], "stop", [*podman_args, name], sleep=0)
         for name in names:
-            compose.podman.run([], "rm", [name], sleep=0)
+            await compose.podman.run([], "rm", [name], sleep=0)
     if args.volumes:
         vol_names_to_keep = set()
         for cnt in containers:
@@ -2380,22 +2396,22 @@ def compose_down(compose, args):
                 continue
             vol_names_to_keep.update(get_volume_names(compose, cnt))
         log("keep", vol_names_to_keep)
-        for volume_name in compose.podman.volume_ls():
+        async for volume_name in compose.podman.volume_ls():
             if volume_name in vol_names_to_keep:
                 continue
-            compose.podman.run([], "volume", ["rm", volume_name])
+            await compose.podman.run([], "volume", ["rm", volume_name])
 
     if excluded:
         return
     for pod in compose.pods:
-        compose.podman.run([], "pod", ["rm", pod["name"]], sleep=0)
+        await compose.podman.run([], "pod", ["rm", pod["name"]], sleep=0)
 
 
 @cmd_run(podman_compose, "ps", "show status of containers")
-def compose_ps(compose, args):
+async def compose_ps(compose, args):
     proj_name = compose.project_name
     if args.quiet is True:
-        compose.podman.run(
+        await compose.podman.run(
             [],
             "ps",
             [
@@ -2407,7 +2423,7 @@ def compose_ps(compose, args):
             ],
         )
     else:
-        compose.podman.run(
+        await compose.podman.run(
             [], "ps", ["-a", "--filter", f"label=io.podman.compose.project={proj_name}"]
         )
 
@@ -2417,7 +2433,7 @@ def compose_ps(compose, args):
     "run",
     "create a container similar to a service to run a one-off command",
 )
-def compose_run(compose, args):
+async def compose_run(compose, args):
     create_pods(compose, args)
     compose.assert_services(args.service)
     container_names = compose.container_names_by_service[args.service]
@@ -2483,17 +2499,17 @@ def compose_run(compose, args):
     if args.rm and "restart" in cnt:
         del cnt["restart"]
     # run podman
-    podman_args = container_to_args(compose, cnt, args.detach)
+    podman_args = await container_to_args(compose, cnt, args.detach)
     if not args.detach:
         podman_args.insert(1, "-i")
         if args.rm:
             podman_args.insert(1, "--rm")
-    p = compose.podman.run([], "run", podman_args, sleep=0)
-    sys.exit(p.returncode)
+    p = await compose.podman.run([], "run", podman_args, sleep=0)
+    sys.exit(p)
 
 
 @cmd_run(podman_compose, "exec", "execute a command in a running container")
-def compose_exec(compose, args):
+async def compose_exec(compose, args):
     compose.assert_services(args.service)
     container_names = compose.container_names_by_service[args.service]
     container_name = container_names[args.index - 1]
@@ -2518,11 +2534,11 @@ def compose_exec(compose, args):
     podman_args += [container_name]
     if args.cnt_command is not None and len(args.cnt_command) > 0:
         podman_args += args.cnt_command
-    p = compose.podman.run([], "exec", podman_args, sleep=0)
-    sys.exit(p.returncode)
+    p = await compose.podman.run([], "exec", podman_args, sleep=0)
+    sys.exit(p)
 
 
-def transfer_service_status(compose, args, action):
+async def transfer_service_status(compose, args, action):
     # TODO: handle dependencies, handle creations
     container_names_by_service = compose.container_names_by_service
     if not args.services:
@@ -2537,6 +2553,7 @@ def transfer_service_status(compose, args, action):
         targets = list(reversed(targets))
     podman_args = []
     timeout_global = getattr(args, "timeout", None)
+    tasks = []
     for target in targets:
         if action != "start":
             timeout = timeout_global
@@ -2548,26 +2565,27 @@ def transfer_service_status(compose, args, action):
                 timeout = str_to_seconds(timeout_str)
             if timeout is not None:
                 podman_args.extend(["-t", str(timeout)])
-        compose.podman.run([], action, podman_args + [target], sleep=0)
+        tasks.append(asyncio.create_task(compose.podman.run([], action, podman_args + [target], sleep=0)))
+    await asyncio.gather(*tasks)
 
 
 @cmd_run(podman_compose, "start", "start specific services")
-def compose_start(compose, args):
+async def compose_start(compose, args):
     transfer_service_status(compose, args, "start")
 
 
 @cmd_run(podman_compose, "stop", "stop specific services")
-def compose_stop(compose, args):
+async def compose_stop(compose, args):
     transfer_service_status(compose, args, "stop")
 
 
 @cmd_run(podman_compose, "restart", "restart specific services")
-def compose_restart(compose, args):
+async def compose_restart(compose, args):
     transfer_service_status(compose, args, "restart")
 
 
 @cmd_run(podman_compose, "logs", "show logs from services")
-def compose_logs(compose, args):
+async def compose_logs(compose, args):
     container_names_by_service = compose.container_names_by_service
     if not args.services and not args.latest:
         args.services = container_names_by_service.keys()
@@ -2594,11 +2612,11 @@ def compose_logs(compose, args):
         podman_args.extend(["--until", args.until])
     for target in targets:
         podman_args.append(target)
-    compose.podman.run([], "logs", podman_args)
+    await compose.podman.run([], "logs", podman_args)
 
 
 @cmd_run(podman_compose, "config", "displays the compose file")
-def compose_config(compose, args):
+async def compose_config(compose, args):
     if args.services:
         for service in compose.services:
             print(service)
@@ -2607,7 +2625,7 @@ def compose_config(compose, args):
 
 
 @cmd_run(podman_compose, "port", "Prints the public port for a port binding.")
-def compose_port(compose, args):
+async def compose_port(compose, args):
     # TODO - deal with pod index
     compose.assert_services(args.service)
     containers = compose.container_names_by_service[args.service]
@@ -2635,31 +2653,31 @@ def _published_target(port_string):
 
 
 @cmd_run(podman_compose, "pause", "Pause all running containers")
-def compose_pause(compose, args):
+async def compose_pause(compose, args):
     container_names_by_service = compose.container_names_by_service
     if not args.services:
         args.services = container_names_by_service.keys()
     targets = []
     for service in args.services:
         targets.extend(container_names_by_service[service])
-    compose.podman.run([], "pause", targets)
+    await compose.podman.run([], "pause", targets)
 
 
 @cmd_run(podman_compose, "unpause", "Unpause all running containers")
-def compose_unpause(compose, args):
+async def compose_unpause(compose, args):
     container_names_by_service = compose.container_names_by_service
     if not args.services:
         args.services = container_names_by_service.keys()
     targets = []
     for service in args.services:
         targets.extend(container_names_by_service[service])
-    compose.podman.run([], "unpause", targets)
+    await compose.podman.run([], "unpause", targets)
 
 
 @cmd_run(
     podman_compose, "kill", "Kill one or more running containers with a specific signal"
 )
-def compose_kill(compose, args):
+async def compose_kill(compose, args):
     # to ensure that the user did not execute the command by mistake
     if not args.services and not args.all:
         print(
@@ -2680,15 +2698,14 @@ def compose_kill(compose, args):
             targets.extend(container_names_by_service[service])
         for target in targets:
             podman_args.append(target)
-        compose.podman.run([], "kill", podman_args)
-
-    if args.services:
+        await compose.podman.run([], "kill", podman_args)
+    elif args.services:
         targets = []
         for service in args.services:
             targets.extend(container_names_by_service[service])
         for target in targets:
             podman_args.append(target)
-        compose.podman.run([], "kill", podman_args)
+        await compose.podman.run([], "kill", podman_args)
 
 
 @cmd_run(
@@ -2696,7 +2713,7 @@ def compose_kill(compose, args):
     "stats",
     "Display percentage of CPU, memory, network I/O, block I/O and PIDs for services.",
 )
-def compose_stats(compose, args):
+async def compose_stats(compose, args):
     container_names_by_service = compose.container_names_by_service
     if not args.services:
         args.services = container_names_by_service.keys()
@@ -2717,7 +2734,7 @@ def compose_stats(compose, args):
         podman_args.append(target)
 
     try:
-        compose.podman.run([], "stats", podman_args)
+        await compose.podman.run([], "stats", podman_args)
     except KeyboardInterrupt:
         pass
 
@@ -3075,6 +3092,15 @@ def compose_ps_parse(parser):
     )
 
 
+@cmd_parse(podman_compose, ["build", "pull", "up"])
+def compose_build_pull_parse(parser):
+    parser.add_argument(
+        "--parallel",
+        type=int,
+        default=os.environ.get("PODMAN_PARALLEL", sys.maxsize)
+    )
+
+
 @cmd_parse(podman_compose, ["build", "up"])
 def compose_build_up_parse(parser):
     parser.add_argument(
@@ -3201,9 +3227,9 @@ def compose_stats_parse(parser):
     )
 
 
-def main():
-    podman_compose.run()
+async def main():
+    await podman_compose.run()
 
 
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())
diff --git a/tests/test_podman_compose.py b/tests/test_podman_compose.py
index 14e80b2c..0984431a 100644
--- a/tests/test_podman_compose.py
+++ b/tests/test_podman_compose.py
@@ -33,6 +33,8 @@ def test_podman_compose_extends_w_file_subdir():
         "podman",
         "container",
         "ps",
+        "--sort",
+        "status",
         "--all",
         "--format",
         '"{{.Image}}"',
@@ -51,14 +53,16 @@ def test_podman_compose_extends_w_file_subdir():
     # check container was created and exists
     out, _, returncode = capture(command_check_container)
     assert 0 == returncode
-    assert out == b'"localhost/subdir_test:me"\n'
+    assert b'"localhost/subdir_test:me"\n' in out
     out, _, returncode = capture(command_down)
     # cleanup test image(tags)
     assert 0 == returncode
+    print('ok')
     # check container did not exists anymore
     out, _, returncode = capture(command_check_container)
+    print(out)
     assert 0 == returncode
-    assert out == b""
+    assert b'"localhost/subdir_test:me"\n' not in out
 
 
 def test_podman_compose_extends_w_empty_service():

From c0bfaac9afbb5ecf3b21fe5346f28606d846dba0 Mon Sep 17 00:00:00 2001
From: Falmarri <463948+Falmarri@users.noreply.github.com>
Date: Sun, 10 Dec 2023 15:26:58 -0800
Subject: [PATCH 2/4] Fix a couple issues and update docs

Signed-off-by: Falmarri <463948+Falmarri@users.noreply.github.com>
---
 .coveragerc                          |  2 ++
 .github/workflows/pytest.yml         |  9 +++++---
 CONTRIBUTING.md                      | 34 +++++++++++++++++-----------
 podman_compose.py                    | 10 ++++++--
 setup.py                             |  1 +
 test-requirements.txt                |  1 -
 tests/test_podman_compose.py         |  3 ++-
 tests/test_podman_compose_config.py  |  4 ++--
 tests/test_podman_compose_include.py |  3 ++-
 tests/test_podman_compose_up_down.py |  6 +++--
 10 files changed, 48 insertions(+), 25 deletions(-)
 create mode 100644 .coveragerc

diff --git a/.coveragerc b/.coveragerc
new file mode 100644
index 00000000..ba792219
--- /dev/null
+++ b/.coveragerc
@@ -0,0 +1,2 @@
+[run] 
+parallel=True
diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml
index d9620538..3cc3fad9 100644
--- a/.github/workflows/pytest.yml
+++ b/.github/workflows/pytest.yml
@@ -21,9 +21,10 @@ jobs:
         python-version: "3.10"
     - name: Install dependencies
       run: |
+        apt update && apt install podman
         python -m pip install --upgrade pip
-        pip install flake8 pytest
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+        if [ -f test-requirements.txt ]; then pip install -r test-requirements.txt; fi
     - name: Lint with flake8
       run: |
         # stop the build if there are Python syntax errors or undefined names
@@ -32,5 +33,7 @@ jobs:
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Test with pytest
       run: |
-        python -m pytest ./pytests
-
+        coverage run --source podman_compose -m pytest ./pytests
+        python -m pytest ./tests
+        coverage combine
+        coverage report
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 5cd371be..3aca3222 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -39,7 +39,15 @@ $ pre-commit install
 ```shell
 $ pre-commit run --all-files
 ```
-6. Commit your code to your fork's branch. 
+6. Run code coverage
+```shell
+coverage run --source podman_compose -m pytest ./pytests
+python -m pytest ./tests
+coverage combine
+coverage report
+coverage html
+```
+7. Commit your code to your fork's branch. 
    - Make sure you include a `Signed-off-by` message in your commits. Read [this guide](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits) to learn how to sign your commits 
    - In the commit message reference the Issue ID that your code fixes and a brief description of the changes. Example: `Fixes #516: allow empty network`
 7. Open a PR to `containers/podman-compose:devel` and wait for a maintainer to review your work.
@@ -48,18 +56,18 @@ $ pre-commit run --all-files
 
 To add a command you need to add a function that is decorated
 with `@cmd_run` passing the compose instance, command name and
-description. the wrapped function should accept two arguments
-the compose instance and the command-specific arguments (resulted
-from python's `argparse` package) inside that command you can
-run PodMan like this `compose.podman.run(['inspect', 'something'])`
-and inside that function you can access `compose.pods`
-and `compose.containers` ...etc.
-Here is an example
+description. This function must be declared `async` the wrapped 
+function should accept two arguments the compose instance and 
+the command-specific arguments (resulted from python's `argparse` 
+package) inside that command you can run PodMan like this 
+`await compose.podman.run(['inspect', 'something'])`and inside 
+that function you can access `compose.pods` and `compose.containers` 
+...etc. Here is an example
 
 ```
 @cmd_run(podman_compose, 'build', 'build images defined in the stack')
-def compose_build(compose, args):
-    compose.podman.run(['build', 'something'])
+async def compose_build(compose, args):
+    await compose.podman.run(['build', 'something'])
 ```
 
 ## Command arguments parsing
@@ -90,10 +98,10 @@ do something like:
 
 ```
 @cmd_run(podman_compose, 'up', 'up desc')
-def compose_up(compose, args):
-    compose.commands['down'](compose, args)
+async def compose_up(compose, args):
+    await compose.commands['down'](compose, args)
     # or
-    compose.commands['down'](argparse.Namespace(foo=123))
+    await compose.commands['down'](argparse.Namespace(foo=123))
 ```
 
 
diff --git a/podman_compose.py b/podman_compose.py
index ee661298..973ddb2f 100755
--- a/podman_compose.py
+++ b/podman_compose.py
@@ -2248,7 +2248,8 @@ async def compose_up(compose: PodmanCompose, args):
     if not args.no_build:
         # `podman build` does not cache, so don't always build
         build_args = argparse.Namespace(if_not_exists=(not args.build), **args.__dict__)
-        ret = await compose.commands["build"](compose, build_args)
+        if await compose.commands["build"](compose, build_args) != 0:
+            log("Build command failed")
 
     hashes = (
         (await compose.podman.output(
@@ -2317,7 +2318,12 @@ async def compose_up(compose: PodmanCompose, args):
             log("** skipping: ", cnt["name"])
             continue
 
-        tasks.add(asyncio.create_task(compose.podman.run([], "start", ["-a", cnt["name"]], wait=True, sleep=None, log_formatter=log_formatter), name=cnt["_service"]))
+        tasks.add(
+            asyncio.create_task(
+                compose.podman.run([], "start", ["-a", cnt["name"]], sleep=None, log_formatter=log_formatter),
+                name=cnt["_service"]
+            )
+        )
 
     exit_code = 0
     for task in asyncio.as_completed(tasks):
diff --git a/setup.py b/setup.py
index 5222c143..4994d331 100644
--- a/setup.py
+++ b/setup.py
@@ -45,6 +45,7 @@
             "black",
             "pylint",
             "pre-commit",
+            "coverage"
         ]
     }
     # test_suite='tests',
diff --git a/test-requirements.txt b/test-requirements.txt
index 5a204264..55c142a6 100644
--- a/test-requirements.txt
+++ b/test-requirements.txt
@@ -3,7 +3,6 @@
 # process, which may cause wedges in the gate later.
 
 coverage
-pytest-cov
 pytest
 tox
 black
diff --git a/tests/test_podman_compose.py b/tests/test_podman_compose.py
index 0984431a..5be6629b 100644
--- a/tests/test_podman_compose.py
+++ b/tests/test_podman_compose.py
@@ -21,7 +21,8 @@ def test_podman_compose_extends_w_file_subdir():
     main_path = Path(__file__).parent.parent
 
     command_up = [
-        "python3",
+        "coverage",
+        "run",
         str(main_path.joinpath("podman_compose.py")),
         "-f",
         str(main_path.joinpath("tests", "extends_w_file_subdir", "docker-compose.yml")),
diff --git a/tests/test_podman_compose_config.py b/tests/test_podman_compose_config.py
index 2f879ba2..96f9814d 100644
--- a/tests/test_podman_compose_config.py
+++ b/tests/test_podman_compose_config.py
@@ -22,7 +22,7 @@ def test_config_no_profiles(podman_compose_path, profile_compose_file):
     :param podman_compose_path: The fixture used to specify the path to the podman compose file.
     :param profile_compose_file: The fixtued used to specify the path to the "profile" compose used in the test.
     """
-    config_cmd = ["python3", podman_compose_path, "-f", profile_compose_file, "config"]
+    config_cmd = ["coverage", "run", podman_compose_path, "-f", profile_compose_file, "config"]
 
     out, _, return_code = capture(config_cmd)
     assert return_code == 0
@@ -61,7 +61,7 @@ def test_config_profiles(
     :param expected_services: Dictionary used to model the expected "enabled" services in the profile.
         Key = service name, Value = True if the service is enabled, otherwise False.
     """
-    config_cmd = ["python3", podman_compose_path, "-f", profile_compose_file]
+    config_cmd = ["coverage", "run", podman_compose_path, "-f", profile_compose_file]
     config_cmd.extend(profiles)
 
     out, _, return_code = capture(config_cmd)
diff --git a/tests/test_podman_compose_include.py b/tests/test_podman_compose_include.py
index c9867f56..0fec7734 100644
--- a/tests/test_podman_compose_include.py
+++ b/tests/test_podman_compose_include.py
@@ -20,7 +20,8 @@ def test_podman_compose_include():
     main_path = Path(__file__).parent.parent
 
     command_up = [
-        "python3",
+        "coverage",
+        "run",
         str(main_path.joinpath("podman_compose.py")),
         "-f",
         str(main_path.joinpath("tests", "include", "docker-compose.yaml")),
diff --git a/tests/test_podman_compose_up_down.py b/tests/test_podman_compose_up_down.py
index 833604e4..a06a2c40 100644
--- a/tests/test_podman_compose_up_down.py
+++ b/tests/test_podman_compose_up_down.py
@@ -27,7 +27,8 @@ def teardown(podman_compose_path, profile_compose_file):
     yield
 
     down_cmd = [
-        "python3",
+        "coverage",
+        "run",
         podman_compose_path,
         "--profile",
         "profile-1",
@@ -59,7 +60,8 @@ def teardown(podman_compose_path, profile_compose_file):
 )
 def test_up(podman_compose_path, profile_compose_file, profiles, expected_services):
     up_cmd = [
-        "python3",
+        "coverage",
+        "run",
         podman_compose_path,
         "-f",
         profile_compose_file,

From d4b646bb289c84ff5c4380a27df753f4ba3e72b4 Mon Sep 17 00:00:00 2001
From: Falmarri <463948+Falmarri@users.noreply.github.com>
Date: Sun, 10 Dec 2023 23:46:20 -0800
Subject: [PATCH 3/4] Add tests to make sure all async paths are covered

Not at 100% yet. But upped code coverage significantly and covered major
async calls.

Signed-off-by: Falmarri <463948+Falmarri@users.noreply.github.com>
---
 podman_compose.py                  | 274 +++++++++++++++--------------
 tests/deps/README.md               |   2 +-
 tests/deps/docker-compose.yaml     |   3 +-
 tests/test_podman_compose.py       |  19 +-
 tests/test_podman_compose_tests.py | 180 +++++++++++++++++++
 5 files changed, 332 insertions(+), 146 deletions(-)
 create mode 100644 tests/test_podman_compose_tests.py

diff --git a/podman_compose.py b/podman_compose.py
index 973ddb2f..3295bc32 100755
--- a/podman_compose.py
+++ b/podman_compose.py
@@ -21,6 +21,7 @@
 import signal
 
 import shlex
+from asyncio import Task
 
 try:
     from shlex import quote as cmd_quote
@@ -82,7 +83,13 @@ def try_float(i, fallback=None):
 
 
 def log(*msgs, sep=" ", end="\n"):
+    try:
+        current_task = asyncio.current_task()
+    except RuntimeError:
+        current_task = None
     line = (sep.join([str(msg) for msg in msgs])) + end
+    if current_task and not current_task.get_name().startswith("Task"):
+        line = f"[{current_task.get_name()}] " + line
     sys.stderr.write(line)
     sys.stderr.flush()
 
@@ -1151,27 +1158,29 @@ def flat_deps(services, with_extends=False):
 
 
 class Podman:
-    def __init__(self, compose, podman_path="podman", dry_run=False):
+    def __init__(self, compose, podman_path="podman", dry_run=False, semaphore: asyncio.Semaphore = asyncio.Semaphore(sys.maxsize)):
         self.compose = compose
         self.podman_path = podman_path
         self.dry_run = dry_run
+        self.semaphore = semaphore
 
     async def output(self, podman_args, cmd="", cmd_args=None):
-        cmd_args = cmd_args or []
-        xargs = self.compose.get_podman_args(cmd) if cmd else []
-        cmd_ls = [self.podman_path, *podman_args, cmd] + xargs + cmd_args
-        log(cmd_ls)
-        p = await asyncio.subprocess.create_subprocess_exec(
-            *cmd_ls,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE
-        )
+        async with self.semaphore:
+            cmd_args = cmd_args or []
+            xargs = self.compose.get_podman_args(cmd) if cmd else []
+            cmd_ls = [self.podman_path, *podman_args, cmd] + xargs + cmd_args
+            log(cmd_ls)
+            p = await asyncio.subprocess.create_subprocess_exec(
+                *cmd_ls,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE
+            )
 
-        stdout_data, stderr_data = await p.communicate()
-        if p.returncode == 0:
-            return stdout_data
-        else:
-            raise subprocess.CalledProcessError(p.returncode, " ".join(cmd_ls), stderr_data)
+            stdout_data, stderr_data = await p.communicate()
+            if p.returncode == 0:
+                return stdout_data
+            else:
+                raise subprocess.CalledProcessError(p.returncode, " ".join(cmd_ls), stderr_data)
 
     def exec(
         self,
@@ -1190,62 +1199,62 @@ async def run(
         podman_args,
         cmd="",
         cmd_args=None,
-        wait=True,
-        sleep=1,
         log_formatter=None,
         *,
         # Intentionally mutable default argument to hold references to tasks
         task_reference=set()
-    ):
-        cmd_args = list(map(str, cmd_args or []))
-        xargs = self.compose.get_podman_args(cmd) if cmd else []
-        cmd_ls = [self.podman_path, *podman_args, cmd] + xargs + cmd_args
-        log(" ".join([str(i) for i in cmd_ls]))
-        if self.dry_run:
-            return None
-        if log_formatter is not None:
-
-            async def format_out(stdout):
-                while True:
-                    l = await stdout.readline()
-                    if l:
-                        print(log_formatter, l.decode('utf-8'), end='')
-                    if stdout.at_eof():
-                        break
-
-            # read, write = os.pipe()
-            # Pipe podman process output through log_formatter (which can add colored prefix)
-            p = await asyncio.subprocess.create_subprocess_exec(
-                *cmd_ls, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )  # pylint: disable=consider-using-with
-
-            # This is hacky to make the tasks not get garbage collected
-            # https://github.com/python/cpython/issues/91887
-            out_t = asyncio.create_task(format_out(p.stdout))
-            task_reference.add(out_t)
-            out_t.add_done_callback(task_reference.discard)
-
-            err_t = asyncio.create_task(format_out(p.stderr))
-            task_reference.add(err_t)
-            err_t.add_done_callback(task_reference.discard)
-        else:
-            p = await asyncio.subprocess.create_subprocess_exec(*cmd_ls)  # pylint: disable=consider-using-with
+    ) -> int:
+        async with self.semaphore:
+            cmd_args = list(map(str, cmd_args or []))
+            xargs = self.compose.get_podman_args(cmd) if cmd else []
+            cmd_ls = [self.podman_path, *podman_args, cmd] + xargs + cmd_args
+            log(" ".join([str(i) for i in cmd_ls]))
+            if self.dry_run:
+                return None
+
+            if log_formatter is not None:
+
+                async def format_out(stdout):
+                    while True:
+                        l = await stdout.readline()
+                        if l:
+                            print(log_formatter, l.decode('utf-8'), end='')
+                        if stdout.at_eof():
+                            break
+
+                p = await asyncio.subprocess.create_subprocess_exec(
+                    *cmd_ls, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+                )  # pylint: disable=consider-using-with
+
+                # This is hacky to make the tasks not get garbage collected
+                # https://github.com/python/cpython/issues/91887
+                out_t = asyncio.create_task(format_out(p.stdout))
+                task_reference.add(out_t)
+                out_t.add_done_callback(task_reference.discard)
+
+                err_t = asyncio.create_task(format_out(p.stderr))
+                task_reference.add(err_t)
+                err_t.add_done_callback(task_reference.discard)
+
+            else:
+                p = await asyncio.subprocess.create_subprocess_exec(*cmd_ls)  # pylint: disable=consider-using-with
 
-        if wait:
             try:
                 exit_code = await p.wait()
-            except asyncio.CancelledError:
+            except asyncio.CancelledError as e:
+                log(f"Sending termination signal")
                 p.terminate()
-                exit_code = await p.wait()
-
-            log("exit code:", exit_code)
+                try:
+                    async with asyncio.timeout(10):
+                        exit_code = await p.wait()
+                except TimeoutError:
+                    log(f"container did not shut down after 10 seconds, killing")
+                    p.kill()
+                    exit_code = await p.wait()
+
+            log(f"exit code: {exit_code}")
             return exit_code
 
-        if sleep:
-            log(f"Sleep {sleep}")
-            await asyncio.sleep(sleep)
-        return p
-
     async def volume_ls(self, proj=None):
         if not proj:
             proj = self.compose.project_name
@@ -1520,7 +1529,8 @@ async def run(self):
                 if args.dry_run is False:
                     log(f"Binary {podman_path} has not been found.")
                     sys.exit(1)
-        self.podman = Podman(self, podman_path, args.dry_run)
+        self.podman = Podman(self, podman_path, args.dry_run, asyncio.Semaphore(args.parallel))
+
         if not args.dry_run:
             # just to make sure podman is running
             try:
@@ -1921,6 +1931,11 @@ def _init_global_parser(parser):
             help="No action; perform a simulation of commands",
             action="store_true",
         )
+        parser.add_argument(
+            "--parallel",
+            type=int,
+            default=os.environ.get("COMPOSE_PARALLEL_LIMIT", sys.maxsize)
+        )
 
 
 podman_compose = PodmanCompose()
@@ -1979,7 +1994,7 @@ async def compose_version(compose, args):
         print(json.dumps(res))
         return
     print("podman-compose version", __version__)
-    await compose.podman.run(["--version"], "", [], sleep=0)
+    await compose.podman.run(["--version"], "", [])
 
 
 def is_local(container: dict) -> bool:
@@ -2023,8 +2038,8 @@ async def compose_systemd(compose, args):
                     f.write(f"{k}={v}\n")
         print(f"writing [{fn}]: done.")
         print("\n\ncreating the pod without starting it: ...\n\n")
-        process = subprocess.run([script, "up", "--no-start"], check=False)
-        print("\nfinal exit code is ", process.returncode)
+        process = await asyncio.subprocess.create_subprocess_exec(script, ["up", "--no-start"])
+        print("\nfinal exit code is ", process)
         username = getpass.getuser()
         print(
             f"""
@@ -2096,13 +2111,7 @@ async def compose_pull(compose, args):
         local_images = {cnt["image"] for cnt in img_containers if is_local(cnt)}
         images -= local_images
 
-    sem = asyncio.Semaphore(args.parallel)
-
-    async def _pull(image: str):
-        async with sem:
-            return await compose.podman.run([], "pull", [image], sleep=0)
-
-    await asyncio.gather(*[_pull(image) for image in images])
+    await asyncio.gather(*[compose.podman.run([], "pull", [image]) for image in images])
 
 
 @cmd_run(podman_compose, "push", "push stack images")
@@ -2113,7 +2122,7 @@ async def compose_push(compose, args):
             continue
         if services and cnt["_service"] not in services:
             continue
-        await compose.podman.run([], "push", [cnt["image"]], sleep=0)
+        await compose.podman.run([], "push", [cnt["image"]])
 
 
 async def build_one(compose, args, cnt):
@@ -2177,29 +2186,24 @@ async def build_one(compose, args, cnt):
             )
         )
     build_args.append(ctx)
-    status = await compose.podman.run([], "build", build_args, sleep=0)
+    status = await compose.podman.run([], "build", build_args)
     return status
 
 
 @cmd_run(podman_compose, "build", "build stack images")
 async def compose_build(compose, args):
     tasks = []
-    sem = asyncio.Semaphore(args.parallel)
-
-    async def safe_build_task(cnt):
-        async with sem:
-            return await build_one(compose, args, cnt)
 
     if args.services:
         container_names_by_service = compose.container_names_by_service
         compose.assert_services(args.services)
         for service in args.services:
             cnt = compose.container_by_name[container_names_by_service[service][0]]
-            tasks.append(asyncio.create_task(safe_build_task(cnt)))
+            tasks.append(asyncio.create_task(build_one(compose, args, cnt)))
 
     else:
         for cnt in compose.containers:
-            tasks.append(asyncio.create_task(safe_build_task(cnt)))
+            tasks.append(asyncio.create_task(build_one(compose, args, cnt)))
 
     status = 0
     for t in asyncio.as_completed(tasks):
@@ -2283,7 +2287,7 @@ async def compose_up(compose: PodmanCompose, args):
             continue
         podman_args = await container_to_args(compose, cnt, detached=args.detach)
         subproc = await compose.podman.run([], podman_command, podman_args)
-        if podman_command == "run" and subproc:
+        if podman_command == "run" and subproc is not None:
             await compose.podman.run([], "start", [cnt["name"]])
     if args.no_start or args.detach or args.dry_run:
         return
@@ -2320,20 +2324,28 @@ async def compose_up(compose: PodmanCompose, args):
 
         tasks.add(
             asyncio.create_task(
-                compose.podman.run([], "start", ["-a", cnt["name"]], sleep=None, log_formatter=log_formatter),
+                compose.podman.run([], "start", ["-a", cnt["name"]], log_formatter=log_formatter),
                 name=cnt["_service"]
             )
         )
 
     exit_code = 0
-    for task in asyncio.as_completed(tasks):
-        t = await task
+    exiting = False
+    while tasks:
+        done, tasks = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
         if args.abort_on_container_exit:
-            [_.cancel() for _ in tasks if not _.cancelling() and not _.cancelled()]
-            if task.get_name() == exit_code_from:
-                exit_code = t
+            if not exiting:
+                # If 2 containers exit at the exact same time, the cancellation of the other ones cause the status
+                # to overwrite. Sleeping for 1 seems to fix this and make it match docker-compose
+                await asyncio.sleep(1)
+                [_.cancel() for _ in tasks if not _.cancelling() and not _.cancelled()]
+            t: Task
+            exiting = True
+            for t in done:
+                if t.get_name() == exit_code_from:
+                    exit_code = t.result()
 
-    sys.exit(exit_code)
+    return exit_code
 
 
 def get_volume_names(compose, cnt):
@@ -2360,6 +2372,8 @@ async def compose_down(compose, args):
     timeout_global = getattr(args, "timeout", None)
     containers = list(reversed(compose.containers))
 
+    down_tasks = []
+
     for cnt in containers:
         if cnt["_service"] in excluded:
             continue
@@ -2370,11 +2384,12 @@ async def compose_down(compose, args):
             timeout = str_to_seconds(timeout_str)
         if timeout is not None:
             podman_stop_args.extend(["-t", str(timeout)])
-        await compose.podman.run([], "stop", [*podman_stop_args, cnt["name"]], sleep=0)
+        down_tasks.append(asyncio.create_task(compose.podman.run([], "stop", [*podman_stop_args, cnt["name"]]), name=cnt["name"]))
+    await asyncio.gather(*down_tasks)
     for cnt in containers:
         if cnt["_service"] in excluded:
             continue
-        await compose.podman.run([], "rm", [cnt["name"]], sleep=0)
+        await compose.podman.run([], "rm", [cnt["name"]])
     if args.remove_orphans:
         names = (
             (await compose.podman.output(
@@ -2392,9 +2407,9 @@ async def compose_down(compose, args):
             .splitlines()
         )
         for name in names:
-            await compose.podman.run([], "stop", [*podman_args, name], sleep=0)
+            await compose.podman.run([], "stop", [*podman_args, name])
         for name in names:
-            await compose.podman.run([], "rm", [name], sleep=0)
+            await compose.podman.run([], "rm", [name])
     if args.volumes:
         vol_names_to_keep = set()
         for cnt in containers:
@@ -2402,7 +2417,7 @@ async def compose_down(compose, args):
                 continue
             vol_names_to_keep.update(get_volume_names(compose, cnt))
         log("keep", vol_names_to_keep)
-        async for volume_name in compose.podman.volume_ls():
+        for volume_name in await compose.podman.volume_ls():
             if volume_name in vol_names_to_keep:
                 continue
             await compose.podman.run([], "volume", ["rm", volume_name])
@@ -2410,28 +2425,23 @@ async def compose_down(compose, args):
     if excluded:
         return
     for pod in compose.pods:
-        await compose.podman.run([], "pod", ["rm", pod["name"]], sleep=0)
+        await compose.podman.run([], "pod", ["rm", pod["name"]])
 
 
 @cmd_run(podman_compose, "ps", "show status of containers")
 async def compose_ps(compose, args):
     proj_name = compose.project_name
+    ps_args = ["-a", "--filter", f"label=io.podman.compose.project={proj_name}"]
     if args.quiet is True:
-        await compose.podman.run(
-            [],
-            "ps",
-            [
-                "-a",
-                "--format",
-                "{{.ID}}",
-                "--filter",
-                f"label=io.podman.compose.project={proj_name}",
-            ],
-        )
-    else:
-        await compose.podman.run(
-            [], "ps", ["-a", "--filter", f"label=io.podman.compose.project={proj_name}"]
-        )
+        ps_args.extend(["--format", "{{.ID}}"])
+    elif args.format:
+        ps_args.extend(["--format", args.format])
+
+    await compose.podman.run(
+        [],
+        "ps",
+        ps_args,
+    )
 
 
 @cmd_run(
@@ -2459,17 +2469,19 @@ async def compose_run(compose, args):
                 no_start=False,
                 no_cache=False,
                 build_arg=[],
+                parallel=1,
+                remove_orphans=True
             )
         )
-        compose.commands["up"](compose, up_args)
+        await compose.commands["up"](compose, up_args)
 
     build_args = argparse.Namespace(
         services=[args.service],
         if_not_exists=(not args.build),
         build_arg=[],
-        **args.__dict__,
+        **args.__dict__
     )
-    compose.commands["build"](compose, build_args)
+    await compose.commands["build"](compose, build_args)
 
     # adjust one-off container options
     name0 = "{}_{}_tmp{}".format(
@@ -2510,7 +2522,7 @@ async def compose_run(compose, args):
         podman_args.insert(1, "-i")
         if args.rm:
             podman_args.insert(1, "--rm")
-    p = await compose.podman.run([], "run", podman_args, sleep=0)
+    p = await compose.podman.run([], "run", podman_args)
     sys.exit(p)
 
 
@@ -2540,7 +2552,7 @@ async def compose_exec(compose, args):
     podman_args += [container_name]
     if args.cnt_command is not None and len(args.cnt_command) > 0:
         podman_args += args.cnt_command
-    p = await compose.podman.run([], "exec", podman_args, sleep=0)
+    p = await compose.podman.run([], "exec", podman_args)
     sys.exit(p)
 
 
@@ -2571,23 +2583,23 @@ async def transfer_service_status(compose, args, action):
                 timeout = str_to_seconds(timeout_str)
             if timeout is not None:
                 podman_args.extend(["-t", str(timeout)])
-        tasks.append(asyncio.create_task(compose.podman.run([], action, podman_args + [target], sleep=0)))
+        tasks.append(asyncio.create_task(compose.podman.run([], action, podman_args + [target])))
     await asyncio.gather(*tasks)
 
 
 @cmd_run(podman_compose, "start", "start specific services")
 async def compose_start(compose, args):
-    transfer_service_status(compose, args, "start")
+    await transfer_service_status(compose, args, "start")
 
 
 @cmd_run(podman_compose, "stop", "stop specific services")
 async def compose_stop(compose, args):
-    transfer_service_status(compose, args, "stop")
+    await transfer_service_status(compose, args, "stop")
 
 
 @cmd_run(podman_compose, "restart", "restart specific services")
 async def compose_restart(compose, args):
-    transfer_service_status(compose, args, "restart")
+    await transfer_service_status(compose, args, "restart")
 
 
 @cmd_run(podman_compose, "logs", "show logs from services")
@@ -3098,15 +3110,6 @@ def compose_ps_parse(parser):
     )
 
 
-@cmd_parse(podman_compose, ["build", "pull", "up"])
-def compose_build_pull_parse(parser):
-    parser.add_argument(
-        "--parallel",
-        type=int,
-        default=os.environ.get("PODMAN_PARALLEL", sys.maxsize)
-    )
-
-
 @cmd_parse(podman_compose, ["build", "up"])
 def compose_build_up_parse(parser):
     parser.add_argument(
@@ -3215,12 +3218,6 @@ def compose_stats_parse(parser):
         type=int,
         help="Time in seconds between stats reports (default 5)",
     )
-    parser.add_argument(
-        "-f",
-        "--format",
-        type=str,
-        help="Pretty-print container statistics to JSON or using a Go template",
-    )
     parser.add_argument(
         "--no-reset",
         help="Disable resetting the screen between intervals",
@@ -3233,6 +3230,15 @@ def compose_stats_parse(parser):
     )
 
 
+@cmd_parse(podman_compose, ["ps", "stats"])
+def compose_format_parse(parser):
+    parser.add_argument(
+        "-f",
+        "--format",
+        type=str,
+        help="Pretty-print container statistics to JSON or using a Go template",
+    )
+
 async def main():
     await podman_compose.run()
 
diff --git a/tests/deps/README.md b/tests/deps/README.md
index bde213aa..f689ed50 100644
--- a/tests/deps/README.md
+++ b/tests/deps/README.md
@@ -1,4 +1,4 @@
 
 ```
-podman-compose run --rm sleep /bin/sh -c 'wget -O - http://localhost:8000/hosts'
+podman-compose run --rm sleep /bin/sh -c 'wget -O - http://web:8000/hosts'
 ```
diff --git a/tests/deps/docker-compose.yaml b/tests/deps/docker-compose.yaml
index 0f06bbd4..77993235 100644
--- a/tests/deps/docker-compose.yaml
+++ b/tests/deps/docker-compose.yaml
@@ -9,7 +9,8 @@ services:
     sleep:
       image: busybox
       command: ["/bin/busybox", "sh", "-c", "sleep 3600"]
-      depends_on: "web"
+      depends_on:
+        - "web"
       tmpfs:
         - /run
         - /tmp
diff --git a/tests/test_podman_compose.py b/tests/test_podman_compose.py
index 5be6629b..453e79ad 100644
--- a/tests/test_podman_compose.py
+++ b/tests/test_podman_compose.py
@@ -31,14 +31,14 @@ def test_podman_compose_extends_w_file_subdir():
     ]
 
     command_check_container = [
-        "podman",
-        "container",
+        "coverage",
+        "run",
+        str(main_path.joinpath("podman_compose.py")),
+        "-f",
+        str(main_path.joinpath("tests", "extends_w_file_subdir", "docker-compose.yml")),
         "ps",
-        "--sort",
-        "status",
-        "--all",
         "--format",
-        '"{{.Image}}"',
+        '{{.Image}}',
     ]
 
     command_down = [
@@ -52,18 +52,17 @@ def test_podman_compose_extends_w_file_subdir():
     out, _, returncode = capture(command_up)
     assert 0 == returncode
     # check container was created and exists
-    out, _, returncode = capture(command_check_container)
+    out, err, returncode = capture(command_check_container)
     assert 0 == returncode
-    assert b'"localhost/subdir_test:me"\n' in out
+    assert b'localhost/subdir_test:me\n' == out
     out, _, returncode = capture(command_down)
     # cleanup test image(tags)
     assert 0 == returncode
     print('ok')
     # check container did not exists anymore
     out, _, returncode = capture(command_check_container)
-    print(out)
     assert 0 == returncode
-    assert b'"localhost/subdir_test:me"\n' not in out
+    assert b'' == out
 
 
 def test_podman_compose_extends_w_empty_service():
diff --git a/tests/test_podman_compose_tests.py b/tests/test_podman_compose_tests.py
new file mode 100644
index 00000000..378e5bda
--- /dev/null
+++ b/tests/test_podman_compose_tests.py
@@ -0,0 +1,180 @@
+"""
+test_podman_compose_up_down.py
+
+Tests the podman compose up and down commands used to create and remove services.
+"""
+# pylint: disable=redefined-outer-name
+import os
+import time
+
+from test_podman_compose import capture
+
+
+def test_exit_from(podman_compose_path, test_path):
+    up_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "exit-from", "docker-compose.yaml"),
+        "up"
+    ]
+
+    out, _, return_code = capture(up_cmd + ["--exit-code-from", "sh1"])
+    assert return_code == 1
+
+    out, _, return_code = capture(up_cmd + ["--exit-code-from", "sh2"])
+    assert return_code == 2
+
+
+def test_run(podman_compose_path, test_path):
+    """
+    This will test depends_on as well
+    """
+    run_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "deps", "docker-compose.yaml"),
+        "run",
+        "--rm",
+        "sleep",
+        "/bin/sh",
+        "-c",
+        "wget -q -O - http://web:8000/hosts"
+    ]
+
+    out, _, return_code = capture(run_cmd)
+    assert b'127.0.0.1\tlocalhost' in out
+
+    # Run it again to make sure we can run it twice. I saw an issue where a second run, with the container left up,
+    # would fail
+    run_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "deps", "docker-compose.yaml"),
+        "run",
+        "--rm",
+        "sleep",
+        "/bin/sh",
+        "-c",
+        "wget -q -O - http://web:8000/hosts"
+    ]
+
+    out, _, return_code = capture(run_cmd)
+    assert b'127.0.0.1\tlocalhost' in out
+    assert return_code == 0
+
+    # This leaves a container running. Not sure it's intended, but it matches docker-compose
+    down_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "deps", "docker-compose.yaml"),
+        "down",
+    ]
+
+    out, _, return_code = capture(run_cmd)
+    assert return_code == 0
+
+
+def test_up_with_ports(podman_compose_path, test_path):
+
+
+    up_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "ports", "docker-compose.yml"),
+        "up",
+        "-d",
+        "--force-recreate"
+    ]
+
+    down_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "ports", "docker-compose.yml"),
+        "down",
+        "--volumes"
+    ]
+
+    try:
+        out, _, return_code = capture(up_cmd)
+        assert return_code == 0
+
+
+    finally:
+        out, _, return_code = capture(down_cmd)
+        assert return_code == 0
+
+
+def test_down_with_vols(podman_compose_path, test_path):
+
+    up_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "vol", "docker-compose.yaml"),
+        "up",
+        "-d"
+    ]
+
+    down_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "vol", "docker-compose.yaml"),
+        "down",
+        "--volumes"
+    ]
+
+    try:
+        out, _, return_code = capture(["podman", "volume", "create", "my-app-data"])
+        assert return_code == 0
+        out, _, return_code = capture(["podman", "volume", "create", "actual-name-of-volume"])
+        assert return_code == 0
+
+        out, _, return_code = capture(up_cmd)
+        assert return_code == 0
+
+        capture(["podman", "inspect", "volume", ""])
+
+    finally:
+        out, _, return_code = capture(down_cmd)
+        capture(["podman", "volume", "rm", "my-app-data"])
+        capture(["podman", "volume", "rm", "actual-name-of-volume"])
+        assert return_code == 0
+
+
+def test_down_with_orphans(podman_compose_path, test_path):
+
+    container_id, _ , return_code = capture(["podman", "run", "--rm", "-d", "busybox", "/bin/busybox", "httpd", "-f", "-h", "/etc/", "-p", "8000"])
+
+    down_cmd = [
+        "coverage",
+        "run",
+        podman_compose_path,
+        "-f",
+        os.path.join(test_path, "ports", "docker-compose.yml"),
+        "down",
+        "--volumes",
+        "--remove-orphans"
+    ]
+
+    out, _, return_code = capture(down_cmd)
+    assert return_code == 0
+
+    _, _, exists = capture(["podman", "container", "exists", container_id.decode("utf-8")])
+
+    assert exists == 1
+

From 8d8fa54855ce7eb73e802ef06e9f48645a30e2ac Mon Sep 17 00:00:00 2001
From: Falmarri <463948+Falmarri@users.noreply.github.com>
Date: Wed, 27 Dec 2023 11:54:21 -0800
Subject: [PATCH 4/4] Fixup tests

Signed-off-by: Falmarri <463948+Falmarri@users.noreply.github.com>
---
 .github/workflows/pytest.yml | 2 +-
 test-requirements.txt        | 1 +
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml
index 3cc3fad9..1adcb953 100644
--- a/.github/workflows/pytest.yml
+++ b/.github/workflows/pytest.yml
@@ -21,7 +21,7 @@ jobs:
         python-version: "3.10"
     - name: Install dependencies
       run: |
-        apt update && apt install podman
+        sudo apt update && apt install podman
         python -m pip install --upgrade pip
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
         if [ -f test-requirements.txt ]; then pip install -r test-requirements.txt; fi
diff --git a/test-requirements.txt b/test-requirements.txt
index 55c142a6..9c99e91f 100644
--- a/test-requirements.txt
+++ b/test-requirements.txt
@@ -6,3 +6,4 @@ coverage
 pytest
 tox
 black
+flake8
\ No newline at end of file