[STYLE] Cleanup for release (- WIP PR #454 -)

Changes in file multicast/__init__.py:
 * added some documentation
 * style tweaks

Changes in file multicast/__main__.py:
 * added some documentation
 * style tweaks

Changes in file multicast/recv.py:
 * added some documentation
 * style tweaks

Changes in file multicast/skt.py:
 * style tweaks

Changes in file tests/MulticastUDPClient.py:
 * style tweaks

Changes in file tests/__init__.py:
 * added some documentation
 * style tweaks

Changes in file tests/context.py:
 * added some documentation
 * style tweaks

Changes in file tests/test_extra.py:
 * added some documentation

Changes in file tests/test_fuzz.py:
 * added some documentation

Changes in file tests/test_hear_server.py:
 * added some documentation
 * style tweaks

Changes in file tests/test_hear_server_activate.py:
 * added some documentation
 * style tweaks

Changes in file tests/test_usage.py:
 * added some documentation
 * style tweaks

Author: reactive-firewall

multicast/__init__.py

@@ -35,15 +35,15 @@
 	"""exceptions""",
 	"""exceptions.CommandExecutionError""",  # skipcq: PYL-E0603 -- imports ok
 	"""exceptions.ShutdownCommandReceived""",  # skipcq: PYL-E0603 -- imports ok
-	"""exceptions.get_exit_code_from_exception""",
-	"""exceptions.exit_on_exception""",
+	"""exceptions.get_exit_code_from_exception""",  # skipcq: PYL-E0603 -- imports ok
+	"""exceptions.exit_on_exception""",  # skipcq: PYL-E0603 -- imports ok
 	"""get_exit_code_from_exception""",
 	"""exit_on_exception""",
 	"""env""",
 	"""skt""",
-	"""skt.__package__""",
-	"""skt.__module__""",
-	"""skt.__name__""",
+	"""skt.__package__""",  # skipcq: PYL-E0603 -- shadow imports ok
+	"""skt.__module__""",  # skipcq: PYL-E0603 -- shadow imports ok
+	"""skt.__name__""",  # skipcq: PYL-E0603 -- shadow imports ok
 	"""skt.__file__""",  # skipcq: PYL-E0603 -- shadow imports ok
 	"""skt.genSocket""",  # skipcq: PYL-E0603 -- shadow imports ok
 	"""skt.genSocket.__func__""",  # skipcq: PYL-E0603 -- shadow imports ok

multicast/__main__.py

@@ -245,6 +245,24 @@ class McastNope(mtool):
 
 	@classmethod
 	def setupArgs(cls, parser):
+		"""
+		Sets up command-line arguments for the trivial subclass implementation of mtool.
+
+		This method is intended to be overridden by subclasses to define
+		specific command-line arguments. It takes a parser object as an
+		argument, which is typically an instance of `argparse.ArgumentParser`.
+
+		Args:
+			parser (argparse.ArgumentParser): The argument parser to which
+												the arguments should be added.
+
+		Returns:
+			None: This method does not return a value.
+
+		Note:
+			This is trivial implementation make this an optional abstract method. Subclasses may
+			choose to implement it or leave it as a no-op.
+		"""
 		pass  # skipcq: PTC-W0049 - optional abstract method
 
 	@staticmethod
@@ -396,7 +414,14 @@ def setupArgs(cls, parser):
 		Will attempt to add hear args.
 
 		Both HEAR and RECV use the same arguments, and are differentiated only by the global,
-		'--daemon' argument during dispatch.
+		'--daemon' argument during dispatch. The remaining arguments are:
+
+			| Arguments | Notes |
+			|-----------|------------------------------------------------------------|
+			| --deamon  | Enable use of HEAR, otherwise use RECV if omitted          |
+			| --group   | multicast group (ip address) to bind-to for the udp socket |
+			| --groups  | multicast groups to join (should include the bind group)   |
+			| --port    | The UDP port number to listen/filter on for the udp socket |
 
 		Testing:
 
@@ -582,6 +607,12 @@ class McastDispatch(mtool):
 
 	@classmethod
 	def setupArgs(cls, parser) -> None:
+		"""
+		Will attempt to add each subcomponent's args.
+
+		As the dispatch tool, this is more of a proxy implementation to call the sub-components
+		own setupArgs.
+		"""
 		if parser is not None:  # pragma: no branch
 			for sub_tool in sorted(TASK_OPTIONS.keys()):
 				sub_parser = parser.add_parser(sub_tool, help="...")
@@ -619,12 +650,12 @@ def doStep(self, *args, **kwargs) -> tuple:
 			(argz, _) = type(self).parseArgs(*args)
 			service_cmd = argz.cmd_tool
 			argz.__dict__.__delitem__("cmd_tool")
-			_TOOL_MSG = (self.useTool(service_cmd, **argz.__dict__))
+			_TOOL_MSG = (self.useTool(service_cmd, **argz.__dict__))  # skipcq: PYL-C0325
 			if _TOOL_MSG[0]:
 				_response = (0, _TOOL_MSG)
 			else:
 				_response = (70, _TOOL_MSG)
-				if (sys.stdout.isatty()):  # pragma: no cover
+				if sys.stdout.isatty():  # pragma: no cover
 					print(str(_TOOL_MSG))
 		except Exception as _cause:  # pragma: no branch
 			exit_code = exceptions.get_exit_code_from_exception(_cause)

multicast/recv.py

@@ -481,7 +481,7 @@ def tryrecv(msgbuffer: list, chunk: bytes, sock: _socket.socket) -> str:
 
 	"""
 	chunk = sock.recv(multicast._MCAST_DEFAULT_BUFFER_SIZE)  # skipcq: PYL-W0212 - module ok
-	if not (chunk is None):  # pragma: no branch
+	if not (chunk is None):  # skipcq: PYL-C0325 -- avoids flake E714 anti-pattern
 		msgbuffer += str(chunk, encoding='utf8')  # pragma: no cover
 		chunk = None  # pragma: no cover
 	return msgbuffer
@@ -504,14 +504,14 @@ def recvstep(msgbuffer: list, chunk: bytes, sock: _socket.socket) -> str:
 	try:
 		msgbuffer = tryrecv(msgbuffer, chunk, sock)
 	except KeyboardInterrupt:  # pragma: no branch
-		if (sys.stdout.isatty()):  # pragma: no cover
+		if sys.stdout.isatty():  # pragma: no cover
 			module_logger.warning("User Interrupted")
 	except OSError:  # pragma: no branch
 		module_logger.debug("[CWE-440] Nothing happened. There seems to have been an OS error.")
 	finally:
 		if sock:  # pragma: no branch
 			sock = multicast.endSocket(sock)
-	if not (chunk is None):  # pragma: no branch
+	if not (chunk is None):  # skipcq: PYL-C0325 -- avoids flake E714 anti-pattern
 		msgbuffer += str(chunk, encoding='utf8')  # pragma: no cover
 		chunk = None  # pragma: no cover
 	# about 969 bytes in base64 encoded as chars

multicast/skt.py

@@ -205,7 +205,7 @@ def genSocket() -> _socket.socket:
 	"""
 	sock = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM, _socket.IPPROTO_UDP)
 	sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1)
