Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ classifiers = [
dependencies = [
"psutil",
"six>=1.9.0",
"testgres.common>=0.0.3,<1.0.0",
"testgres.common @ git+https://github.com/postgrespro/testgres.common.git@1.0.0",
]

[project.urls]
Expand Down
120 changes: 102 additions & 18 deletions src/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,38 +3,122 @@
from testgres.common.exceptions import TestgresException
from testgres.common.exceptions import InvalidOperationException
import six
import typing


T_CMD = typing.Union[str, list]
T_OUT_DATA = typing.Union[str, bytes]
T_ERR_DATA = typing.Union[str, bytes]


class ExecUtilException(TestgresException):
def __init__(self, message=None, command=None, exit_code=0, out=None, error=None):
super(ExecUtilException, self).__init__(message)
_description: typing.Optional[str]
_command: typing.Optional[T_CMD]
_exit_code: typing.Optional[int]
_out: typing.Optional[T_OUT_DATA]
_error: typing.Optional[T_ERR_DATA]

def __init__(
self,
message: typing.Optional[str] = None,
command: typing.Optional[T_CMD] = None,
exit_code: typing.Optional[int] = None,
out: typing.Optional[T_OUT_DATA] = None,
error: typing.Optional[T_ERR_DATA] = None,
):
assert message is None or type(message) == str # noqa: E721
assert command is None or type(command) in [str, list] # noqa: E721
assert exit_code is None or type(exit_code) == int # noqa: E721
assert out is None or type(out) in [str, bytes] # noqa: E721
assert error is None or type(error) in [str, bytes] # noqa: E721

self.message = message
self.command = command
self.exit_code = exit_code
self.out = out
self.error = error
super().__init__(message)

def __str__(self):
self._description = message
self._command = command
self._exit_code = exit_code
self._out = out
self._error = error

@property
def message(self) -> str:
msg = []

if self.message:
msg.append(self.message)
if self._description:
msg.append(self._description)

if self.command:
command_s = ' '.join(self.command) if isinstance(self.command, list) else self.command
if self._command:
command_s = ' '.join(self._command) if isinstance(self._command, list) else self._command
msg.append(u'Command: {}'.format(command_s))

if self.exit_code:
msg.append(u'Exit code: {}'.format(self.exit_code))
if self._exit_code:
msg.append(u'Exit code: {}'.format(self._exit_code))

if self.error:
msg.append(u'---- Error:\n{}'.format(self.error))
if self._error:
msg.append(u'---- Error:\n{}'.format(self._error))

if self.out:
if self._out:
msg.append(u'---- Out:\n{}'.format(self.out))

return self.convert_and_join(msg)
r = self.convert_and_join(msg)
assert type(r) == str # noqa: E721
return r

@property
def description(self) -> typing.Optional[str]:
assert self._description is None or type(self._description) == str # noqa: E721
return self._description

@property
def command(self) -> typing.Optional[T_CMD]:
assert self._command is None or type(self._command) in [str, list] # noqa: E721
return self._command

@property
def exit_code(self) -> typing.Optional[int]:
assert self._exit_code is None or type(self._exit_code) == int # noqa: E721
return self._exit_code

@property
def out(self) -> typing.Optional[T_OUT_DATA]:
assert self._out is None or type(self._out) in [str, bytes] # noqa: E721
return self._out

@property
def error(self) -> typing.Optional[T_ERR_DATA]:
assert self._error is None or type(self._error) in [str, bytes] # noqa: E721
return self._error

def __repr__(self) -> str:
args = []

if self._description is not None:
args.append(("message", self._description))

if self._command is not None:
args.append(("command", self._command))

if self._exit_code is not None:
args.append(("exit_code", self._exit_code))

if self._out is not None:
args.append(("out", self._out))

if self._error is not None:
args.append(("error", self._error))

assert type(self) == ExecUtilException # noqa: E721
assert __class__ == ExecUtilException # noqa: E721

result = "{}(".format(__class__.__name__)
sep = ""
for a in args:
if a[1] is not None:
result += sep + a[0] + "=" + repr(a[1])
sep = ", "
continue
result += ")"
return result

@staticmethod
def convert_and_join(msg_list):
Expand Down
2 changes: 1 addition & 1 deletion tests/requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ pytest
pytest-xdist
psutil
six
testgres.common>=0.0.2,<1.0.0
git+https://github.com/postgrespro/testgres.common.git@1.0.0
black
flake8
flake8-pyproject
3 changes: 2 additions & 1 deletion tests/test_os_ops_remote.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,8 @@ def test_rmdirs__try_to_delete_file(self, os_ops: OsOperations):

assert os.path.exists(path)
assert type(x.value) == ExecUtilException # noqa: E721
assert x.value.message == "Utility exited with non-zero code (20). Error: `cannot remove '" + path + "': it is not a directory`"
assert x.value.description == "Utility exited with non-zero code (20). Error: `cannot remove '" + path + "': it is not a directory`"
assert x.value.message.startswith(x.value.description)
assert type(x.value.error) == str # noqa: E721
assert x.value.error.strip() == "cannot remove '" + path + "': it is not a directory"
assert type(x.value.exit_code) == int # noqa: E721
Expand Down
136 changes: 136 additions & 0 deletions tests/units/exceptions/ExecUtilException/test_set001__constructor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
from src.exceptions import ExecUtilException


class TestSet001_Constructor:
def test_001__default(self):
e = ExecUtilException()
assert e.source is None
assert e.message == ""
assert e.description is None
assert e.command is None
assert e.exit_code is None
assert e.out is None
assert e.error is None
assert str(e) == ""
assert repr(e) == "ExecUtilException()"
return

def test_002__description(self):
e = ExecUtilException("operation description")
assert e.source is None
assert e.message == "operation description"
assert e.description == "operation description"
assert e.command is None
assert e.exit_code is None
assert e.out is None
assert e.error is None
assert str(e) == "operation description"
assert repr(e) == "ExecUtilException(message='operation description')"
return

def test_003__commandList(self):
e = ExecUtilException(command=["ls", "."])
assert e.source is None
assert e.message == "Command: ls ."
assert e.description is None
assert e.command == ["ls", "."]
assert e.exit_code is None
assert e.out is None
assert e.error is None
assert str(e) == "Command: ls ."
assert repr(e) == "ExecUtilException(command=['ls', '.'])"
return

def test_004__commandStr(self):
e = ExecUtilException(command="ls /home")
assert e.source is None
assert e.message == "Command: ls /home"
assert e.description is None
assert e.command == "ls /home"
assert e.exit_code is None
assert e.out is None
assert e.error is None
assert str(e) == "Command: ls /home"
assert repr(e) == "ExecUtilException(command='ls /home')"
return

def test_005__exit_code(self):
e = ExecUtilException(exit_code=123)
assert e.source is None
assert e.message == "Exit code: 123"
assert e.description is None
assert e.command is None
assert e.exit_code == 123
assert e.out is None
assert e.error is None
assert str(e) == "Exit code: 123"
assert repr(e) == "ExecUtilException(exit_code=123)"
return

def test_006__outBytes(self):
e = ExecUtilException(out=b'abcdefg\n123456')
assert e.source is None
assert e.message == "---- Out:\nb'abcdefg\\n123456'"
assert e.description is None
assert e.command is None
assert e.exit_code is None
assert e.out == b'abcdefg\n123456'
assert e.error is None
assert str(e) == "---- Out:\nb'abcdefg\\n123456'"
assert repr(e) == "ExecUtilException(out=b'abcdefg\\n123456')"
return

def test_007__outStr(self):
e = ExecUtilException(out='abcdefg\n123456')
assert e.source is None
assert e.message == "---- Out:\nabcdefg\n123456"
assert e.description is None
assert e.command is None
assert e.exit_code is None
assert e.out == 'abcdefg\n123456'
assert e.error is None
assert str(e) == "---- Out:\nabcdefg\n123456"
assert repr(e) == "ExecUtilException(out='abcdefg\\n123456')"
return

def test_008__errorBytes(self):
e = ExecUtilException(error=b'abcdefg\n123456')
assert e.source is None
assert e.message == "---- Error:\nb'abcdefg\\n123456'"
assert e.description is None
assert e.command is None
assert e.exit_code is None
assert e.out is None
assert e.error == b'abcdefg\n123456'
assert str(e) == "---- Error:\nb'abcdefg\\n123456'"
assert repr(e) == "ExecUtilException(error=b'abcdefg\\n123456')"
return

def test_009__errorStr(self):
e = ExecUtilException(error='abcdefg\n123456')
assert e.source is None
assert e.message == "---- Error:\nabcdefg\n123456"
assert e.description is None
assert e.command is None
assert e.exit_code is None
assert e.out is None
assert e.error == 'abcdefg\n123456'
assert str(e) == "---- Error:\nabcdefg\n123456"
assert repr(e) == "ExecUtilException(error='abcdefg\\n123456')"
return

def test_010__all(self):
e = ExecUtilException('descr', ['rm', 'me'], -1, 'out\n123456', b'error\n321')

expected_msg = "descr\nCommand: rm me\nExit code: -1\n---- Error:\nb'error\\n321'\n---- Out:\nout\n123456"

assert e.source is None
assert e.message == expected_msg
assert e.description == "descr"
assert e.command == ['rm', 'me']
assert e.exit_code == -1
assert e.out == 'out\n123456'
assert e.error == b'error\n321'
assert str(e) == expected_msg
assert repr(e) == "ExecUtilException(message='descr', command=['rm', 'me'], exit_code=-1, out='out\\n123456', error=b'error\\n321')"
return