Linting/Refactor for Tests

podman_compose.py

@@ -1155,15 +1155,16 @@ async def output(self, podman_args, cmd="", cmd_args=None):
  xargs = self.compose.get_podman_args(cmd) if cmd else []
  cmd_ls = [self.podman_path, *podman_args, cmd] + xargs + cmd_args
  log(cmd_ls)
+ # pylint false positive - https://github.com/pylint-dev/pylint/issues/1469
+ # pylint: disable=no-member
  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)
+ raise subprocess.CalledProcessError(p.returncode, " ".join(cmd_ls), stderr_data)
 
  def exec(
  self,
@@ -1177,6 +1178,7 @@ def exec(
  log(" ".join([str(i) for i in cmd_ls]))
  os.execlp(self.podman_path, *cmd_ls)
 
+ # pylint: disable=dangerous-default-value
  async def run(
  self,
  podman_args,
@@ -1205,6 +1207,8 @@ async def format_out(stdout):
  if stdout.at_eof():
  break
 
+ # pylint false positive - https://github.com/pylint-dev/pylint/issues/1469
+ # pylint: disable=no-member
  p = await asyncio.subprocess.create_subprocess_exec(
  *cmd_ls, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
  ) # pylint: disable=consider-using-with
@@ -1220,7 +1224,8 @@ async def format_out(stdout):
  err_t.add_done_callback(task_reference.discard)
 
  else:
- p = await asyncio.subprocess.create_subprocess_exec(*cmd_ls) # pylint: disable=consider-using-with
+ # pylint false positive - https://github.com/pylint-dev/pylint/issues/1469
+ p = await asyncio.subprocess.create_subprocess_exec(*cmd_ls) # pylint: disable=consider-using-with,no-member
 
  try:
  exit_code = await p.wait()
@@ -1230,11 +1235,11 @@ async def format_out(stdout):
  try:
  exit_code = await wait_with_timeout(p.wait(), 10)
  except TimeoutError:
- log("container did not shut down after 10 seconds, killing")
+ log("Container did not shut down after 10 seconds, killing")
  p.kill()
  exit_code = await p.wait()
 
- log(f"exit code: {exit_code}")
+ log(f"Exit code: {exit_code}")
  return exit_code
 
  async def volume_ls(self, proj=None):
@@ -1444,6 +1449,13 @@ def dotenv_to_dict(dotenv_path):
 ]
 
 
+class CalledNonCoroutineError(Exception):
+ def __init__(self):
+ self.message = "Attempted to call a non-asynchronous function"
+ def __str__(self):
+ return str(self.message)
+
+
 class PodmanCompose:
  def __init__(self):
  self.podman = None
@@ -1905,9 +1917,8 @@ def __init__(self, compose, cmd_name, cmd_desc=None):
  def __call__(self, func):
  def wrapped(*args, **kw):
  return func(*args, **kw)
-
  if not asyncio.iscoroutinefunction(func):
- raise Exception("Command must be async")
+ raise CalledNonCoroutineError
  wrapped._compose = self.compose
  # Trim extra indentation at start of multiline docstrings.
  wrapped.desc = self.cmd_desc or re.sub(r"^\s+", "", func.__doc__)
@@ -1989,6 +2000,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")
+ # pylint false positive - https://github.com/pylint-dev/pylint/issues/1469
+ # pylint: disable=no-member
  process = await asyncio.subprocess.create_subprocess_exec(script, ["up", "--no-start"])
  print("\nfinal exit code is ", process)
  username = getpass.getuser()
@@ -2284,7 +2297,7 @@ async def compose_up(compose: PodmanCompose, args):
  # 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()]
+ [_.cancel() for _ in tasks if not _.cancelling() and not _.cancelled()] # pylint: disable=expression-not-assigned
  t: Task
  exiting = True
  for t in done: