[COVERAGE] improving coverage slightly

multicast/hear.py

@@ -264,43 +264,8 @@
 					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):
@@ -333,12 +298,12 @@
 
 	def handle(self):
 		data = self.request[0].strip()
-		socket = self.request[1]
+		sock = self.request[1]
 		print(str("{} SAYS: {} to {}").format(
 			self.client_address[0], str(data), "ALL"
 		))
 		if data is not None:
-			myID = str(socket.getsockname()[0])
+			myID = str(sock.getsockname()[0])
 			print(
 				str("{me} HEAR: [{you} SAID {what}]").format(
 					me=myID, you=self.client_address, what=str(data)

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,14 +205,17 @@ class MCastClient(object):  # skipcq: PYL-R0205
 
 	"""
 
+	__module__ = """tests.MulticastUDPClient.MCastClient"""
+
 	_group_addr = None
 	"""The multicast group address."""
 
 	_source_port = None
 	"""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.
@@ -208,8 +291,8 @@ 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.
+		"""
+		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
@@ -265,8 +348,42 @@ def say(address, port, conn, msg):
 		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()
+		socket = self.request[1]
+		print(str("{} wrote:").format(self.client_address[0]))
+		print(data)
+		socket.sendto(data.upper(), self.client_address)
+
+
 def main():
-	"""The main test operations.
+	"""
+	The main test operations.
 
 	Testing:
 

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

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.

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