almalinux/pungi
15
2
Fork 0
Code Issues Pull Requests Releases Activity

Protect against decoding errors with subprocess text mode

Browse Source 
All these are calling subprocess in 'text mode', where it will
try to decode stdout/stderr using the default encoding (utf-8
for us). If it doesn't decode, subprocess will raise an exception
and kobo doesn't handle it, it just passes it along to us, so
things blow up - see https://pagure.io/releng/issue/12474 . To
avoid this, let's set `errors="replace"`, which tells the decoder
to replace invalid data with ? characters. This way we should get
as much of the output as can be read, and no crashes.

We also replace `universal_newlines=True` with `text=True` as
the latter is shorter, clearer, and what Python 3 subprocess
wants us to use, it considers `universal_newlines` to just be
a backwards-compatibility thing - "The universal_newlines argument
is equivalent to text and is provided for backwards compatibility"

Signed-off-by: Adam Williamson <awilliam@redhat.com>
Merges: https://pagure.io/pungi/pull-request/1812
(cherry picked from commit 2d16a3af004f61cf41e4eb2e5e694bb46a5d3cda)
This commit is contained in:
Adam Williamson 2025-01-15 11:00:33 -08:00 committed by Stepan Oksanichenko
parent e59566feb2
commit f3dcb036a5
17 changed files with 132 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
pungi/__init__.py
View File

@ -16,7 +16,8 @@ def get_full_version():
proc = subprocess.Popen( proc = subprocess.Popen(
["git", "--git-dir=%s/.git" % location, "describe", "--tags"], ["git", "--git-dir=%s/.git" % location, "describe", "--tags"],
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
) )
output, _ = proc.communicate() output, _ = proc.communicate()
return re.sub(r"-1.fc\d\d?", "", output.strip().replace("pungi-", "")) return re.sub(r"-1.fc\d\d?", "", output.strip().replace("pungi-", ""))
@ -24,7 +25,7 @@ def get_full_version():
import subprocess import subprocess
proc = subprocess.Popen( proc = subprocess.Popen(
["rpm", "-q", "pungi"], stdout=subprocess.PIPE, universal_newlines=True ["rpm", "-q", "pungi"], stdout=subprocess.PIPE, text=True, errors="replace"
) )
(output, err) = proc.communicate() (output, err) = proc.communicate()
if not err: if not err:

3
pungi/notifier.py
View File

@ -104,7 +104,8 @@ class PungiNotifier(object):
workdir=workdir, workdir=workdir,
return_stdout=False, return_stdout=False,
show_cmd=True, show_cmd=True,
universal_newlines=True, text=True,
errors="replace",
logfile=logfile, logfile=logfile,
) )
if ret != 0: if ret != 0:

6
pungi/ostree/tree.py
View File

@ -64,7 +64,8 @@ class Tree(OSTree):
show_cmd=True, show_cmd=True,
stdout=True, stdout=True,
logfile=log_file, logfile=log_file,
universal_newlines=True, text=True,
errors="replace",
) )
finally: finally:
os.umask(oldumask) os.umask(oldumask)
@ -77,7 +78,8 @@ class Tree(OSTree):
show_cmd=True, show_cmd=True,
stdout=True, stdout=True,
logfile=log_file, logfile=log_file,
universal_newlines=True, text=True,
errors="replace",
) )
def _update_ref(self): def _update_ref(self):

2
pungi/phases/ostree_container.py
View File

@ -139,7 +139,7 @@ class OSTreeContainerThread(WorkerThread):
"--version=%s" % version, "--version=%s" % version,
] ]
_, runroot_script = shortcuts.run(cmd, universal_newlines=True) _, runroot_script = shortcuts.run(cmd, text=True, errors="replace")
default_packages = ["ostree", "rpm-ostree", "selinux-policy-targeted"] default_packages = ["ostree", "rpm-ostree", "selinux-policy-targeted"]
additional_packages = config.get("runroot_packages", []) additional_packages = config.get("runroot_packages", [])

11
pungi/util.py
View File

