diff --git a/pyaptly/command.py b/pyaptly/command.py index fcbe4ee..12bbfc2 100644 --- a/pyaptly/command.py +++ b/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 diff --git a/pyaptly/conftest.py b/pyaptly/conftest.py index 6d8157a..6eda42a 100644 --- a/pyaptly/conftest.py +++ b/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() diff --git a/pyaptly/main.py b/pyaptly/main.py index 1e39b80..f1319cf 100755 --- a/pyaptly/main.py +++ b/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 diff --git a/pyaptly/mirror.py b/pyaptly/mirror.py index 1a3992a..1466e39 100644 --- a/pyaptly/mirror.py +++ b/pyaptly/mirror.py @@ -1,6 +1,5 @@ """Create and update mirrors in aptly.""" import logging -import subprocess from . import state_reader, util @@ -47,8 +46,8 @@ 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 = ( @@ -56,7 +55,7 @@ def add_gpg_keys(mirror_config): "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) diff --git a/pyaptly/state_reader.py b/pyaptly/state_reader.py index 0a6273f..078f475 100644 --- a/pyaptly/state_reader.py +++ b/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) diff --git a/pyaptly/tests/test_dateround.py b/pyaptly/tests/test_dateround.py index 8a61935..8b60eaa 100644 --- a/pyaptly/tests/test_dateround.py +++ b/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]) diff --git a/pyaptly/tests/test_helpers.py b/pyaptly/tests/test_helpers.py index 81eb9b3..4ae0e0f 100644 --- a/pyaptly/tests/test_helpers.py +++ b/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(): diff --git a/pyaptly/util.py b/pyaptly/util.py index af6f086..64a041a 100644 --- a/pyaptly/util.py +++ b/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