Enhance test coverage reporting and refactor test_hear_cleanup module

.github/workflows/Tests.yml

@@ -117,6 +117,8 @@ jobs:
       uses: codecov/codecov-action@0565863a31f2c772f9f0395002a31e3f06189574 # v5.4.0
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
+        job_code: ${{ join( [github.run_id, github.run_number], '-' ) }}
+        os: ${{ matrix.os }}
         files: ./coverage.xml
         directory: .
         flags: multicast,${{ matrix.os }},${{ matrix.python-version }}
@@ -267,7 +269,9 @@ jobs:
       uses: codecov/codecov-action@0565863a31f2c772f9f0395002a31e3f06189574 # v5.4.0
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
-        files: ./test-reports/coverage_supplement.xml
+        job_code: ${{ join( [github.run_id, github.run_number], '-' ) }}
+        os: ${{ matrix.os }}
+        files: ./test-reports/coverage_supplement.xml,./test-reports/coverage.xml
         directory: .
         flags: multicast,${{ matrix.os }},${{ matrix.python-version }}
         name: multicast-github-${{ matrix.os }}-${{ matrix.python-version }}

tests/check_integration_coverage

@@ -144,9 +144,15 @@ fi
 
 TEST_MCAST_PORT=$(((RANDOM % 16384) + 49152));
 TEST_MCAST_GROUP='224.0.0.1' ;
-PY_3_CMD=$(command -v python3);
+COVERAGE_DATA_FILE="./coverage_integration" ;
+if [[ -x "${PYTHON}" ]] ; then
+	PY_3_CMD="${PYTHON}"
+else
+	PY_3_CMD=$(command -v python3);
+fi ;
+printf "%s\n" "::debug:: Will Use ${PY_3_CMD}"
 # shellcheck disable=SC2086
-COVERAGE_CMD="${PY_3_CMD} -m coverage run --source=multicast -p -m"
+COVERAGE_CMD="${PY_3_CMD} -m coverage run -p --source=multicast -m";
 export COVERAGE_CMD
 
 # start test-suite
@@ -155,7 +161,8 @@ make -j1 -f Makefile test-reports || EXIT_CODE=3 ;
 printf "%s\n\n" "Start of Log:" > ./"${LOG_FILE}" ; wait ;
 printf "%s\n\n" "Will use host-port: [${TEST_MCAST_GROUP} : ${TEST_MCAST_PORT}]" >> ./"${LOG_FILE}" ; wait ;
 printf "%s\n\n" "Start of Integration Test one-shot:" >> ./"${LOG_FILE}" ; wait ;
-( sleep 1 ; "${COVERAGE_CMD} multicast --use-std RECV --port ${TEST_MCAST_PORT} --groups ${TEST_MCAST_GROUP} --group ${TEST_MCAST_GROUP}" 2>./"${ERR_FILE}" & sleep 6 ; kill -9 $! 2>/dev/null ) >> ./"${LOG_FILE}" & true ;
+# shellcheck disable=SC2086
+( sleep 1 ; ${COVERAGE_CMD} multicast --use-std RECV --port "${TEST_MCAST_PORT}" --groups "${TEST_MCAST_GROUP}" --group "${TEST_MCAST_GROUP}" 2>./"${ERR_FILE}" & sleep 6 ; kill -9 $! 2>/dev/null ) >> ./"${LOG_FILE}" & true ;
 
 for PCOUNT in $(seq 1 5) ; do
 	# shellcheck disable=SC2086
@@ -184,14 +191,15 @@ printf "%s\n\n" "End of Integration Test daemon" >> ./"${LOG_FILE}" ; wait ;
 
 # cleanup from test-suite
 