-	if (hasattr(_socket, 'IP_MULTICAST_TTL')):  # pragma: no branch
+	if hasattr(_socket, 'IP_MULTICAST_TTL'):  # pragma: no branch
 		sock.setsockopt(_socket.IPPROTO_IP, _socket.IP_MULTICAST_TTL, _MCAST_DEFAULT_TTL)
 	sock.settimeout(_MCAST_DEFAULT_TTL)
 	return sock

tests/MulticastUDPClient.py

@@ -309,7 +309,7 @@ def say(address: str, port: int, sock: socket.socket, msg: str) -> None:  # prag
 		sock.sendto(bytes(msg + "\n", "utf-8"), (address, port))  # pragma: no cover
 		received = str(sock.recv(1024), "utf-8")  # pragma: no cover
 		sp = " " * 4  # pragma: no cover
-		if (sys.stdout.isatty()):  # pragma: no cover
+		if sys.stdout.isatty():  # pragma: no cover
 			print(f"Sent: {sp}{msg}")  # skipcq: PYL-C0209  -  must remain compatible
 			print(f"Received: {received}")  # skipcq: PYL-C0209  -  must remain compatible
 

tests/__init__.py

@@ -83,6 +83,32 @@
 	import logging
 	try:
 		class ANSIColors:
+			"""
+			A class that defines ANSI color codes for terminal output.
+
+			This class contains ANSI escape sequences for various colors and a
+			mapping of logging levels to their corresponding colors. It also
+			includes a nested class for a colored stream handler that formats
+			log messages with the appropriate colors based on their severity.
+
+			Attributes:
+				BLACK (str): ANSI code for black text.
+				RED (str): ANSI code for red text.
+				GREEN (str): ANSI code for green text.
+				YELLOW (str): ANSI code for yellow text.
+				BLUE (str): ANSI code for blue text.
+				MAGENTA (str): ANSI code for magenta text.
+				CYAN (str): ANSI code for cyan text.
+				GREY (str): ANSI code for grey text.
+				AMBER (str): ANSI code for amber text.
+				REDBG (str): ANSI code for red background.
+				ENDC (str): ANSI code to reset text formatting.
+
+				logging_color (dict): A dictionary mapping logging levels to their
+										corresponding ANSI color codes.
+				logging_level (dict): A dictionary mapping logging levels to their
+										corresponding logging module constants.
+			"""
 			# Define ANSI color codes
 			BLACK = """\033[30m"""
 			RED = """\033[31m"""
