multicast.hear

Provides multicast HEAR Features.

Provides server-side functionality for listening to multicast messages. It implements a UDP server that can receive and process multicast messages continuously.

Classes: McastServer: UDP server implementation for multicast communication. HearUDPHandler: Request handler for processing multicast messages. McastHEAR: Main tool class for HEAR operations.

Caution: See details regarding dynamic imports documented in this module.

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

    Testcase 0: Multicast should be importable.

            >>> import multicast
            >>> multicast.hear is not None
            True
            >>> multicast.__doc__ is not None
            True
            >>>

    Testcase 1: Recv should be automatically imported.
            A: Test that the multicast component is initialized.
            B: Test that the hear component is initialized.
            C: Test that the hear component has __doc__

            >>> multicast is not None
            True
            >>> multicast.hear is not None
            True
            >>> multicast.hear.__doc__ is not None
            True
            >>> type(multicast.hear.__doc__) == type(str(''''''))
            True
            >>>

    Testcase 2: Recv should be detailed with some metadata.
            A: Test that the __MAGIC__ variables are initialized.
            B: Test that the __MAGIC__ variables are strings.

            >>> multicast.hear is not None
            True
            >>> multicast.hear.__module__ is not None
            True
            >>> multicast.hear.__package__ is not None
            True
            >>> type(multicast.hear.__doc__) == type(multicast.recv.__module__)
            True
            >>>

    Testcase 3: main should return an int.
            A: Test that the multicast component is initialized.
            B: Test that the hear component is initialized.
            C: Test that the main(HEAR) function-flow is initialized.
            D: Test that the main(HEAR) function-flow returns an int 0-256.

            >>> multicast.__main__ is not None
            True
            >>> multicast.__main__.main is not None
            True
            >>> tst_fxtr_args = ['''HEAR''', '''--port=1234''']
            >>> (test_fixture, ignored_value) = multicast.__main__.main(tst_fxtr_args)
            >>> test_fixture is not None
            True
            >>> type(test_fixture) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...int...>
            >>> int(test_fixture) >= int(0)
            True
            >>> type(test_fixture) is type(0)
            True
            >>> int(test_fixture) < int(256)
            True
            >>> (int(test_fixture) >= int(0)) and (int(test_fixture) < int(256))
            True
            >>>

    Testcase 3: Class McastHEAR should be stable.
            A: Test that the hear component is initialized.
            B: Tests that McastHEAR instantiates without error, as in trivial use-case.

            >>> multicast.hear is not None
            True
            >>> hear = multicast.hear.McastHEAR()
            >>> isinstance(hear, multicast.hear.McastHEAR)
            True
            >>>

Module Contents

Classes

McastServer	Generic Subclasses socketserver.UDPServer for handling ‘–daemon’ function.
HearUDPHandler	Subclass of socketserver.BaseRequestHandler for handling the HEAR function.
McastHEAR	Provides the HEAR tooling by subclassing multicast.mtool.

Data

__package__	Names the package of this program.
__module__	Names the module of this program.
__file__	Names the file of this component.
__name__	Names this component.
module_logger

API

multicast.hear.__package__

‘multicast’

Names the package of this program.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>>

Testcase 1: Hear should be automatically imported.

    >>> multicast.hear.__package__ is not None
    True
    >>>
    >>> multicast.hear.__package__ == multicast.__package__
    True
    >>>

multicast.hear.__module__

‘multicast’

Names the module of this program.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>>

Testcase 1: Hear should be automatically imported.

    >>> multicast.hear.__module__ is not None
    True
    >>>

multicast.hear.__file__

‘multicast/hear.py’

Names the file of this component.

multicast.hear.__name__

‘multicast.hear’

Names this component.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>>

Testcase 1: Hear should be automatically imported.

    >>> multicast.hear.__name__ is not None
    True
    >>>

multicast.hear.module_logger

‘getLogger(…)’

