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.

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
    >>>

class multicast.hear.McastServer(server_address, RequestHandlerClass, bind_and_activate=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

Constructor. May be extended, do not override.

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

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.