diff --git a/pycommons/base/concurrent/exception.py b/pycommons/base/concurrent/exception.py new file mode 100644 index 0000000..da16c3e --- /dev/null +++ b/pycommons/base/concurrent/exception.py @@ -0,0 +1,7 @@ +from ..exception import CommonsException + + +class ConcurrentException(CommonsException): + """ + Raised for Concurrent Operations + """ diff --git a/pycommons/base/exception/__init__.py b/pycommons/base/exception/__init__.py index f14a4ae..5455d32 100644 --- a/pycommons/base/exception/__init__.py +++ b/pycommons/base/exception/__init__.py @@ -1,15 +1,68 @@ -__all__ = ["NoSuchElementError", "IllegalStateException"] +__all__ = [ + "CommonsException", + "CommonsRuntimeException", + "NoSuchElementError", + "IllegalStateException", +] +import traceback +from types import TracebackType +from typing import Optional, Any -class NoSuchElementError(RuntimeError): + +class CommonsException(Exception): + def __init__( # pylint: disable=W1113 + self, message: Optional[str] = None, cause: Optional[Exception] = None, *args: Any + ) -> None: + super().__init__(*args) + if cause is not None: + self.__cause__ = cause + self._message = message + + def get_cause(self) -> Optional[BaseException]: + return self.__cause__ + + def get_traceback(self) -> Optional[TracebackType]: + return self.__traceback__ + + def get_message(self) -> Optional[str]: + return self._message + + def print_traceback(self) -> None: + traceback.print_tb(self.get_traceback()) + + +class CommonsRuntimeException(RuntimeError): + def __init__( # pylint: disable=W1113 + self, message: Optional[str] = None, cause: Optional[Exception] = None, *args: Any + ) -> None: + super().__init__(*args) + if cause is not None: + self.__cause__ = cause + self._message = message + + def get_cause(self) -> Optional[BaseException]: + return self.__cause__ + + def get_traceback(self) -> Optional[TracebackType]: + return self.__traceback__ + + def get_message(self) -> Optional[str]: + return self._message + + def print_traceback(self) -> None: + traceback.print_tb(self.get_traceback()) + + +class NoSuchElementError(CommonsRuntimeException): """ Raised when an object is expected to be present but is not in real. Extends RuntimeError as this error happens during runtime. """ -class IllegalStateException(RuntimeError): +class IllegalStateException(CommonsRuntimeException): """ - Runtime Error raised when the program's state is not in a expected state or the code execution + Runtime Error raised when the program's state is not in an expected state or the code execution should never have reached this state """ diff --git a/tests/parametrized.py b/tests/parametrized.py index b59f57f..852175e 100644 --- a/tests/parametrized.py +++ b/tests/parametrized.py @@ -1,10 +1,10 @@ import dataclasses import functools -from typing import Any, Optional, List +from typing import Any, Optional @dataclasses.dataclass -class TestData: +class CommonsTestData: data: Any expected: Any @@ -17,7 +17,7 @@ def get_message(self): ) -def cases(testcases: List[TestData]): +def cases(*testcases: CommonsTestData): def decorator(f): @functools.wraps(f) def wrapped(self): diff --git a/tests/pycommons/base/test_exception.py b/tests/pycommons/base/test_exception.py new file mode 100644 index 0000000..d8765c1 --- /dev/null +++ b/tests/pycommons/base/test_exception.py @@ -0,0 +1,59 @@ +from unittest import TestCase + +from pycommons.base.exception import CommonsException, CommonsRuntimeException +from tests.parametrized import CommonsTestData, cases + + +class TestCommonsException(TestCase): + @cases( + CommonsTestData(data=CommonsException, expected=None), + CommonsTestData(data=CommonsRuntimeException, expected=None), + ) + def test_initialize_without_params(self, test_data: CommonsTestData): + try: + raise test_data.data() + except test_data.data as exc: + self.assertEqual(test_data.expected, exc.get_message()) + self.assertIsNone(exc.get_cause()) + self.assertIsNotNone(exc.get_traceback()) + exc.print_traceback() + + @cases( + CommonsTestData(data=CommonsException, expected="Some error occurred"), + CommonsTestData(data=CommonsRuntimeException, expected="Some error occurred"), + ) + def test_initialize_with_message(self, test_data: CommonsTestData): + try: + raise test_data.data(message="Some error occurred") + except test_data.data as exc: + self.assertEqual(test_data.expected, exc.get_message()) + self.assertIsNone(exc.get_cause()) + self.assertIsNotNone(exc.get_traceback()) + + @cases( + CommonsTestData(data=CommonsException, expected="Some error occurred"), + CommonsTestData(data=CommonsRuntimeException, expected="Some error occurred"), + ) + def test_initialize_with_cause(self, test_data: CommonsTestData): + _cause = Exception("cause exception") + try: + raise test_data.data(message="Some error occurred", cause=_cause) + except test_data.data as exc: + self.assertEqual(test_data.expected, exc.get_message()) + self.assertEqual(_cause, exc.get_cause()) + self.assertEqual(_cause, exc.__cause__) + self.assertIsNotNone(exc.get_traceback()) + + @cases( + CommonsTestData(data=CommonsException, expected="Some error occurred"), + CommonsTestData(data=CommonsRuntimeException, expected="Some error occurred"), + ) + def test_initialize_with_cause2(self, test_data: CommonsTestData): + _cause = Exception("cause exception") + try: + raise test_data.data(message="Some error occurred") from _cause + except test_data.data as exc: + self.assertEqual(test_data.expected, exc.get_message()) + self.assertEqual(_cause, exc.get_cause()) + self.assertEqual(_cause, exc.__cause__) + self.assertIsNotNone(exc.get_traceback())