class multicast.hear.McastServer(server_address: tuple, RequestHandlerClass: type, bind_and_activate: bool = True)

Bases: socketserver.UDPServer

Generic Subclasses socketserver.UDPServer for handling ‘–daemon’ function.

Basically simplifies testing by allowing a trivial echo back (case-insensitive) of string data, after printing the sender’s ip out.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>> multicast.hear is not None
    True
    >>> from multicast.hear import McastServer as McastServer
    >>>

Testcase 1: McastServer should be automatically imported.

    >>> McastServer.__name__ is not None
    True
    >>>

Initialization

Initialize a new instance of the McastServer.

Creates a new UDP server for multicast communication and sets up an appropriate logger based on the server address provided. May be extended, do not override.

Returns: None

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import socketserver
    >>> import multicast
    >>>

Testcase 0: Basic initialization of McastServer. A: Test that McastServer can be initialized with minimal arguments. B: Test that the resulting instance is of the correct type.

    >>> server = multicast.hear.McastServer(('224.0.0.1', 12345), None)
    >>> isinstance(server, multicast.hear.McastServer)
    True
    >>> isinstance(server, socketserver.UDPServer)
    True
    >>> server.server_close()  # Clean up
    >>>

Testcase 1: Server initialization with logger name extraction. A: Test that the server extracts the logger name from server_address. B: Test that the logger is properly initialized.

    >>> test_addr = ('239.0.0.9', 23456)
    >>> server = multicast.hear.McastServer(test_addr, None)
    >>> server.logger is not None
    True
    >>> server.logger.name.endswith('239.0.0.9')
    True
    >>> server.server_close()  # Clean up
    >>>

__log_handle__

‘multicast.hear.McastServer’

Names this server’s Logger.

Basically just the prefix of the logger’s name. Subclasses should override.

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

    Testcase 0: Multicast should be importable.

            >>> import multicast
            >>>

    Testcase 1: McastServer should be automatically imported.

            >>> multicast.hear.McastServer.__name__ is not None
            True
            >>> multicast.hear.McastServer.__log_handle__ is not None
            True
            >>>

_sync_logger() → None

Synchronize the logger instance with the bound socket address.

This internal method updates the instance’s logger attribute based on the current socket address. It extracts the address component from the socket’s bound address and uses it to create a hierarchical logger name in the format ‘multicast.hear.McastServer.[address]’.

If no valid address is found, it falls back to the base McastServer logger. This method is typically called after server_bind() to ensure the logger reflects the actual bound socket address.

Note: This is an internal method and should not be called directly from outside the class.

Args: None

Returns: None

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import types
    >>> import logging
    >>> import multicast
    >>>

Testcase 1: Method exists and takes expected arguments. A: Test method exists in McastServer class. B: Test method signature does not accept arguments.

    >>> from multicast.hear import McastServer
    >>> hasattr(McastServer, '_sync_logger')
    True
    >>> import inspect
    >>> len(inspect.signature(McastServer._sync_logger).parameters) - 1  # Remove 'self'
    0
    >>>

Testcase 2: Method handles case where socket has a valid address. A: Use a mock socket with a valid address. B: Verify logger name is properly formatted.

    >>> # Setup a server instance with mock socket
    >>> server = multicast.hear.McastServer(('239.0.0.9', 51234), None)
    >>> # Create mock socket with valid address
    >>> mock_socket = types.SimpleNamespace()
    >>> mock_socket.getsockname = lambda: ('239.0.0.9', 51234)
    >>> server.socket = mock_socket
    >>> # Call method and verify logger name
    >>> server._sync_logger()
    >>> server.logger.name
    'multicast.hear.McastServer.239.0.0.9'
    >>>

Testcase 3: Method handles case where socket address name component is None. A: Use a mock socket with None as the address component. B: Verify logger falls back to base logger name.

    >>> # Setup a server instance
    >>> server = multicast.hear.McastServer(('239.0.0.9', 51234), None)
    >>> # Create mock socket with None as the address
    >>> mock_socket = types.SimpleNamespace()
    >>> mock_socket.getsockname = lambda: (None, 5678)
    >>> server.socket = mock_socket
    >>> # Call method and verify logger name
    >>> server._sync_logger()
    >>> server.logger.name
    'multicast.hear.McastServer'
    >>>

Testcase 4: Method handles case where socket address has special formatting. A: Use a mock socket with an IPv6 address. B: Verify logger name incorporates the address correctly.

    >>> import multicast
    >>> from multicast.hear import McastServer
    >>> import types
    >>> import logging
    >>> # Setup a server instance
    >>> server = McastServer(('239.0.0.9', 51234), None)
    >>> # Create mock socket with IPv6 address format
    >>> mock_socket = types.SimpleNamespace()
    >>> mock_socket.getsockname = lambda: ('2001:db8::1', 5678)
    >>> server.socket = mock_socket
    >>> # Call method and verify logger name
    >>> server._sync_logger()
    >>> server.logger.name
    'multicast.hear.McastServer.2001:db8::1'
    >>>

property logger: logging.Logger

Getter for the logger attribute of McastServer.

This property provides access to the server’s internal logger instance. The logger name is determined during initialization and may be updated by calling _sync_logger when the server’s socket address changes.

Returns: logging.Logger – The logger instance associated with this server.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

    >>> import multicast
    >>> import logging
    >>>

Testcase 0: Verify logger accessibility. A: Test that the logger property exists. B: Test that it’s accessible from an McastServer instance.

    >>> server = multicast.hear.McastServer(("239.0.0.9", 0), multicast.hear.HearUDPHandler)
    >>> hasattr(server, 'logger')
    True
    >>> server.server_close()  # Clean up
    >>>

Testcase 1: Verify logger type. A: Test that the logger property returns a logging.Logger instance.

    >>> server = multicast.hear.McastServer(("239.0.0.9", 0), multicast.hear.HearUDPHandler)
    >>> isinstance(server.logger, logging.Logger)
    True
    >>> server.server_close()  # Clean up
    >>>

Testcase 2: Verify logger name. A: Test that the logger name includes the server class name. B: Test that it’s properly formatted.

    >>> server = multicast.hear.McastServer(("239.0.0.9", 0), multicast.hear.HearUDPHandler)
    >>> server.logger.name.startswith('multicast.hear.McastServer')
    True
    >>> server.server_close()  # Clean up
    >>>

server_activate()

Activate the server to begin handling requests.

Overrides the base class method to set up the server after binding.

Returns: None

open_for_request()

Prepare the server to accept requests.

Overrides the base class method to set up a new listening UDP socket before the server starts processing requests.

UDP Sockets are considered ephemeral. Sequentially, the old socket is recycled, or replaced, yielding a fungable socket, with the same port and bound ip, which is then used to join the same multicast group(s), at which point the new socket has transparently replaced the old socket.

Returns: None

server_bind()

Bind the server to the specified address.

Overrides the base class method to handle multicast group binding.

Returns: None

close_request(request)

Clean up after handling a request.

Overrides the base class method to call open_for_request to close and regenerate the UDP socket, in addition to closing the request as normal.

Args: request: The request object to close.

Returns: None

handle_error(request, client_address)

Handle errors that occur during request processing.

Overrides the base class method to handle requests with STOP in them, resulting in a graceful server shutdown. Otherwise forwards the call to super.

Args: request: The request being handled when the error occurred. client_address: The client address associated with the request.

Returns: None

class multicast.hear.HearUDPHandler(request, client_address, server)

Bases: socketserver.BaseRequestHandler

Subclass of socketserver.BaseRequestHandler for handling the HEAR function.

