feat: import CalledProcessError for users of run_command (no merge, nl)

pyaptly/command.py

@@ -1,11 +1,10 @@
 """Commands with dependencies."""
 import collections
 import logging
-import subprocess
 
 from frozendict import frozendict
 
-from . import state_reader
+from . import state_reader, util
 
 lg = logging.getLogger(__name__)
 
@@ -94,9 +93,18 @@ def execute(self):
 
  if not Command.pretend_mode:
  lg.debug("Running command: %s", " ".join(self.cmd))
- self._finished = bool(subprocess.check_call(self.cmd))
+ # It seems the intention of the original code is to a redo a command if it
+ # scheduled multiple times and fails the first time. But there is also:
+
+ # `commands = set([c for c in commands if c is not None])`
+
+ # which prevents that. I guess the feature is currently not needed.
+ # So I decided to change that. For now we fail hard if a `Command` fails.
+ # I guess we will see in production what happens.
+ util.run_command(self.cmd, check=True)
  else:
  lg.info("Pretending to run command: %s", " ".join(self.cmd))
+ self._finished = True
 
  return self._finished
 

pyaptly/conftest.py

@@ -8,8 +8,6 @@
 
 import freezegun
 import pytest
-import tomli
-import yaml
 
 from pyaptly import main, state_reader, util
 
