diff --git a/multicast/hear.py b/multicast/hear.py index 4685bf5f..893b8fae 100644 --- a/multicast/hear.py +++ b/multicast/hear.py @@ -264,43 +264,8 @@ def kill_func(a_server): a_server.shutdown() end_thread = threading.Thread(name="Kill_Thread", target=kill_func, args=[self]) end_thread.start() - super(McastServer, self).handle_error(request, client_address) - - -class MyUDPHandler(socketserver.BaseRequestHandler): - """ - Subclasses socketserver.BaseRequestHandler for handling echo function. - - Basically simplifies testing by allowing a trivial echo back (case-insensitive) of string - data, after printing the sender's ip out. - - Minimal Acceptance Testing: - - First set up test fixtures by importing multicast. - - Testcase 0: Multicast should be importable. - - >>> import multicast - >>> multicast.hear is not None - True - >>> from multicast.hear import MyUDPHandler as MyUDPHandler - >>> - - Testcase 1: MyUDPHandler should be automatically imported. - - >>> MyUDPHandler.__name__ is not None - True - >>> - - - """ - - def handle(self): - data = self.request[0].strip() - socket = self.request[1] - print(str("{} wrote:").format(self.client_address[0])) - print(data) - socket.sendto(data.upper(), self.client_address) + else: + super(McastServer, self).handle_error(request, client_address) class HearUDPHandler(socketserver.BaseRequestHandler): @@ -332,30 +297,30 @@ class HearUDPHandler(socketserver.BaseRequestHandler): """ def handle(self): - data = self.request[0].strip() - socket = self.request[1] + (data, sock) = self.request + _sim_data_str = data.strip().replace('\r', '').replace('%', '%%') print(str("{} SAYS: {} to {}").format( - self.client_address[0], str(data), "ALL" + self.client_address[0], str(_sim_data_str), "ALL" )) - if data is not None: - myID = str(socket.getsockname()[0]) + if _sim_data_str is not None: + myID = str(sock.getsockname()[0]) print( str("{me} HEAR: [{you} SAID {what}]").format( - me=myID, you=self.client_address, what=str(data) + me=myID, you=self.client_address, what=str(_sim_data_str) ) ) print( str("{me} SAYS [ HEAR [ {what} SAID {you} ] from {me} ]").format( - me=myID, you=self.client_address, what=str(data) + me=myID, you=self.client_address, what=str(_sim_data_str) ) ) send.McastSAY()._sayStep( # skipcq: PYL-W0212 - module ok self.client_address[0], self.client_address[1], str("HEAR [ {what} SAID {you} ] from {me}").format( - me=myID, you=self.client_address, what=data.upper() + me=myID, you=self.client_address, what=_sim_data_str.upper() ) ) - if """STOP""" in str(data): + if """STOP""" in str(_sim_data_str): raise RuntimeError("SHUTDOWN") from None diff --git a/tests/MulticastUDPClient.py b/tests/MulticastUDPClient.py index 76aadcbd..b11505e9 100644 --- a/tests/MulticastUDPClient.py +++ b/tests/MulticastUDPClient.py @@ -80,12 +80,92 @@ # even if the above stated remedy fails of its essential purpose. ################################################################################ -import socket -import random +__module__ = """tests.MulticastUDPClient""" +"""This is a testing related stand-alone utilities module.""" + + +try: + import sys + if sys.__name__ is None: # pragma: no branch + raise ImportError("[CWE-758] OMG! we could not import sys! ABORT. ABORT.") from None +except Exception as badErr: # pragma: no branch + baton = ImportError(badErr, str("[CWE-758] Test module failed completely.")) + baton.module = __module__ + baton.path = __file__ + baton.__cause__ = badErr + raise baton from badErr + + +try: + if 'os' not in sys.modules: + import os + else: # pragma: no branch + os = sys.modules["""os"""] +except Exception as badErr: # pragma: no branch + baton = ImportError(badErr, str("[CWE-758] Test module failed completely.")) + baton.module = __module__ + baton.path = __file__ + baton.__cause__ = badErr + raise baton from badErr + + +try: + if 'functools' not in sys.modules: + import functools + else: # pragma: no branch + functools = sys.modules["""functools"""] +except Exception as badErr: # pragma: no branch + baton = ImportError(badErr, str("[CWE-758] Test module failed completely.")) + baton.module = __module__ + baton.path = __file__ + baton.__cause__ = badErr + raise baton from badErr + + +try: + if 'socket' not in sys.modules: + import socket + else: # pragma: no branch + socket = sys.modules["""socket"""] +except Exception as badErr: # pragma: no branch + baton = ImportError(badErr, str("[CWE-758] Test module failed completely.")) + baton.module = __module__ + baton.path = __file__ + baton.__cause__ = badErr + raise baton from badErr + + +try: + if 'socketserver' not in sys.modules: + import socketserver + else: # pragma: no branch + socketserver = sys.modules["""socketserver"""] +except Exception as badErr: # pragma: no branch + baton = ImportError(badErr, str("[CWE-758] Test module failed completely.")) + baton.module = __module__ + baton.path = __file__ + baton.__cause__ = badErr + raise baton from badErr + + +try: + if 'random' not in sys.modules: + import random + else: # pragma: no branch + random = sys.modules["""random"""] +except Exception as badErr: # pragma: no branch + baton = ImportError(badErr, str("[CWE-758] Test module failed completely.")) + baton.module = __module__ + baton.path = __file__ + baton.__cause__ = badErr + raise baton from badErr class MCastClient(object): # skipcq: PYL-R0205 - """For use as a test fixture. A trivial implementation of a socket-based object with a function + """ + For use as a test fixture. + + A trivial implementation of a socket-based object with a function named say. The say function of this class performs a send and recv on a given socket and then prints out simple diognostics about the content sent and any response received. @@ -125,6 +205,8 @@ class MCastClient(object): # skipcq: PYL-R0205 """ + __module__ = """tests.MulticastUDPClient.MCastClient""" + _group_addr = None """The multicast group address.""" @@ -132,7 +214,8 @@ class MCastClient(object): # skipcq: PYL-R0205 """The source port for the client.""" def __init__(self, *args, **kwargs): - """Initialize a MCastClient object with optional group address and source port. + """ + Initialize a MCastClient object with optional group address and source port. The client can be initialized with or without specifying a group address and source port. If no source port is provided, a random port between 50000 and 59999 is generated. @@ -207,9 +290,9 @@ def __init__(self, *args, **kwargs): ) @staticmethod - def say(address, port, conn, msg): - """Send a message to a specified multicast address and port, - then receive and print the response. + def say(address, port, sock, msg): + """ + Send a message to a specified multicast address and port, then receive and print it. This function sends a UTF-8 encoded message to the specified multicast address and port using the provided connection. It then waits for a response, decodes it, and prints both @@ -218,7 +301,7 @@ def say(address, port, conn, msg): Args: address (str): The multicast group address to send the message to. port (int): The port number to send the message to. - conn (socket.socket): The socket connection to use for sending and receiving. + sock (socket.socket): The socket connection to use for sending and receiving. msg (str): The message to be sent. Returns: @@ -259,14 +342,48 @@ def say(address, port, conn, msg): for multicast communication. """ - conn.sendto(bytes(msg + "\n", "utf-8"), (address, port)) - received = str(conn.recv(1024), "utf-8") + sock.sendto(bytes(msg + "\n", "utf-8"), (address, port)) + received = str(sock.recv(1024), "utf-8") print(str("Sent: {}").format(msg)) # skipcq: PYL-C0209 - must remain compatible print(str("Received: {}").format(received)) # skipcq: PYL-C0209 - must remain compatible +class MyUDPHandler(socketserver.BaseRequestHandler): + """ + Subclasses socketserver.BaseRequestHandler to handle echo functionality. + + Simplifies testing by echoing back the received string data in uppercase, + after printing the sender's IP address. + + Meta Testing: + + First set up test fixtures by importing test context. + + >>> import tests.MulticastUDPClient as MulticastUDPClient + >>> from MulticastUDPClient import MyUDPHandler as MyUDPHandler + >>> + + Testcase 1: MyUDPHandler should be automatically imported. + + >>> MyUDPHandler.__name__ is not None + True + >>> + + """ + + __module__ = """tests.MulticastUDPClient.MyUDPHandler""" + + def handle(self): + data = self.request[0].strip() + sock = self.request[1] + print(str("{} wrote:").format(self.client_address[0])) + print(data) + sock.sendto(data.upper(), self.client_address) + + def main(): - """The main test operations. + """ + The main test operations. Testing: diff --git a/tests/context.py b/tests/context.py index 13447f9a..66c7041f 100644 --- a/tests/context.py +++ b/tests/context.py @@ -69,6 +69,15 @@ raise ModuleNotFoundError("[CWE-440] OS Failed to import.") from err +try: + if 'random' not in sys.modules: + import random + else: # pragma: no branch + random = sys.modules["""random"""] +except Exception as err: # pragma: no branch + raise ModuleNotFoundError("[CWE-440] Random Failed to import.") from err + + try: if 'unittest' not in sys.modules: import unittest @@ -826,12 +835,27 @@ def setUpClass(cls): """Overrides unittest.TestCase.setUpClass(cls) to set up thepython test fixture.""" cls._thepython = getPythonCommand() + @staticmethod + def _always_generate_random_port_WHEN_called(): + """ + Generates a pseudo-random port number within the dynamic/private port range. + + This method returns a random port number between 49152 and 65535, + compliant with RFC 6335, suitable for temporary testing purposes to + avoid port conflicts. + + Returns: + int: A random port number between 49152 and 65535. + """ + return random.randint(49152, 65535) + def setUp(self): """Overrides unittest.TestCase.setUp(unittest.TestCase). Defaults is to skip test if class is missing thepython test fixture. """ if not self._thepython: self.skipTest(str("""No python cmd to test with!""")) + self._the_test_port = self._always_generate_random_port_WHEN_called() def _get_package_version(self): """ @@ -869,6 +893,13 @@ def test_finds_python_WHEN_testing(self): self.fail(str("""No python cmd to test with!""")) self.test_absolute_truth_and_meaning() + def tearDown(self): + """Overrides unittest.TestCase.tearDown(unittest.TestCase). + Defaults is to reset the random port test fixture. + """ + if self._the_test_port: + self._the_test_port = None + @classmethod def tearDownClass(cls): """Overrides unittest.TestCase.tearDownClass(cls) to clean up thepython test fixture.""" diff --git a/tests/test_fuzz.py b/tests/test_fuzz.py index 1feacc9f..d1ba302d 100644 --- a/tests/test_fuzz.py +++ b/tests/test_fuzz.py @@ -42,10 +42,11 @@ from context import unittest from hypothesis import given, strategies as st from context import Process - import random as _random + from context import random as _random except Exception as err: raise ImportError("[CWE-758] Failed to import test context") from err + class HypothesisTestSuite(context.BasicUsageTestSuite): """ A test suite that uses Hypothesis to perform fuzz testing on the multicast sender and receiver. diff --git a/tests/test_usage.py b/tests/test_usage.py index 59bd925a..58d3c806 100644 --- a/tests/test_usage.py +++ b/tests/test_usage.py @@ -69,7 +69,6 @@ from context import unittest from context import subprocess from context import Process - import random as _random except Exception as err: raise ImportError("[CWE-758] Failed to import test context") from err @@ -342,16 +341,11 @@ def test_hear_works_WHEN_say_works(self): 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() + _fixture_port_num = self._the_test_port try: self.assertIsNotNone(_fixture_port_num) self.assertEqual(type(_fixture_port_num), type(int(0)))