16954025989

Committed 14 Aug 2025 02:00AM UTC coverage: 97.128%. Remained the same

Build # 16954025989

Build Type

push

github

Committed by

reactive-firewall

Commit Message

[HOTFIX] Fix for CI-CD regression

Changes in file .github/workflows/CI-CHGLOG.yml:
 * hotfix for regression of download-artifacts action logic after version bump

Run Details

127 of 133 branches covered (95.49%)

Branch coverage included in aggregate %.

2173 of 2235 relevant lines covered (97.23%)

6.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.87

/tests/test_hear_data_processing.py

#! /usr/bin/env python3
# -*- coding: utf-8 -*-

# Multicast Python Module (Testing)
# ..................................
# Copyright (c) 2017-2025, Mr. Walls
# ..................................
# Licensed under MIT (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# ..........................................
# https://github.com/reactive-firewall-org/multicast/tree/HEAD/LICENSE.md
# ..........................................
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

__module__ = "tests"

try:
        try:
                import context
        except ImportError as _cause:  # pragma: no branch
                del _cause  # skipcq - cleanup any error vars early
                from . import context
        if not hasattr(context, '__name__') or not context.__name__:  # pragma: no branch
                raise ModuleNotFoundError("[CWE-758] Failed to import context") from None
        else:
                from context import multicast  # pylint: disable=cyclic-import - skipcq: PYL-R0401
                from context import unittest
                from unittest.mock import MagicMock
                from context import Process
except Exception as baton:
        raise ImportError("[CWE-758] Failed to import test context") from baton


