multicast.__main__

Main entry point for the multicast package.

This module provides the command-line interface and core functionalities for multicast communication. It handles command dispatching and execution of multicast operations.

Classes: McastNope: No-operation implementation for testing and validation. McastRecvHearDispatch: Handler for receiving multicast messages. McastDispatch: Main dispatcher for multicast operations.

Functions: main(*argv): Main entry point for CLI operations.

Example: >>> import multicast.main as main >>> exit_code, _ = main.main([‘NOOP’]) >>> isinstance(exit_code, int) True

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

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

            >>> import multicast as _multicast
            >>>
            >>> import _multicast.__main__
            >>>

            >>> _multicast.__doc__ is not None
            True
            >>>

            >>> _multicast.__main__.__doc__ is not None
            True
            >>>

            >>> _multicast.__version__ is not None
            True
            >>>

    Testcase 0: multicast.__main__ should have a doctests.

            >>> import _multicast.__main__
            >>>

            >>> _multicast.__main__.__module__ is not None
            True
            >>>

            >>> _multicast.__main__.__doc__ is not None
            True
            >>>

Module Contents

Classes

McastNope

The trivial implementation of mtool.

McastRecvHearDispatch

The McastRecvHearDispatch class handles receiving and dispatching multicast messages.

McastDispatch

The McastDispatch class is the main entry point for dispatching multicast tasks.

Functions

main

Do main event stuff.

cli

Do main console script stuff.

Data

__all__

__package__

The package of this component.

__module__

The module of this component.

__file__

The file of this component.

TASK_OPTIONS

The callable function tasks of this program. will add.

API

multicast.__main__.__all__[source]

[‘McastNope’, ‘McastRecvHearDispatch’, ‘McastDispatch’, ‘main’, ‘TASK_OPTIONS’]

multicast.__main__.__package__[source]

‘multicast’

The package of this component.

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

    Testcase 0: Multicast should be importable.

            >>> import multicast
            >>>

    Testcase 1: __main__ should be automatically imported.

            >>> multicast.__main__.__package__ is not None
            True
            >>>
            >>> multicast.__main__.__package__ == multicast.__package__
            True
            >>>
multicast.__main__.__module__[source]

‘multicast.main

The module of this component.

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

    Testcase 0: Multicast should be importable.

            >>> import multicast
            >>>

    Testcase 1: __main__ should be automatically imported.

            >>> multicast.__main__.__module__ is not None
            True
            >>>
multicast.__main__.__file__[source]

‘multicast/main.py’

The file of this component.

class multicast.__main__.McastNope[source][source]

Bases: multicast.mtool

The trivial implementation of mtool.

Testing:

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

            >>> import multicast.__main__ as _multicast
            >>> _multicast.McastNope is not None
            True
            >>>

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

            >>> _multicast.McastNope is not None
            True
            >>> _multicast.McastNope is not None
            True
            >>> _multicast.McastNope.__module__ is not None
            True
            >>> _multicast.McastNope.__proc__ is not None
            True
            >>> _multicast.McastNope.__prologue__ is not None
            True
            >>>


    Testcase 2: parseArgs should return a namespace.
            A: Test that the multicast.mtool component is initialized.
            B: Test that the multicast.mtool.parseArgs component is initialized.

            >>> multicast.mtool is not None
            True
            >>> _multicast.McastNope.parseArgs is not None
            True
            >>> tst_fxtr_args = ['''NOOP''', '''--port=1234''', '''--iface=127.0.0.1''']
            >>> test_fixture = _multicast.McastNope.parseArgs(tst_fxtr_args)
            >>> test_fixture is not None
            True
            >>> type(test_fixture) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...tuple...>
            >>> tst_args_2 = ['''NOOP''', '''--junk''', '''--more-trash=stuff''']
            >>> (test_fixture_2, test_ignore_extras) = _multicast.McastNope.parseArgs(tst_args_2)
            >>> test_fixture_2 is not None
            True
            >>> type(test_fixture_2) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...Namespace...>
            >>>
__module__[source]

‘multicast.main

__name__[source]

‘multicast.main.McastNope’

__proc__[source]

‘NOOP’

__prologue__[source]

‘No Operation.’

classmethod setupArgs(parser)[source][source]

Sets up command-line arguments for the trivial subclass implementation of mtool.

This method is intended to be overridden by subclasses to define specific command-line arguments. It takes a parser object as an argument, which is typically an instance of argparse.ArgumentParser.

Args: parser (argparse.ArgumentParser): The argument parser to which the arguments should be added.

Returns: None: This method does not return a value.

Note: This is trivial implementation make this an optional abstract method. Subclasses may choose to implement it or leave it as a no-op.

static NoOp(*args, **kwargs)[source][source]

Do Nothing.

The meaning of Nothing. This function should be self-explanitory; it does ‘no operation’ i.e. nothing.

This serves as a placeholder when no specific operation is required.

Args: *args: Variable length argument list (unused). **kwargs: Arbitrary keyword arguments (unused).

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

    >>> import multicast.__main__
    >>>

Testcase 0: multicast.main should have a McastNope class.

    >>> import multicast.__main__
    >>> multicast.__main__.McastNope is not None
    True
    >>>

Testcase 1: multicast.NoOp should return None.

    >>> import multicast.__main__
    >>> multicast.__main__.McastNope.NoOp() is None
    True
    >>> multicast.__main__.McastNope.NoOp() is not None
    False
    >>>
    >>> multicast.__main__.McastNope.NoOp("Junk")
    None
    >>>
doStep(*args, **kwargs)[source][source]

Overrides the doStep method from mtool to perform no action.

This method calls the NoOp function with the provided arguments and returns the result. This serves as a placeholder or default action when no specific operation is required.

Args: *args: Positional arguments passed to NoOp. **kwargs: Keyword arguments passed to NoOp.

Args: *args: Variable length argument list (unused). **kwargs: Arbitrary keyword arguments (unused).

Returns: tuple: A “tuple” set to None.

class multicast.__main__.McastRecvHearDispatch[source][source]

Bases: multicast.mtool

The McastRecvHearDispatch class handles receiving and dispatching multicast messages.

This class listens for multicast messages on a specified group and port, and dispatches them to the appropriate handler. It is designed to work with both command-line tools and programmatic interfaces.

Testing:

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

            >>> import multicast.__main__ as _multicast
            >>> _multicast.McastNope is not None
            True
            >>>

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

            >>> multicast.__main__ is not None
            True
            >>> multicast.__main__.McastNope is not None
            True
            >>> multicast.recv.McastRECV.__module__ is not None
            True
            >>> multicast.recv.McastRECV.__proc__ is not None
            True
            >>> multicast.recv.McastRECV.__epilogue__ is not None
            True
            >>> multicast.recv.McastRECV.__prologue__ is not None
            True
            >>>


    Testcase 2: parseArgs should return a namespace.
            A: Test that the multicast.mtool component is initialized.
            B: Test that the multicast.mtool.parseArgs component is initialized.

            >>> multicast.mtool is not None
            True
            >>> _multicast.McastRecvHearDispatch.parseArgs is not None
            True
            >>> tst_fxtr_args = ['''NOOP''', '''--port=1234''', '''--iface=127.0.0.1''']
            >>> test_fixture = _multicast.McastRecvHearDispatch.parseArgs(tst_fxtr_args)
            >>> test_fixture is not None
            True
            >>> type(test_fixture) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...tuple...>
            >>> tst_args_2 = ['''NOOP''', '''--junk''', '''--more-trash=stuff''']
            >>> (test_fixture_2, t_ig_ext) = _multicast.McastRecvHearDispatch.parseArgs(tst_args_2)
            >>> test_fixture_2 is not None
            True
            >>> type(test_fixture_2) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...Namespace...>
            >>>
__module__[source]

‘multicast.main

__name__[source]

‘multicast.main.McastRecvHearDispatch’

__proc__[source]

‘HEAR’

__epilogue__ = <Multiline-String>[source]
__prologue__[source]

‘Python Multicast Receiver. Primitives for a listener for multicast data.’

classmethod setupArgs(parser)[source][source]

Will attempt to add hear args.

Both HEAR and RECV use the same arguments, and are differentiated only by the global, ‘–daemon’ argument during dispatch. The remaining arguments are:

    | Arguments | Notes |
    |-----------|------------------------------------------------------------|
    | --deamon  | Enable use of HEAR, otherwise use RECV if omitted          |
    | --group   | multicast group (ip address) to bind-to for the udp socket |
    | --groups  | multicast groups to join (should include the bind group)   |
    | --port    | The UDP port number to listen/filter on for the udp socket |

