diff --git a/.deepsource.toml b/.deepsource.toml index fb7fdfd3..21e0a085 100644 --- a/.deepsource.toml +++ b/.deepsource.toml @@ -32,6 +32,7 @@ enabled = true dependency_file_paths = [ "requirements.txt", "tests/requirements.txt", + "docs/requirements.txt", "setup.py" ] diff --git a/README.md b/README.md index 73330d78..ceac51a9 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,8 @@ Next-steps and bug-fixes are tracked [Here](https://github.com/users/reactive-fi ## License +### Copyright (c) 2021-2024, Mr. Walls + This project is licensed under the MIT License. See the [LICENSE.md](LICENSE.md) file for details. [![License - MIT](https://img.shields.io/github/license/reactive-firewall/multicast.svg?maxAge=3600)](https://github.com/reactive-firewall/multicast/blob/stable/LICENSE.md) diff --git a/docs/FAQ.md b/docs/FAQ.md index 80861c35..581a7366 100644 --- a/docs/FAQ.md +++ b/docs/FAQ.md @@ -62,7 +62,7 @@ python3 -m multicast SAY --mcast-group 224.1.1.2 --port 59595 --message "Hello W [Here is how it is tested right now](https://github.com/reactive-firewall/multicast/blob/cdd577549c0bf7c2bcf85d1b857c86135778a9ed/tests/test_usage.py#L251-L554) ```python3 -import multicast as multicast +import multicast from multiprocessing import Process as Process # set up some stuff diff --git a/docs/USAGE.md b/docs/USAGE.md index aa71910a..318df865 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -51,7 +51,7 @@ p.start() and elsewhere (like another function or even module) for the sender: ```python3 -# assuming already did 'import multicast as multicast' +# assuming already did 'import multicast' _fixture_SAY_args = [ """--port""", _fixture_PORT_arg, diff --git a/docs/conf.py b/docs/conf.py index c2449721..a5f909e9 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -239,7 +239,7 @@ # see https://myst-parser.readthedocs.io/en/latest/syntax/roles-and-directives.html#syntax-directives # be more like GFM with style -myst_enable_extensions = set(["tasklist", "strikethrough", "fieldlist"]) +myst_enable_extensions = ("tasklist", "strikethrough", "fieldlist") # for GFM diagrams and interoperability with other Markdown renderers myst_fence_as_directive = ("mermaid", "suggestion", "note") diff --git a/multicast/__init__.py b/multicast/__init__.py index 855bf7e6..7a671aac 100644 --- a/multicast/__init__.py +++ b/multicast/__init__.py @@ -105,6 +105,16 @@ __doc__ = __prologue__ + """ + Dynamic Imports: + The sub-modules within "multicast" are interdependent, requiring access to each other's + functionalities. These statements import sub-modules of "multicast" and assign them to + aliases that match their sub-module names, facilitating organized access to these + components. + While the multicast alias is the same as the multicast module name, this pattern should + serve to reinforce the Multicast module's namespace, especially when dealing with dynamic + imports and to maintain consistency across different parts of the code. + + Minimal Acceptance Testing: First set up test fixtures by importing multicast. @@ -564,12 +574,12 @@ def doStep(self, *args): # pragma: no cover try: if 'multicast.skt' not in sys.modules: - from . import skt + from . import skt as skt # pylint: disable=cyclic-import - skipcq: PLY-R0401, PYL-C0414 else: # pragma: no branch skt = sys.modules["""multicast.skt"""] except Exception as importErr: del importErr - import multicast.skt as skt + import multicast.skt as skt # pylint: disable=cyclic-import - skipcq: PLY-R0401 genSocket = skt.genSocket @@ -580,39 +590,39 @@ def doStep(self, *args): # pragma: no cover try: if 'multicast.recv' not in sys.modules: - from . import recv as recv + from . import recv as recv # pylint: disable=cyclic-import - skipcq: PLY-R0401 else: # pragma: no branch recv = sys.modules["""multicast.recv"""] except Exception as importErr: del importErr - import multicast.recv as recv + import multicast.recv as recv # pylint: disable=cyclic-import - skipcq: PLY-R0401 try: if 'multicast.send' not in sys.modules: - from . import send as send + from . import send as send # pylint: disable=cyclic-import - skipcq: PLY-R0401 else: # pragma: no branch send = sys.modules["""multicast.send"""] except Exception as importErr: del importErr - import multicast.send as send + import multicast.send as send # pylint: disable=cyclic-import - skipcq: PLY-R0401 try: if 'multicast.hear' not in sys.modules: - from . import hear as hear + from . import hear as hear # pylint: disable=cyclic-import - skipcq: PLY-R0401 else: # pragma: no branch hear = sys.modules["""multicast.hear"""] except Exception as importErr: del importErr - import multicast.hear as hear + import multicast.hear as hear # pylint: disable=cyclic-import - skipcq: PLY-R0401 try: if """multicast.__main__""" in sys.modules: # pragma: no cover __main__ = sys.modules["""multicast.__main__"""] except Exception: - import multicast.__main__ as __main__ + import multicast.__main__ as __main__ # pylint: disable=cyclic-import - skipcq: PLY-R0401 if __name__ in u'__main__': diff --git a/multicast/__main__.py b/multicast/__main__.py index b3ad074f..f3b7e667 100644 --- a/multicast/__main__.py +++ b/multicast/__main__.py @@ -19,37 +19,39 @@ """The Main Entrypoint. +Caution: See details regarding dynamic imports [documented](../__init__.py) in this module. + Minimal Acceptance Testing: First set up test fixtures by importing multicast. - >>> import multicast as multicast + >>> import multicast as _multicast >>> - >>> import multicast.__main__ + >>> import _multicast.__main__ >>> - >>> multicast.__doc__ is not None + >>> _multicast.__doc__ is not None True >>> - >>> multicast.__main__.__doc__ is not None + >>> _multicast.__main__.__doc__ is not None True >>> - >>> multicast.__version__ is not None + >>> _multicast.__version__ is not None True >>> Testcase 0: multicast.__main__ should have a doctests. - >>> import multicast.__main__ + >>> import _multicast.__main__ >>> - >>> multicast.__main__.__module__ is not None + >>> _multicast.__main__.__module__ is not None True >>> - >>> multicast.__main__.__doc__ is not None + >>> _multicast.__main__.__doc__ is not None True >>> @@ -76,80 +78,80 @@ try: - from . import sys as sys # skipcq: PYL-C0414 + from . import sys as _sys except Exception: # Throw more relevant Error raise ImportError(str("[CWE-440] Error Importing Python")) try: - if 'multicast.__version__' not in sys.modules: + if 'multicast.__version__' not in _sys.modules: from . import __version__ as __version__ # skipcq: PYL-C0414 else: # pragma: no branch - __version__ = sys.modules["""multicast.__version__"""] + __version__ = _sys.modules["""multicast.__version__"""] except Exception as importErr: del importErr - import multicast.__version__ as __version__ # noqa - used by --version argument. + import multicast.__version__ as __version__ # noqa. skipcq - used by --version argument. try: - if 'multicast._MCAST_DEFAULT_PORT' not in sys.modules: + if 'multicast._MCAST_DEFAULT_PORT' not in _sys.modules: from . import _MCAST_DEFAULT_PORT as _MCAST_DEFAULT_PORT # skipcq: PYL-C0414 else: # pragma: no branch - _MCAST_DEFAULT_PORT = sys.modules["""multicast._MCAST_DEFAULT_PORT"""] + _MCAST_DEFAULT_PORT = _sys.modules["""multicast._MCAST_DEFAULT_PORT"""] except Exception as importErr: del importErr - import multicast._MCAST_DEFAULT_PORT as _MCAST_DEFAULT_PORT + import multicast._MCAST_DEFAULT_PORT as _MCAST_DEFAULT_PORT # skipcq - used by port argument. try: - if 'multicast._MCAST_DEFAULT_GROUP' not in sys.modules: + if 'multicast._MCAST_DEFAULT_GROUP' not in _sys.modules: from . import _MCAST_DEFAULT_GROUP as _MCAST_DEFAULT_GROUP # skipcq: PYL-C0414 else: # pragma: no branch - _MCAST_DEFAULT_GROUP = sys.modules["""multicast._MCAST_DEFAULT_GROUP"""] + _MCAST_DEFAULT_GROUP = _sys.modules["""multicast._MCAST_DEFAULT_GROUP"""] except Exception as importErr: del importErr - import multicast._MCAST_DEFAULT_GROUP as _MCAST_DEFAULT_GROUP + import multicast._MCAST_DEFAULT_GROUP as _MCAST_DEFAULT_GROUP # skipcq - used by group arg. try: - if 'multicast.mtool' not in sys.modules: + if 'multicast.mtool' not in _sys.modules: from . import mtool as mtool # skipcq: PYL-C0414 else: # pragma: no branch - mtool = sys.modules["""multicast.mtool"""] + mtool = _sys.modules["""multicast.mtool"""] except Exception as importErr: del importErr - import multicast.mtool as mtool + import multicast.mtool as mtool # noqa - used by all arguments' CMD (sub-command). try: - if 'multicast.recv' not in sys.modules: - from . import recv as recv # skipcq: PYL-C0414 + if 'multicast.recv' not in _sys.modules: + from . import recv as recv # pylint: disable=useless-import-alias - skipcq: PYL-C0414 else: # pragma: no branch - recv = sys.modules["""multicast.recv"""] + recv = _sys.modules["""multicast.recv"""] except Exception as importErr: del importErr - import multicast.recv as recv + import multicast.recv as recv # pylint: disable=useless-import-alias - skipcq: PYL-C0414 try: - if 'multicast.send' not in sys.modules: - from . import send as send # skipcq: PYL-C0414 + if 'multicast.send' not in _sys.modules: + from . import send as send # pylint: disable=useless-import-alias - skipcq: PYL-C0414 else: # pragma: no branch - send = sys.modules["""multicast.send"""] + send = _sys.modules["""multicast.send"""] except Exception as importErr: del importErr - import multicast.send as send + import multicast.send as send # pylint: disable=useless-import-alias - skipcq: PYL-C0414 try: - if 'multicast.hear' not in sys.modules: - from . import hear as hear # skipcq: PYL-C0414 + if 'multicast.hear' not in _sys.modules: + from . import hear as hear # pylint: disable=useless-import-alias - skipcq: PYL-C0414 else: # pragma: no branch - hear = sys.modules["""multicast.hear"""] + hear = _sys.modules["""multicast.hear"""] except Exception as importErr: del importErr - import multicast.hear as hear + import multicast.hear as hear # pylint: disable=useless-import-alias - skipcq: PYL-C0414 class McastNope(mtool): @@ -447,12 +449,12 @@ def doStep(self, *args, **kwargs): # More boiler-plate-code -TASK_OPTIONS = dict({ - 'NOOP': McastNope(), - 'RECV': McastRecvHearDispatch(), - 'SAY': send.McastSAY(), - 'HEAR': McastRecvHearDispatch(), -}) +TASK_OPTIONS = { + """NOOP""": McastNope(), + """RECV""": McastRecvHearDispatch(), + """SAY""": send.McastSAY(), + """HEAR""": McastRecvHearDispatch(), +} """The callable function tasks of this program. will add.""" @@ -501,13 +503,13 @@ def doStep(self, *args): _TOOL_MSG = (self.useTool(service_cmd, **argz.__dict__)) if _TOOL_MSG[0]: __EXIT_MSG = (0, _TOOL_MSG) - elif (sys.stdout.isatty()): # pragma: no cover + elif (_sys.stdout.isatty()): # pragma: no cover print(_TOOL_MSG) except Exception as inerr: # pragma: no branch w = str("WARNING - An error occurred while") w += str(" handling the arguments.") w += str(" Refused.") - if (sys.stdout.isatty()): # pragma: no cover + if (_sys.stdout.isatty()): # pragma: no cover print(w) print(str(inerr)) print(str(inerr.args)) @@ -516,7 +518,7 @@ def doStep(self, *args): except BaseException: # pragma: no branch e = str("CRITICAL - An error occurred while handling") e += str(" the dispatch.") - if (sys.stdout.isatty()): # pragma: no cover + if (_sys.stdout.isatty()): # pragma: no cover print(str(e)) __EXIT_MSG = (3, "STOP") return __EXIT_MSG # noqa @@ -536,7 +538,7 @@ def main(*argv): = 3: Any undefined (but assumed erroneous) state > 0: implicitly erroneous and treated same as abs(exit_code) would be. - param iterable - argv - the array of arguments. Usually sys.argv[1:] + param iterable - argv - the array of arguments. Usually _sys.argv[1:] returns int - the Namespace parsed with the key-value pairs. Minimal Acceptance Testing: @@ -601,8 +603,8 @@ def main(*argv): if __name__ in '__main__': __EXIT_CODE = (2, "NoOp") - if (sys.argv is not None) and (len(sys.argv) > 1): - __EXIT_CODE = main(sys.argv[1:]) - elif (sys.argv is not None): + if (_sys.argv is not None) and (len(_sys.argv) > 1): + __EXIT_CODE = main(_sys.argv[1:]) + elif (_sys.argv is not None): __EXIT_CODE = main([str(__name__), """-h"""]) exit(__EXIT_CODE[0]) diff --git a/multicast/hear.py b/multicast/hear.py index a4adc51a..e01a83c4 100644 --- a/multicast/hear.py +++ b/multicast/hear.py @@ -19,6 +19,8 @@ """Provides multicast HEAR Features. +Caution: See details regarding dynamic imports [documented](../__init__.py) in this module. + Minimal Acceptance Testing: First set up test fixtures by importing multicast. @@ -160,28 +162,28 @@ """ try: - import sys - if 'multicast' not in sys.modules: - from . import multicast as multicast # skipcq: PYL-C0414 + import sys as _sys + if 'multicast' not in _sys.modules: + from . import multicast as multicast # pylint: disable=cyclic-import - skipcq: PLY-C0414 else: # pragma: no branch - multicast = sys.modules["""multicast"""] + multicast = _sys.modules["""multicast"""] _BLANK = multicast._BLANK - from . import recv as recv # skipcq: PYL-C0414 - from . import send as send # skipcq: PYL-C0414 + from . import recv as recv # pylint: disable=useless-import-alias - skipcq: PYL-C0414 + from . import send as send # pylint: disable=useless-import-alias - skipcq: PYL-C0414 except Exception as importErr: del importErr - import multicast as multicast # skipcq: PYL-C0414 + import multicast as multicast # pylint: disable=cyclic-import - skipcq: PLY-R0401, PYL-C0414 try: import socketserver - from socketserver import threading as threading # skipcq: PYL-C0414 - from multicast import argparse as argparse # skipcq: PYL-C0414 - from multicast import unicodedata as unicodedata # skipcq: PYL-C0414 - from multicast import socket as socket # skipcq: PYL-C0414 - from multicast import struct as struct # skipcq: PYL-C0414 + import threading as _threading + from multicast import argparse as _argparse + from multicast import unicodedata as _unicodedata + from multicast import socket as _socket + from multicast import struct as _struct depends = [ - unicodedata, socket, struct, argparse + _unicodedata, _socket, _struct, _argparse ] for unit in depends: try: @@ -249,7 +251,7 @@ def handle_error(self, request, client_address): def kill_func(a_server): if a_server is not None: a_server.shutdown() - end_thread = threading.Thread(name="Kill_Thread", target=kill_func, args=[self]) + end_thread = _threading.Thread(name="Kill_Thread", target=kill_func, args=[self]) end_thread.start() super(McastServer, self).handle_error(request, client_address) diff --git a/multicast/recv.py b/multicast/recv.py index ec8adb7d..97a3ead2 100644 --- a/multicast/recv.py +++ b/multicast/recv.py @@ -32,6 +32,8 @@ """multicast RECV Features. +Caution: See details regarding dynamic imports [documented](../__init__.py) in this module. + Minimal Acceptance Testing: First set up test fixtures by importing multicast. @@ -175,22 +177,22 @@ try: import sys if 'multicast' not in sys.modules: - from . import multicast as multicast # skipcq: PYL-C0414 + from . import multicast as multicast # pylint: disable=cyclic-import - skipcq: PYL-C0414 else: # pragma: no branch multicast = sys.modules["""multicast"""] _BLANK = multicast._BLANK except Exception as importErr: del importErr - import multicast as multicast # skipcq: PYL-C0414 + import multicast as multicast # pylint: disable=cyclic-import - skipcq: PLY-R0401, PYL-C0414 try: - from multicast import argparse as argparse # skipcq: PYL-C0414 - from multicast import unicodedata as unicodedata # skipcq: PYL-C0414 - from multicast import socket as socket # skipcq: PYL-C0414 - from multicast import struct as struct # skipcq: PYL-C0414 + from multicast import argparse as _argparse + from multicast import unicodedata as _unicodedata + from multicast import socket as _socket + from multicast import struct as _struct depends = [ - unicodedata, socket, struct, argparse + _unicodedata, _socket, _struct, _argparse ] for unit in depends: if unit.__name__ is None: # pragma: no branch @@ -261,12 +263,12 @@ def joinstep(groups, port, iface=None, bind_group=None, isock=None): try: sock.bind(('224.0.0.1' if bind_group is None else bind_group, port)) for group in groups: - mreq = struct.pack( + mreq = _struct.pack( '4sl' if iface is None else '4s4s', - socket.inet_aton(group), - socket.INADDR_ANY if iface is None else socket.inet_aton(iface) + _socket.inet_aton(group), + _socket.INADDR_ANY if iface is None else _socket.inet_aton(iface) ) - sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + sock.setsockopt(_socket.IPPROTO_IP, _socket.IP_ADD_MEMBERSHIP, mreq) except Exception as err: # pragma: no branch raise NotImplementedError("""[CWE-440] Not Implemented.""", err) # pragma: no cover return sock @@ -397,6 +399,7 @@ def setupArgs(cls, parser): Testcase 0: First set up test fixtures by importing multicast. >>> import multicast + >>> import argparse as _argparse >>> multicast.recv is not None True >>> multicast.recv.McastRECV is not None @@ -435,7 +438,7 @@ def setupArgs(cls, parser): True >>> multicast.__main__.main is not None True - >>> tst_fxtr_args = argparse.ArgumentParser(prog="testcase") + >>> tst_fxtr_args = _argparse.ArgumentParser(prog="testcase") >>> multicast.recv.McastRECV.setupArgs(parser=tst_fxtr_args) >>> @@ -527,5 +530,5 @@ def doStep(self, *args, **kwargs): print(str(response)) print(multicast._BLANK) _result = (len(response) > 0) is True - return tuple((_result, None if not _result else response)) + return tuple((_result, None if not _result else response)) # skipcq: PTC-W0020 - intended diff --git a/multicast/send.py b/multicast/send.py index b37ba1d5..94036967 100644 --- a/multicast/send.py +++ b/multicast/send.py @@ -33,6 +33,8 @@ """Python Multicast Broadcaster. +Caution: See details regarding dynamic imports [documented](../__init__.py) in this module. + Minimal Acceptance Testing: First set up test fixtures by importing multicast. @@ -138,22 +140,22 @@ try: import sys if 'multicast' not in sys.modules: - from . import multicast as multicast # skipcq: PYL-C0414 + from . import multicast as multicast # pylint: disable=cyclic-import - skipcq: PYL-C0414 else: # pragma: no branch multicast = sys.modules["""multicast"""] _BLANK = multicast._BLANK except Exception as importErr: del importErr - import multicast as multicast + import multicast as multicast # pylint: disable=cyclic-import - skipcq: PLY-R0401, PYL-C0414 try: - from multicast import argparse as argparse # skipcq: PYL-C0414 - from multicast import unicodedata as unicodedata # skipcq: PYL-C0414 - from multicast import socket as socket # skipcq: PYL-C0414 - from multicast import struct as struct # skipcq: PYL-C0414 + from multicast import argparse as _argparse # skipcq: PYL-C0414 + from multicast import unicodedata as _unicodedata # skipcq: PYL-C0414 + from multicast import socket as _socket # skipcq: PYL-C0414 + from multicast import struct as _struct # skipcq: PYL-C0414 depends = [ - unicodedata, socket, struct, argparse + _unicodedata, _socket, _struct, _argparse ] for unit in depends: try: @@ -291,9 +293,12 @@ def _sayStep(group, port, data): The actual magic is handled here. """ - sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM, _socket.IPPROTO_UDP) try: - sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, multicast._MCAST_DEFAULT_TTL) + sock.setsockopt( + _socket.IPPROTO_IP, _socket.IP_MULTICAST_TTL, + multicast._MCAST_DEFAULT_TTL # skipcq: PYL-W0212 - use of module-wide var. + ) sock.sendto(data.encode('utf8'), (group, port)) finally: multicast.endSocket(sock) diff --git a/multicast/skt.py b/multicast/skt.py index 7c4f2960..8c92a9da 100644 --- a/multicast/skt.py +++ b/multicast/skt.py @@ -34,6 +34,8 @@ NOT intended for DIRECT use! +Caution: See details regarding dynamic imports [documented](../__init__.py) in this module. + Minimal Acceptance Testing: First set up test fixtures by importing multicast. diff --git a/tests/MulticastUDPClient.py b/tests/MulticastUDPClient.py index 29cb3ad4..775a7b54 100644 --- a/tests/MulticastUDPClient.py +++ b/tests/MulticastUDPClient.py @@ -206,7 +206,8 @@ def __init__(self, *args, **kwargs): ) ) - def say(self, address, port, conn, msg): + @staticmethod + def say(address, port, conn, msg): """Send a message to a specified multicast address and port, then receive and print the response. @@ -260,8 +261,8 @@ def say(self, address, port, conn, msg): """ conn.sendto(bytes(msg + "\n", "utf-8"), (address, port)) received = str(conn.recv(1024), "utf-8") - print("Sent: {}".format(msg)) - print("Received: {}".format(received)) + print(str("Sent: {}").format(msg)) # skipcq: PYL-C0209 - must remain compatible + print(str("Received: {}").format(received)) # skipcq: PYL-C0209 - must remain compatible def main(): diff --git a/tests/__init__.py b/tests/__init__.py index 96cedc51..7c98c4db 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -19,6 +19,12 @@ """Multicast Testing Module. + Robust imports: These statements import the entire "multicast" module, + allowing access to all its functionalities within the test environment. + This may be flagged as an intentional cyclic import by pylint. + See warning about cyclic-imports + [here](https://pylint.pycqa.org/en/latest/user_guide/messages/refactor/cyclic-import.html) + Testing: Testcase 0: Load tests fixtures @@ -72,7 +78,7 @@ try: if 'multicast' not in sys.modules: - import multicast + import multicast # pylint: disable=cyclic-import - skipcq: PLY-R0401 else: # pragma: no branch multicast = sys.modules["""multicast"""] except Exception: # pragma: no branch diff --git a/tests/context.py b/tests/context.py index 81a81563..76deb4d1 100644 --- a/tests/context.py +++ b/tests/context.py @@ -22,6 +22,13 @@ __name__ = """tests.context""" # skipcq: PYL-W0622 __doc__ = """ + + Robust imports: These statements import the entire "multicast" module, + allowing access to all its functionalities within the test environment. + This can be flagged as an intentional + [cyclic-import](https://pylint.pycqa.org/en/latest/user_guide/messages/refactor/cyclic-import.html) + warning. + Context for Testing. Meta Tests - Fixtures: @@ -91,9 +98,9 @@ try: if 'multicast' not in sys.modules: - import multicast + import multicast # pylint: disable=cyclic-import - skipcq: PLY-R0401 else: # pragma: no branch - multicast = sys.modules["""multicast"""] + multicast = sys.modules["""multicast"""] # pylint: disable=cyclic-import except Exception: # pragma: no branch raise ImportError("[CWE-440] Python Multicast Repo Failed to import.") @@ -144,13 +151,12 @@ def getCoverageCommand(): First set up test fixtures by importing test context. - >>> import tests.context as context + >>> import tests.context as _context >>> Testcase 1: function should have a output. - >>> import tests.context as context - >>> context.getCoverageCommand() is None + >>> _context.getCoverageCommand() is None False >>> @@ -180,18 +186,18 @@ def __check_cov_before_py(): First set up test fixtures by importing test context. - >>> import tests.context + >>> import tests.context as _context >>> Testcase 1: function should have a output. - >>> tests.context.__check_cov_before_py() is not None + >>> _context.__check_cov_before_py() is not None True >>> Testcase 2: function should have a string output of python or coverage. - >>> _test_fixture = tests.context.__check_cov_before_py() + >>> _test_fixture = _context.__check_cov_before_py() >>> isinstance(_test_fixture, type(str(""))) True >>> (str("python") in _test_fixture) or (str("coverage") in _test_fixture) @@ -224,12 +230,12 @@ def getPythonCommand(): First set up test fixtures by importing test context. - >>> import tests.context + >>> import tests.context as _context >>> Testcase 1: function should have a output. - >>> tests.context.getPythonCommand() is not None + >>> _context.getPythonCommand() is not None True >>> @@ -265,36 +271,36 @@ def checkCovCommand(args=[None]): First set up test fixtures by importing test context. - >>> import tests.context as context + >>> import tests.context as _context >>> Testcase 1: Function should return unmodified arguments if 'coverage' is missing. - >>> context.checkCovCommand(["python", "script.py"]) + >>> _context.checkCovCommand(["python", "script.py"]) ['python', 'script.py'] Testcase 2: Function should modify arguments when 'coverage' is the first argument. A.) Missing 'run' - >>> checkCovCommand(["coverage", "script.py"]) #doctest: +ELLIPSIS + >>> _context.checkCovCommand(["coverage", "script.py"]) #doctest: +ELLIPSIS ['...', 'run', '-p', '--context=Integration', '--source=multicast', 'script.py'] Testcase 3: Function should modify arguments when 'coverage run' is in the first argument. A.) NOT missing 'run' - >>> checkCovCommand(["coverage run", "script.py"]) #doctest: +ELLIPSIS + >>> _context.checkCovCommand(["coverage run", "script.py"]) #doctest: +ELLIPSIS ['...', 'run', '-p', '--context=Integration', '--source=multicast', 'script.py'] Testcase 4: Function should handle coverage command with full path. - >>> checkCovCommand(["/usr/bin/coverage", "test.py"]) #doctest: +ELLIPSIS + >>> _context.checkCovCommand(["/usr/bin/coverage", "test.py"]) #doctest: +ELLIPSIS ['...', 'run', '-p', '--context=Integration', '--source=multicast', 'test.py'] Testcase 5: Function should handle coverage invoked via sys.executable. - >>> import sys - >>> test_fixture = [str("{} -m coverage run").format(sys.executable), "test.py"] - >>> context.checkCovCommand(test_fixture) #doctest: +ELLIPSIS + >>> import sys as _sys + >>> test_fixture = [str("{} -m coverage run").format(_sys.executable), "test.py"] + >>> _context.checkCovCommand(test_fixture) #doctest: +ELLIPSIS [..., '-m', 'coverage', 'run', '-p', '...', '--source=multicast', 'test.py'] @@ -418,38 +424,39 @@ def checkPythonCommand(args, stderr=None): First set up test fixtures by importing test context. - >>> import tests.context + >>> import sys as _sys + >>> import tests.context as _context >>> Testcase 1: Function should have an output when provided valid arguments. - >>> test_fixture_1 = [str(sys.executable), '-c', 'print("Hello, World!")'] - >>> tests.context.checkPythonCommand(test_fixture_1) + >>> test_fixture_1 = [str(_sys.executable), '-c', 'print("Hello, World!")'] + >>> _context.checkPythonCommand(test_fixture_1) 'Hello, World!\\n' Testcase 2: Function should capture stderr when specified. - >>> import subprocess + >>> import subprocess as _subprocess >>> test_args_2 = [ - ... str(sys.executable), '-c', 'import sys; print("Error", file=sys.stderr)' + ... str(_sys.executable), '-c', 'import sys; print("Error", file=sys.stderr)' ... ] >>> - >>> tests.context.checkPythonCommand(test_args_2, stderr=subprocess.STDOUT) + >>> _context.checkPythonCommand(test_args_2, stderr=_subprocess.STDOUT) 'Error\\n' Testcase 3: Function should handle exceptions and return output. - >>> test_fixture_e = [str(sys.executable), '-c', 'raise ValueError("Test error")'] - >>> tests.context.checkPythonCommand( - ... test_fixture_e, stderr=subprocess.STDOUT + >>> test_fixture_e = [str(_sys.executable), '-c', 'raise ValueError("Test error")'] + >>> _context.checkPythonCommand( + ... test_fixture_e, stderr=_subprocess.STDOUT ... ) #doctest: +ELLIPSIS 'Traceback (most recent call last):\\n...ValueError...' Testcase 4: Function should return the output as a string. - >>> test_fixture_s = [str(sys.executable), '-c', 'print(b"Bytes output")'] - >>> isinstance(tests.context.checkPythonCommand( - ... test_fixture_s, stderr=subprocess.STDOUT + >>> test_fixture_s = [str(_sys.executable), '-c', 'print(b"Bytes output")'] + >>> isinstance(_context.checkPythonCommand( + ... test_fixture_s, stderr=_subprocess.STDOUT ... ), str) True @@ -507,7 +514,7 @@ def debugBlob(blob=None): First set up test fixtures by importing test context. - >>> import tests.context + >>> import tests.context as _context >>> >>> norm_fixture = "Example Sample" @@ -573,7 +580,7 @@ def debugtestError(someError): First set up test fixtures by importing test context. - >>> import tests.context + >>> import tests.context as _context >>> >>> err_fixture = RuntimeError(\"Example Error\") @@ -648,7 +655,7 @@ def debugUnexpectedOutput(expectedOutput, actualOutput, thepython): First set up test fixtures by importing test context. - >>> import tests.context + >>> import tests.context as _context >>> >>> expected_fixture = "" @@ -658,7 +665,7 @@ def debugUnexpectedOutput(expectedOutput, actualOutput, thepython): Testcase 1: function should have a output. - >>> tests.context.debugUnexpectedOutput( + >>> _context.debugUnexpectedOutput( ... expected_fixture, unexpected_fixture, python_fixture ... ) #doctest: -DONT_ACCEPT_BLANKLINE @@ -676,7 +683,7 @@ def debugUnexpectedOutput(expectedOutput, actualOutput, thepython): Testcase 2: function should have a output even with bad input. - >>> tests.context.debugUnexpectedOutput( + >>> _context.debugUnexpectedOutput( ... expected_fixture, unexpected_fixture, None ... ) #doctest: -DONT_ACCEPT_BLANKLINE @@ -718,10 +725,10 @@ class BasicUsageTestSuite(unittest.TestCase): First set up test fixtures by importing test context. - >>> import tests.context + >>> import tests.context as _context >>> - >>> class TestCaseFixture(tests.context.BasicUsageTestSuite): + >>> class TestCaseFixture(_context.BasicUsageTestSuite): ... pass >>> >>> TestCaseFixture diff --git a/tests/test_usage.py b/tests/test_usage.py index e14ac26c..9bbd9cbe 100644 --- a/tests/test_usage.py +++ b/tests/test_usage.py @@ -21,6 +21,7 @@ """ Tests of integration by usage. + Caution: See details about Robust Imports documented in tests.context. Meta tests.test_usage.BasicIntegrationTestSuite @@ -64,10 +65,11 @@ if context.__name__ is None: raise ImportError("[CWE-758] Failed to import context") else: - from context import multicast as multicast # skipcq: PYL-C0414 + from context import multicast # pylint: disable=cyclic-import - skipcq: PLY-R0401 from context import unittest from context import subprocess from context import Process + import random as _random except Exception: raise ImportError("[CWE-758] Failed to import test context") @@ -139,7 +141,7 @@ def test_aborts_WHEN_calling_multicast_GIVEN_invalid_tools(self): self.assertIsNone(test_fixture) self.assertTupleEqual( tst_dispatch.useTool(tst_in), - tuple((False, None)), + tuple((False, None)), # skipcq: PTC-W0020 - This is test-code. fail_fixture ) theResult = True @@ -253,7 +255,7 @@ def test_hear_is_stable_WHEN_calling_multicast_GIVEN_invalid_tool(self): try: self.assertTupleEqual( multicast.__main__.main(["HEAR", "--hex"]), - tuple((0, (True, (False, None)))) + tuple((0, (True, (False, None)))) # skipcq: PTC-W0020 - This is test-code. ) theResult = True except Exception as err: @@ -268,10 +270,10 @@ def test_noop_stable_WHEN_calling_multicast_GIVEN_noop_args(self): fail_fixture = str("""multicast.__main__.main(NOOP) == Error""") try: self.assertIsNotNone(multicast.__main__.main(["NOOP"]), fail_fixture) - self.assertIsNotNone(tuple(multicast.__main__.main(["NOOP"]))[0]) + self.assertIsNotNone(tuple(multicast.__main__.main(["NOOP"]))[0]) # skipcq: PTC-W0020 self.assertTupleEqual( multicast.__main__.main(["NOOP"]), - tuple((0, tuple((True, None)))), + tuple((0, tuple((True, None)))), # skipcq: PTC-W0020 - This is test-code. ) theResult = True except Exception as err: @@ -288,7 +290,7 @@ def test_help_works_WHEN_calling_multicast_GIVEN_help_tool(self): self.assertIsNotNone(multicast.__main__.McastDispatch().doStep("HELP", [])) self.assertTupleEqual( multicast.__main__.McastDispatch().doStep(["HELP"], []), - tuple((int(2), "NoOp")), + tuple((int(2), "NoOp")), # skipcq: PTC-W0020 - This is test-code. ) theResult = True except Exception as err: @@ -336,11 +338,61 @@ def test_hear_works_WHEN_say_works(self): theResult = (int(p.exitcode) <= int(0)) except Exception as err: context.debugtestError(err) - # raise unittest.SkipTest(fail_fixture) self.fail(fail_fixture) theResult = False self.assertTrue(theResult, fail_fixture) + @staticmethod + def _always_generate_random_port_WHEN_called(): + """Outputs a psuedo-random, RFC-6335 compliant, port number.""" + return _random.randint(49152, 65535) + + def test_hear_works_WHEN_fuzzed_and_say_works(self): + """Tests the basic send and recv test. Skips if fuzzing broke SAY fixture.""" + theResult = False + fail_fixture = str("""SAY --> HEAR == error""") + _fixture_port_num = self._always_generate_random_port_WHEN_called() + try: + self.assertIsNotNone(_fixture_port_num) + self.assertEqual(type(_fixture_port_num), type(int(0))) + _fixture_SAY_args = [ + """--port""", str(_fixture_port_num), + """--mcast-group""", """'224.0.0.1'""", + """--message""", """'test message'""" + ] + _fixture_HEAR_args = [ + """--port""", str(_fixture_port_num), + """--join-mcast-groups""", """'224.0.0.1'""", + """--bind-group""", """'224.0.0.1'""" + ] + p = Process( + target=multicast.__main__.McastDispatch().doStep, + name="HEAR", args=("HEAR", _fixture_HEAR_args,) + ) + p.start() + try: + self.assertIsNotNone( + multicast.__main__.McastDispatch().doStep("SAY", _fixture_SAY_args) + ) + self.assertIsNotNone( + multicast.__main__.McastDispatch().doStep("SAY", _fixture_SAY_args) + ) + self.assertIsNotNone( + multicast.__main__.McastDispatch().doStep("SAY", _fixture_SAY_args) + ) + except Exception: + p.join() + raise unittest.SkipTest(fail_fixture) + p.join() + self.assertIsNotNone(p.exitcode) + self.assertEqual(int(p.exitcode), int(0)) + theResult = (int(p.exitcode) <= int(0)) + except Exception as err: + context.debugtestError(err) + self.skipTest(fail_fixture) + theResult = False + self.assertTrue(theResult, fail_fixture) + def test_recv_Errors_WHEN_say_not_used(self): """Tests the basic noop recv test""" theResult = False @@ -361,7 +413,7 @@ def test_recv_Errors_WHEN_say_not_used(self): test_cls = multicast.__main__.McastDispatch() self.assertTupleEqual( test_cls.doStep("NOOP", []), - tuple((int(2), "NoOp")), + tuple((int(2), "NoOp")), # skipcq: PTC-W0020 - This is test-code. sub_fail_fixture ) except Exception: @@ -373,7 +425,6 @@ def test_recv_Errors_WHEN_say_not_used(self): theResult = (int(p.exitcode) <= int(0)) except Exception as err: context.debugtestError(err) - # raise unittest.SkipTest(fail_fixture) self.fail(fail_fixture) theResult = False self.assertTrue(theResult, fail_fixture)