@context.markWithMetaTag("mat", "hear")
class RecvDataProcessingTestSuite(context.BasicUsageTestSuite):
        """
        A test suite that validates the multicast sender and receiver's handling of empty data.

        Test cases:
                - Sending empty binary data.
                - Sending empty data followed by a stop command.
        """

        __module__ = "tests.test_hear_data_processing"

        __name__ = "tests.test_hear_data_processing.RecvDataProcessingTestSuite"

        def test_multicast_sender_with_no_data(self) -> None:
                """
                Tests the multicast sender and receiver with Empty binary data.

                """
                theResult = False
                fail_fixture = "SAY -X] RECV? != error"
                _fixture_port_num = self._always_generate_random_port_WHEN_called()
                _fixture_mcast_addr = "224.0.0.1"
                try:
                        self.assertIsNotNone(_fixture_port_num)
                        self.assertIsInstance(_fixture_port_num, int)
                        self.assertIsNotNone(_fixture_mcast_addr)
                        _fixture_HEAR_args = [
                                "--port",
                                str(_fixture_port_num),
                                "--groups",
                                f"'{_fixture_mcast_addr}'",
                                "--group",
                                f"'{_fixture_mcast_addr}'",
                        ]
                        p = Process(
                                target=multicast.__main__.main, name="RECV", args=(
                                        "RECV",
                                        _fixture_HEAR_args,
                                )
                        )
                        p.start()
                        self.assertIsNotNone(p)
                        try:
                                sender = multicast.send.McastSAY()
                                self.assertIsNotNone(sender)
                                sender(group=_fixture_mcast_addr, port=_fixture_port_num, ttl=1, data=b'')
                                self.assertIsNotNone(p)
                                self.assertTrue(p.is_alive(), fail_fixture)
                        except Exception as _root_cause:
                                p.join(3)
                                if p.is_alive():
                                        p.terminate()
                                        p.close()
                                raise unittest.SkipTest(fail_fixture) from _root_cause
                        p.join(5)
                        self.assertFalse(p.is_alive(), "RESOURCE LEAK.")
                        self.assertIsNotNone(p.exitcode)
                        self.assertEqual(int(p.exitcode), int(0))
                        theResult = (int(p.exitcode) == int(0))
                except Exception as _cause:
                        context.debugtestError(_cause)
                        self.fail(fail_fixture)
                        theResult = False
                self.assertTrue(theResult, fail_fixture)

        def test_multicast_sender_with_no_data_before_follow_by_stop(self) -> None:
                """
                Tests the multicast sender and receiver with Empty binary data, followed by a stop.

                """
                theResult = False
                fail_fixture = "SAY -X] HEAR? != error"
                _fixture_port_num = self._always_generate_random_port_WHEN_called()
                _fixture_mcast_addr = "224.0.0.1"
                try:
                        self.assertIsNotNone(_fixture_port_num)
                        self.assertIsInstance(_fixture_port_num, int)
                        _fixture_HEAR_args = [
                                "--port",
                                str(_fixture_port_num),
                                "--groups",
                                f"'{_fixture_mcast_addr}'",
                                "--group",
                                f"'{_fixture_mcast_addr}'",
                        ]
                        p = Process(
                                target=multicast.__main__.main,
                                name="HEAR",
                                args=(
                                        "--daemon",
                                        "HEAR",
                                        _fixture_HEAR_args,
                                )
                        )
                        p.start()
                        self.assertIsNotNone(p)
                        try:
                                sender = multicast.send.McastSAY()
                                self.assertIsNotNone(sender)
                                sender(group=_fixture_mcast_addr, port=_fixture_port_num, ttl=1, data=b'')
                                self.assertIsNotNone(p)
                                self.assertTrue(p.is_alive(), fail_fixture)
                                while p.is_alive():
                                        sender(group=_fixture_mcast_addr, port=_fixture_port_num, data=["STOP"])
                                        p.join(1)
                                self.assertFalse(p.is_alive(), "HEAR ignored STOP")
                        except Exception as _root_cause:
                                p.join(3)
                                if p.is_alive():
                                        p.terminate()
                                        p.close()
                                raise unittest.SkipTest(fail_fixture) from _root_cause
                        p.join(5)
                        self.assertFalse(p.is_alive(), "RESOURCE LEAK.")
                        self.assertIsNotNone(p.exitcode, "Unexpected None == Exit-Code.")
                        self.assertEqual(int(p.exitcode), int(0), f"Unexpected Exit-Code: {p.exitcode}.")
                        theResult = (int(p.exitcode) >= int(0))
                except unittest.SkipTest as baton:
                        raise unittest.SkipTest() from baton
                except Exception as _cause:
                        context.debugtestError(_cause)
                        self.fail(fail_fixture)
                        theResult = False
                self.assertTrue(theResult, fail_fixture)


class HearHandleNoneDataTestSuite(context.BasicUsageTestSuite):
        """
        A test suite that uses MagicMock to perform light testing of the default handler for HEAR.

        """

        __module__ = "tests.test_hear_data_processing"

        __name__ = "tests.test_hear_data_processing.HearHandleNoneDataTestSuite"

        def test_handle_none_data(self) -> None:
                """Test that HearUDPHandler properly handles None data without raising exceptions.

                This test verifies that:
                        1. The handler initializes correctly with None request data
                        2. The handle() method executes without errors
                        3. The handler properly processes the None data case
                """
                _fixture_port_num = self._always_generate_random_port_WHEN_called()
                self.assertIsNotNone(_fixture_port_num)
                self.assertIsInstance(_fixture_port_num, int)
                handler = multicast.hear.HearUDPHandler(
                        request=(None, None), client_address=('224.0.0.1', _fixture_port_num), server=None
                )
                # Mock the socket to prevent actual network calls
                handler.request = (None, MagicMock())
                mock_socket = handler.request[1]
                handler.handle()
                # Verify that the handler processed the None data case correctly
                self.assertEqual(
                        mock_socket.method_calls, [], "Socket should not be used when data is None"
                )

        def test_handle_with_invalid_utf8_data(self) -> None:
                """Test that HearUDPHandler silently ignores invalid UTF-8 data.

                This test verifies that:
                        1. The handler continues processing when receiving invalid UTF-8 data
                        2. No exception is raised
                        3. The handler silently ignores the decoding error
                """
                _fixture_port_num = self._always_generate_random_port_WHEN_called()
                self.assertIsNotNone(_fixture_port_num)
                self.assertIsInstance(_fixture_port_num, int)
                _fixture_client_addr = ("224.0.0.1", _fixture_port_num)
                data = b'\xff\xfe\xfd\xfc'  # Invalid UTF-8 bytes
                sock = multicast.genSocket()
                handler = multicast.hear.HearUDPHandler(
                        request=(data, sock), client_address=_fixture_client_addr, server=None
                )
                try:
                        # Mock the processing method
                        handler._process = MagicMock()
                        # Should silently ignore invalid UTF-8 data
                        handler.handle()  # If no exception is raised, the test passes
                        # Verify handler state after processing invalid data
                        self.assertIsNone(handler.server)  # Server should remain None
                        self.assertEqual(handler.client_address, _fixture_client_addr)
                        # Verify no data was processed
                        handler._process.assert_not_called()
                        # Test with different invalid UTF-8 sequences
                        for invalid_data in [b'\xff', b'\xfe\xff', b'\xff\xff\xff']:
                                handler.request = (invalid_data, sock)
                                handler.handle()
                                handler._process.assert_not_called()
                except Exception as _cause:
                        self.fail(f"Handler raised an unexpected exception: {_cause}")
                finally:
                        # Clean up socket
                        multicast.endSocket(sock)