Basically simplifies testing by allowing a simple HEAR back (case-insensitive) of string data, after printing the sender’s ip out.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>> multicast.hear is not None
    True
    >>> from multicast.hear import HearUDPHandler as HearUDPHandler
    >>>

Testcase 1: HearUDPHandler should be automatically imported.

    >>> HearUDPHandler.__name__ is not None
    True
    >>>

Initialization

__name__

‘multicast.hear.HearUDPHandler’

Names this handler type.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>>

Testcase 1: HearUDPHandler should be automatically imported.

    >>> multicast.hear.HearUDPHandler.__name__ is not None
    True
    >>>

handle()

Handles incoming UDP requests in the HEAR functionality.

Overrides the base class method to define how incoming data is handled.

By default: Processes the incoming data from the client, logs the messages, and sends a response back. If the data contains the keyword “STOP”, it raises a RuntimeError to initiate server shutdown. Silently ignores any UnicodeDecodeError when decoding data. Returns early if data or socket is None.

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

    >>> import multicast
    >>>

    Testcase 0: Ensure `HearUDPHandler` can be imported.

    >>> import multicast
    >>> from multicast.hear import HearUDPHandler
    >>> HearUDPHandler.__name__ is not None
    True
    >>>

    Testcase 1: Verify the `handle` method exists.

    >>> handler = HearUDPHandler(
    ...     request=('Test data', None), client_address=('192.0.2.1', 51111), server=None
    ... )
    >>> hasattr(handler, 'handle')
    True
    >>>

    Testcase 2: `handle` requires valid requests or ignores input.

    >>> handler.request = ("No-Op", None)
    >>> handler.client_address = ("192.0.2.2", 51234)
    >>> handler.handle() is None
    True
    >>>

    Testcase 3: `handle` requires valid requests or ignores input.

    >>> tst_fixture_sock = multicast.genSocket()
    >>> handler.request = ("The Test", tst_fixture_sock)
    >>> handler.client_address = ("224.0.1.2", 51234)
    >>> handler.handle() is None
    True
    >>>
    >>> multicast.endSocket(tst_fixture_sock)
    >>>

class multicast.hear.McastHEAR

Bases: multicast.multicast.mtool

Provides the HEAR tooling by subclassing multicast.mtool.

This class sets up a multicast server that listens for messages and processes them accordingly.

    Testing:

    Testcase 0: First set up test fixtures by importing multicast.

            >>> import multicast
            >>> multicast.hear is not None
            True
            >>> multicast._MCAST_DEFAULT_PORT is not None
            True
            >>> multicast._MCAST_DEFAULT_GROUP is not None
            True
            >>> multicast._MCAST_DEFAULT_TTL is not None
            True
            >>> multicast.hear.McastHEAR is not None
            True
            >>>

    Testcase 2: Recv should be detailed with some metadata.
            A: Test that the __MAGIC__ variables are initialized.
            B: Test that the __MAGIC__ variables are strings.

            >>> multicast.hear is not None
            True
            >>> multicast.hear.McastHEAR is not None
            True
            >>> multicast.hear.McastHEAR.__module__ is not None
            True
            >>> multicast.hear.McastHEAR.__proc__ is not None
            True
            >>> multicast.hear.McastHEAR.__epilogue__ is not None
            True
            >>> multicast.hear.McastHEAR.__prologue__ is not None
            True
            >>>

__module__

‘multicast.hear’

__name__

‘multicast.hear.McastHEAR’

__proc__

‘HEAR’

__epilogue__ = <Multiline-String>

__prologue__

‘Python Multicast Server for multicast input.’

classmethod setupArgs(parser)

doStep(*args, **kwargs)

Execute the HEAR operation for multicast communication.

Overrides the doStep method from mtool to set up a server that listens for multicast messages and processes them accordingly.

Args: *args: Variable length argument list containing command-line arguments. **kwargs: Arbitrary keyword arguments.

Returns: tuple: A tuple containing a status indicator and an optional result message.