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
7 changes: 7 additions & 0 deletions pycommons/base/concurrent/exception.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
from ..exception import CommonsException


class ConcurrentException(CommonsException):
"""
Raised for Concurrent Operations
"""
61 changes: 57 additions & 4 deletions pycommons/base/exception/__init__.py
Original file line number Diff line number Diff line change
@@ -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
"""
6 changes: 3 additions & 3 deletions tests/parametrized.py
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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):
Expand Down
59 changes: 59 additions & 0 deletions tests/pycommons/base/test_exception.py
Original file line number Diff line number Diff line change
@@ -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())