if __name__ == '__main__':
        unittest.main()

1	#! /usr/bin/env python3
2	# -- coding: utf-8 --
3
4	# Multicast Python Module (Testing)
5	# ..................................
6	# Copyright (c) 2017-2025, Mr. Walls
7	# ..................................
8	# Licensed under MIT (the "License");
9	# you may not use this file except in compliance with the License.
10	# You may obtain a copy of the License at
11	# ..........................................
12	# https://github.com/reactive-firewall-org/multicast/tree/HEAD/LICENSE.md
13	# ..........................................
14	# Unless required by applicable law or agreed to in writing, software
15	# distributed under the License is distributed on an "AS IS" BASIS,
16	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	# See the License for the specific language governing permissions and
18	# limitations under the License.
19
20	__module__ = "tests"	4✔
21
22	try:	4✔
23	try:	4✔
24	import context	4✔
25	except ImportError as _cause: # pragma: no branch
26	del _cause # skipcq - cleanup any error vars early
27	from . import context
28	if not hasattr(context, '__name__') or not context.__name__: # pragma: no branch	4✔
29	raise ModuleNotFoundError("[CWE-758] Failed to import context") from None
30	else:
31	from context import multicast # pylint: disable=cyclic-import - skipcq: PYL-R0401	4✔
32	from context import unittest	4✔
33	from unittest.mock import MagicMock	4✔
34	from context import Process	4✔
35	except Exception as baton:
36	raise ImportError("[CWE-758] Failed to import test context") from baton
37
38
39	@context.markWithMetaTag("mat", "hear")	4✔
40	class RecvDataProcessingTestSuite(context.BasicUsageTestSuite):	4✔
41	"""
42	A test suite that validates the multicast sender and receiver's handling of empty data.
43
44	Test cases:
45	- Sending empty binary data.
46	- Sending empty data followed by a stop command.
47	"""
48
49	__module__ = "tests.test_hear_data_processing"	4✔
50
51	__name__ = "tests.test_hear_data_processing.RecvDataProcessingTestSuite"	4✔
52
53	def test_multicast_sender_with_no_data(self) -> None:	4✔
54	"""
55	Tests the multicast sender and receiver with Empty binary data.
56
57	"""
58	theResult = False	4✔
59	fail_fixture = "SAY -X] RECV? != error"	4✔
60	_fixture_port_num = self._always_generate_random_port_WHEN_called()	4✔
61	_fixture_mcast_addr = "224.0.0.1"	4✔
62	try:	4✔
63	self.assertIsNotNone(_fixture_port_num)	4✔
64	self.assertIsInstance(_fixture_port_num, int)	4✔
65	self.assertIsNotNone(_fixture_mcast_addr)	4✔
66	_fixture_HEAR_args = [	4✔
67	"--port",
68	str(_fixture_port_num),
69	"--groups",
70	f"'{_fixture_mcast_addr}'",
71	"--group",
72	f"'{_fixture_mcast_addr}'",
73	]
74	p = Process(	4✔
75	target=multicast.__main__.main, name="RECV", args=(
76	"RECV",
77	_fixture_HEAR_args,
78	)
79	)
80	p.start()	4✔
81	self.assertIsNotNone(p)	4✔
82	try:	4✔
83	sender = multicast.send.McastSAY()	4✔
84	self.assertIsNotNone(sender)	4✔
85	sender(group=_fixture_mcast_addr, port=_fixture_port_num, ttl=1, data=b'')	4✔
86	self.assertIsNotNone(p)	4✔
87	self.assertTrue(p.is_alive(), fail_fixture)	4✔
88	except Exception as _root_cause:
89	p.join(3)
90	if p.is_alive():
91	p.terminate()
92	p.close()
93	raise unittest.SkipTest(fail_fixture) from _root_cause
94	p.join(5)	4✔
95	self.assertFalse(p.is_alive(), "RESOURCE LEAK.")	4✔
96	self.assertIsNotNone(p.exitcode)
97	self.assertEqual(int(p.exitcode), int(0))
98	theResult = (int(p.exitcode) == int(0))
99	except Exception as _cause:
100	context.debugtestError(_cause)
101	self.fail(fail_fixture)
102	theResult = False
103	self.assertTrue(theResult, fail_fixture)	4✔
104
105	def test_multicast_sender_with_no_data_before_follow_by_stop(self) -> None:	4✔
106	"""
107	Tests the multicast sender and receiver with Empty binary data, followed by a stop.
108
109	"""
110	theResult = False	4✔
111	fail_fixture = "SAY -X] HEAR? != error"	4✔
112	_fixture_port_num = self._always_generate_random_port_WHEN_called()	4✔
113	_fixture_mcast_addr = "224.0.0.1"	4✔
114	try:	4✔
115	self.assertIsNotNone(_fixture_port_num)	4✔
116	self.assertIsInstance(_fixture_port_num, int)	4✔
117	_fixture_HEAR_args = [	4✔
118	"--port",
119	str(_fixture_port_num),
120	"--groups",
121	f"'{_fixture_mcast_addr}'",
122	"--group",
123	f"'{_fixture_mcast_addr}'",
124	]
125	p = Process(	4✔
126	target=multicast.__main__.main,
127	name="HEAR",
128	args=(
129	"--daemon",
130	"HEAR",
131	_fixture_HEAR_args,
132	)
133	)
134	p.start()	4✔
135	self.assertIsNotNone(p)	4✔
136	try:	4✔
137	sender = multicast.send.McastSAY()	4✔
138	self.assertIsNotNone(sender)	4✔
139	sender(group=_fixture_mcast_addr, port=_fixture_port_num, ttl=1, data=b'')	4✔
140	self.assertIsNotNone(p)	4✔
141	self.assertTrue(p.is_alive(), fail_fixture)	4✔
142	while p.is_alive():	4✔
143	sender(group=_fixture_mcast_addr, port=_fixture_port_num, data=["STOP"])	4✔
144	p.join(1)	4✔
145	self.assertFalse(p.is_alive(), "HEAR ignored STOP")	4✔
146	except Exception as _root_cause:
147	p.join(3)
148	if p.is_alive():
149	p.terminate()
150	p.close()
151	raise unittest.SkipTest(fail_fixture) from _root_cause
152	p.join(5)	4✔
153	self.assertFalse(p.is_alive(), "RESOURCE LEAK.")	4✔
154	self.assertIsNotNone(p.exitcode, "Unexpected None == Exit-Code.")
155	self.assertEqual(int(p.exitcode), int(0), f"Unexpected Exit-Code: {p.exitcode}.")
156	theResult = (int(p.exitcode) >= int(0))
157	except unittest.SkipTest as baton:	×
158	raise unittest.SkipTest() from baton	×
159	except Exception as _cause:
160	context.debugtestError(_cause)
161	self.fail(fail_fixture)
162	theResult = False
163	self.assertTrue(theResult, fail_fixture)	4✔
164
165
166	class HearHandleNoneDataTestSuite(context.BasicUsageTestSuite):	4✔
167	"""
168	A test suite that uses MagicMock to perform light testing of the default handler for HEAR.
169
170	"""
171
172	__module__ = "tests.test_hear_data_processing"	4✔
173
174	__name__ = "tests.test_hear_data_processing.HearHandleNoneDataTestSuite"	4✔
175
176	def test_handle_none_data(self) -> None:	4✔
177	"""Test that HearUDPHandler properly handles None data without raising exceptions.
178
179	This test verifies that:
180	1. The handler initializes correctly with None request data
181	2. The handle() method executes without errors
182	3. The handler properly processes the None data case
183	"""
184	_fixture_port_num = self._always_generate_random_port_WHEN_called()	4✔
185	self.assertIsNotNone(_fixture_port_num)	4✔
186	self.assertIsInstance(_fixture_port_num, int)	4✔
187	handler = multicast.hear.HearUDPHandler(	4✔
188	request=(None, None), client_address=('224.0.0.1', _fixture_port_num), server=None
189	)
190	# Mock the socket to prevent actual network calls
191	handler.request = (None, MagicMock())	4✔
192	mock_socket = handler.request[1]	4✔
193	handler.handle()	4✔
194	# Verify that the handler processed the None data case correctly
195	self.assertEqual(	4✔
196	mock_socket.method_calls, [], "Socket should not be used when data is None"
197	)
198
199	def test_handle_with_invalid_utf8_data(self) -> None:	4✔
200	"""Test that HearUDPHandler silently ignores invalid UTF-8 data.
201
202	This test verifies that:
203	1. The handler continues processing when receiving invalid UTF-8 data
204	2. No exception is raised
205	3. The handler silently ignores the decoding error
206	"""
207	_fixture_port_num = self._always_generate_random_port_WHEN_called()	4✔
208	self.assertIsNotNone(_fixture_port_num)	4✔
209	self.assertIsInstance(_fixture_port_num, int)	4✔
210	_fixture_client_addr = ("224.0.0.1", _fixture_port_num)	4✔
211	data = b'\xff\xfe\xfd\xfc' # Invalid UTF-8 bytes	4✔
212	sock = multicast.genSocket()	4✔
213	handler = multicast.hear.HearUDPHandler(	4✔
214	request=(data, sock), client_address=_fixture_client_addr, server=None
215	)
216	try:	4✔
217	# Mock the processing method
218	handler._process = MagicMock()	4✔
219	# Should silently ignore invalid UTF-8 data
220	handler.handle() # If no exception is raised, the test passes
221	# Verify handler state after processing invalid data
222	self.assertIsNone(handler.server) # Server should remain None	4✔
223	self.assertEqual(handler.client_address, _fixture_client_addr)	4✔
224	# Verify no data was processed
225	handler._process.assert_not_called()	4✔
226	# Test with different invalid UTF-8 sequences
227	for invalid_data in [b'\xff', b'\xfe\xff', b'\xff\xff\xff']:	4✔
228	handler.request = (invalid_data, sock)	4✔
229	handler.handle()	4✔
230	handler._process.assert_not_called()	4✔
231	except Exception as _cause:
232	self.fail(f"Handler raised an unexpected exception: {_cause}")
233	finally:
234	# Clean up socket
235	multicast.endSocket(sock)	4✔
236
237
238	if __name__ == '__main__':
239	unittest.main()

reactive-firewall-org / multicast / 16954025989

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous