diff --git a/docs/FAQ.md b/docs/FAQ.md index 6f71ef9e..9ad0e61d 100644 --- a/docs/FAQ.md +++ b/docs/FAQ.md @@ -79,8 +79,8 @@ import random # for random port # set up some stuff _fixture_PORT_arg = int(random.SystemRandom().randint(49152, 65535)) -_fixture_mcast_GRP_arg = """224.0.0.1""" # only use dotted notation for multicast group addresses -_fixture_host_BIND_arg = """224.0.0.1""" +_fixture_mcast_GRP_arg = "224.0.0.1" # only use dotted notation for multicast group addresses +_fixture_host_BIND_arg = "224.0.0.1" # spawn a listening proc @@ -107,15 +107,15 @@ inputHandler() from multiprocessing import Process as Process _fixture_HEAR_kwargs = { - """is_daemon""": True, - """port""": _fixture_PORT_arg, - """group""": _fixture_host_BIND_arg + "is_daemon": True, + "port": _fixture_PORT_arg, + "group": _fixture_host_BIND_arg } p = Process( target=multicast.hear.McastHEAR().doStep, name="HEAR", kwargs=_fixture_HEAR_kwargs ) -p.daemon = _fixture_HEAR_kwargs["""is_daemon"""] +p.daemon = _fixture_HEAR_kwargs["is_daemon"] p.start() # ... use CTL+C (or signal 2) to shutdown the server 'p' @@ -128,9 +128,9 @@ _and elsewhere (like another function or even module) for the sender:_ # assuming already did 'import multicast as multicast' _fixture_SAY_args = [ - """--port""", _fixture_PORT_arg, - """--group""", _fixture_mcast_GRP_arg, - """--message""", """'test message'""" + "--port", _fixture_PORT_arg, + "--group", _fixture_mcast_GRP_arg, + "--message", "'test message'" ] try: multicast.__main__.McastDispatch().doStep(["SAY", _fixture_SAY_args]) diff --git a/docs/USAGE.md b/docs/USAGE.md index 82a86bd5..a11eb0e3 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -13,8 +13,8 @@ from multiprocessing import Process # set up some stuff _fixture_PORT_arg = int(59595) # Valid multicast addresses range from 224.0.0.0 to 239.255.255.255 -_fixture_mcast_GRP_arg = """224.0.0.1""" # only use dotted notation for multicast group addresses -_fixture_host_BIND_arg = """224.0.0.1""" # only use dotted notation for multicast group addresses +_fixture_mcast_GRP_arg = "224.0.0.1" # only use dotted notation for multicast group addresses +_fixture_host_BIND_arg = "224.0.0.1" # only use dotted notation for multicast group addresses _fixture_host_IFACE_arg = None # Initial value representing no specific interface _fixture_HEAR_args = [ "--port", _fixture_PORT_arg, @@ -38,7 +38,7 @@ def print_loop_stub(func, iterations=5): @print_loop_stub def inputHandle(): test_RCEV = multicast.recv.McastRECV() - buffer_string = str("""""") + buffer_string = str() (didWork, result) = test_RCEV.doStep( groups=[_fixture_mcast_GRP_arg], port=_fixture_PORT_arg, @@ -69,9 +69,9 @@ _and elsewhere (like another function or even module) for the sender:_ # assuming already did 'import multicast' _fixture_SAY_args = [ - """--port""", _fixture_PORT_arg, - """--group""", _fixture_mcast_GRP_arg, - """--message""", """'test message'""" + "--port", _fixture_PORT_arg, + "--group", _fixture_mcast_GRP_arg, + "--message", "'test message'" ] try: multicast.__main__.McastDispatch().doStep("SAY", _fixture_SAY_args) diff --git a/docs/conf.py b/docs/conf.py index d78f44e5..ba88962c 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -43,7 +43,7 @@ # If extensions (or modules to document with autodoc) are in another directory, # add these directories to sys.path here. If the directory is relative to the # documentation root, use os.path.abspath to make it absolute, like shown here. -sys.path.insert(0, os.path.abspath("""..""")) +sys.path.insert(0, os.path.abspath("..")) from docs.utils import _validate_git_ref # noqa from docs.utils import slugify_header # noqa @@ -58,8 +58,8 @@ variable is not set. """ -sys.path.insert(1, os.path.abspath("""multicast""")) -sys.path.insert(1, os.path.abspath("""tests""")) +sys.path.insert(1, os.path.abspath("multicast")) +sys.path.insert(1, os.path.abspath("tests")) # -- General configuration ----------------------------------------------------- @@ -71,18 +71,18 @@ # for md us 'autodoc2' (pip install sphinx-autodoc2) # for rst use 'sphinx.ext.autodoc' extensions = [ - """sphinx.ext.napoleon""", - """autodoc2""", - """sphinx.ext.autosectionlabel""", - """sphinx.ext.githubpages""", - """myst_parser""", - """sphinx_design""", - """sphinx.ext.autosummary""", - """sphinx.ext.doctest""", - """sphinx.ext.todo""", - """sphinx.ext.linkcode""", - """sphinx.ext.viewcode""", - """sphinx.ext.intersphinx""", + "sphinx.ext.napoleon", + "autodoc2", + "sphinx.ext.autosectionlabel", + "sphinx.ext.githubpages", + "myst_parser", + "sphinx_design", + "sphinx.ext.autosummary", + "sphinx.ext.doctest", + "sphinx.ext.todo", + "sphinx.ext.linkcode", + "sphinx.ext.viewcode", + "sphinx.ext.intersphinx", ] # for md auto-docs @@ -179,15 +179,15 @@ pygments_style_dark = "github-dark" pygments_options = { - """tabsize""": 4, - """stripall""": False, - """encoding""": "utf-8", + "tabsize": 4, + "stripall": False, + "encoding": "utf-8", } pygments_yaml_options = { - """tabsize""": 2, - """stripall""": True, - """encoding""": "utf-8", + "tabsize": 2, + "stripall": True, + "encoding": "utf-8", } highlight_options = { @@ -421,19 +421,19 @@ linkcode_url_prefix: str = f"https://github.com/reactive-firewall/{project}" -suffix = """/issues/%s""" +suffix = "/issues/%s" extlinks = { - """issue""": ( + "issue": ( f"{linkcode_url_prefix}/{suffix}", - """issue #%s""" + "issue #%s" ) } # try to link with official python3 documentation. # see https://www.sphinx-doc.org/en/master/usage/extensions/intersphinx.html for more intersphinx_mapping = { - """python""": ("""https://docs.python.org/3""", (None, """python-inv.txt""")) + "python": ("https://docs.python.org/3", (None, "python-inv.txt")) } diff --git a/multicast/__init__.py b/multicast/__init__.py index ceb9194a..10819322 100644 --- a/multicast/__init__.py +++ b/multicast/__init__.py @@ -64,7 +64,7 @@ """hear.McastHEAR""", ] -__package__ = """multicast""" # skipcq: PYL-W0622 +__package__ = "multicast" # skipcq: PYL-W0622 """The package of this program. Minimal Acceptance Testing: @@ -80,7 +80,7 @@ """ -__module__ = """multicast""" +__module__ = "multicast" """The module of this program. Minimal Acceptance Testing: @@ -96,7 +96,7 @@ """ -__name__ = """multicast""" # skipcq: PYL-W0622 +__name__ = "multicast" # skipcq: PYL-W0622 """The name of this program. Minimal Acceptance Testing: @@ -114,7 +114,7 @@ global __version__ # skipcq: PYL-W0604 -__version__ = """2.0.4""" +__version__ = "2.0.4" """The version of this program. Minimal Acceptance Testing: @@ -265,7 +265,7 @@ global _MCAST_DEFAULT_GROUP # skipcq: PYL-W0604 -_MCAST_DEFAULT_GROUP = """224.0.0.1""" +_MCAST_DEFAULT_GROUP = "224.0.0.1" """Arbitrary group to use by default, though any mcst grp would work. The Value of "224.0.0.1" is chosen as a default multicast group as per RFC-5771 @@ -378,7 +378,7 @@ from . import exceptions # pylint: disable=cyclic-import - skipcq: PYL-R0401, PYL-C0414 else: # pragma: no branch global exceptions # skipcq: PYL-W0604 - exceptions = sys.modules["""multicast.exceptions"""] + exceptions = sys.modules["multicast.exceptions"] EXIT_CODES = exceptions.EXIT_CODES """See multicast.exceptions.EXIT_CODES.""" @@ -395,12 +395,12 @@ exit_on_exception = exceptions.exit_on_exception """See multicast.exceptions.exit_on_exception function.""" -if 'multicast.env' not in sys.modules: +if "multicast.env" not in sys.modules: # pylint: disable=cyclic-import - skipcq: PYL-R0401, PYL-C0414 from . import env # pylint: disable=cyclic-import - skipcq: PYL-R0401, PYL-C0414 else: # pragma: no branch global env # skipcq: PYL-W0604 - env = sys.modules["""multicast.env"""] + env = sys.modules["multicast.env"] _config = env.load_config() @@ -439,13 +439,13 @@ class mtool(abc.ABC): """ - __module__ = """multicast""" + __module__ = "multicast" __proc__ = None - __prologue__ = """Add a prologue here.""" + __prologue__ = "Add a prologue here." - __epilogue__ = """Add an epilogue here.""" + __epilogue__ = "Add an epilogue here." @classmethod def buildArgs(cls, calling_parser_group): @@ -498,7 +498,7 @@ def buildArgs(cls, calling_parser_group): add_help=False ) group = calling_parser_group.add_mutually_exclusive_group(required=False) - group.add_argument('-h', '--help', action='help') + group.add_argument("-h", "--help", action="help") group.add_argument( "-v", "--version", @@ -506,13 +506,13 @@ def buildArgs(cls, calling_parser_group): version=str("%(prog)s {version}").format(version=str(__version__)) ) calling_parser_group.add_argument( - """--use-std""", dest='is_std', default=False, action='store_true' + "--use-std", dest="is_std", default=False, action="store_true" ) calling_parser_group.add_argument( - """--daemon""", dest='is_daemon', default=False, action='store_true' + "--daemon", dest="is_daemon", default=False, action="store_true" ) subparsers = calling_parser_group.add_subparsers( - title="Tools", dest="cmd_tool", help=str("""Sub-Commands."""), metavar="CMD" + title="Tools", dest="cmd_tool", help=str("Sub-Commands."), metavar="CMD" ) if mtool.__class__.__subclasscheck__(mtool, cls): # pragma: no branch cls.setupArgs(subparsers) @@ -662,11 +662,11 @@ def doStep(self, *args, **kwargs): # pragma: no cover raise NotImplementedError("Subclasses must implement this method.") -if 'multicast.skt' not in sys.modules: +if "multicast.skt" not in sys.modules: from . import skt as skt # pylint: disable=cyclic-import - skipcq: PYL-R0401, PYL-C0414 else: # pragma: no branch global skt # skipcq: PYL-W0604 - skt = sys.modules["""multicast.skt"""] + skt = sys.modules["multicast.skt"] genSocket = skt.genSocket """See multicast.skt.genSocket.""" @@ -674,32 +674,32 @@ def doStep(self, *args, **kwargs): # pragma: no cover endSocket = skt.endSocket """See multicast.skt.endSocket.""" -if 'multicast.recv' not in sys.modules: +if "multicast.recv" not in sys.modules: from . import recv as recv # pylint: disable=cyclic-import - skipcq: PYL-R0401, PYL-C0414 else: # pragma: no branch global recv # skipcq: PYL-W0604 - recv = sys.modules["""multicast.recv"""] + recv = sys.modules["multicast.recv"] -if 'multicast.send' not in sys.modules: +if "multicast.send" not in sys.modules: from . import send as send # pylint: disable=cyclic-import - skipcq: PYL-R0401, PYL-C0414 else: # pragma: no branch global send # skipcq: PYL-W0604 - send = sys.modules["""multicast.send"""] + send = sys.modules["multicast.send"] -if 'multicast.hear' not in sys.modules: +if "multicast.hear" not in sys.modules: from . import hear as hear # pylint: disable=cyclic-import - skipcq: PYL-R0401, PYL-C0414 else: # pragma: no branch global hear # skipcq: PYL-W0604 - hear = sys.modules["""multicast.hear"""] + hear = sys.modules["multicast.hear"] try: - if """multicast.__main__""" in sys.modules: # pragma: no cover + if "multicast.__main__" in sys.modules: # pragma: no cover global __main__ # skipcq: PYL-W0604 - __main__ = sys.modules["""multicast.__main__"""] + __main__ = sys.modules["multicast.__main__"] except Exception: import multicast.__main__ as __main__ # pylint: disable=cyclic-import - skipcq: PYL-R0401 -if __name__ in u'__main__': +if __name__ in u"__main__": __EXIT_CODE = 2 if __main__.main is not None: __EXIT_CODE = __main__.main(sys.argv[1:]) diff --git a/multicast/__main__.py b/multicast/__main__.py index 686bb5ee..3537c8bd 100644 --- a/multicast/__main__.py +++ b/multicast/__main__.py @@ -84,7 +84,7 @@ """TASK_OPTIONS""", ] -__package__ = """multicast""" # skipcq: PYL-W0622 +__package__ = "multicast" # skipcq: PYL-W0622 """ The package of this component. @@ -108,7 +108,7 @@ """ -__module__ = """multicast.__main__""" # skipcq: PYL-W0622 +__module__ = "multicast.__main__" # skipcq: PYL-W0622 """ The module of this component. @@ -129,7 +129,7 @@ """ -__file__ = """multicast/__main__.py""" +__file__ = "multicast/__main__.py" """The file of this component.""" try: @@ -138,46 +138,46 @@ # Throw more relevant Error raise ImportError(str("[CWE-440] Error Importing Python")) from impErr -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__"] -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"] -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"] -if 'multicast.exceptions' not in sys.modules: +if "multicast.exceptions" not in sys.modules: # pylint: disable=useless-import-alias - skipcq: PYL-C0414 from . import exceptions as exceptions # skipcq: PYL-C0414 else: # pragma: no branch - exceptions = sys.modules["""multicast.exceptions"""] + exceptions = sys.modules["multicast.exceptions"] -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"] -if 'multicast.recv' not in sys.modules: +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"] -if 'multicast.send' not in sys.modules: +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"] -if 'multicast.hear' not in sys.modules: +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"] class McastNope(mtool): @@ -235,13 +235,13 @@ class McastNope(mtool): """ - __module__ = """multicast.__main__""" + __module__ = "multicast.__main__" - __name__ = """multicast.__main__.McastNope""" + __name__ = "multicast.__main__.McastNope" - __proc__ = """NOOP""" # NOT "Nope" rather "NoOp" + __proc__ = "NOOP" # NOT "Nope" rather "NoOp" - __prologue__ = """No Operation.""" + __prologue__ = "No Operation." @classmethod def setupArgs(cls, parser): @@ -377,18 +377,18 @@ class McastRecvHearDispatch(mtool): """ - __module__ = """multicast.__main__""" + __module__ = "multicast.__main__" - __name__ = """multicast.__main__.McastRecvHearDispatch""" + __name__ = "multicast.__main__.McastRecvHearDispatch" - __proc__ = """HEAR""" + __proc__ = "HEAR" __epilogue__ = """Generally speaking you want to bind to one of the groups you joined in this module/instance, but it is also possible to bind to group which is added by some other programs (like another python program instance of this) """ - __prologue__ = """Python Multicast Receiver. Primitives for a listener for multicast data.""" + __prologue__ = "Python Multicast Receiver. Primitives for a listener for multicast data." @classmethod def setupArgs(cls, parser): @@ -476,34 +476,34 @@ def setupArgs(cls, parser): """ if parser is not None: # pragma: no branch parser.add_argument( - """--port""", + "--port", type=int, default=_MCAST_DEFAULT_PORT # skipcq: PYL-W0212 - module ok ) - __tmp_help = """local interface to use for listening to multicast data; """ - __tmp_help += """if unspecified, any one interface may be chosen.""" - parser.add_argument("""--iface""", default=None, help=str(__tmp_help)) - __tmp_help = """multicast group (ip address) to bind-to for the udp socket; """ - __tmp_help += """should be one of the multicast groups joined globally """ - __tmp_help += """(not necessarily joined in this python program) """ - __tmp_help += """in the interface specified by --iface. """ + __tmp_help = "local interface to use for listening to multicast data; " + __tmp_help += "if unspecified, any one interface may be chosen." + parser.add_argument("--iface", default=None, help=str(__tmp_help)) + __tmp_help = "multicast group (ip address) to bind-to for the udp socket; " + __tmp_help += "should be one of the multicast groups joined globally " + __tmp_help += "(not necessarily joined in this python program) " + __tmp_help += "in the interface specified by --iface. " __tmp_help += f"If unspecified, bind-to {_MCAST_DEFAULT_GROUP} " - __tmp_help += """(all addresses (all multicast addresses) of that interface)""" + __tmp_help += "(all addresses (all multicast addresses) of that interface)" parser.add_argument( - """--group""", + "--group", default=_MCAST_DEFAULT_GROUP, # skipcq: PYL-W0212 - module ok help=str(__tmp_help) ) - __tmp_help = """multicast groups (ip addresses) to join globally; """ - __tmp_help += """should be one of the multicast groups joined globally """ - __tmp_help += """by the interface specified by --iface. """ - __tmp_help += """If unspecified, or supplied an empty list, the default """ - __tmp_help += """implementation will join """ + __tmp_help = "multicast groups (ip addresses) to join globally; " + __tmp_help += "should be one of the multicast groups joined globally " + __tmp_help += "by the interface specified by --iface. " + __tmp_help += "If unspecified, or supplied an empty list, the default " + __tmp_help += "implementation will join " __tmp_help += f"{_MCAST_DEFAULT_GROUP} (all addresses (all multicast addresses) " - __tmp_help += """of that interface) instead of not joining. NOTE: If you really need """ - __tmp_help += """to NOT join the multicast group you should instead use the sockets """ - __tmp_help += """module directly, as this module does not support such a use-case.""" - parser.add_argument("""--groups""", default=[], nargs="*", help=str(__tmp_help)) + __tmp_help += "of that interface) instead of not joining. NOTE: If you really need " + __tmp_help += "to NOT join the multicast group you should instead use the sockets " + __tmp_help += "module directly, as this module does not support such a use-case." + parser.add_argument("--groups", default=[], nargs="*", help=str(__tmp_help)) @staticmethod def _help_daemon_dispatch(*args, **kwargs): @@ -556,10 +556,10 @@ def doStep(self, *args, **kwargs): # More boiler-plate-code TASK_OPTIONS = { - """NOOP""": McastNope(), - """RECV""": McastRecvHearDispatch(), - """SAY""": send.McastSAY(), - """HEAR""": McastRecvHearDispatch(), + "NOOP": McastNope(), + "RECV": McastRecvHearDispatch(), + "SAY": send.McastSAY(), + "HEAR": McastRecvHearDispatch(), } """The callable function tasks of this program. will add.""" @@ -574,12 +574,12 @@ class McastDispatch(mtool): """ - __proc__ = """multicast""" + __proc__ = "multicast" - __prologue__ = """The Main Entrypoint.""" + __prologue__ = "The Main Entrypoint." __epilogue__ = str( - """When called from the command line the __main__ component handles the CLI dispatch.""" + "When called from the command line the __main__ component handles the CLI dispatch." ) @classmethod @@ -620,7 +620,7 @@ def doStep(self, *args, **kwargs): try: (argz, _) = type(self).parseArgs(*args) service_cmd = argz.cmd_tool - argz.__dict__.__delitem__("""cmd_tool""") + argz.__dict__.__delitem__("cmd_tool") _TOOL_MSG = (self.useTool(service_cmd, **argz.__dict__)) if _TOOL_MSG[0]: __EXIT_MSG = (0, _TOOL_MSG) @@ -759,5 +759,5 @@ def main(*argv): 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_CODE = main([str(__name__), "-h"]) exit(__EXIT_CODE[0]) # skipcq: PYL-R1722 - intentionally allow overwriteing exit for testing diff --git a/multicast/env.py b/multicast/env.py index 616e959b..027570cb 100644 --- a/multicast/env.py +++ b/multicast/env.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__package__ = """multicast""" # skipcq: PYL-W0622 +__package__ = "multicast" # skipcq: PYL-W0622 """ The package of this component. @@ -41,7 +41,7 @@ """ -__module__ = """multicast.env""" +__module__ = "multicast.env" """ The module of this component. @@ -62,10 +62,10 @@ """ -__file__ = """multicast/env.py""" +__file__ = "multicast/env.py" """The file of this component.""" -__name__ = """multicast.env""" # skipcq: PYL-W0622 +__name__ = "multicast.env" # skipcq: PYL-W0622 """The name of this component. Minimal Acceptance Testing: @@ -289,7 +289,7 @@ def load_port() -> int: port = _MCAST_DEFAULT_PORT # Validate and potentially update port if validate_port(port): - globals()["""_MCAST_DEFAULT_PORT"""] = port + globals()["_MCAST_DEFAULT_PORT"] = port else: warnings.warn( f"Port {port} is outside valid range (49152-65535), using default {_MCAST_DEFAULT_PORT}", @@ -403,7 +403,7 @@ def load_group() -> ipaddress.IPv4Address: group = os.getenv("MULTICAST_GROUP", _MCAST_DEFAULT_GROUP) # Validate and potentially update group if validate_multicast_address(group): - globals()["""_MCAST_DEFAULT_GROUP"""] = group + globals()["_MCAST_DEFAULT_GROUP"] = group else: warnings.warn( f"Invalid multicast group {group}, using default {_MCAST_DEFAULT_GROUP}", stacklevel=2 @@ -490,7 +490,7 @@ def load_TTL() -> int: ttl = _MCAST_DEFAULT_TTL # Validate and potentially update TTL if validate_ttl(ttl): - globals()["""_MCAST_DEFAULT_TTL"""] = ttl + globals()["_MCAST_DEFAULT_TTL"] = ttl else: warnings.warn( f"TTL {ttl} is outside valid range (1-126), using default {_MCAST_DEFAULT_TTL}", diff --git a/multicast/exceptions.py b/multicast/exceptions.py index 1935d8dc..ee74d097 100644 --- a/multicast/exceptions.py +++ b/multicast/exceptions.py @@ -124,7 +124,7 @@ """ -__package__ = """multicast""" # skipcq: PYL-W0622 +__package__ = "multicast" # skipcq: PYL-W0622 """ The package of this program. @@ -148,7 +148,7 @@ """ -__module__ = """multicast.exceptions""" +__module__ = "multicast.exceptions" """ The module of this program. @@ -169,10 +169,10 @@ """ -__file__ = """multicast/exceptions.py""" +__file__ = "multicast/exceptions.py" """The file of this component.""" -__name__ = """multicast.exceptions""" # skipcq: PYL-W0622 +__name__ = "multicast.exceptions" # skipcq: PYL-W0622 """The name of this component. Minimal Acceptance Testing: @@ -231,9 +231,9 @@ class CommandExecutionError(RuntimeError): 1 """ - __module__ = """multicast.exceptions""" + __module__ = "multicast.exceptions" - __name__ = """multicast.exceptions.CommandExecutionError""" + __name__ = "multicast.exceptions.CommandExecutionError" def __init__(self, *args, **kwargs): """ @@ -331,9 +331,9 @@ class ShutdownCommandReceived(RuntimeError): 'Custom shutdown message.' """ - __module__ = """multicast.exceptions""" + __module__ = "multicast.exceptions" - __name__ = """multicast.exceptions.ShutdownCommandReceived""" + __name__ = "multicast.exceptions.ShutdownCommandReceived" def __init__(self, *args, **kwargs): """ diff --git a/multicast/hear.py b/multicast/hear.py index 1855db7a..6ba6dcfa 100644 --- a/multicast/hear.py +++ b/multicast/hear.py @@ -110,7 +110,7 @@ """ -__package__ = """multicast""" # skipcq: PYL-W0622 +__package__ = "multicast" # skipcq: PYL-W0622 """Names the package of this program. Minimal Acceptance Testing: @@ -134,7 +134,7 @@ """ -__module__ = """multicast""" +__module__ = "multicast" """Names the module of this program. Minimal Acceptance Testing: @@ -155,10 +155,10 @@ """ -__file__ = """multicast/hear.py""" +__file__ = "multicast/hear.py" """Names the file of this component.""" -__name__ = """multicast.hear""" # skipcq: PYL-W0622 +__name__ = "multicast.hear" # skipcq: PYL-W0622 """Names this component. Minimal Acceptance Testing: @@ -184,7 +184,7 @@ if 'multicast' not in _sys.modules: from . import multicast as multicast # pylint: disable=cyclic-import - skipcq: PYL-C0414 else: # pragma: no branch - multicast = _sys.modules["""multicast"""] + multicast = _sys.modules["multicast"] _BLANK = multicast._BLANK # skipcq: PYL-W0212 - module ok # skipcq from . import recv as recv # pylint: disable=useless-import-alias - skipcq: PYL-C0414 @@ -332,7 +332,7 @@ def handle_error(self, request, client_address): None """ print(str("handle_error")) - if request is not None and request[0] is not None and """STOP""" in str(request[0]): + if request is not None and request[0] is not None and "STOP" in str(request[0]): def kill_func(a_server): """ Terminate the server. @@ -460,7 +460,7 @@ def handle(self): self.client_address[0], self.client_address[1], f"HEAR [ {data.upper()} SAID {self.client_address} ] from {me}" # noqa ) - if """STOP""" in str(data): + if "STOP" in str(data): raise multicast.exceptions.ShutdownCommandReceived("SHUTDOWN") from None @@ -508,18 +508,18 @@ class McastHEAR(multicast.mtool): """ - __module__ = """multicast.hear""" + __module__ = "multicast.hear" - __name__ = """multicast.hear.McastHEAR""" + __name__ = "multicast.hear.McastHEAR" - __proc__ = """HEAR""" + __proc__ = "HEAR" __epilogue__ = """Generally speaking you want to bind to one of the groups you joined in this module/instance, but it is also possible to bind to group which is added by some other programs (like another python program instance of this) """ - __prologue__ = """Python Multicast Server for multicast input.""" + __prologue__ = "Python Multicast Server for multicast input." @classmethod def setupArgs(cls, parser): diff --git a/multicast/recv.py b/multicast/recv.py index 13a9790c..e7b0bbcc 100644 --- a/multicast/recv.py +++ b/multicast/recv.py @@ -103,7 +103,7 @@ """ -__package__ = """multicast""" # skipcq: PYL-W0622 +__package__ = "multicast" # skipcq: PYL-W0622 """The package of this program. Minimal Acceptance Testing: @@ -126,7 +126,7 @@ """ -__module__ = """multicast""" +__module__ = "multicast" """The module of this program. Minimal Acceptance Testing: @@ -146,10 +146,10 @@ """ -__file__ = """multicast/recv.py""" +__file__ = "multicast/recv.py" """The file of this component.""" -__name__ = """multicast.recv""" # skipcq: PYL-W0622 +__name__ = "multicast.recv" # skipcq: PYL-W0622 """The name of this component. Minimal Acceptance Testing: @@ -175,7 +175,7 @@ # skipcq from . import multicast as multicast # pylint: disable=cyclic-import - skipcq: PYL-C0414 else: # pragma: no branch - multicast = sys.modules["""multicast"""] + multicast = sys.modules["multicast"] _BLANK = multicast._BLANK # skipcq: PYL-W0212 - module ok except Exception as importErr: del importErr # skipcq - cleanup any error leaks early @@ -407,7 +407,7 @@ def recvstep(msgbuffer, chunk, sock): except KeyboardInterrupt: # pragma: no branch if (sys.stdout.isatty()): # pragma: no cover print(multicast._BLANK) # skipcq: PYL-W0212 - module ok - print(str("""User Interrupted""")) + print(str("User Interrupted")) except OSError: # pragma: no branch if (sys.stdout.isatty()): # pragma: no cover print(multicast._BLANK) # skipcq: PYL-W0212 - module ok @@ -462,18 +462,18 @@ class McastRECV(multicast.mtool): """ - __module__ = """multicast.recv""" + __module__ = "multicast.recv" - __name__ = """multicast.recv.McastRECV""" + __name__ = "multicast.recv.McastRECV" - __proc__ = """RECV""" + __proc__ = "RECV" __epilogue__ = """Generally speaking you want to bind to one of the groups you joined in this module/instance, but it is also possible to bind to group which is added by some other programs (like another python program instance of this) """ - __prologue__ = """Python Multicast Receiver. Primitives for a listener for multicast data.""" + __prologue__ = "Python Multicast Receiver. Primitives for a listener for multicast data." @classmethod def setupArgs(cls, parser): diff --git a/multicast/send.py b/multicast/send.py index 066068e6..178060c9 100644 --- a/multicast/send.py +++ b/multicast/send.py @@ -66,7 +66,7 @@ """ -__package__ = """multicast""" # skipcq: PYL-W0622 +__package__ = "multicast" # skipcq: PYL-W0622 """The package of this program. Minimal Acceptance Testing: @@ -89,7 +89,7 @@ """ -__module__ = """multicast""" +__module__ = "multicast" """The module of this program. Minimal Acceptance Testing: @@ -109,10 +109,10 @@ """ -__file__ = """multicast/send.py""" +__file__ = "multicast/send.py" """The file of this component.""" -__name__ = """multicast.send""" # skipcq: PYL-W0622 - Ensures the correct name value. +__name__ = "multicast.send" # skipcq: PYL-W0622 - Ensures the correct name value. """The name of this component. Minimal Acceptance Testing: @@ -138,7 +138,7 @@ # skipcq from . import multicast as multicast # pylint: disable=cyclic-import - skipcq: PYL-C0414 else: # pragma: no branch - multicast = sys.modules["""multicast"""] + multicast = sys.modules["multicast"] _BLANK = multicast._BLANK # skipcq: PYL-W0212 - module ok except Exception as importErr: del importErr # skipcq - cleanup any error leaks early @@ -200,13 +200,13 @@ class McastSAY(multicast.mtool): """ - __module__ = """multicast.send""" + __module__ = "multicast.send" - __name__ = """multicast.send.McastSAY""" + __name__ = "multicast.send.McastSAY" - __proc__ = """SAY""" + __proc__ = "SAY" - __prologue__ = """Python Multicast Broadcaster.""" + __prologue__ = "Python Multicast Broadcaster." @classmethod def setupArgs(cls, parser): @@ -277,27 +277,27 @@ def setupArgs(cls, parser): """ if parser is not None: # pragma: no branch parser.add_argument( - """--port""", + "--port", type=int, default=multicast._MCAST_DEFAULT_PORT # skipcq: PYL-W0212 - module ok ) parser.add_argument( - """--group""", + "--group", default=multicast._MCAST_DEFAULT_GROUP # skipcq: PYL-W0212 - module ok ) parser.add_argument( - """--groups""", + "--groups", required=False, nargs="*", - dest="""groups""", - help="""multicast groups (ip addrs) to listen to join.""" + dest="groups", + help="multicast groups (ip addrs) to listen to join." ) parser.add_argument( - """-m""", - """--message""", + "-m", + "--message", nargs="+", - dest="""data""", - default=str("""PING from {name}: group: {group}, port: {port}""") + dest="data", + default=str("PING from {name}: group: {group}, port: {port}") ) @staticmethod @@ -363,7 +363,7 @@ def doStep(self, *args, **kwargs): _result = _result and self._sayStep(group, port, chunk) elif isinstance(data, list): # Join multiple arguments into a single string - message = str(""" """).join(data) + message = str(" ").join(data) _result = self._sayStep(group, port, message) else: message = data.decode('utf8') if isinstance(data, bytes) else str(data) diff --git a/multicast/skt.py b/multicast/skt.py index ffae417c..6052887b 100644 --- a/multicast/skt.py +++ b/multicast/skt.py @@ -70,7 +70,7 @@ """ -__package__ = """multicast""" # skipcq: PYL-W0622 +__package__ = "multicast" # skipcq: PYL-W0622 """The package of this program. Minimal Acceptance Testing: @@ -93,7 +93,7 @@ """ -__module__ = """multicast.skt""" +__module__ = "multicast.skt" """The module of this program. Minimal Acceptance Testing: @@ -113,10 +113,10 @@ """ -__file__ = """multicast/skt.py""" +__file__ = "multicast/skt.py" """The file of this component.""" -__name__ = """multicast.skt""" # skipcq: PYL-W0622 +__name__ = "multicast.skt" # skipcq: PYL-W0622 """The name of this component. Minimal Acceptance Testing: diff --git a/tests/MulticastUDPClient.py b/tests/MulticastUDPClient.py index ec74058e..056b1fc1 100644 --- a/tests/MulticastUDPClient.py +++ b/tests/MulticastUDPClient.py @@ -80,7 +80,7 @@ # even if the above stated remedy fails of its essential purpose. ################################################################################ -__module__ = """tests""" +__module__ = "tests" """This is a testing related stand-alone utilities module. This module provides test fixtures and utilities for testing multicast communication. @@ -103,7 +103,7 @@ """ -__name__ = """tests.MulticastUDPClient""" # skipcq: PYL-W0622 +__name__ = "tests.MulticastUDPClient" # skipcq: PYL-W0622 try: import sys @@ -170,7 +170,7 @@ class MCastClient(object): # skipcq: PYL-R0205 """ - __module__ = """tests.MulticastUDPClient.MCastClient""" + __module__ = "tests.MulticastUDPClient.MCastClient" _group_addr = None """The multicast group address.""" @@ -240,10 +240,10 @@ def __init__(self, *args, **kwargs): """ # skipcq: TCV-002 - if str("""grp_addr""") in kwargs: - self._group_addr = kwargs.get("""grp_addr""", None) # skipcq: PTC-W0039 - ensure None - if str("""src_port""") in kwargs: - self._source_port = kwargs.get("""src_port""", 0) + if str("grp_addr") in kwargs: + self._group_addr = kwargs.get("grp_addr", None) # skipcq: PTC-W0039 - ensure None + if str("src_port") in kwargs: + self._source_port = kwargs.get("src_port", 0) else: self._source_port = int( 50000 + ( @@ -337,7 +337,7 @@ class MyUDPHandler(socketserver.BaseRequestHandler): """ - __module__ = """tests.MulticastUDPClient.MyUDPHandler""" + __module__ = "tests.MulticastUDPClient.MyUDPHandler" # skipcq: TCV-002 def handle(self): @@ -429,7 +429,7 @@ def main(): tsts_fxr = MCastClient() print(str((HOST, PORT))) tsts_fxr.say(HOST, PORT, sock, data) - tsts_fxr.say(HOST, PORT, sock, str("""STOP""")) + tsts_fxr.say(HOST, PORT, sock, str("STOP")) if __name__ == "__main__": diff --git a/tests/__init__.py b/tests/__init__.py index d6e413c7..a447707f 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -49,9 +49,9 @@ """ -__package__ = """tests""" # skipcq: PYL-W0622 +__package__ = "tests" # skipcq: PYL-W0622 -__module__ = """tests""" +__module__ = "tests" try: import sys @@ -64,7 +64,7 @@ if 'multicast' not in sys.modules: import multicast # pylint: disable=cyclic-import - skipcq: PYL-R0401 else: # pragma: no branch - multicast = sys.modules["""multicast"""] + multicast = sys.modules["multicast"] except Exception as err: # pragma: no branch raise ImportError("[CWE-440] multicast Failed to import.") from err @@ -138,7 +138,7 @@ if 'tests.context' not in sys.modules: from tests import context else: # pragma: no branch - context = sys.modules["""tests.context"""] + context = sys.modules["tests.context"] except ImportError as _cause: # pragma: no branch raise ImportError("[CWE-440] context Failed to import.") from _cause @@ -204,7 +204,7 @@ def loadDocstringsFromModule(module): if 'doctest' not in sys.modules: import doctest else: # pragma: no branch - doctest = sys.modules["""doctest"""] + doctest = sys.modules["doctest"] except Exception as _cause: # pragma: no branch raise ImportError("[CWE-440] doctest Failed to import.") from _cause finder = doctest.DocTestFinder(verbose=True, recurse=True, exclude_empty=True) diff --git a/tests/context.py b/tests/context.py index 091e8372..ce349b5c 100644 --- a/tests/context.py +++ b/tests/context.py @@ -16,9 +16,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" -__name__ = """tests.context""" # skipcq: PYL-W0622 +__name__ = "tests.context" # skipcq: PYL-W0622 __doc__ = """Test context and environment setup module. @@ -103,7 +103,7 @@ if 'Process' not in sys.modules: from multiprocessing import Process as Process # skipcq: PYL-C0414 else: # pragma: no branch - Process = sys.modules["""Process"""] + Process = sys.modules["Process"] except ImportError as err: # pragma: no branch raise ModuleNotFoundError("[CWE-440] Process Failed to import.") from err @@ -117,7 +117,7 @@ import packaging from packaging import version else: # pragma: no branch - packaging = sys.modules["""packaging"""] + packaging = sys.modules["packaging"] from packaging import version except ImportError as err: # pragma: no branch raise ModuleNotFoundError("[CWE-440] packaging.version Failed to import.") from err @@ -126,7 +126,7 @@ if 'multicast' not in sys.modules: import multicast # pylint: disable=cyclic-import - skipcq: PYL-R0401 else: # pragma: no branch - multicast = sys.modules["""multicast"""] # pylint: disable=cyclic-import + multicast = sys.modules["multicast"] # pylint: disable=cyclic-import except Exception as err: # pragma: no branch raise ImportError("[CWE-440] Multicast Python Module Failed to import.") from err @@ -134,7 +134,7 @@ if 'tests.profiling' not in sys.modules: import tests.profiling as profiling else: # pragma: no branch - profiling = sys.modules["""tests.profiling"""] + profiling = sys.modules["tests.profiling"] except ImportError as err: # pragma: no branch raise ModuleNotFoundError("[CWE-440] profiling Failed to import.") from err @@ -142,7 +142,7 @@ if 'multicast.exceptions' not in sys.modules: import multicast.exceptions else: # pragma: no branch - multicast.exceptions = sys.modules["""multicast.exceptions"""] + multicast.exceptions = sys.modules["multicast.exceptions"] from multicast.exceptions import CommandExecutionError except ImportError as err: # pragma: no branch raise ModuleNotFoundError("[CWE-440] Test Exceptions Failed to import.") from err @@ -372,7 +372,7 @@ def checkCovCommand(*args): # skipcq: PYL-W0102 - [] != [None] i += 2 else: # pragma: no branch args[0] = str(getCoverageCommand()) - extra_args = ["""run""", """-p""", """--context=Integration""", """--source=multicast"""] + extra_args = ["run", "-p", "--context=Integration", "--source=multicast"] # PEP-279 - see https://www.python.org/dev/peps/pep-0279/ for k, ktem in enumerate(extra_args): offset = i + k @@ -447,7 +447,7 @@ def checkStrOrByte(theInput): theOutput = theInput try: if isinstance(theInput, bytes): - theOutput = theInput.decode("""UTF-8""") + theOutput = theInput.decode("UTF-8") except UnicodeDecodeError: # pragma: no branch theOutput = bytes(theInput) return theOutput @@ -754,7 +754,7 @@ def check_exec_command_has_output(test_case, someArgs): returns True if has output and False otherwise. """ theResult = False - fail_msg_fixture = str("""Expecting output: CLI test had no output.""") + fail_msg_fixture = str("Expecting output: CLI test had no output.") try: if (test_case._thepython is not None): try: @@ -898,11 +898,11 @@ class BasicUsageTestSuite(unittest.TestCase): """ - __module__ = """tests.context""" + __module__ = "tests.context" - __name__ = """tests.context.BasicUsageTestSuite""" + __name__ = "tests.context.BasicUsageTestSuite" - NO_PYTHON_ERROR = """No python cmd to test with!""" # skipcq: TCV-002 + NO_PYTHON_ERROR = "No python cmd to test with!" # skipcq: TCV-002 """Error message used when Python command is not available for testing. This constant is used across multiple test methods to maintain consistency @@ -911,7 +911,7 @@ class BasicUsageTestSuite(unittest.TestCase): @classmethod def setUpClass(cls): - """Overrides unittest.TestCase.setUpClass(cls) to set up thepython test fixture.""" + "Overrides unittest.TestCase.setUpClass(cls) to set up thepython test fixture." cls._thepython = getPythonCommand() @staticmethod @@ -950,25 +950,25 @@ def _should_get_package_version_WHEN_valid(self): """ try: - self.assertIsNotNone(multicast.__module__, """Version will be efectivly None.""") - self.assertIsNotNone(multicast.__version__, """Version is not valid.""") + self.assertIsNotNone(multicast.__module__, "Version will be efectivly None.") + self.assertIsNotNone(multicast.__version__, "Version is not valid.") _raw_version_fixture = multicast.__version__ - self.assertIsInstance(_raw_version_fixture, str, """Version is not a string.""") + self.assertIsInstance(_raw_version_fixture, str, "Version is not a string.") # Strip custom tags # stuff like: mcast_version = mcast_version.replace("-hotfix", "").replace("-hf", "") # Refactor alpha/beta tags parsed_version = version.parse(_raw_version_fixture) - self.assertIsNotNone(parsed_version, """Version is not valid.""") - self.assertIsInstance(parsed_version, version.Version, """Version is not valid.""") + self.assertIsNotNone(parsed_version, "Version is not valid.") + self.assertIsInstance(parsed_version, version.Version, "Version is not valid.") self.assertTrue( len(parsed_version.release) >= 2, - """Version must have at least major.minor components.""" + "Version must have at least major.minor components." ) return parsed_version except ImportError: - self.fail("""Failed to import the multicast package to retrieve version.""") + self.fail("Failed to import the multicast package to retrieve version.") - @unittest.skipUnless(True, """Insanitty Test. Good luck debugging.""") + @unittest.skipUnless(True, "Insanitty Test. Good luck debugging.") def test_absolute_truth_and_meaning(self): """Test case 0: Insanitty Test.""" assert True diff --git a/tests/profiling.py b/tests/profiling.py index 55aa01e9..79f21bc7 100644 --- a/tests/profiling.py +++ b/tests/profiling.py @@ -27,7 +27,7 @@ # .......................................... # NO ASSOCIATION -__module__ = """tests.profiling""" +__module__ = "tests.profiling" """This is pythonrepo testing module Template.""" try: @@ -43,7 +43,7 @@ if 'os' not in sys.modules: import os else: # pragma: no branch - os = sys.modules["""os"""] + os = sys.modules["os"] except Exception as badErr: # pragma: no branch baton = ModuleNotFoundError(badErr, str("[CWE-758] Test module failed completely.")) baton.module = __module__ @@ -55,7 +55,7 @@ if 'functools' not in sys.modules: import functools else: # pragma: no branch - functools = sys.modules["""functools"""] + functools = sys.modules["functools"] except Exception as badErr: # pragma: no branch baton = ModuleNotFoundError(badErr, str("[CWE-758] Test module failed completely.")) baton.module = __module__ @@ -78,7 +78,7 @@ if 'cProfile' not in sys.modules: import cProfile else: # pragma: no branch - cProfile = sys.modules["""cProfile"""] + cProfile = sys.modules["cProfile"] except Exception as badErr: # pragma: no branch baton = ModuleNotFoundError(badErr, str("[CWE-758] Test module failed completely.")) baton.module = __module__ diff --git a/tests/run_selective.py b/tests/run_selective.py index a87ac932..fa6f9f08 100755 --- a/tests/run_selective.py +++ b/tests/run_selective.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" import sys import argparse diff --git a/tests/test_basic.py b/tests/test_basic.py index e07bdbf4..a7f583c5 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: try: @@ -63,7 +63,7 @@ class BasicTestSuite(context.BasicUsageTestSuite): platforms. """ - __module__ = """tests.test_basic""" + __module__ = "tests.test_basic" @unittest.skipUnless(True, "Insanitty Test. Good luck debugging.") def test_absolute_truth_and_meaning(self): @@ -122,7 +122,7 @@ def test_IsNone_WHEN_given_corner_case_input(self): (_ignored_code, test_fixture) = tst_dispatch.useTool(tst_in) self.assertIsNone( test_fixture, - str("""multicast.McastDispatch().useTool({}) == ERROR""").format(str(tst_in)) + str("multicast.McastDispatch().useTool({}) == ERROR").format(str(tst_in)) ) theResult = True except Exception: diff --git a/tests/test_build.py b/tests/test_build.py index 4699e86a..dd7899d8 100644 --- a/tests/test_build.py +++ b/tests/test_build.py @@ -35,7 +35,7 @@ """ -__module__ = """tests""" +__module__ = "tests" try: try: @@ -59,7 +59,7 @@ @context.markWithMetaTag("mat", "mat_build") class BuildPEP517TestSuite(BasicUsageTestSuite): - __module__ = """tests.test_build""" + __module__ = "tests.test_build" def test_build_with_pep517(self): """ @@ -87,7 +87,7 @@ def test_build_with_pep517(self): # Build the source distribution theBuildtxt = context.checkPythonCommand(build_arguments, stderr=subprocess.STDOUT) self.assertIn(str("running build"), str(theBuildtxt)) - self.assertIn(str("""Successfully built"""), str(theBuildtxt)) + self.assertIn(str("Successfully built"), str(theBuildtxt)) # Verify that the dist directory contains the expected files dist_dir = os.path.join(os.getcwd(), 'dist') pkg_version = str(self._should_get_package_version_WHEN_valid()) diff --git a/tests/test_deps.py b/tests/test_deps.py index 4e73b992..e8cf92aa 100644 --- a/tests/test_deps.py +++ b/tests/test_deps.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: try: @@ -50,7 +50,7 @@ class BuildRequirementsTxtTestSuite(context.BasicUsageTestSuite): """Test cases for 'tests/requirements.txt'.""" - __module__ = """tests.test_deps""" + __module__ = "tests.test_deps" def test_requirements_path_exists(self): """Test that 'tests/requirements.txt' exists.""" @@ -103,7 +103,7 @@ def test_requirements_installation(self): result = context.checkPythonCommand( [context.getPythonCommand(), '-m pip', 'install', '-r', 'tests/requirements.txt'] ) - self.assertIsNotNone(result, str("""Failed to install requirements!""")) + self.assertIsNotNone(result, str("Failed to install requirements!")) # leave this part diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py index 784af2c7..ba87b0c9 100644 --- a/tests/test_exceptions.py +++ b/tests/test_exceptions.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: """Handle imports with CWE-758 mitigation. @@ -55,9 +55,9 @@ class ExceptionsTestSuite(BasicUsageTestSuite): initialization with different arguments and proper error propagation. """ - __module__ = """tests.test_exceptions""" + __module__ = "tests.test_exceptions" - __name__ = """tests.test_exceptions.ExceptionsTestSuite""" + __name__ = "tests.test_exceptions.ExceptionsTestSuite" def test_command_execution_error_with_args(self): """ diff --git a/tests/test_fuzz.py b/tests/test_fuzz.py index e34d9568..824ff0ac 100644 --- a/tests/test_fuzz.py +++ b/tests/test_fuzz.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: """Handle imports with CWE-758 mitigation. @@ -121,9 +121,9 @@ class HypothesisTestSuite(BasicUsageTestSuite): without fragmentation. """ - __module__ = """tests.test_fuzz""" + __module__ = "tests.test_fuzz" - __name__ = """tests.test_fuzz.HypothesisTestSuite""" + __name__ = "tests.test_fuzz.HypothesisTestSuite" @given(st.binary(min_size=1, max_size=1472)) @settings(deadline=None) @@ -144,28 +144,28 @@ def test_multicast_sender_with_random_data(self, data): - If the receiver process encounters an error, the test is skipped. """ theResult = False - fail_fixture = str("""SAY --> HEAR == error""") + fail_fixture = str("SAY --> HEAR == error") _fixture_port_num = self._always_generate_random_port_WHEN_called() try: self.assertIsNotNone(_fixture_port_num) self.assertIsInstance(_fixture_port_num, int) _fixture_SAY_args = [ - """SAY""", - """--port""", + "SAY", + "--port", str(_fixture_port_num), - """--group""", - """'224.0.0.1'""", - """--message""", + "--group", + "'224.0.0.1'", + "--message", f"'{data}'" ] _fixture_HEAR_args = [ - """HEAR""", - """--port""", + "HEAR", + "--port", str(_fixture_port_num), - """--groups""", - """'224.0.0.1'""", - """--group""", - """'224.0.0.1'""" + "--groups", + "'224.0.0.1'", + "--group", + "'224.0.0.1'" ] p = Process( target=multicast.__main__.McastDispatch().doStep, @@ -209,7 +209,7 @@ def test_invalid_Error_WHEN_cli_called_GIVEN_invalid_fuzz_input(self, text): - The CLI output includes the invalid input text """ theResult = False - fail_fixture = str("""XZY? --> Multicast != error""") + fail_fixture = str("XZY? --> Multicast != error") if (self._thepython is not None): try: args = [str(self._thepython), str("-m"), str("multicast"), str(text)] @@ -251,8 +251,8 @@ def test_say_works_WHEN_using_stdin_GIVEN_alnum_of_any_size_fuzz_input(self, tex self.assertIsNotNone(_fixture_port_num) self.assertEqual(type(_fixture_port_num), type(int(0))) _fixture_HEAR_kwargs = { - """port""": _fixture_port_num, - """group""": """224.0.0.1""" + "port": _fixture_port_num, + "group": "224.0.0.1" } self.assertIsNotNone(_fixture_HEAR_kwargs) p = Process( diff --git a/tests/test_hear_cleanup.py b/tests/test_hear_cleanup.py index ce62b847..2009498c 100644 --- a/tests/test_hear_cleanup.py +++ b/tests/test_hear_cleanup.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: try: @@ -47,9 +47,9 @@ class HearCleanupTestSuite(context.BasicUsageTestSuite): protocols. """ - __module__ = """tests.test_hear_cleanup""" + __module__ = "tests.test_hear_cleanup" - __name__ = """tests.test_hear_cleanup.HearCleanupTestSuite""" + __name__ = "tests.test_hear_cleanup.HearCleanupTestSuite" # Class-level constants QUICK_JOIN_TIMEOUT = 1 # Quick check for process termination @@ -74,14 +74,14 @@ def test_cleanup_on_exit(self): - No lingering processes or sockets """ theResult = False - fail_fixture = str("""STOP --> HEAR == error""") + fail_fixture = str("STOP --> HEAR == error") _fixture_port_num = self._the_test_port try: self.assertIsNotNone(_fixture_port_num) self.assertEqual(type(_fixture_port_num), type(int(0))) _fixture_HEAR_kwargs = { - """port""": _fixture_port_num, - """group""": """224.0.0.1""" + "port": _fixture_port_num, + "group": "224.0.0.1" } self.assertIsNotNone(_fixture_HEAR_kwargs) p = Process( diff --git a/tests/test_hear_data_processing.py b/tests/test_hear_data_processing.py index a2c8d3c0..809fd15d 100644 --- a/tests/test_hear_data_processing.py +++ b/tests/test_hear_data_processing.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: try: @@ -43,9 +43,9 @@ class RecvDataProcessingTestSuite(context.BasicUsageTestSuite): """ - __module__ = """tests.test_hear_data_processing""" + __module__ = "tests.test_hear_data_processing" - __name__ = """tests.test_hear_data_processing.RecvDataProcessingTestSuite""" + __name__ = "tests.test_hear_data_processing.RecvDataProcessingTestSuite" def test_multicast_sender_with_no_data(self): """ @@ -53,18 +53,18 @@ def test_multicast_sender_with_no_data(self): """ theResult = False - fail_fixture = str("""SAY -X] RECV? != error""") + fail_fixture = str("SAY -X] RECV? != error") _fixture_port_num = self._always_generate_random_port_WHEN_called() try: self.assertIsNotNone(_fixture_port_num) self.assertIsInstance(_fixture_port_num, int) _fixture_HEAR_args = [ - """--port""", + "--port", str(_fixture_port_num), - """--groups""", - """'224.0.0.1'""", - """--group""", - """'224.0.0.1'""" + "--groups", + "'224.0.0.1'", + "--group", + "'224.0.0.1'" ] p = Process( target=multicast.__main__.main, name="RECV", args=( @@ -87,7 +87,7 @@ def test_multicast_sender_with_no_data(self): p.close() raise unittest.SkipTest(fail_fixture) from _cause p.join(5) - self.assertFalse(p.is_alive(), """RESOURCE LEAK.""") + self.assertFalse(p.is_alive(), "RESOURCE LEAK.") self.assertIsNotNone(p.exitcode) self.assertEqual(int(p.exitcode), int(0)) theResult = (int(p.exitcode) == int(0)) @@ -103,18 +103,18 @@ def test_multicast_sender_with_no_data_before_follow_by_stop(self): """ theResult = False - fail_fixture = str("""SAY -X] HEAR? != error""") + fail_fixture = str("SAY -X] HEAR? != error") _fixture_port_num = self._always_generate_random_port_WHEN_called() try: self.assertIsNotNone(_fixture_port_num) self.assertIsInstance(_fixture_port_num, int) _fixture_HEAR_args = [ - """--port""", + "--port", str(_fixture_port_num), - """--groups""", - """'224.0.0.1'""", - """--group""", - """'224.0.0.1'""" + "--groups", + "'224.0.0.1'", + "--group", + "'224.0.0.1'" ] p = Process( target=multicast.__main__.main, @@ -136,7 +136,7 @@ def test_multicast_sender_with_no_data_before_follow_by_stop(self): while p.is_alive(): sender(group="224.0.0.1", port=_fixture_port_num, data=["STOP"]) p.join(1) - self.assertFalse(p.is_alive(), """HEAR ignored STOP""") + self.assertFalse(p.is_alive(), "HEAR ignored STOP") except Exception as _cause: p.join(3) if p.is_alive(): @@ -144,7 +144,7 @@ def test_multicast_sender_with_no_data_before_follow_by_stop(self): p.close() raise unittest.SkipTest(fail_fixture) from _cause p.join(5) - self.assertFalse(p.is_alive(), """RESOURCE LEAK.""") + self.assertFalse(p.is_alive(), "RESOURCE LEAK.") self.assertIsNotNone(p.exitcode, "Unexpected None == Exit-Code.") self.assertEqual(int(p.exitcode), int(0), f"Unexpected Exit-Code: {p.exitcode}.") theResult = (int(p.exitcode) >= int(0)) @@ -163,9 +163,9 @@ class HearHandleNoneDataTestSuite(context.BasicUsageTestSuite): """ - __module__ = """tests.test_hear_data_processing""" + __module__ = "tests.test_hear_data_processing" - __name__ = """tests.test_hear_data_processing.HearHandleNoneDataTestSuite""" + __name__ = "tests.test_hear_data_processing.HearHandleNoneDataTestSuite" def test_handle_none_data(self): """Test that HearUDPHandler properly handles None data without raising exceptions. diff --git a/tests/test_hear_keyboard_interrupt.py b/tests/test_hear_keyboard_interrupt.py index 89ac2839..9e4c185d 100644 --- a/tests/test_hear_keyboard_interrupt.py +++ b/tests/test_hear_keyboard_interrupt.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: """Handle imports with CWE-758 mitigation. @@ -59,7 +59,7 @@ class TestHearKeyboardInterrupt(BasicUsageTestSuite): SIGINT signals by cleaning up resources and exiting gracefully with the expected status code (130). """ - __module__ = """tests.test_hear_keyboard_interrupt""" + __module__ = "tests.test_hear_keyboard_interrupt" # Constants for test configuration STARTUP_DELAY_SECONDS = 1 # Allow server to start @@ -70,7 +70,7 @@ class TestHearKeyboardInterrupt(BasicUsageTestSuite): def test_hear_keyboard_interrupt(self): """Tests the special hear and stop test""" theResult = False - fail_fixture = str("""C^INT --> HEAR == error""") + fail_fixture = str("C^INT --> HEAR == error") _fixture_port_num = self._the_test_port try: self.assertIsNotNone(_fixture_port_num) @@ -78,15 +78,15 @@ def test_hear_keyboard_interrupt(self): _fixture_cmd = f"{str(sys.executable)} -m coverage run -p --context=Integration" _fixture_HEAR_args = [ _fixture_cmd, - """--source=multicast""", - """-m""", - """multicast""", - """--daemon""", - """HEAR""", - """--port""", + "--source=multicast", + "-m", + "multicast", + "--daemon", + "HEAR", + "--port", str(_fixture_port_num), - """--group""", - """224.0.0.1""" + "--group", + "224.0.0.1" ] self.assertIsNotNone(_fixture_HEAR_args) process = subprocess.Popen( diff --git a/tests/test_hear_server.py b/tests/test_hear_server.py index 7cc2e9dd..9f2b41f4 100644 --- a/tests/test_hear_server.py +++ b/tests/test_hear_server.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: try: @@ -38,9 +38,9 @@ @context.markWithMetaTag("mat", "mat_hear") class McastHearTestSuite(context.BasicUsageTestSuite): - __module__ = """tests.test_hear_server""" + __module__ = "tests.test_hear_server" - __name__ = """tests.test_hear_server.McastHearTestSuite""" + __name__ = "tests.test_hear_server.McastHearTestSuite" @staticmethod def get_default_ip(): @@ -71,9 +71,9 @@ def get_default_ip(): class McastServerTestSuite(McastHearTestSuite): - __module__ = """tests.test_hear_server""" + __module__ = "tests.test_hear_server" - __name__ = """tests.test_hear_server.McastServerTestSuite""" + __name__ = "tests.test_hear_server.McastServerTestSuite" def test_handle_error_without_stop_in_request(self): """ @@ -83,7 +83,7 @@ def test_handle_error_without_stop_in_request(self): the STOP command and cleans up resources. """ theResult = False - fail_fixture = str("""Mock(BLAH) --> Handler-HEAR == error""") + fail_fixture = str("Mock(BLAH) --> Handler-HEAR == error") _fixture_port_num = self._the_test_port try: self.assertIsNotNone(_fixture_port_num) @@ -108,7 +108,7 @@ def test_handle_error_without_stop_in_request(self): def test_handle_error_with_none_request(self): theResult = False - fail_fixture = str("""Mock(EMPTY) --X Handler-HEAR != Safe""") + fail_fixture = str("Mock(EMPTY) --X Handler-HEAR != Safe") _fixture_port_num = self._the_test_port try: self.assertIsNotNone(_fixture_port_num) @@ -134,12 +134,12 @@ def test_handle_error_with_none_request(self): class HearUDPHandlerTestSuite(McastHearTestSuite): - __module__ = """tests.test_hear_server""" + __module__ = "tests.test_hear_server" - __name__ = """tests.test_hear_server.HearUDPHandlerTestSuite""" + __name__ = "tests.test_hear_server.HearUDPHandlerTestSuite" def test_handle_with_none_data_and_sock(self): - fail_fixture = str("""Handler(None, None) --> HEAR == error""") + fail_fixture = str("Handler(None, None) --> HEAR == error") _fixture_port_num = self._the_test_port self.assertIsNotNone(_fixture_port_num) self.assertIsInstance(_fixture_port_num, int) diff --git a/tests/test_hear_server_activate.py b/tests/test_hear_server_activate.py index 27d5c1c8..d5868293 100644 --- a/tests/test_hear_server_activate.py +++ b/tests/test_hear_server_activate.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: try: @@ -44,9 +44,9 @@ class McastServerActivateTestSuite(context.BasicUsageTestSuite): of the multicast server, including socket setup and cleanup procedures. """ - __module__ = """tests.test_hear_server_activate""" + __module__ = "tests.test_hear_server_activate" - __name__ = """tests.test_hear_server_activate.McastServerActivateTestSuite""" + __name__ = "tests.test_hear_server_activate.McastServerActivateTestSuite" def test_server_activate(self): """ diff --git a/tests/test_install_requires.py b/tests/test_install_requires.py index 150ad287..7cb0b8f4 100644 --- a/tests/test_install_requires.py +++ b/tests/test_install_requires.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: try: @@ -41,7 +41,7 @@ @context.markWithMetaTag("mat", "mat_build") class ParseRequirementsTestSuite(BasicUsageTestSuite): - __module__ = """tests.test_install_requires""" + __module__ = "tests.test_install_requires" requirements_file = None """stores the temporary requirements file path for testing""" @@ -49,7 +49,7 @@ class ParseRequirementsTestSuite(BasicUsageTestSuite): def setUp(self): super(ParseRequirementsTestSuite, self).setUp() # Create a temporary requirements file for testing - self.requirements_file = """test_requirements.txt""" + self.requirements_file = "test_requirements.txt" def tearDown(self): """Clean up the temporary requirements file""" @@ -65,19 +65,19 @@ def write_requirements(self, content): def test_simple_version_constraint(self): """Test parsing a simple version constraint.""" - self.write_requirements("""package>=1.0\n""") + self.write_requirements("package>=1.0\n") install_requires = parse_requirements_for_install_requires( readFile(self.requirements_file) ) - self.assertEqual(install_requires, ["""package>=1.0"""]) + self.assertEqual(install_requires, ["package>=1.0"]) def test_multiple_version_constraints(self): """Test parsing multiple version constraints.""" - self.write_requirements("""package>=1.0,!=1.5,<2.0\n""") + self.write_requirements("package>=1.0,!=1.5,<2.0\n") install_requires = parse_requirements_for_install_requires( readFile(self.requirements_file) ) - self.assertEqual(install_requires, ["""package>=1.0"""]) + self.assertEqual(install_requires, ["package>=1.0"]) def test_comments_and_empty_lines(self): """Test handling comments and empty lines.""" @@ -92,7 +92,7 @@ def test_comments_and_empty_lines(self): install_requires = parse_requirements_for_install_requires( readFile(self.requirements_file) ) - self.assertEqual(install_requires, ["""package>=1.0"""]) + self.assertEqual(install_requires, ["package>=1.0"]) def test_options_and_urls_ignored(self): """Test that options and URLs are ignored.""" @@ -111,7 +111,7 @@ def test_options_and_urls_ignored(self): def test_malformed_lines(self): """Test handling of malformed requirement lines.""" - self.write_requirements("""bad_package==\n""") + self.write_requirements("bad_package==\n") install_requires = parse_requirements_for_install_requires( readFile(self.requirements_file) ) @@ -119,7 +119,7 @@ def test_malformed_lines(self): def test_nonexistent_requirements_file(self): """Test behavior when requirements file does not exist.""" - _test_fixture = """nonexistent.txt""" + _test_fixture = "nonexistent.txt" install_requires = parse_requirements_for_install_requires(readFile(_test_fixture)) self.assertEqual(install_requires, []) diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 379461db..52d207ee 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -17,7 +17,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -__module__ = """tests""" +__module__ = "tests" try: try: @@ -42,7 +42,7 @@ @context.markWithMetaTag("mat", "mat_build") class ManifestInclusionTestSuite(BasicUsageTestSuite): - __module__ = """tests.test_manifest""" + __module__ = "tests.test_manifest" def setUp(self): super(ManifestInclusionTestSuite, self).setUp() @@ -94,20 +94,20 @@ def test_sdist_includes_required_files(self): the sdist and checking if the required files are present in the tar archive. """ members, pkg_version = self._build_sdist_and_get_members() - package_prefix = str("""multicast-{}""").format(pkg_version) + package_prefix = str("multicast-{}").format(pkg_version) expected_files = [ - str("""{}/README.md""").format(package_prefix), - str("""{}/LICENSE.md""").format(package_prefix), - str("""{}/requirements.txt""").format(package_prefix), - str("""{}/setup.py""").format(package_prefix), - str("""{}/MANIFEST.in""").format(package_prefix), - str("""{}/setup.cfg""").format(package_prefix), - str("""{}/multicast/__init__.py""").format(package_prefix), - str("""{}/multicast/__main__.py""").format(package_prefix), - str("""{}/multicast/skt.py""").format(package_prefix), - str("""{}/multicast/recv.py""").format(package_prefix), - str("""{}/multicast/send.py""").format(package_prefix), - str("""{}/multicast/hear.py""").format(package_prefix), + str("{}/README.md").format(package_prefix), + str("{}/LICENSE.md").format(package_prefix), + str("{}/requirements.txt").format(package_prefix), + str("{}/setup.py").format(package_prefix), + str("{}/MANIFEST.in").format(package_prefix), + str("{}/setup.cfg").format(package_prefix), + str("{}/multicast/__init__.py").format(package_prefix), + str("{}/multicast/__main__.py").format(package_prefix), + str("{}/multicast/skt.py").format(package_prefix), + str("{}/multicast/recv.py").format(package_prefix), + str("{}/multicast/send.py").format(package_prefix), + str("{}/multicast/hear.py").format(package_prefix), # Include other important files and directories ] for expected_file in expected_files: @@ -124,11 +124,11 @@ def test_sdist_excludes_unwanted_files(self): by building the sdist and verifying that these files are absent from the tar archive. """ members, pkg_version = self._build_sdist_and_get_members() - package_prefix = str("""multicast-{}""").format(pkg_version) + package_prefix = str("multicast-{}").format(pkg_version) unwanted_files = [ - str("""{}/.gitignore""").format(package_prefix), - str("""{}/.github/""").format(package_prefix), - str("""{}/tests/""").format(package_prefix), + str("{}/.gitignore").format(package_prefix), + str("{}/.github/").format(package_prefix), + str("{}/tests/").format(package_prefix), # Exclude other files or directories as specified in MANIFEST.in ] for unwanted_file in unwanted_files: diff --git a/tests/test_usage.py b/tests/test_usage.py index be0f1762..a570a8c0 100644 --- a/tests/test_usage.py +++ b/tests/test_usage.py @@ -43,7 +43,7 @@ """ -__module__ = """tests""" +__module__ = "tests" try: try: @@ -118,14 +118,14 @@ class MulticastTestSuite(context.BasicUsageTestSuite): or the testing environment rather than actual bugs in the code. """ - __module__ = """tests.test_usage""" + __module__ = "tests.test_usage" - __name__ = """tests.test_usage.MulticastTestSuite""" + __name__ = "tests.test_usage.MulticastTestSuite" def test_aborts_WHEN_calling_multicast_GIVEN_invalid_tools(self): """Tests the impossible state for CLI tools given bad tools""" theResult = False - fail_fixture = str("""multicast.__main__.McastDispatch().useTool(JUNK) == error""") + fail_fixture = str("multicast.__main__.McastDispatch().useTool(JUNK) == error") tst_dispatch = multicast.__main__.McastDispatch() test_junk_values = ["", "NoSuchTool", None] try: @@ -158,7 +158,7 @@ def test_say_is_stable_WHEN_calling_multicast_GIVEN_say_tool(self): """ theResult = False fail_fixture = str( - """multicast.__main__.McastDispatch().useTool(SAY, message) != valid exit(0..3)""" + "multicast.__main__.McastDispatch().useTool(SAY, message) != valid exit(0..3)" ) try: with self.assertRaises(SystemExit) as rtn_val_a: @@ -193,7 +193,7 @@ def test_say_is_stable_WHEN_calling_multicast_GIVEN_say_tool(self): def test_recv_aborts_WHEN_calling_multicast_GIVEN_invalid_args(self): """Tests the message argument for failure given invalid input""" theResult = False - fail_fixture = str("""multicast.__main__.McastDispatch().useTool(RECV, junk) != exit(1)""") + fail_fixture = str("multicast.__main__.McastDispatch().useTool(RECV, junk) != exit(1)") try: with self.assertRaises(SystemExit) as rtn_val_c: _ = multicast.__main__.McastDispatch().doStep(["RECV", "--port", "test"]) @@ -226,7 +226,7 @@ def test_recv_aborts_WHEN_calling_multicast_GIVEN_invalid_args(self): def test_hear_aborts_WHEN_calling_multicast_GIVEN_invalid_args(self): """Tests the message argument for failure given invalid input""" theResult = False - fail_fixture = str("""multicast.__main__.McastDispatch().useTool(HEAR, junk) != exit(2)""") + fail_fixture = str("multicast.__main__.McastDispatch().useTool(HEAR, junk) != exit(2)") try: with self.assertRaises(SystemExit) as rtn_val_e: _ = __main__.main(["HEAR", "--port", "test"]) @@ -258,7 +258,7 @@ def test_hear_aborts_WHEN_calling_multicast_GIVEN_invalid_args(self): def test_hear_ignores_WHEN_calling_multicast_GIVEN_invalid_args(self): """Tests the group argument for new auto-default behavior given None.""" theResult = False - fail_fixture = str("""multicast.__main__.main(HEAR, group=None) == ERROR""") + fail_fixture = str("multicast.__main__.main(HEAR, group=None) == ERROR") try: (rtn_val_f, tst_err_rslt_f) = __main__.main( ["HEAR", "--group", "None", "--iface=None"] @@ -280,7 +280,7 @@ def test_hear_ignores_WHEN_calling_multicast_GIVEN_invalid_args(self): def test_hear_is_stable_WHEN_calling_multicast_GIVEN_invalid_tool(self): """Tests the hexdump argument for failure given future tools""" theResult = False - fail_fixture = str("""multicast.__main__.McastDispatch().useTool(HEAR, hex) == error""") + fail_fixture = str("multicast.__main__.McastDispatch().useTool(HEAR, hex) == error") try: self.assertTupleEqual( multicast.__main__.main(["HEAR", "--hex"]), @@ -296,7 +296,7 @@ def test_hear_is_stable_WHEN_calling_multicast_GIVEN_invalid_tool(self): def test_noop_stable_WHEN_calling_multicast_GIVEN_noop_args(self): """Tests the NOOP state for multicast given bad input""" theResult = False - fail_fixture = str("""multicast.__main__.main(NOOP) == Error""") + fail_fixture = str("multicast.__main__.main(NOOP) == Error") try: self.assertIsNotNone(multicast.__main__.main(["NOOP"]), fail_fixture) self.assertIsNotNone(multicast.__main__.main(["NOOP"])[0]) # skipcq: PTC-W0020 @@ -314,7 +314,7 @@ def test_noop_stable_WHEN_calling_multicast_GIVEN_noop_args(self): def test_help_works_WHEN_calling_multicast_GIVEN_help_tool(self): """Tests the HELP argument for help usage""" theResult = False - fail_fixture = str("""multicast.__main__.McastDispatch().useTool(HELP, []) == Empty""") + fail_fixture = str("multicast.__main__.McastDispatch().useTool(HELP, []) == Empty") try: with self.assertRaises(SystemExit) as rtn_val_h: multicast.__main__.McastDispatch().doStep(["HELP"]) @@ -329,41 +329,41 @@ def test_help_works_WHEN_calling_multicast_GIVEN_help_tool(self): def test_hear_works_WHEN_say_works(self): """Tests the basic send and recv test""" theResult = False - fail_fixture = str("""SAY --> HEAR == error""") - sub_fail_fixture = str("""SAY X-> HEAR == Error X-> HEAR :: (Error in SAY)""") + fail_fixture = str("SAY --> HEAR == error") + sub_fail_fixture = str("SAY X-> HEAR == Error X-> HEAR :: (Error in SAY)") try: _fixture_SAY_args = [ - """--port""", - """59991""", - """--group""", - """'224.0.0.1'""", - """--message""", - """'test message'""" + "--port", + "59991", + "--group", + "'224.0.0.1'", + "--message", + "'test message'" ] _fixture_HEAR_args = [ - """--port""", - """59991""", - """--groups""", - """'224.0.0.1'""", - """--group""", - """'224.0.0.1'""" + "--port", + "59991", + "--groups", + "'224.0.0.1'""", + "--group", + "'224.0.0.1'" ] p = Process( target=multicast.__main__.McastDispatch().doStep, name="HEAR", - args=(["""HEAR""", _fixture_HEAR_args]) + args=(["HEAR", _fixture_HEAR_args]) ) p.start() try: tst_fixture_sendDispatch = multicast.__main__.McastDispatch() self.assertIsNotNone( - tst_fixture_sendDispatch.doStep(["""SAY""", _fixture_SAY_args]) + tst_fixture_sendDispatch.doStep(["SAY", _fixture_SAY_args]) ) self.assertIsNotNone( - tst_fixture_sendDispatch.doStep(["""SAY""", _fixture_SAY_args]) + tst_fixture_sendDispatch.doStep(["SAY", _fixture_SAY_args]) ) self.assertIsNotNone( - tst_fixture_sendDispatch.doStep(["""SAY""", _fixture_SAY_args]) + tst_fixture_sendDispatch.doStep(["SAY", _fixture_SAY_args]) ) except Exception as _cause: p.join() @@ -383,32 +383,32 @@ def test_hear_works_WHEN_say_works(self): 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""") + fail_fixture = str("SAY --> HEAR == error") _fixture_port_num = self._the_test_port try: self.assertIsNotNone(_fixture_port_num) self.assertEqual(type(_fixture_port_num), type(int(0))) _fixture_SAY_args = [ - """--port""", + "--port", str(_fixture_port_num), - """--group""", - """'224.0.0.1'""", - """--message""", - """'test message'""" + "--group", + "'224.0.0.1'", + "--message", + "'test message'" ] _fixture_HEAR_args = [ - """HEAR""", - """--port""", + "HEAR", + "--port", str(_fixture_port_num), - """--groups""", - """'224.0.0.1'""", - """--group""", - """'224.0.0.1'""" + "--groups", + "'224.0.0.1'", + "--group", + "'224.0.0.1'" ] p = Process( target=multicast.__main__.McastDispatch().doStep, name="HEAR", - args=(["""HEAR""", _fixture_HEAR_args]) + args=(["HEAR", _fixture_HEAR_args]) ) p.start() try: @@ -439,7 +439,7 @@ def test_hear_works_WHEN_fuzzed_and_say_works(self): def test_say_works_WHEN_using_stdin(self): """Tests the basic send with streamed input test case.""" theResult = False - fail_fixture = str("""STDIN --> SAY == error""") + fail_fixture = str("STDIN --> SAY == error") _fixture_port_num = self._the_test_port try: say = multicast.send.McastSAY() @@ -463,16 +463,16 @@ def test_say_works_WHEN_using_stdin(self): def test_recv_Errors_WHEN_say_not_used(self): """Tests the basic noop recv test""" theResult = False - fail_fixture = str("""NOOP --> RECV != error""") - sub_fail_fixture = str("""NOOP X-> RECV == Error X-> RECV :: (Error in NOOP)""") + fail_fixture = str("NOOP --> RECV != error") + sub_fail_fixture = str("NOOP X-> RECV == Error X-> RECV :: (Error in NOOP)") try: _fixture_RECV_args = [ - """--port""", - """59992""", - """--groups""", - """'224.0.0.1'""", - """--group""", - """'224.0.0.1'""" + "--port", + "59992", + "--groups", + "'224.0.0.1'", + "--group", + "'224.0.0.1'" ] p = Process( target=multicast.__main__.McastDispatch().doStep, @@ -547,19 +547,19 @@ class BasicIntegrationTestSuite(context.BasicUsageTestSuite): ``` """ - __module__ = """tests.test_usage""" + __module__ = "tests.test_usage" - __name__ = """tests.test_usage.BasicIntegrationTestSuite""" + __name__ = "tests.test_usage.BasicIntegrationTestSuite" def setUp(self): super(self.__class__, self).setUp() # skipcq: PYL-E1003 - this is more polymorphic if (self._thepython is None): - self.skipTest(str("""No python cmd to test with!""")) + self.skipTest(str("No python cmd to test with!")) def test_prints_usage_WHEN_called_GIVEN_help_argument(self): """Test case for multicast.__main__ help.""" theResult = False - fail_fixture = str("""multicast.__main__(--help) == not helpful""") + fail_fixture = str("multicast.__main__(--help) == not helpful") try: if (self._thepython is not None): theOutputtxt = context.checkPythonCommand( @@ -580,12 +580,12 @@ def test_prints_usage_WHEN_called_GIVEN_help_argument(self): del err # skipcq - cleanup any error leaks early self.fail(fail_fixture) theResult = False - self.assertTrue(theResult, str("""Could Not find usage from multicast --help""")) + self.assertTrue(theResult, str("Could Not find usage from multicast --help")) def test_prints_usage_WHEN_called_GIVEN_cmd_and_help_argument(self): """Test case for multicast HEAR|RECV|SAY help.""" theResult = None - fail_fixture = str("""multicast.__main__(--help) == not helpful""") + fail_fixture = str("multicast.__main__(--help) == not helpful") try: if (self._thepython is not None): for test_case in [".__main__", ""]: @@ -610,7 +610,7 @@ def test_prints_usage_WHEN_called_GIVEN_cmd_and_help_argument(self): del err # skipcq - cleanup any error leaks early self.fail(fail_fixture) theResult = False - self.assertTrue(theResult, str("""Could Not find usage from multicast CMD --help""")) + self.assertTrue(theResult, str("Could Not find usage from multicast CMD --help")) def test_equivilant_response_WHEN_absolute_vs_implicit(self): """Test case for multicast vs multicast.__main__""" @@ -637,7 +637,7 @@ def test_equivilant_response_WHEN_absolute_vs_implicit(self): err = None del err # skipcq - cleanup any error leaks early theResult = False - self.assertTrue(theResult, str("""Could Not swap multicast for multicast.__main__""")) + self.assertTrue(theResult, str("Could Not swap multicast for multicast.__main__")) def test_prints_version_WHEN_called_GIVEN_version_argument(self): """Test for result from --version argument: python -m multicast.* --version """ @@ -659,12 +659,12 @@ def test_prints_version_WHEN_called_GIVEN_version_argument(self): err = None del err # skipcq - cleanup any error leaks early theResult = False - self.assertTrue(theResult, str("""Could Not find version from multicast --version""")) + self.assertTrue(theResult, str("Could Not find version from multicast --version")) def test_Usage_Error_WHEN_the_help_command_is_called(self): """Test case for multicast* --help.""" theResult = False - fail_fixture = str("""multicast --help == not helpful""") + fail_fixture = str("multicast --help == not helpful") try: if (self._thepython is not None): for test_case in [".__main__", ""]: @@ -684,8 +684,8 @@ def test_Usage_Error_WHEN_the_help_command_is_called(self): theOutputtxt = str(repr(bytes(theOutputtxt))) # or simply: self.assertIsNotNone(theOutputtxt) - self.assertIn(str("""usage:"""), str(theOutputtxt)) - if (str("""usage:""") in str(theOutputtxt)): + self.assertIn(str("usage:"), str(theOutputtxt)) + if (str("usage:") in str(theOutputtxt)): theResult = True or theResult else: theResult = False @@ -698,7 +698,7 @@ def test_Usage_Error_WHEN_the_help_command_is_called(self): del err # skipcq - cleanup any error leaks early self.fail(fail_fixture) theResult = False - self.assertTrue(theResult, str("""Could Not find usage from multicast --help""")) + self.assertTrue(theResult, str("Could Not find usage from multicast --help")) def test_Usage_Error_WHEN_the_help_sub_command_is_called(self): """ @@ -713,7 +713,7 @@ def test_Usage_Error_WHEN_the_help_sub_command_is_called(self): - command: the sub-command being tested """ theResult = False - fail_fixture = str("""multicast [HEAR|RECV] --help == not helpful""") + fail_fixture = str("multicast [HEAR|RECV] --help == not helpful") try: TestCase = namedtuple("TestCase", ["mode", "command"]) inner_fixtures = [ @@ -746,8 +746,8 @@ def test_Usage_Error_WHEN_the_help_sub_command_is_called(self): theOutputtxt = str(repr(bytes(theOutputtxt))) # or simply: self.assertIsNotNone(theOutputtxt) - self.assertIn(str("""usage:"""), str(theOutputtxt)) - if (str("""usage:""") in str(theOutputtxt)): + self.assertIn(str("usage:"), str(theOutputtxt)) + if (str("usage:") in str(theOutputtxt)): theResult = True and theResult else: theResult = False @@ -760,7 +760,7 @@ def test_Usage_Error_WHEN_the_help_sub_command_is_called(self): del err # skipcq - cleanup any error leaks early self.fail(fail_fixture) theResult = False - self.assertTrue(theResult, str("""Could Not find usage from multicast --help""")) + self.assertTrue(theResult, str("Could Not find usage from multicast --help")) def test_profile_WHEN_the_noop_command_is_called(self): """Test case template for profiling""" @@ -811,7 +811,7 @@ def test_stable_WHEN_the_noop_command_is_called(self): err = None del err # skipcq - cleanup any error leaks early theResult = False - self.assertTrue(theResult, str("""Could Not handle multicast NOOP""")) + self.assertTrue(theResult, str("Could Not handle multicast NOOP")) def test_invalid_Error_WHEN_cli_called_GIVEN_bad_input(self): """Test case template for invalid input to multicast CLI.""" @@ -836,7 +836,7 @@ def test_invalid_Error_WHEN_cli_called_GIVEN_bad_input(self): err = None del err # skipcq - cleanup any error leaks early theResult = False - self.assertTrue(theResult, str("""Could Not handle negative inputs""")) + self.assertTrue(theResult, str("Could Not handle negative inputs")) if __name__ == '__main__':