chore: switch to toml (no merge)

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/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():