@ -649,7 +649,11 @@ def run_unmount_cmd(cmd, max_retries=10, path=None, logger=None):
""" """
for i in range(max_retries): for i in range(max_retries):
proc = subprocess.Popen( proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
errors="replace",
) )
out, err = proc.communicate() out, err = proc.communicate()
if proc.returncode == 0: if proc.returncode == 0:
@ -671,7 +675,8 @@ def run_unmount_cmd(cmd, max_retries=10, path=None, logger=None):
c, c,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, stderr=subprocess.STDOUT,
universal_newlines=True, text=True,
errors="replace",
) )
out, _ = proc.communicate() out, _ = proc.communicate()
logger.debug( logger.debug(
@ -876,7 +881,7 @@ def git_ls_remote(baseurl, ref, credential_helper=None):
if credential_helper: if credential_helper:
cmd.extend(["-c", "credential.useHttpPath=true"]) cmd.extend(["-c", "credential.useHttpPath=true"])
cmd.extend(["-c", "credential.helper=%s" % credential_helper]) cmd.extend(["-c", "credential.helper=%s" % credential_helper])
return run(cmd + ["ls-remote", baseurl, ref], universal_newlines=True) return run(cmd + ["ls-remote", baseurl, ref], text=True, errors="replace")
def get_tz_offset(): def get_tz_offset():

8
pungi/wrappers/iso.py
View File

@ -227,7 +227,7 @@ def get_checkisomd5_cmd(iso_path, just_print=False):
def get_checkisomd5_data(iso_path, logger=None): def get_checkisomd5_data(iso_path, logger=None):
cmd = get_checkisomd5_cmd(iso_path, just_print=True) cmd = get_checkisomd5_cmd(iso_path, just_print=True)
retcode, output = run(cmd, universal_newlines=True) retcode, output = run(cmd, text=True, errors="replace")
items = [line.strip().rsplit(":", 1) for line in output.splitlines()] items = [line.strip().rsplit(":", 1) for line in output.splitlines()]
items = dict([(k, v.strip()) for k, v in items]) items = dict([(k, v.strip()) for k, v in items])
md5 = items.get(iso_path, "") md5 = items.get(iso_path, "")
@ -283,13 +283,13 @@ def get_manifest_cmd(iso_name, xorriso=False, output_file=None):
def get_volume_id(path, xorriso=False): def get_volume_id(path, xorriso=False):
if xorriso: if xorriso:
cmd = ["xorriso", "-indev", path] cmd = ["xorriso", "-indev", path]
retcode, output = run(cmd, universal_newlines=True) retcode, output = run(cmd, text=True, errors="replace")
for line in output.splitlines(): for line in output.splitlines():
if line.startswith("Volume id"): if line.startswith("Volume id"):
return line.split("'")[1] return line.split("'")[1]
else: else:
cmd = ["isoinfo", "-d", "-i", path] cmd = ["isoinfo", "-d", "-i", path]
retcode, output = run(cmd, universal_newlines=True) retcode, output = run(cmd, text=True, errors="replace")
for line in output.splitlines(): for line in output.splitlines():
line = line.strip() line = line.strip()
@ -500,7 +500,7 @@ def mount(image, logger=None, use_guestmount=True):
else: else:
env = {} env = {}
cmd = ["mount", "-o", "loop", image, mount_dir] cmd = ["mount", "-o", "loop", image, mount_dir]
ret, out = run(cmd, env=env, can_fail=True, universal_newlines=True) ret, out = run(cmd, env=env, can_fail=True, text=True, errors="replace")
if ret != 0: if ret != 0:
# The mount command failed, something is wrong. # The mount command failed, something is wrong.
# Log the output and raise an exception. # Log the output and raise an exception.

11
pungi/wrappers/kojiwrapper.py
View File

@ -295,7 +295,8 @@ class KojiWrapper(object):
show_cmd=True, show_cmd=True,
env=env, env=env,
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
) )
# Look for first line that contains only a number. This is the ID of # Look for first line that contains only a number. This is the ID of
@ -431,7 +432,7 @@ class KojiWrapper(object):
while True: while True:
retcode, output = run( retcode, output = run(
cmd, can_fail=True, logfile=logfile, universal_newlines=True cmd, can_fail=True, logfile=logfile, text=True, errors="replace"
) )
if retcode == 0 or not ( if retcode == 0 or not (
@ -464,7 +465,8 @@ class KojiWrapper(object):
logfile=log_file, logfile=log_file,
env=env, env=env,
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
) )
match = re.search(r"Created task: (\d+)", output) match = re.search(r"Created task: (\d+)", output)
@ -848,7 +850,8 @@ def get_buildroot_rpms(compose, task_id):
# local # local
retcode, output = run( retcode, output = run(
"rpm -qa --qf='%{name}-%{version}-%{release}.%{arch}\n'", "rpm -qa --qf='%{name}-%{version}-%{release}.%{arch}\n'",
universal_newlines=True, text=True,
errors="replace",
) )
for i in output.splitlines(): for i in output.splitlines():
if not i: if not i:

3
pungi/wrappers/scm.py
View File

@ -56,7 +56,8 @@ class ScmBase(kobo.log.LoggingBase):
workdir=cwd, workdir=cwd,
can_fail=True, can_fail=True,
stdin_data="", stdin_data="",
universal_newlines=True, text=True,
errors="replace",
) )
if retcode != 0: if retcode != 0:
self.log_error("Output was: %r" % output) self.log_error("Output was: %r" % output)

5
pungi_utils/patch_iso.py
View File

@ -25,7 +25,7 @@ from pungi.wrappers import iso
def sh(log, cmd, *args, **kwargs): def sh(log, cmd, *args, **kwargs):
log.info("Running: %s", " ".join(shlex.quote(x) for x in cmd)) log.info("Running: %s", " ".join(shlex.quote(x) for x in cmd))
ret, out = shortcuts.run(cmd, *args, universal_newlines=True, **kwargs) ret, out = shortcuts.run(cmd, *args, text=True, errors="replace", **kwargs)
if out: if out:
log.debug("%s", out) log.debug("%s", out)
return ret, out return ret, out
@ -35,7 +35,8 @@ def get_lorax_dir(default="/usr/share/lorax"):
try: try:
_, out = shortcuts.run( _, out = shortcuts.run(
["python3", "-c" "import pylorax; print(pylorax.find_templates())"], ["python3", "-c" "import pylorax; print(pylorax.find_templates())"],
universal_newlines=True, text=True,
errors="replace",
) )
return out.strip() return out.strip()
except Exception: except Exception:

3
pungi_utils/unified_isos.py
View File

@ -394,7 +394,8 @@ class UnifiedISO(object):
iso.get_mkisofs_cmd( iso.get_mkisofs_cmd(
iso_path, [source_dir], volid=volid, exclude=["./lost+found"] iso_path, [source_dir], volid=volid, exclude=["./lost+found"]
), ),
universal_newlines=True, text=True,
errors="replace",
) )
# implant MD5 # implant MD5

6
tests/test_iso_wrapper.py
View File

@ -63,7 +63,8 @@ class TestIsoUtils(unittest.TestCase):
[ [
mock.call( mock.call(
["/usr/bin/checkisomd5", "--md5sumonly", "dummy.iso"], ["/usr/bin/checkisomd5", "--md5sumonly", "dummy.iso"],
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -79,7 +80,8 @@ class TestIsoUtils(unittest.TestCase):
[ [
mock.call( mock.call(
["/usr/bin/checkisomd5", "--md5sumonly", "dummy.iso"], ["/usr/bin/checkisomd5", "--md5sumonly", "dummy.iso"],
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )

66
tests/test_koji_wrapper.py
View File

@ -529,7 +529,8 @@ class RunrootKojiWrapperTest(KojiWrapperBaseTestCase):
buffer_size=-1, buffer_size=-1,
logfile=None, logfile=None,
show_cmd=True, show_cmd=True,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -558,7 +559,8 @@ class RunrootKojiWrapperTest(KojiWrapperBaseTestCase):
buffer_size=-1, buffer_size=-1,
logfile=None, logfile=None,
show_cmd=True, show_cmd=True,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -582,7 +584,8 @@ class RunrootKojiWrapperTest(KojiWrapperBaseTestCase):
buffer_size=-1, buffer_size=-1,
logfile=None, logfile=None,
show_cmd=True, show_cmd=True,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -615,7 +618,8 @@ class RunrootKojiWrapperTest(KojiWrapperBaseTestCase):
buffer_size=-1, buffer_size=-1,
logfile=None, logfile=None,
show_cmd=True, show_cmd=True,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -641,7 +645,8 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
show_cmd=True, show_cmd=True,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -671,7 +676,8 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
"PYTHONUNBUFFERED": "1", "PYTHONUNBUFFERED": "1",
}, },
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -694,7 +700,8 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
logfile="logfile", logfile="logfile",
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -717,7 +724,8 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
logfile=None, logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -740,7 +748,8 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
logfile=None, logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -765,13 +774,15 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
logfile=None, logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"], ["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True, can_fail=True,
logfile=None, logfile=None,
universal_newlines=True, text=True,
errors="replace",
), ),
], ],
) )
@ -795,13 +806,15 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
logfile=None, logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"], ["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True, can_fail=True,
logfile=None, logfile=None,
universal_newlines=True, text=True,
errors="replace",
), ),
], ],
) )
@ -826,25 +839,29 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
logfile=None, logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"], ["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True, can_fail=True,
logfile=None, logfile=None,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"], ["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True, can_fail=True,
logfile=None, logfile=None,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"], ["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True, can_fail=True,
logfile=None, logfile=None,
universal_newlines=True, text=True,
errors="replace",
), ),
], ],
) )
@ -870,19 +887,22 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
logfile=None, logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"], ["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True, can_fail=True,
logfile=None, logfile=None,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"], ["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True, can_fail=True,
logfile=None, logfile=None,
universal_newlines=True, text=True,
errors="replace",
), ),
], ],
) )
@ -907,13 +927,15 @@ class RunBlockingCmdTest(KojiWrapperBaseTestCase):
logfile=None, logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"}, env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1, buffer_size=-1,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"], ["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True, can_fail=True,
logfile=None, logfile=None,
universal_newlines=True, text=True,
errors="replace",
), ),
], ],
) )

