1717# See the License for the specific language governing permissions and
1818# limitations under the License.
1919
20+ """
21+ Test module for verifying cleanup behavior of the multicast hearing mechanism.
22+
23+ This module contains test suites that verify proper resource cleanup and process
24+ termination when the multicast hearing process receives shutdown signals.
25+ """
26+
2027__module__ = "tests"
2128
2229try :
30+ """Handle imports with CWE-758 mitigation.
31+
32+ This implementation uses a nested try-except pattern to:
33+ 1. Attempt direct context import
34+ 2. Fallback to relative import
35+ 3. Validate context module integrity
36+ 4. Import required dependencies
37+
38+ References:
39+ - CWE-758: Reliance on Undefined, Unspecified, or Implementation-Defined Behavior
40+ """
2341 try :
2442 import context
2543 except Exception as _ : # pragma: no branch
@@ -51,12 +69,39 @@ class HearCleanupTestSuite(context.BasicUsageTestSuite):
5169
5270 __name__ = "tests.test_hear_cleanup.HearCleanupTestSuite"
5371
54- # Class-level constants
55- QUICK_JOIN_TIMEOUT = 1 # Quick check for process termination
56- ERROR_JOIN_TIMEOUT = 3 # Timeout when handling errors
57- FINAL_JOIN_TIMEOUT = 15 # Final wait for process cleanup
72+ # Constants for test configuration
73+ STOP_DELAY_SECONDS : int = 1
74+ """
75+ Time to wait for server cleanup after sending `STOP`.
76+
77+ Must be > 0 to ensure server has an opportunity to handle messages.
78+ """
79+
80+ KILL_DELAY_SECONDS : int = 3
81+ """
82+ Average time to wait for process completion after sending `STOP` before sending `SIGKILL`.
83+
84+ Should be sufficient for handling `STOP` messages but not too long.
85+ """
86+
87+ PROCESS_TIMEOUT_SECONDS : int = 15
88+ """
89+ Maximum time to wait for process completion after sending `STOP`.
5890
59- def test_cleanup_on_exit (self ):
91+ Should be sufficient for cleanup but not too long.
92+ """
93+
94+ EXPECTED_STOP_EXIT_CODE : int = 0
95+ """
96+ Expected exit code when process receives `STOP` messages.
97+
98+ `0` = `success` as per POSIX convention.
99+ """
100+
101+ TEST_MULTICAST_GROUP : str = "224.0.0.1"
102+ """Standard multicast group address for testing."""
103+
104+ def test_cleanup_on_exit (self ) -> None :
60105 """Test proper cleanup of McastHEAR when receiving STOP message.
61106
62107 Prerequisites:
@@ -73,15 +118,15 @@ def test_cleanup_on_exit(self):
73118 - Process exits with code 0
74119 - No lingering processes or sockets
75120 """
76- theResult = False
77- fail_fixture = "STOP --> HEAR == error"
78- _fixture_port_num = self ._the_test_port
121+ theResult : bool = False
122+ fail_fixture : str = "STOP --> HEAR == error"
123+ _fixture_port_num : int = self ._the_test_port
79124 try :
80125 self .assertIsNotNone (_fixture_port_num )
81126 self .assertEqual (type (_fixture_port_num ), type (int (0 )))
82127 _fixture_HEAR_kwargs = {
83128 "port" : _fixture_port_num ,
84- "group" : "224.0.0.1" ,
129+ "group" : self . TEST_MULTICAST_GROUP ,
85130 }
86131 self .assertIsNotNone (_fixture_HEAR_kwargs )
87132 p = Process (
@@ -95,25 +140,32 @@ def test_cleanup_on_exit(self):
95140 sender = multicast .send .McastSAY ()
96141 self .assertIsNotNone (sender )
97142 while p .is_alive ():
98- sender (group = "224.0.0.1" , port = _fixture_port_num , ttl = 1 , data = "STOP Test" )
99- p .join (self .QUICK_JOIN_TIMEOUT )
143+ sender (
144+ group = self .TEST_MULTICAST_GROUP , port = _fixture_port_num ,
145+ ttl = 1 , data = "STOP Test" ,
146+ )
147+ p .join (self .STOP_DELAY_SECONDS )
100148 self .assertFalse (p .is_alive ())
101149 except Exception as _cause :
102- p .join (self .ERROR_JOIN_TIMEOUT )
150+ p .join (self .KILL_DELAY_SECONDS )
103151 if p .is_alive ():
104152 p .terminate ()
105153 p .close ()
106154 raise unittest .SkipTest (fail_fixture ) from _cause
107- p .join (self .FINAL_JOIN_TIMEOUT )
155+ p .join (self .PROCESS_TIMEOUT_SECONDS )
108156 self .assertIsNotNone (p .exitcode )
109- self .assertEqual (int (p .exitcode ), int (0 ))
110- theResult = (int (p .exitcode ) <= int (0 ))
157+ self .assertEqual (
158+ int (p .exitcode ),
159+ int (self .EXPECTED_STOP_EXIT_CODE ),
160+ "CEP-8 VIOLATION."
161+ )
162+ theResult = (int (p .exitcode ) <= int (self .EXPECTED_STOP_EXIT_CODE ))
111163 except Exception as err :
112164 context .debugtestError (err )
113165 self .fail (fail_fixture )
114166 theResult = False
115167 self .assertTrue (theResult , fail_fixture )
116168
117169
118- if __name__ == ' __main__'
170+ if __name__ == " __main__"
119171 unittest .main ()
0 commit comments