Skip to content

Commit

Permalink
Make decoding a parameter
Browse files Browse the repository at this point in the history
Make the `decode` an optional, and just return the string as bytes
unless the developer has passed an encoding.

It is false to assume an encoding, and auto-resolving the encoding is
unreliable.
  • Loading branch information
rustydb committed Aug 9, 2023
1 parent cdae3a6 commit 6ef5db9
Show file tree
Hide file tree
Showing 2 changed files with 76 additions and 16 deletions.
44 changes: 36 additions & 8 deletions libcsm/os.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,8 @@ class _CLI:
An object to abstract the return result from ``run_command``.
"""
_stdout = ''
_stderr = ''
_stdout = b''
_stderr = b''
_return_code = None
_duration = None

Expand Down Expand Up @@ -87,8 +87,8 @@ def _run(self) -> None:
LOG.error('Could not find command for given args: %s', self.args)
else:
try:
self._stdout = stdout.decode('utf8')
self._stderr = stderr.decode('utf8')
self._stdout = stdout
self._stderr = stderr
self._return_code = command.returncode
except UnicodeDecodeError as error:
self._stderr = error
Expand All @@ -105,14 +105,14 @@ def _run(self) -> None:
)

@property
def stdout(self) -> str:
def stdout(self) -> [str, bytes]:
"""
``stdout`` from the command.
"""
return self._stdout

@property
def stderr(self) -> str:
def stderr(self) -> [str, bytes]:
"""
``stderr`` from the command.
"""
Expand All @@ -132,6 +132,26 @@ def duration(self) -> float:
"""
return self._duration

def decode(self, charset: str) -> None:
"""
Decodes ``self._stdout`` and ``self._stderr`` with the given ``charset``.
:param charset: The character set to decode with.
"""
if not isinstance(self._stdout, str):
try:
self._stdout = self._stdout.decode(charset)
except ValueError as error:
LOG.error("Command output was requested to be decoded as"
" %s but failed: %s", charset, error)
raise error
if not isinstance(self._stderr, str):
try:
self._stderr = self._stderr.decode(charset)
except ValueError as error:
LOG.error("Command output was requested to be decoded as"
" %s but failed: %s", charset, error)
raise error


@contextmanager
def chdir(directory: str, create: bool = False) -> None:
Expand Down Expand Up @@ -171,7 +191,9 @@ def chdir(directory: str, create: bool = False) -> None:
def run_command(
args: [list, str],
in_shell: bool = False,
silence: bool = False, ) -> _CLI:
silence: bool = False,
charset: str = None,
) -> _CLI:
"""
Runs a given command or list of commands by instantiating a ``CLI`` object.
Expand All @@ -185,6 +207,9 @@ def run_command(
:param args: List of arguments to run, can also be a string. If a string,
:param in_shell: Whether to use a shell when invoking the command.
:param silence: Tells this not to output the command to console.
:param charset: Returns the command ``stdout`` and ``stderr`` as a
string instead of bytes, and decoded with the given
``charset``.
"""
args_string = [str(x) for x in args]
if not silence:
Expand All @@ -193,4 +218,7 @@ def run_command(
' '.join(args_string),
in_shell
)
return _CLI(args_string, shell=in_shell)
result = _CLI(args_string, shell=in_shell)
if charset:
result.decode(charset)
return result
48 changes: 40 additions & 8 deletions libcsm/tests/test_os.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
"""
from os import getcwd
from subprocess import Popen

import pytest
import mock

from libcsm.os import run_command
Expand Down Expand Up @@ -55,18 +57,48 @@ def test_run_command(self) -> None:
assert isinstance(result.return_code, int)
assert result.stdout or result.stderr

def test_run_command_unicode_error(self) -> None:
def test_run_command_decode(self) -> None:
"""
Assert that UnicodeDecodeErrors are caught by the run_command
and a non-zero return code is recieved.
Assert that a ``_CLI`` object successfully decodes its ``stdout`` and
``stderr``.
"""
# test with an invalid utf-8 byte
command = ['ls', '-l']
shell_command = 'ls -l'
bad_command = 'foo!!!'
no_shell_result = run_command(shell_command)
shell_result = run_command(
shell_command,
in_shell=True,
charset='utf-8'
)
bad_result = run_command(bad_command)
command_result = run_command(command)
for result in [no_shell_result, shell_result, bad_result,
command_result]:
assert result.duration > 0.0
assert isinstance(result.return_code, int)
assert result.stdout or result.stderr

def test_run_command_decode_unicode_error(self) -> None:
"""
Assert that ``UnicodeDecodeErrors`` are caught by the ``run_command``
and a non-zero return code is received.
"""
# Test using a CP1252 (windows ostensibly) character "œ"
byte_string = b'\x9c'
with mock.patch.object(Popen, 'communicate', return_value=(byte_string, '')):
result = run_command(['ls','-l'])
assert isinstance(result.stderr, UnicodeDecodeError)
assert result.return_code == 1
with mock.patch.object(
Popen,
'communicate',
return_value=(byte_string, byte_string)
):
result = run_command(['ls', '-l'])
assert result.duration > 0.0
assert isinstance(result.return_code, int)
assert result.stdout or result.stderr
assert isinstance(result.stdout, bytes)
assert isinstance(result.stderr, bytes)
with pytest.raises(UnicodeDecodeError):
run_command(['ls', '-l'], charset='utf-8')

def test_chdir(self) -> None:
"""
Expand Down

0 comments on commit 6ef5db9

Please sign in to comment.