-COVERAGE_DATE_FILE="./coverage_integration"
-
-$(command -v python3) -m coverage combine --data-file="${COVERAGE_DATE_FILE}" 2>/dev/null || EXIT_CODE=2 ;
-if [[ -r "${COVERAGE_DATE_FILE}" ]] ; then
-	$(command -v python3) -m coverage report -m --data-file="${COVERAGE_DATE_FILE}" 2>/dev/null || EXIT_CODE=2 ;
-	$(command -v python3) -m coverage xml -o ./test-reports/coverage_supplement.xml --include=multicast/* --data-file="${COVERAGE_DATE_FILE}" || EXIT_CODE=2 ;
+# shellcheck disable=SC2086
+${PY_3_CMD} -m coverage combine --data-file="${COVERAGE_DATA_FILE}" ./.coverage.* || EXIT_CODE=2 ;
+if [[ -r "${COVERAGE_DATA_FILE}" ]] ; then
+	# shellcheck disable=SC2086
+	${PY_3_CMD} -m coverage report -m --data-file="${COVERAGE_DATA_FILE}" 2>/dev/null || EXIT_CODE=2 ;
+	# shellcheck disable=SC2086
+	${PY_3_CMD} -m coverage xml -o ./test-reports/coverage_supplement.xml --include=multicast/* --data-file="${COVERAGE_DATA_FILE}" || EXIT_CODE=2 ;
 else
-	printf "%s\n" "Coverage collection FAILED!" 2>/dev/null || EXIT_CODE=2 ;
+	printf "%s\n" "::error file='${0}',title='No Coverage':: Coverage collection FAILED!" >>./"${ERR_FILE}" || EXIT_CODE=2 ;
 fi ;
 wait ;
 cp -f ./"${LOG_FILE}" ./test-reports/integration_data_log.log 2>/dev/null ; wait ;

tests/test_hear_cleanup.py

@@ -17,9 +17,27 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+"""
+Test module for verifying cleanup behavior of the multicast hearing mechanism.
+
+This module contains test suites that verify proper resource cleanup and process
+termination when the multicast hearing process receives shutdown signals.
+"""
+
 __module__ = "tests"
 
 try:
+	"""Handle imports with CWE-758 mitigation.
+
+	This implementation uses a nested try-except pattern to:
+	1. Attempt direct context import
+	2. Fallback to relative import
+	3. Validate context module integrity
+	4. Import required dependencies
+
+	References:
+	- CWE-758: Reliance on Undefined, Unspecified, or Implementation-Defined Behavior
+	"""
 	try:
 		import context
 	except Exception as _:  # pragma: no branch
@@ -51,12 +69,39 @@ class HearCleanupTestSuite(context.BasicUsageTestSuite):
 
 	__name__ = "tests.test_hear_cleanup.HearCleanupTestSuite"
 
-	# Class-level constants
-	QUICK_JOIN_TIMEOUT = 1  # Quick check for process termination
-	ERROR_JOIN_TIMEOUT = 3  # Timeout when handling errors
-	FINAL_JOIN_TIMEOUT = 15  # Final wait for process cleanup
+	# Constants for test configuration
+	STOP_DELAY_SECONDS: int = 1
+	"""
+	Time to wait for server cleanup after sending `STOP`.
+
+	Must be > 0 to ensure server has an opportunity to handle messages.
+	"""
+
+	KILL_DELAY_SECONDS: int = 3
+	"""
+	Average time to wait for process completion after sending `STOP` before sending `SIGKILL`.
+
+	Should be sufficient for handling `STOP` messages but not too long.
+	"""
+
+	PROCESS_TIMEOUT_SECONDS: int = 15
+	"""
+	Maximum time to wait for process completion after sending `STOP`.
 
-	def test_cleanup_on_exit(self):
+	Should be sufficient for cleanup but not too long.
+	"""
+
+	EXPECTED_STOP_EXIT_CODE: int = 0
+	"""
+	Expected exit code when process receives `STOP` messages.
+
+	`0` = `success` as per POSIX convention.
+	"""
+
+	TEST_MULTICAST_GROUP: str = "224.0.0.1"
+	"""Standard multicast group address for testing."""
+
+	def test_cleanup_on_exit(self) -> None:
 		"""Test proper cleanup of McastHEAR when receiving STOP message.
 
 		Prerequisites:
@@ -73,15 +118,15 @@ def test_cleanup_on_exit(self):
 			- Process exits with code 0
 			- No lingering processes or sockets
 		"""
-		theResult = False
-		fail_fixture = "STOP --> HEAR == error"
-		_fixture_port_num = self._the_test_port
+		theResult: bool = False
+		fail_fixture: str = "STOP --> HEAR == error"
+		_fixture_port_num: int = 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",
+				"group": self.TEST_MULTICAST_GROUP,
 			}
 			self.assertIsNotNone(_fixture_HEAR_kwargs)
 			p = Process(
@@ -95,25 +140,32 @@ def test_cleanup_on_exit(self):
 				sender = multicast.send.McastSAY()
 				self.assertIsNotNone(sender)
 				while p.is_alive():
-					sender(group="224.0.0.1", port=_fixture_port_num, ttl=1, data="STOP Test")
-					p.join(self.QUICK_JOIN_TIMEOUT)
+					sender(
+						group=self.TEST_MULTICAST_GROUP, port=_fixture_port_num,
+						ttl=1, data="STOP Test",
+					)
+					p.join(self.STOP_DELAY_SECONDS)
 				self.assertFalse(p.is_alive())
 			except Exception as _cause:
-				p.join(self.ERROR_JOIN_TIMEOUT)
+				p.join(self.KILL_DELAY_SECONDS)
 				if p.is_alive():
 					p.terminate()
 					p.close()
 				raise unittest.SkipTest(fail_fixture) from _cause
-			p.join(self.FINAL_JOIN_TIMEOUT)
+			p.join(self.PROCESS_TIMEOUT_SECONDS)
 			self.assertIsNotNone(p.exitcode)
-			self.assertEqual(int(p.exitcode), int(0))
-			theResult = (int(p.exitcode) <= int(0))
+			self.assertEqual(
+				int(p.exitcode),
+				int(self.EXPECTED_STOP_EXIT_CODE),
+				"CEP-8 VIOLATION.",
+			)
+			theResult = (int(p.exitcode) <= int(self.EXPECTED_STOP_EXIT_CODE))
 		except Exception as err:
 			context.debugtestError(err)
 			self.fail(fail_fixture)
 			theResult = False
 		self.assertTrue(theResult, fail_fixture)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
 	unittest.main()