Skip to content

[PATCH] specialize exception hear (- WIP #154 -) #179

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Nov 13, 2024
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 docs/conf.py
Original file line number Diff line number Diff line change
Expand Up @@ -237,7 +237,7 @@
# html_use_opensearch = ''

# This is the file name suffix for HTML files (e.g. ".xhtml").
#html_file_suffix = "html"
# html_file_suffix = "html"

# Output file base name for HTML help builder.
htmlhelp_basename = "multicast_doc"
Expand Down
105 changes: 102 additions & 3 deletions multicast/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -254,6 +254,105 @@ def __init__(self, *args, **kwargs):
self.exit_code = exit_code


class ShutdownCommandReceived(RuntimeError):
"""
Exception raised when a 'STOP' command is received during the HEAR operation.

This exception signals a graceful shutdown of the multicast server.

Attributes:
message (str): Description of the exception.

Testing:

Testcase 1: Initialization with default message.

>>> exc = ShutdownCommandReceived()
>>> isinstance(exc, RuntimeError)
True
>>> exc.message
'SHUTDOWN'

Testcase 2: Initialization with custom message.

>>> exc = ShutdownCommandReceived("Custom shutdown message.")
>>> exc.message
'Custom shutdown message.'
"""

__module__ = """multicast.exceptions"""

__name__ = """multicast.exceptions.ShutdownCommandReceived"""

def __init__(self, message="SHUTDOWN", *args, **kwargs):
"""
Initialize the ShutdownCommandReceived exception.

The ShutdownComandRecived exception is used by the default handler
for the HEAR servers, to instruct the HEAR server to shutdown gracefully.

Parameters:
message (str): Optional custom message for the exception.
*args: Additional positional arguments passed to RuntimeError.
**kwargs: Additional keyword arguments passed to RuntimeError.

Raises:
TypeError: If message is not a string.

Testing:

Testcase 1: Default initialization.

>>> exc = ShutdownCommandReceived()
>>> isinstance(exc, ShutdownCommandReceived)
True
>>> exc.message
'SHUTDOWN'

Testcase 2: Initialization with custom message.

>>> exc = ShutdownCommandReceived("Custom shutdown message.")
>>> isinstance(exc, ShutdownCommandReceived)
True
>>> exc.message
'Custom shutdown message.'

Testcase 3: Invalid message type.

>>> exc = ShutdownCommandReceived(123) # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
TypeError: message must be a string

Testcase 4: Exit code verification.

>>> exc = ShutdownCommandReceived()
>>> exc.exit_code == 143 # Verify SIGTERM exit code
True

Testcase 5: Error propagation with exit_on_exception.

>>> @exit_on_exception
... def test_func():
... raise ShutdownCommandReceived()
>>> test_func() # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
SystemExit: 143

Testcase 6: Error message propagation.

>>> try:
... raise ShutdownCommandReceived("Custom message")
... except ShutdownCommandReceived as e:
... str(e) == "Custom message"
True
"""
if not isinstance(message, str):
raise TypeError("[CWE-573] message must be a string")
super().__init__(message, *args, **kwargs)
self.message = message
self.exit_code = 143 # Use SIGTERM exit code for graceful shutdown


EXIT_CODES = {
0: (None, 'Success'),
1: (RuntimeError, 'General Error'),
Expand Down Expand Up @@ -478,7 +577,7 @@ def wrapper(*args, **kwargs):

# skipcq
__all__ = [
"""__package__""", """__module__""", """__name__""", "__doc__", # skipcq: PYL-E0603
"CommandExecutionError", "EXCEPTION_EXIT_CODES", "EXIT_CODES",
"get_exit_code_from_exception", "exit_on_exception"
"""__package__""", """__module__""", """__name__""", """__doc__""", # skipcq: PYL-E0603
"""CommandExecutionError""", """EXCEPTION_EXIT_CODES""", """EXIT_CODES""",
"""get_exit_code_from_exception""", """exit_on_exception""", """ShutdownCommandReceived"""
]
2 changes: 1 addition & 1 deletion multicast/hear.py
Original file line number Diff line number Diff line change
Expand Up @@ -438,7 +438,7 @@ def handle(self):
)
)
if """STOP""" in str(data):
raise RuntimeError("SHUTDOWN") from None
raise multicast.exceptions.ShutdownCommandReceived("SHUTDOWN") from None


class McastHEAR(multicast.mtool):
Expand Down
7 changes: 5 additions & 2 deletions tox.ini
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,9 @@ commands =
basepython =
flake: python3
deps =
flake: flake8>=5.0.0
mccabe>=0.7.0
flake8-comprehensions>=3.15.0
{[base]deps}
commands = flake8 --ignore=W191,W391 --verbose --max-line-length=100 --count --exclude=site-packages,activate_this.py

Expand All @@ -179,8 +182,8 @@ whitelist_externals = make
passenv =
{[base]passenv}
deps =
docs: sphinx>=5.2
flake8>=2.5.4
docs: sphinx>=7.3
flake8>=5.0.0
mccabe>=0.6.1
pyflakes>=1.1.0
pep8>=1.7.0
Expand Down
Loading