3
tests/test_notifier.py
View File

@ -70,7 +70,8 @@ class TestNotifier(unittest.TestCase):
can_fail=True, can_fail=True,
return_stdout=False, return_stdout=False,
workdir=None, workdir=None,
universal_newlines=True, text=True,
errors="replace",
show_cmd=True, show_cmd=True,
logfile=self.logfile, logfile=self.logfile,
) )

6
tests/test_ostree_script.py
View File

@ -67,7 +67,8 @@ class OstreeTreeScriptTest(helpers.PungiTestCase):
logfile=self.topdir + "/logs/Atomic/create-ostree-repo.log", logfile=self.topdir + "/logs/Atomic/create-ostree-repo.log",
show_cmd=True, show_cmd=True,
stdout=True, stdout=True,
universal_newlines=True, text=True,
errors="replace",
) )
] ]
+ extra_calls, + extra_calls,
@ -136,7 +137,8 @@ class OstreeTreeScriptTest(helpers.PungiTestCase):
logfile=self.topdir + "/logs/Atomic/ostree-summary.log", logfile=self.topdir + "/logs/Atomic/ostree-summary.log",
show_cmd=True, show_cmd=True,
stdout=True, stdout=True,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )

2
tests/test_patch_iso.py
View File

@ -34,7 +34,7 @@ class TestSh(unittest.TestCase):
patch_iso.sh(log, ["ls"], foo="bar") patch_iso.sh(log, ["ls"], foo="bar")
self.assertEqual( self.assertEqual(
mock_run.call_args_list, mock_run.call_args_list,
[mock.call(["ls"], foo="bar", universal_newlines=True)], [mock.call(["ls"], foo="bar", text=True, errors="replace")],
) )
self.assertEqual(log.info.call_args_list, [mock.call("Running: %s", "ls")]) self.assertEqual(log.info.call_args_list, [mock.call("Running: %s", "ls")])
self.assertEqual(log.debug.call_args_list, [mock.call("%s", "ok")]) self.assertEqual(log.debug.call_args_list, [mock.call("%s", "ok")])

2
tests/test_unified_isos.py
View File

@ -610,7 +610,7 @@ class TestCreateiso(PungiTestCase):
self.assertEqual( self.assertEqual(
run.mock_calls, run.mock_calls,
[ [
mock.call(self.mkisofs_cmd, universal_newlines=True), mock.call(self.mkisofs_cmd, text=True, errors="replace"),
mock.call(iso.get_implantisomd5_cmd.return_value), mock.call(iso.get_implantisomd5_cmd.return_value),
mock.call(iso.get_manifest_cmd.return_value), mock.call(iso.get_manifest_cmd.return_value),
] ]

60
tests/test_util.py
View File

@ -25,7 +25,8 @@ class TestGitRefResolver(unittest.TestCase):
self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE") self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE")
run.assert_called_once_with( run.assert_called_once_with(
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"], ["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
universal_newlines=True, text=True,
errors="replace",
) )
@mock.patch("pungi.util.run") @mock.patch("pungi.util.run")
@ -39,7 +40,8 @@ class TestGitRefResolver(unittest.TestCase):
self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE") self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE")
run.assert_called_once_with( run.assert_called_once_with(
GIT_WITH_CREDS + ["ls-remote", "https://git.example.com/repo.git", "HEAD"], GIT_WITH_CREDS + ["ls-remote", "https://git.example.com/repo.git", "HEAD"],
universal_newlines=True, text=True,
errors="replace",
) )
@mock.patch("pungi.util.run") @mock.patch("pungi.util.run")
@ -53,7 +55,8 @@ class TestGitRefResolver(unittest.TestCase):
self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE") self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE")
run.assert_called_once_with( run.assert_called_once_with(
["git", "ls-remote", "https://git.example.com/repo.git", "refs/heads/f24"], ["git", "ls-remote", "https://git.example.com/repo.git", "refs/heads/f24"],
universal_newlines=True, text=True,
errors="replace",
) )
def test_resolve_ref_with_commit_id(self): def test_resolve_ref_with_commit_id(self):
@ -72,7 +75,8 @@ class TestGitRefResolver(unittest.TestCase):
self.assertEqual(ref, "CAFEBABE") self.assertEqual(ref, "CAFEBABE")
run.assert_called_once_with( run.assert_called_once_with(
["git", "ls-remote", "https://git.example.com/repo.git", "master"], ["git", "ls-remote", "https://git.example.com/repo.git", "master"],
universal_newlines=True, text=True,
errors="replace",
) )
@mock.patch("pungi.util.run") @mock.patch("pungi.util.run")
@ -84,7 +88,8 @@ class TestGitRefResolver(unittest.TestCase):
self.assertEqual(ref, "CAFEBABE") self.assertEqual(ref, "CAFEBABE")
run.assert_called_once_with( run.assert_called_once_with(
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"], ["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
universal_newlines=True, text=True,
errors="replace",
) )
@mock.patch("pungi.util.run") @mock.patch("pungi.util.run")
@ -110,7 +115,8 @@ class TestGitRefResolver(unittest.TestCase):
run.assert_called_once_with( run.assert_called_once_with(
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"], ["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
universal_newlines=True, text=True,
errors="replace",
) )
@mock.patch("pungi.util.run") @mock.patch("pungi.util.run")
@ -121,7 +127,8 @@ class TestGitRefResolver(unittest.TestCase):
run.assert_called_once_with( run.assert_called_once_with(
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"], ["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
universal_newlines=True, text=True,
errors="replace",
) )
self.assertEqual(url, "https://git.example.com/repo.git?#CAFEBABE") self.assertEqual(url, "https://git.example.com/repo.git?#CAFEBABE")
@ -133,7 +140,8 @@ class TestGitRefResolver(unittest.TestCase):
run.assert_called_once_with( run.assert_called_once_with(
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"], ["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
universal_newlines=True, text=True,
errors="replace",
) )
self.assertEqual(url, "git+https://git.example.com/repo.git#CAFEBABE") self.assertEqual(url, "git+https://git.example.com/repo.git#CAFEBABE")
@ -153,7 +161,8 @@ class TestGitRefResolver(unittest.TestCase):
"https://git.example.com/repo.git", "https://git.example.com/repo.git",
"refs/heads/my-branch", "refs/heads/my-branch",
], ],
universal_newlines=True, text=True,
errors="replace",
) )
self.assertIn("ref does not exist in remote repo", str(ctx.exception)) self.assertIn("ref does not exist in remote repo", str(ctx.exception))
@ -171,7 +180,8 @@ class TestGitRefResolver(unittest.TestCase):
[ [
mock.call( mock.call(
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"], ["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
universal_newlines=True, text=True,
errors="replace",
) )
] ]
* 2, * 2,
@ -600,7 +610,8 @@ class TestUnmountCmd(unittest.TestCase):
cmd, cmd,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -621,7 +632,8 @@ class TestUnmountCmd(unittest.TestCase):
cmd, cmd,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
) )
], ],
) )
@ -643,7 +655,8 @@ class TestUnmountCmd(unittest.TestCase):
cmd, cmd,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
) )
] ]
* 3, * 3,
@ -668,7 +681,8 @@ class TestUnmountCmd(unittest.TestCase):
cmd, cmd,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
) )
] ]
* 3, * 3,
@ -707,37 +721,43 @@ class TestUnmountCmd(unittest.TestCase):
cmd, cmd,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
cmd, cmd,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
cmd, cmd,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["ls", "-lA", "/path"], ["ls", "-lA", "/path"],
stderr=subprocess.STDOUT, stderr=subprocess.STDOUT,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["fuser", "-vm", "/path"], ["fuser", "-vm", "/path"],
stderr=subprocess.STDOUT, stderr=subprocess.STDOUT,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
), ),
mock.call( mock.call(
["lsof", "+D", "/path"], ["lsof", "+D", "/path"],
stderr=subprocess.STDOUT, stderr=subprocess.STDOUT,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True, text=True,
errors="replace",
), ),
] ]
self.assertEqual(mockPopen.call_args_list, expected) self.assertEqual(mockPopen.call_args_list, expected)