@@ -112,7 +138,30 @@ class ANSIColors:
 		}
 
 		class ColoredStreamHandler(logging.StreamHandler):
+			"""
+			A custom logging handler that outputs with color formatting based on the log level.
+
+			This handler checks if the output stream is a terminal and applies
+			the appropriate ANSI color codes to the log messages. If the output
+			is not a terminal, it outputs plain text without color.
+
+			Methods:
+				emit(record: logging.LogRecord) -> None:
+					Formats and writes the log message to the stream with
+					appropriate color based on the log level.
+			"""
+
 			def emit(self, record: logging.LogRecord) -> None:
+				"""
+				Formats and writes the log message to the stream with color.
+
+				Args:
+					record (logging.LogRecord): The log record containing
+												information about the log event.
+
+				Raises:
+					ValueError: If the log level is invalid or not recognized.
+				"""
 				# Get the log level as a string
 				loglevel = record.levelname.lower()
 				# Validate the log level
@@ -429,6 +478,21 @@ def get_test_suite(
 
 	# Helper function to add test cases to suite
 	def add_test_cases(test_cases: list) -> None:
+		"""
+		Adds a list of test cases to the test suite.
+
+		This function iterates over the provided list of test cases. If a test case
+		is an instance of unittest.TestSuite, it adds the entire suite to the
+		main suite. If the test case is a class, it loads the tests from that
+		class and adds them to the main suite.
+
+		Args:
+			test_cases (list): A list of test cases or test suites to be added
+								to the main test suite.
+
+		Returns:
+			None: This function does not return a value.
+		"""
 		for test_case in test_cases:
 			if isinstance(test_case, unittest.TestSuite):
 				# Handle doctests

tests/context.py

@@ -181,7 +181,7 @@ def markWithMetaTag(*marks: str) -> callable:
 	except ImportError:
 		pytest_available = False
 
-	def decorator(cls) -> any:
+	def decorator(cls) -> any:  # skipcq: PY-D0003 -- decorator ok
 		if pytest_available:
 			for mark in marks:
 				cls = pytest.mark.__getattr__(mark)(cls)
@@ -263,7 +263,8 @@ def __check_cov_before_py():
 		thepython = str(f"{str(thecov)} run -p")  # skipcq: TCV-002
 	else:  # pragma: no branch
 		try:
-			import coverage as coverage
+			# pylint: disable=cyclic-import - skipcq: PYL-R0401, PYL-C0414
+			import coverage as coverage  # skipcq: PYL-C0414
 			if coverage.__name__ is not None:
 				thepython = str("{} -m coverage run -p").format(str(sys.executable))
 		except Exception:
@@ -1072,7 +1073,7 @@ def setUp(self) -> None:
 			self.skipTest(self.NO_PYTHON_ERROR)  # skipcq: TCV-002
 		self._the_test_port = self._always_generate_random_port_WHEN_called()
 
-	def _should_get_package_version_WHEN_valid(self) -> None:
+	def _should_get_package_version_WHEN_valid(self) -> packaging.version.Version:
 		"""
 		Retrieve the current version of the package.
 
@@ -1085,6 +1086,7 @@ def _should_get_package_version_WHEN_valid(self) -> None:
 			ImportError -- If the multicast package cannot be imported.
 
 		"""
+		parsed_version: packaging.version.Version = None
 		try:
 			self.assertIsNotNone(multicast.__module__, "Version will be effectively None.")
 			self.assertIsNotNone(multicast.__version__, "Version is not valid.")
@@ -1100,9 +1102,10 @@ def _should_get_package_version_WHEN_valid(self) -> None:
 				len(parsed_version.release) >= 2,
 				"Version must have at least major.minor components."
 			)
-			return parsed_version
 		except ImportError:
 			self.fail("Failed to import the multicast package to retrieve version.")
+		finally:
+			return parsed_version
 
 	@unittest.skipUnless(True, "Insanity Test. Good luck debugging.")
 	def test_absolute_truth_and_meaning(self) -> None:

tests/test_extra.py

@@ -107,7 +107,7 @@ def onlyIfHasDocs(has_docs: bool) -> callable:
 		True
 
 	"""
-	def decorator(cls: callable) -> callable:
+	def decorator(cls: callable) -> callable:  # skipcq: PY-D0003 -- decorator ok
 		if not has_docs:
 			# Create an empty class with a method that returns None
 			return type(cls.__name__, (object,), {

tests/test_fuzz.py

@@ -91,7 +91,7 @@ def onlyIfHasHypothesis(has_hypothesis: bool) -> callable:
 		callable: A decorator function that returns either the original class or a dummy class
 		with a placeholder method, depending on the has_hypothesis flag.
 	"""
-	def decorator(cls: callable) -> callable:
+	def decorator(cls: callable) -> callable:  # skipcq: PY-D0003 -- decorator ok
 		if not has_hypothesis:
 			# Create an empty class with a method that returns None
 			return type(cls.__name__, (object,), {