15576550319

Committed 11 Jun 2025 04:56AM UTC coverage: 29.011% (-67.5%) from 96.484%

Build # 15576550319

Build Type

push

github

Committed by

reactive-firewall

Commit Message

[MERGE] Merged and resolved PR #439

Run Details

38 of 131 branches covered (29.01%)

Branch coverage included in aggregate %.

0 of 1 new or added line in 1 file covered. (0.0%)

1451 existing lines in 22 files now uncovered.

622 of 2144 relevant lines covered (29.01%)

1.16 hits per line

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

0.0

/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
UNCOV 20	__module__ = "tests"	×
21
UNCOV 22	try:	×
UNCOV 23	try:	×
UNCOV 24	import context	×
25	except ImportError as _cause: # pragma: no branch
26	del _cause # skipcq - cleanup any error vars early
27	from . import context
UNCOV 28	if not hasattr(context, '__name__') or not context.__name__: # pragma: no branch	×
29	raise ModuleNotFoundError("[CWE-758] Failed to import context") from None
30	else:
UNCOV 31	from context import multicast # pylint: disable=cyclic-import - skipcq: PYL-R0401	×
UNCOV 32	from context import unittest	×
UNCOV 33	from unittest.mock import MagicMock	×
UNCOV 34	from context import Process	×
35	except Exception as baton:
36	raise ImportError("[CWE-758] Failed to import test context") from baton
37
38
UNCOV 39	@context.markWithMetaTag("mat", "hear")	×
UNCOV 40	class RecvDataProcessingTestSuite(context.BasicUsageTestSuite):	×
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
UNCOV 49	__module__ = "tests.test_hear_data_processing"	×
50
UNCOV 51	__name__ = "tests.test_hear_data_processing.RecvDataProcessingTestSuite"	×
52
UNCOV 53	def test_multicast_sender_with_no_data(self) -> None:	×
54	"""
55	Tests the multicast sender and receiver with Empty binary data.
56
57	"""
UNCOV 58	theResult = False	×
UNCOV 59	fail_fixture = "SAY -X] RECV? != error"	×
UNCOV 60	_fixture_port_num = self._always_generate_random_port_WHEN_called()	×
UNCOV 61	_fixture_mcast_addr = "224.0.0.1"	×
UNCOV 62	try:	×
UNCOV 63	self.assertIsNotNone(_fixture_port_num)	×
UNCOV 64	self.assertIsInstance(_fixture_port_num, int)	×
UNCOV 65	self.assertIsNotNone(_fixture_mcast_addr)	×
UNCOV 66	_fixture_HEAR_args = [	×
67	"--port",
68	str(_fixture_port_num),
69	"--groups",
70	f"'{_fixture_mcast_addr}'",
71	"--group",
72	f"'{_fixture_mcast_addr}'",
73	]
UNCOV 74	p = Process(	×
75	target=multicast.__main__.main, name="RECV", args=(
76	"RECV",
77	_fixture_HEAR_args,
78	)
79	)
UNCOV 80	p.start()	×
UNCOV 81	self.assertIsNotNone(p)	×
UNCOV 82	try:	×
UNCOV 83	sender = multicast.send.McastSAY()	×
UNCOV 84	self.assertIsNotNone(sender)	×
UNCOV 85	sender(group=_fixture_mcast_addr, port=_fixture_port_num, ttl=1, data=b'')	×
UNCOV 86	self.assertIsNotNone(p)	×
UNCOV 87	self.assertTrue(p.is_alive(), fail_fixture)	×
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
UNCOV 94	p.join(5)	×
UNCOV 95	self.assertFalse(p.is_alive(), "RESOURCE LEAK.")	×
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
UNCOV 103	self.assertTrue(theResult, fail_fixture)	×
104
UNCOV 105	def test_multicast_sender_with_no_data_before_follow_by_stop(self) -> None:	×
106	"""
107	Tests the multicast sender and receiver with Empty binary data, followed by a stop.
108
109	"""
UNCOV 110	theResult = False	×
UNCOV 111	fail_fixture = "SAY -X] HEAR? != error"	×
UNCOV 112	_fixture_port_num = self._always_generate_random_port_WHEN_called()	×
UNCOV 113	_fixture_mcast_addr = "224.0.0.1"	×
UNCOV 114	try:	×
UNCOV 115	self.assertIsNotNone(_fixture_port_num)	×
UNCOV 116	self.assertIsInstance(_fixture_port_num, int)	×
UNCOV 117	_fixture_HEAR_args = [	×
118	"--port",
119	str(_fixture_port_num),
120	"--groups",
121	f"'{_fixture_mcast_addr}'",
122	"--group",
123	f"'{_fixture_mcast_addr}'",
124	]
UNCOV 125	p = Process(	×
126	target=multicast.__main__.main,
127	name="HEAR",
128	args=(
129	"--daemon",
130	"HEAR",
131	_fixture_HEAR_args,
132	)
133	)
UNCOV 134	p.start()	×
UNCOV 135	self.assertIsNotNone(p)	×
UNCOV 136	try:	×
UNCOV 137	sender = multicast.send.McastSAY()	×
UNCOV 138	self.assertIsNotNone(sender)	×
UNCOV 139	sender(group=_fixture_mcast_addr, port=_fixture_port_num, ttl=1, data=b'')	×
UNCOV 140	self.assertIsNotNone(p)	×
UNCOV 141	self.assertTrue(p.is_alive(), fail_fixture)	×
UNCOV 142	while p.is_alive():	×
UNCOV 143	sender(group=_fixture_mcast_addr, port=_fixture_port_num, data=["STOP"])	×
UNCOV 144	p.join(1)	×
UNCOV 145	self.assertFalse(p.is_alive(), "HEAR ignored STOP")	×
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
UNCOV 152	p.join(5)	×
UNCOV 153	self.assertFalse(p.is_alive(), "RESOURCE LEAK.")	×
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
UNCOV 163	self.assertTrue(theResult, fail_fixture)	×
164
165
UNCOV 166	class HearHandleNoneDataTestSuite(context.BasicUsageTestSuite):	×
167	"""
168	A test suite that uses MagicMock to perform light testing of the default handler for HEAR.
169
170	"""
171
UNCOV 172	__module__ = "tests.test_hear_data_processing"	×
173
UNCOV 174	__name__ = "tests.test_hear_data_processing.HearHandleNoneDataTestSuite"	×
175
UNCOV 176	def test_handle_none_data(self) -> None:	×
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	"""
UNCOV 184	_fixture_port_num = self._always_generate_random_port_WHEN_called()	×
UNCOV 185	self.assertIsNotNone(_fixture_port_num)	×
UNCOV 186	self.assertIsInstance(_fixture_port_num, int)	×
UNCOV 187	handler = multicast.hear.HearUDPHandler(	×
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
UNCOV 191	handler.request = (None, MagicMock())	×
UNCOV 192	mock_socket = handler.request[1]	×
UNCOV 193	handler.handle()	×
194	# Verify that the handler processed the None data case correctly
UNCOV 195	self.assertEqual(	×
196	mock_socket.method_calls, [], "Socket should not be used when data is None"
197	)
198
UNCOV 199	def test_handle_with_invalid_utf8_data(self) -> None:	×
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	"""
UNCOV 207	_fixture_port_num = self._always_generate_random_port_WHEN_called()	×
UNCOV 208	self.assertIsNotNone(_fixture_port_num)	×
UNCOV 209	self.assertIsInstance(_fixture_port_num, int)	×
UNCOV 210	_fixture_client_addr = ("224.0.0.1", _fixture_port_num)	×
UNCOV 211	data = b'\xff\xfe\xfd\xfc' # Invalid UTF-8 bytes	×
UNCOV 212	sock = multicast.genSocket()	×
UNCOV 213	handler = multicast.hear.HearUDPHandler(	×
214	request=(data, sock), client_address=_fixture_client_addr, server=None
215	)
UNCOV 216	try:	×
217	# Mock the processing method
UNCOV 218	handler._process = MagicMock()	×
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
UNCOV 222	self.assertIsNone(handler.server) # Server should remain None	×
UNCOV 223	self.assertEqual(handler.client_address, _fixture_client_addr)	×
224	# Verify no data was processed
UNCOV 225	handler._process.assert_not_called()	×
226	# Test with different invalid UTF-8 sequences
UNCOV 227	for invalid_data in [b'\xff', b'\xfe\xff', b'\xff\xff\xff']:	×
UNCOV 228	handler.request = (invalid_data, sock)	×
UNCOV 229	handler.handle()	×
UNCOV 230	handler._process.assert_not_called()	×
231	except Exception as _cause:
232	self.fail(f"Handler raised an unexpected exception: {_cause}")
233	finally:
234	# Clean up socket
UNCOV 235	multicast.endSocket(sock)	×
236
237
238	if __name__ == '__main__':
239	unittest.main()

reactive-firewall-org / multicast / 15576550319

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