Testing:

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

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

    Testcase 1: 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 is initialized.
            D: Test that the main(hear) function returns an int 0-3.

            >>> multicast.hear is not None
            True
            >>> multicast.__main__.main is not None
            True
            >>> tst_fxtr_args = ['''HEAR''', '''--daemon''', '''--port=1234''']
            >>> (test_fixture, junk_ignore) = 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
            >>> int(test_fixture) < int(4)
            True
            >>>

    Testcase 2: setupArgs should not error given valid input.
            A: Test that the multicast component is initialized.
            B: Test that the __main__ component is initialized.
            C: Test that the McastRecvHearDispatch class is initialized.
            D: Test that the setupArgs function returns without error.

            >>> multicast.__main__ is not None
            True
            >>> multicast.__main__.McastRecvHearDispatch is not None
            True
            >>> tst_fxtr_args = argparse.ArgumentParser(prog="testcase")
            >>> multicast.__main__.McastRecvHearDispatch.setupArgs(parser=tst_fxtr_args)
            >>>

    Testcase 3: setupArgs should return None untouched.
            A: Test that the multicast component is initialized.
            B: Test that the __main__ component is initialized.
            C: Test that the McastRecvHearDispatch class is initialized.
            D: Test that the McastRecvHearDispatch.setupArgs() function yields None.

            >>> multicast.__main__ is not None
            True
            >>> multicast.__main__.McastRecvHearDispatch is not None
            True
            >>> multicast.__main__.McastRecvHearDispatch.setupArgs is not None
            True
            >>> tst_fxtr_N = None
            >>> test_fixture = multicast.__main__.McastRecvHearDispatch.setupArgs(tst_fxtr_N)
            >>> test_fixture is not None
            False
            >>> type(test_fixture) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...None...>
            >>> tst_fxtr_N == test_fixture
            True
            >>> tst_fxtr_N is None
            True
            >>>
            >>> test_fixture is None
            True
            >>>
static _help_daemon_dispatch(*args, **kwargs)[source][source]

Helps checking flags for daemon dispatching.

Internal method to check the --daemon option and interpret how it affects the dispatching of sub-commands.

Args: *args: Additional positional arguments. **kwargs: Parsed command-line arguments.

Returns: boolean: True if daemon mode is to be used, otherwise False.

doStep(*args, **kwargs)[source][source]

Executes a multicast step based on the daemon dispatch.

Overrides the doStep method from mtool to determine and execute the correct sub-command based on provided arguments.

This method selects either the McastHEAR or McastRECV class based on the daemon dispatch flag and executes the corresponding step.

The RECV (via McastRECV) is the primitive sub-command to receive a single multicast hunk. The HEAR (via McastHEAR) is equivalent to running RECV in a loop to continually receive multiple hunks. Most use-case will probably want to use HEAR instead of RECV.

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

Returns: tuple: The result of the dispatched sub-command’s doStep method.

multicast.__main__.TASK_OPTIONS[source]

None

The callable function tasks of this program. will add.

class multicast.__main__.McastDispatch[source][source]

Bases: multicast.mtool

The McastDispatch class is the main entry point for dispatching multicast tasks.

It provides a command-line interface for sending, receiving, and listening to multicast messages. The class handles argument parsing and dispatches the appropriate multicast tool based on the provided command.

__proc__: str[source]

‘multicast’

__prologue__: str[source]

‘The Main Entrypoint.’

__epilogue__: str[source]

‘Called from the command line, this main component handles the CLI dispatch.’

classmethod setupArgs(parser) None[source][source]

Will attempt to add each subcomponent’s args.

As the dispatch tool, this is more of a proxy implementation to call the sub-components own setupArgs.

static useTool(tool, **kwargs) tuple[source][source]

Will Handle launching the actual task functions.

doStep(*args, **kwargs) tuple[source][source]

Executes the multicast tool based on parsed arguments.

This method parses the command-line arguments, selects the appropriate multicast tool, and executes it. If an error occurs during argument handling, it prints a warning message.

Args: *args: Command-line arguments for the multicast tool.

Returns: A tuple containing the exit status and the result of the tool execution.

multicast.__main__.main(*argv) tuple[source][source]

Do main event stuff.

Executes the multicast command-line interface, by parsing command-line arguments and dispatching the appropriate multicast operations.