@@ -114,16 +112,7 @@ def test_mirror_create(environment, config, caplog):
  ...
  ```
  """
- config_file = test_base / request.param
- with config_file.open("rb") as f:
- config = tomli.load(f)
- # TODO: remove yaml conversion
- try:
- with tempfile.NamedTemporaryFile(mode="w", encoding="UTF-8", delete=False) as f:
- yaml.dump(config, f)
- yield f.name
- finally:
- Path(f.name).unlink()
+ yield str(test_base / request.param)
 
 
 @pytest.fixture()

pyaptly/main.py

@@ -1,11 +1,9 @@
 """Aptly mirror/snapshot managment automation."""
 import argparse
-import codecs
 import logging
-import subprocess
 import sys
 
-import yaml
+import tomli
 
 from . import command, mirror, publish, repo, snapshot, state_reader
 
@@ -15,29 +13,6 @@
 lg = logging.getLogger(__name__)
 
 
-# TODO remove
-def call_output(args, input_=None):
- """Call command and return output.
-
- :param args: Command to execute
- :type args: list
- :param input_: Input to command
- :type input_: bytes
- """
- p = subprocess.Popen(
- args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE
- )
- output, err = p.communicate(input_)
- if p.returncode != 0:
- raise subprocess.CalledProcessError(
- p.returncode,
- args,
- output,
- err,
- )
- return (output.decode("UTF-8"), err.decode("UTF-8"))
-
-
 def main(argv=None):
  """Define parsers and executes commands.
 
@@ -108,8 +83,8 @@ def main(argv=None):
  _logging_setup = True # noqa
  lg.debug("Args: %s", vars(args))
 
- with codecs.open(args.config, "r", encoding="UTF-8") as cfgfile:
- cfg = yaml.load(cfgfile, Loader=yaml.FullLoader)
+ with open(args.config, "rb") as f:
+ cfg = tomli.load(f)
  state_reader.state.read()
 
  # run function for selected subparser

pyaptly/mirror.py

@@ -1,6 +1,5 @@
 """Create and update mirrors in aptly."""
 import logging
-import subprocess
 
 from . import state_reader, util
 
@@ -47,16 +46,16 @@ def add_gpg_keys(mirror_config):
  key,
  ]
  lg.debug("Adding gpg key with call: %s", key_command)
- subprocess.check_call(key_command)
- except subprocess.CalledProcessError: # pragma: no cover
+ util.run_command(key_command, check=True)
+ except util.CalledProcessError: # pragma: no cover
  url = keys_urls[key]
  if url:
  key_shell = (
  "curl %s | "
  "gpg --no-default-keyring --keyring trustedkeys.gpg "
  "--import"
  ) % url
- subprocess.check_call(["bash", "-c", key_shell])
+ util.run_command(["bash", "-c", key_shell], check=True)
  else:
  raise
  state_reader.state.read_gpg()
@@ -129,7 +128,7 @@ def cmd_mirror_create(cfg, mirror_name, mirror_config):
  aptly_cmd.extend(util.unit_or_list_to_list(mirror_config["components"]))
 
  lg.debug("Running command: %s", " ".join(aptly_cmd))
- subprocess.check_call(aptly_cmd)
+ util.run_command(aptly_cmd, check=True)
 
 
 def cmd_mirror_update(cfg, mirror_name, mirror_config):
@@ -151,4 +150,4 @@ def cmd_mirror_update(cfg, mirror_name, mirror_config):
 
  aptly_cmd.append(mirror_name)
  lg.debug("Running command: %s", " ".join(aptly_cmd))
- subprocess.check_call(aptly_cmd)
+ util.run_command(aptly_cmd, check=True)

pyaptly/state_reader.py

@@ -2,7 +2,7 @@
 import logging
 import re
 
-from . import main
+from . import util
 
 lg = logging.getLogger(__name__)
 
@@ -70,9 +70,8 @@ def read_gpg(self):
  "--list-keys",
  "--with-colons",
  ]
- data, _ = main.call_output(cmd)
- lg.debug("GPG returned: %s", data)
- for line in data.split("\n"):
+ result = util.run_command(cmd, stdout=util.PIPE, check=True)
+ for line in result.stdout.split("\n"):
  field = line.split(":")
  if field[0] in ("pub", "sub"):
  key = field[4]
@@ -87,9 +86,9 @@ def read_publish_map(self):
  re_snap = re.compile(r"\s+[\w\d-]+\:\s([\w\d-]+)\s\[snapshot\]")
  for publish in self.publishes:
  prefix, dist = publish.split(" ")
- data, _ = main.call_output(["aptly", "publish", "show", dist, prefix])
-
- sources = self._extract_sources(data)
+ cmd = ["aptly", "publish", "show", dist, prefix]
+ result = util.run_command(cmd, stdout=util.PIPE, check=True)
+ sources = self._extract_sources(result.stdout)
  matches = [re_snap.match(source) for source in sources]
  snapshots = [match.group(1) for match in matches if match]
  self.publish_map[publish] = set(snapshots)
@@ -105,8 +104,10 @@ def read_snapshot_map(self):
  # match example: test-snapshot [snapshot]
  re_snap = re.compile(r"\s+([\w\d-]+)\s\[snapshot\]")
  for snapshot_outer in self.snapshots:
- data, _ = main.call_output(["aptly", "snapshot", "show", snapshot_outer])
- sources = self._extract_sources(data)
+ cmd = ["aptly", "snapshot", "show", snapshot_outer]
+
+ result = util.run_command(cmd, stdout=util.PIPE, check=True)
+ sources = self._extract_sources(result.stdout)
  matches = [re_snap.match(source) for source in sources]
  snapshots = [match.group(1) for match in matches if match]
  self.snapshot_map[snapshot_outer] = set(snapshots)
@@ -141,9 +142,9 @@ def read_aptly_list(self, type_, list_):
  :param list_: Read into this list
  :param list_: list
  """
- data, _ = main.call_output(["aptly", type_, "list", "-raw"])
- lg.debug("Aptly returned %s: %s", type_, data)
- for line in data.split("\n"):
+ cmd = ["aptly", type_, "list", "-raw"]
+ result = util.run_command(cmd, stdout=util.PIPE, check=True)
+ for line in result.stdout.split("\n"):
  clean_line = line.strip()
  if clean_line:
  list_.add(clean_line)

pyaptly/tests/test_dateround.py

@@ -4,7 +4,7 @@
 import os.path
 
 import pytest
-import yaml
+import tomli
 from hypothesis import given
 from hypothesis.strategies import datetimes, integers, times
 
@@ -173,8 +173,8 @@ def test_daily_examples():
 @pytest.mark.parametrize("config", ["publish-previous.toml"], indirect=True)
 def test_snapshot_spec_to_name(config, test_path, freeze):
  """Test the complete functionality (snapshot_spec_to_name)."""
- with (test_path / config).open("r") as f:
- tyml = yaml.load(f, Loader=yaml.FullLoader)
+ with open(config, "rb") as f:
+ tyml = tomli.load(f)
  snaps = tyml["snapshot"]["superfake-%T"]["merge"]
 
  rounded1 = snapshot.snapshot_spec_to_name(tyml, snaps[0])

pyaptly/tests/test_helpers.py

@@ -1,24 +1,5 @@
 """Testing testing helper functions."""
-import subprocess
-
-from .. import command, main, state_reader
-
-
-def test_call_output_error():
- """Test if call_output raises errors correctly."""
- # TDOD remove
- args = [
- "bash",
- "-c",
- "exit 42",
- ]
- error = False
- try:
- main.call_output(args)
- except subprocess.CalledProcessError as e:
- assert e.returncode == 42
- error = True
- assert error
+from .. import command, state_reader
 
 
 def test_command_dependency_fail():

pyaptly/util.py

@@ -3,8 +3,8 @@
 import logging
 import subprocess
 from pathlib import Path
-from subprocess import DEVNULL, PIPE # noqa: F401
-from typing import Optional, Union
+from subprocess import DEVNULL, PIPE, CalledProcessError # noqa: F401
+from typing import Optional, Sequence
 
 _DEFAULT_KEYSERVER: str = "hkps://keys.openpgp.org"
 _PYTEST_KEYSERVER: Optional[str] = None
@@ -51,7 +51,7 @@ def is_debug_mode():
  return _DEBUG or _PYTEST_DEBUG
 
 
-def run_command(cmd_args: list[Union[str, Path]], *, decode: bool = True, **kwargs):
+def run_command(cmd_args: Sequence[str | Path], *, decode: bool = True, **kwargs):
  """Instrumented subprocess.run for easier debugging.
 
  By default this run command will add `encoding="UTF-8"` to kwargs. Disable
@@ -83,7 +83,7 @@ def run_command(cmd_args: list[Union[str, Path]], *, decode: bool = True, **kwar
  return result
 
 
-def indent_out(output: Union[bytes, str]) -> str:
+def indent_out(output: bytes | str) -> str:
  """Indent command output for nicer logging-messages.
 
  It will convert bytes to strings if need or display the result as bytes if