The main(*args) function in multicast is expected to return a POSIX compatible exit code and optional detail message. Regardless of errors the result as an ‘exit code’ (int) is returned, even if the optional details are not (e.g., tuple(int(exit_code), None)). The only exception is when the error results in exiting the process, which will exit the python runtime with the underlying return codes, instead of returning directly to the then unreachable caller. See multicast.exceptions.exit_on_exception for the mechanisms involved. The expected return codes are as follows: = 0: Any nominal state (i.e. no errors and possibly success) ≥ 1: Any erroneous state (includes simple failure) > 2: Any failed state < 0: Implicitly erroneous and treated the same as abs(exit_code) would be.

Args: *argv: the array of arguments. Usually sys.argv[1:]

Returns: tuple: the underlying exit code int, and optional detail string.

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

            >>> import multicast
            >>> multicast.send is not None
            True
            >>>

    Testcase 0: main should return an int.
            A: Test that the multicast component is initialized.
            B: Test that the send component is initialized.
            C: Test that the send.main function is initialized.
            D: Test that the send.main function returns an int 0-3.

            >>> multicast.send is not None
            True
            >>> multicast.__main__.main is not None
            True
            >>> tst_fxtr_args = ['''SAY''', '''--port=1234''', '''--message''', '''is required''']
            >>> (test_fixture, junk_ignore) = 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
            >>> int(test_fixture) < int(4)
            True
            >>>


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

            >>> multicast.recv is not None
            True
            >>> multicast.__main__.main is not None
            True
            >>> tst_fxtr_args = ['''RECV''', '''--port=1234''', '''--group''', '''224.0.0.1''']
            >>> (test_fixture, junk_ignore) = 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
            >>> int(test_fixture) < int(4)
            True
            >>>


    Testcase 2: main should error with usage.
            A: Test that the multicast component is initialized.
            B: Test that the recv component is initialized.
            C: Test that the main(recv) function is initialized.
            D: Test that the main(recv) function errors with a usage hint by default.

            >>> multicast.recv is not None
            True
            >>> multicast.__main__.main is not None
            True
            >>> (test_fixture, junk_ignore) = multicast.__main__.main() #doctest: +ELLIPSIS
            usage: multicast [-h | -V] [--use-std] [--daemon] CMD ...
            multicast...
            CRITICAL...
            >>> type(test_fixture) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...int...>
            >>> int(test_fixture) >= int(0)
            True
            >>> int(test_fixture) < int(4)
            True
            >>> type(junk_ignore)
            <...str...>
            >>> junk_ignore in "STOP"
            True
            >>>
multicast.__main__.cli() int[source][source]

Do main console script stuff.

Along with main(), cli() provides a main entrypoint for console usage of multicast.

cli() just calls main() with arguments from sys.argv and returns only the exit-code.

Through calling multicast.__main__.main(sys.argv[1:]), cli()

Executes the multicast command-line interface, by parsing command-line arguments and dispatching the appropriate multicast operations.

The main(*args) function in multicast is expected to return a POSIX compatible exit code…

cli() versus main(*args): - The primary difference is the return types, whereas main(*args) returns a tuple, cli() returns only the first element as an int. - The secondary difference between cli() and main(*args) is that main(*args) requires arguments to be passed, whereas cli() will use sys.argv instead.

Except in the case of errors, the result as an ‘exit code’ (int) is returned by cli(). The expected return codes are the same as those from main(*args).

Args: None: Uses sys.argv instead.

Returns: int: the underlying exit code int

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

            >>> import multicast
            >>> multicast.__main__.main is not None
            True
            >>>

    Testcase 0: calls to cli should return an int.
            A: Test that the call to the `cli` function returns an int 0-3.

            >>> tst_argv_args = ['''SAY''', '''--port=1234''', '''--message''', '''is required''']
            >>> sys.argv = tst_argv_args  # normally arguments are automatically already in argv
            >>> test_code = multicast.__main__.cli()
            >>> test_code is not None
            True
            >>> type(test_code) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...int...>
            >>> int(test_code) >= int(0)
            True
            >>> int(test_code) < int(4)
            True
            >>>

    Testcase 1: main should error with usage.
            A: Test that the multicast component is initialized.
            B: Test that the recv component is initialized.
            C: Test that the main(recv) function is initialized.
            D: Test that the main(recv) function errors with a usage hint by default.

            >>> multicast.__main__.main is not None
            True
            >>> test_code = multicast.__main__.cli() #doctest: +ELLIPSIS
            usage: multicast [-h | -V] [--use-std] [--daemon] CMD ...
            multicast...
            CRITICAL...
            >>> type(test_code) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...int...>
            >>> int(test_code) >= int(0)
            True
            >>> int(test_code) < int(4)
            True
            >>>