16584110377

Committed 22 Jul 2025 10:18PM UTC coverage: 97.128%. Remained the same

Build # 16584110377

Build Type

push

github

Committed by

reactive-firewall

Commit Message

[FIX] Initial Tuning of over limited concurancy settings

Changes in file .github/workflows/CI-BUILD.yml:
 * Removed limits reverting to previous LGV

Changes in file .github/workflows/CI-CHGLOG.yml:
 * Removed limits reverting to previous LGV

Changes in file .github/workflows/CI-DOCS.yml:
 * Removed limits reverting to previous LGV

Changes in file .github/workflows/CI-MATs.yml:
 * Removed limits reverting to previous LGV

Changes in file .github/workflows/Tests.yml:
 * Removed limits (kept throttling)

Changes in file .github/workflows/bandit.yml:
 * Removed limits reverting to previous LGV

Changes in file .github/workflows/codeql-analysis.yml:
 * Retuned limit for schedule runs

Changes in file .github/workflows/flake8.yml:
 * Removed limits reverting to previous LGV

Changes in file .github/workflows/makefile-lint.yml:
 * Removed limits reverting to previous LGV

Changes in file .github/workflows/markdown-lint.yml:
 * Removed limits reverting to previous LGV

Changes in file .github/workflows/scorecard.yml:
 * Retuned limit for schedule runs

Changes in file .github/workflows/shellcheck.yml:
 * Removed limits reverting to previous LGV

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

100.0

/tests/MulticastUDPClient.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.

# THIS FILE IS A TEST FILE ONLY.

# Disclaimer of Warranties.
# A. YOU EXPRESSLY ACKNOWLEDGE AND AGREE THAT, TO THE EXTENT PERMITTED BY
#    APPLICABLE LAW, USE OF THIS SHELL SCRIPT AND ANY SERVICES PERFORMED
#    BY OR ACCESSED THROUGH THIS SHELL SCRIPT IS AT YOUR SOLE RISK AND
#    THAT THE ENTIRE RISK AS TO SATISFACTORY QUALITY, PERFORMANCE, ACCURACY AND
#    EFFORT IS WITH YOU.
#
# B. TO THE MAXIMUM EXTENT PERMITTED BY APPLICABLE LAW, THIS SHELL SCRIPT
#    AND SERVICES ARE PROVIDED "AS IS" AND "AS AVAILABLE", WITH ALL FAULTS AND
#    WITHOUT WARRANTY OF ANY KIND, AND THE AUTHOR OF THIS SHELL SCRIPT'S LICENSORS
#    (COLLECTIVELY REFERRED TO AS "THE AUTHOR" FOR THE PURPOSES OF THIS DISCLAIMER)
#    HEREBY DISCLAIM ALL WARRANTIES AND CONDITIONS WITH RESPECT TO THIS SHELL SCRIPT
#    SOFTWARE AND SERVICES, EITHER EXPRESS, IMPLIED OR STATUTORY, INCLUDING, BUT
#    NOT LIMITED TO, THE IMPLIED WARRANTIES AND/OR CONDITIONS OF
#    MERCHANTABILITY, SATISFACTORY QUALITY, FITNESS FOR A PARTICULAR PURPOSE,
#    ACCURACY, QUIET ENJOYMENT, AND NON-INFRINGEMENT OF THIRD PARTY RIGHTS.
#
# C. THE AUTHOR DOES NOT WARRANT AGAINST INTERFERENCE WITH YOUR ENJOYMENT OF THE
#    THE AUTHOR's SOFTWARE AND SERVICES, THAT THE FUNCTIONS CONTAINED IN, OR
#    SERVICES PERFORMED OR PROVIDED BY, THIS SHELL SCRIPT WILL MEET YOUR
#    REQUIREMENTS, THAT THE OPERATION OF THIS SHELL SCRIPT OR SERVICES WILL
#    BE UNINTERRUPTED OR ERROR-FREE, THAT ANY SERVICES WILL CONTINUE TO BE MADE
#    AVAILABLE, THAT THIS SHELL SCRIPT OR SERVICES WILL BE COMPATIBLE OR
#    WORK WITH ANY THIRD PARTY SOFTWARE, APPLICATIONS OR THIRD PARTY SERVICES,
#    OR THAT DEFECTS IN THIS SHELL SCRIPT OR SERVICES WILL BE CORRECTED.
#    INSTALLATION OF THIS THE AUTHOR SOFTWARE MAY AFFECT THE USABILITY OF THIRD
#    PARTY SOFTWARE, APPLICATIONS OR THIRD PARTY SERVICES.
#
# D. YOU FURTHER ACKNOWLEDGE THAT THIS SHELL SCRIPT AND SERVICES ARE NOT
#    INTENDED OR SUITABLE FOR USE IN SITUATIONS OR ENVIRONMENTS WHERE THE FAILURE
#    OR TIME DELAYS OF, OR ERRORS OR INACCURACIES IN, THE CONTENT, DATA OR
#    INFORMATION PROVIDED BY THIS SHELL SCRIPT OR SERVICES COULD LEAD TO
#    DEATH, PERSONAL INJURY, OR SEVERE PHYSICAL OR ENVIRONMENTAL DAMAGE,
#    INCLUDING WITHOUT LIMITATION THE OPERATION OF NUCLEAR FACILITIES, AIRCRAFT
#    NAVIGATION OR COMMUNICATION SYSTEMS, AIR TRAFFIC CONTROL, LIFE SUPPORT OR
#    WEAPONS SYSTEMS.
#
# E. NO ORAL OR WRITTEN INFORMATION OR ADVICE GIVEN BY THE AUTHOR
#    SHALL CREATE A WARRANTY. SHOULD THIS SHELL SCRIPT OR SERVICES PROVE DEFECTIVE,
#    YOU ASSUME THE ENTIRE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION.
#
#    Limitation of Liability.
# F. TO THE EXTENT NOT PROHIBITED BY APPLICABLE LAW, IN NO EVENT SHALL THE AUTHOR
#    BE LIABLE FOR PERSONAL INJURY, OR ANY INCIDENTAL, SPECIAL, INDIRECT OR
#    CONSEQUENTIAL DAMAGES WHATSOEVER, INCLUDING, WITHOUT LIMITATION, DAMAGES
#    FOR LOSS OF PROFITS, CORRUPTION OR LOSS OF DATA, FAILURE TO TRANSMIT OR
#    RECEIVE ANY DATA OR INFORMATION, BUSINESS INTERRUPTION OR ANY OTHER
#    COMMERCIAL DAMAGES OR LOSSES, ARISING OUT OF OR RELATED TO YOUR USE OR
#    INABILITY TO USE THIS SHELL SCRIPT OR SERVICES OR ANY THIRD PARTY
#    SOFTWARE OR APPLICATIONS IN CONJUNCTION WITH THIS SHELL SCRIPT OR
#    SERVICES, HOWEVER CAUSED, REGARDLESS OF THE THEORY OF LIABILITY (CONTRACT,
#    TORT OR OTHERWISE) AND EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE
#    POSSIBILITY OF SUCH DAMAGES. SOME JURISDICTIONS DO NOT ALLOW THE EXCLUSION
#    OR LIMITATION OF LIABILITY FOR PERSONAL INJURY, OR OF INCIDENTAL OR
#    CONSEQUENTIAL DAMAGES, SO THIS LIMITATION MAY NOT APPLY TO YOU. In no event
#    shall THE AUTHOR's total liability to you for all damages (other than as may
#    be required by applicable law in cases involving personal injury) exceed
#    the amount of five dollars ($5.00). The foregoing limitations will apply
#    even if the above stated remedy fails of its essential purpose.
################################################################################

__module__ = "tests"
"""This is a testing related stand-alone utilities module.

This module provides test fixtures and utilities for testing multicast communication.
It includes a basic UDP client implementation and handler for testing multicast
functionality.

Classes:
        MCastClient: Test fixture for multicast client operations.
        MyUDPHandler: UDP request handler for echo functionality.

Functions:
        main: Entry point for test operations.

Example:
        >>> from tests.MulticastUDPClient import MCastClient
        >>> client = MCastClient(grp_addr='224.0.0.1', src_port=59259)
        >>> isinstance(client._source_port, int)
        True
        >>>

"""

__name__ = "tests.MulticastUDPClient"  # skipcq: PYL-W0622

try:
        import sys
        if not hasattr(sys, 'modules') or not sys.modules:  # pragma: no branch
                raise ModuleNotFoundError(
                        "[CWE-440] OMG! sys.modules is not available or empty."
                ) from None
except ImportError as _cause:  # pragma: no branch
        raise ImportError("[CWE-440] Unable to import sys module.") from _cause

try:
        import socket
        import socketserver
except ImportError as _cause:  # pragma: no branch
        raise ImportError("[CWE-758] Test module failed completely.") from _cause

try:
        import random
except ImportError as _cause:  # pragma: no branch
        raise ModuleNotFoundError("[CWE-758] Test module failed to randomize.") from _cause


class MCastClient(object):  # skipcq: PYL-R0205
        """
        For use as a test fixture.

        A trivial implementation of a socket-based object with a function
        named say. The say function of this class performs a send and recv on a given socket and
        then prints out simple diognostics about the content sent and any response received.

        Testing:

                First some test fixtures:

                >>> import socket as socket
                >>> import random as random
                >>>

        Testcase 0: test the class MCastClient is.
                A: Test that the MulticastUDPClient component is importable.
                B: Test that the MCastClient class is importable.

                >>> import tests.MulticastUDPClient
                >>> from MulticastUDPClient import MCastClient as MCastClient
                >>> MCastClient is not None
                True
                >>>

        Testcase 1: Test the class MCastClient has a say function.
                A: Test that the MulticastUDPClient component is importable.
                B: Test that the MCastClient class is importable.
                C: Test that the MCastClient class has the function named say.

                >>> import tests.MulticastUDPClient
                >>> from MulticastUDPClient import MCastClient as MCastClient
                >>> MCastClient is not None
                True
                >>> MCastClient.say is not None
                True
                >>> type(MCastClient.say)
                <class 'function'>
                >>>


        """

        __module__ = "tests.MulticastUDPClient.MCastClient"

        _group_addr = None
        """The multicast group address."""

        _source_port = None
        """The source port for the client."""

        # skipcq: TCV-002
        def __init__(self, *args, **kwargs) -> None:  # pragma: no cover
                """
                Initialize a MCastClient object with optional group address and source port.

                The client can be initialized with or without specifying a group address and source port.
                If no source port is provided, a random port between 50000 and 59999 is generated.

                Args:
                        *args: Variable length argument list (Unused).
                        **kwargs: Arbitrary keyword arguments.
                                - grp_addr (str): The multicast group address.
                                - src_port (int): The source port for the client.

                Meta Testing:

                        First set up test fixtures by importing test context.

                                >>> import tests.MulticastUDPClient as MulticastUDPClient
                                >>> from MulticastUDPClient import MCastClient as MCastClient
                                >>>

                        Testcase 1: Initialization without any arguments.

                                >>> client = MCastClient()
                                >>> 50000 <= client._source_port <= 59999
                                True
                                >>> client._group_addr is None
                                True
                                >>>

                        Testcase 2: Initialization with only group address.

                                >>> tst_args = {}
                                >>> client = MCastClient(grp_addr="224.0.0.1")
                                >>> client._group_addr
                                '224.0.0.1'
                                >>> 50000 <= client._source_port <= 59999
                                True
                                >>>

                        Testcase 3: Initialization with only source port.

                                >>> client = MCastClient(src_port=55555)
                                >>> client._source_port
                                55555
                                >>> client._group_addr is None
                                True
                                >>>

                        Testcase 4: Initialization with both group address and source port.

                                >>> client = MCastClient(grp_addr="224.0.0.2", src_port=55556)
                                >>> client._group_addr
                                '224.0.0.2'
                                >>> client._source_port
                                55556
                                >>>


                """
                # skipcq: TCV-002
                if "grp_addr" in kwargs:  # pragma: no branch
                        self._group_addr = kwargs.get("grp_addr", None)  # skipcq: PTC-W0039 - ensure None
                if "src_port" in kwargs:  # pragma: no branch
                        self._source_port = kwargs.get("src_port", 0)
                else:  # pragma: no branch
                        self._source_port = int(
                                50000 + (
                                        int(random.SystemRandom().randbytes(int(60000).__sizeof__()).hex(), 16) % 9999
                                )
                        )

        # skipcq: TCV-002
        @staticmethod
        def say(address: str, port: int, sock: socket.socket, msg: str) -> None:  # pragma: no cover
                """
                Send a message to a specified multicast address and port, then receive and print it.

                This function sends a UTF-8 encoded message to the specified multicast address and port
                using the provided connection. It then waits for a response, decodes it, and prints both
                the sent and received messages.

                Args:
                        address (str): The multicast group address to send the message to.
                        port (int): The port number to send the message to.
                        sock (socket.socket): The socket connection to use for sending and receiving.
                        msg (str): The message to be sent.

                Returns:
                        None

                Prints:
                        The sent message and the received response.

                Meta Testing:

                        First, set up test fixtures:

                                >>> import unittest.mock
                                >>> from MulticastUDPClient import MCastClient
                                >>>

                        Testcase 1: Test sending and receiving a message.

                                >>> mock_socket = unittest.mock.Mock()
                                >>> mock_socket.recv.return_value = b"Response received"
                                >>> client = MCastClient()
                                >>> client.say("224.0.0.1", 59991, mock_socket, "Test message")
                                Sent:     Test message
                                Received: Response received
                                >>>

                        Testcase 2: Test sending a 'STOP' message.

                                >>> mock_socket.recv.return_value = b"Stopped"
                                >>> client.say("224.0.0.1", 59991, mock_socket, "STOP")
                                Sent:     STOP
                                Received: Stopped
                                >>>


                Note:
                        This function assumes that the connection is already properly configured
                        for multicast communication.

                """
                # skipcq: TCV-002
                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
                        print(f"Sent: {sp}{msg}")  # skipcq: PYL-C0209  -  must remain compatible
                        print(f"Received: {received}")  # skipcq: PYL-C0209  -  must remain compatible


class MyUDPHandler(socketserver.BaseRequestHandler):
        """
        Subclasses socketserver.BaseRequestHandler to handle echo functionality.

        Simplifies testing by echoing back the received string data in uppercase,
        after printing the sender's IP address.

        Meta Testing:

                First set up test fixtures by importing test context.

                        >>> import tests.MulticastUDPClient as MulticastUDPClient
                        >>> from MulticastUDPClient import MyUDPHandler as MyUDPHandler
                        >>>

                Testcase 1: MyUDPHandler should be automatically imported.

                        >>> MyUDPHandler.__name__ is not None
                        True
                        >>>

        """

        __module__ = "tests.MulticastUDPClient.MyUDPHandler"

        # skipcq: TCV-002
        def handle(self) -> None:  # pragma: no cover
                """
                Handle incoming UDP requests.

                This method overrides the `handle` method from `socketserver.BaseRequestHandler`
                to process incoming UDP messages. It receives a message from a client, echoes it
                back in uppercase, and prints diagnostic information.

                Meta Testing:

                        First set up test fixtures by importing test context.

                                >>> import socket
                                >>> import threading
                                >>> from tests.MulticastUDPClient import MyUDPHandler
                                >>>

                        Testcase 1: Test handling a simple message.

                                >>> import socketserver
                                >>> server = socketserver.UDPServer(('localhost', 0), MyUDPHandler)
                                >>> threading.Thread(target=server.serve_forever, daemon=True).start()
                                >>> client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                                >>> client_socket.sendto("hello world\n", server.server_address)
                                >>> data, _ = client_socket.recvfrom(1024)
                                >>> data
                                b'HELLO WORLD\n'
                                >>> server.shutdown()
                                >>>

                        Testcase 2: Test handling an empty message.

                                >>> import socketserver
                                >>> server = socketserver.UDPServer(('localhost', 0), MyUDPHandler)
                                >>> threading.Thread(target=server.serve_forever, daemon=True).start()
                                >>> client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                                >>> client_socket.sendto("\n", server.server_address)
                                >>> data, _ = client_socket.recvfrom(1024)
                                >>> data
                                b'\n'
                                >>> server.shutdown()
                                >>>

                Note:
                        This method assumes that the incoming request tuple contains a string and a socket,
                        as per `socketserver.BaseRequestHandler` for datagram services.

                """
                # skipcq: TCV-002
                data = self.request[0].strip()  # pragma: no cover
                sock = self.request[1]  # pragma: no cover
                print(f"{self.client_address[0]} wrote: ")  # pragma: no cover
                print(data)  # pragma: no cover
                sock.sendto(data.upper(), self.client_address)  # pragma: no cover


# skipcq: TCV-002
def main() -> None:  # pragma: no cover
        """
        The main test operations.

        Testing:

                First some test fixtures:

                >>> import socket as socket
                >>> import random as random
                >>>

        Testcase 0: test the function main is.
                A: Test that the MulticastUDPClient component is importable.
                B: Test that the MulticastUDPClient has a main function.

                >>> import tests.MulticastUDPClient
                >>> tests.MulticastUDPClient.main is not None
                True
                >>> type(tests.MulticastUDPClient.main)
                <class 'function'>
                >>>


        """
        # skipcq: TCV-002
        HOST, PORT = "224.0.0.1", 59991  # pragma: no cover
        data = "TEST This is a test"  # pragma: no cover
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)  # pragma: no cover
        tsts_fxr = MCastClient()  # pragma: no cover
        print(str((HOST, PORT)))  # pragma: no cover
        tsts_fxr.say(HOST, PORT, sock, data)  # pragma: no cover
        tsts_fxr.say(HOST, PORT, sock, "STOP")  # pragma: no cover


if __name__ == "__main__":  # pragma: no branch
        main()  # skipcq: TCV-002
        # skipcq: PYL-R1722
        exit(0)  # skipcq: PYL-R1722 -- intentionally allow overwriteing exit for testing.

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	# THIS FILE IS A TEST FILE ONLY.
21
22	# Disclaimer of Warranties.
23	# A. YOU EXPRESSLY ACKNOWLEDGE AND AGREE THAT, TO THE EXTENT PERMITTED BY
24	# APPLICABLE LAW, USE OF THIS SHELL SCRIPT AND ANY SERVICES PERFORMED
25	# BY OR ACCESSED THROUGH THIS SHELL SCRIPT IS AT YOUR SOLE RISK AND
26	# THAT THE ENTIRE RISK AS TO SATISFACTORY QUALITY, PERFORMANCE, ACCURACY AND
27	# EFFORT IS WITH YOU.
28	#
29	# B. TO THE MAXIMUM EXTENT PERMITTED BY APPLICABLE LAW, THIS SHELL SCRIPT
30	# AND SERVICES ARE PROVIDED "AS IS" AND "AS AVAILABLE", WITH ALL FAULTS AND
31	# WITHOUT WARRANTY OF ANY KIND, AND THE AUTHOR OF THIS SHELL SCRIPT'S LICENSORS
32	# (COLLECTIVELY REFERRED TO AS "THE AUTHOR" FOR THE PURPOSES OF THIS DISCLAIMER)
33	# HEREBY DISCLAIM ALL WARRANTIES AND CONDITIONS WITH RESPECT TO THIS SHELL SCRIPT
34	# SOFTWARE AND SERVICES, EITHER EXPRESS, IMPLIED OR STATUTORY, INCLUDING, BUT
35	# NOT LIMITED TO, THE IMPLIED WARRANTIES AND/OR CONDITIONS OF
36	# MERCHANTABILITY, SATISFACTORY QUALITY, FITNESS FOR A PARTICULAR PURPOSE,
37	# ACCURACY, QUIET ENJOYMENT, AND NON-INFRINGEMENT OF THIRD PARTY RIGHTS.
38	#
39	# C. THE AUTHOR DOES NOT WARRANT AGAINST INTERFERENCE WITH YOUR ENJOYMENT OF THE
40	# THE AUTHOR's SOFTWARE AND SERVICES, THAT THE FUNCTIONS CONTAINED IN, OR
41	# SERVICES PERFORMED OR PROVIDED BY, THIS SHELL SCRIPT WILL MEET YOUR
42	# REQUIREMENTS, THAT THE OPERATION OF THIS SHELL SCRIPT OR SERVICES WILL
43	# BE UNINTERRUPTED OR ERROR-FREE, THAT ANY SERVICES WILL CONTINUE TO BE MADE
44	# AVAILABLE, THAT THIS SHELL SCRIPT OR SERVICES WILL BE COMPATIBLE OR
45	# WORK WITH ANY THIRD PARTY SOFTWARE, APPLICATIONS OR THIRD PARTY SERVICES,
46	# OR THAT DEFECTS IN THIS SHELL SCRIPT OR SERVICES WILL BE CORRECTED.
47	# INSTALLATION OF THIS THE AUTHOR SOFTWARE MAY AFFECT THE USABILITY OF THIRD
48	# PARTY SOFTWARE, APPLICATIONS OR THIRD PARTY SERVICES.
49	#
50	# D. YOU FURTHER ACKNOWLEDGE THAT THIS SHELL SCRIPT AND SERVICES ARE NOT
51	# INTENDED OR SUITABLE FOR USE IN SITUATIONS OR ENVIRONMENTS WHERE THE FAILURE
52	# OR TIME DELAYS OF, OR ERRORS OR INACCURACIES IN, THE CONTENT, DATA OR
53	# INFORMATION PROVIDED BY THIS SHELL SCRIPT OR SERVICES COULD LEAD TO
54	# DEATH, PERSONAL INJURY, OR SEVERE PHYSICAL OR ENVIRONMENTAL DAMAGE,
55	# INCLUDING WITHOUT LIMITATION THE OPERATION OF NUCLEAR FACILITIES, AIRCRAFT
56	# NAVIGATION OR COMMUNICATION SYSTEMS, AIR TRAFFIC CONTROL, LIFE SUPPORT OR
57	# WEAPONS SYSTEMS.
58	#
59	# E. NO ORAL OR WRITTEN INFORMATION OR ADVICE GIVEN BY THE AUTHOR
60	# SHALL CREATE A WARRANTY. SHOULD THIS SHELL SCRIPT OR SERVICES PROVE DEFECTIVE,
61	# YOU ASSUME THE ENTIRE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION.
62	#
63	# Limitation of Liability.
64	# F. TO THE EXTENT NOT PROHIBITED BY APPLICABLE LAW, IN NO EVENT SHALL THE AUTHOR
65	# BE LIABLE FOR PERSONAL INJURY, OR ANY INCIDENTAL, SPECIAL, INDIRECT OR
66	# CONSEQUENTIAL DAMAGES WHATSOEVER, INCLUDING, WITHOUT LIMITATION, DAMAGES
67	# FOR LOSS OF PROFITS, CORRUPTION OR LOSS OF DATA, FAILURE TO TRANSMIT OR
68	# RECEIVE ANY DATA OR INFORMATION, BUSINESS INTERRUPTION OR ANY OTHER
69	# COMMERCIAL DAMAGES OR LOSSES, ARISING OUT OF OR RELATED TO YOUR USE OR
70	# INABILITY TO USE THIS SHELL SCRIPT OR SERVICES OR ANY THIRD PARTY
71	# SOFTWARE OR APPLICATIONS IN CONJUNCTION WITH THIS SHELL SCRIPT OR
72	# SERVICES, HOWEVER CAUSED, REGARDLESS OF THE THEORY OF LIABILITY (CONTRACT,
73	# TORT OR OTHERWISE) AND EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE
74	# POSSIBILITY OF SUCH DAMAGES. SOME JURISDICTIONS DO NOT ALLOW THE EXCLUSION
75	# OR LIMITATION OF LIABILITY FOR PERSONAL INJURY, OR OF INCIDENTAL OR
76	# CONSEQUENTIAL DAMAGES, SO THIS LIMITATION MAY NOT APPLY TO YOU. In no event
77	# shall THE AUTHOR's total liability to you for all damages (other than as may
78	# be required by applicable law in cases involving personal injury) exceed
79	# the amount of five dollars ($5.00). The foregoing limitations will apply
80	# even if the above stated remedy fails of its essential purpose.
81	################################################################################
82
83	__module__ = "tests"	4✔
84	"""This is a testing related stand-alone utilities module.	4✔
85
86	This module provides test fixtures and utilities for testing multicast communication.
87	It includes a basic UDP client implementation and handler for testing multicast
88	functionality.
89
90	Classes:
91	MCastClient: Test fixture for multicast client operations.
92	MyUDPHandler: UDP request handler for echo functionality.
93
94	Functions:
95	main: Entry point for test operations.
96
97	Example:
98	>>> from tests.MulticastUDPClient import MCastClient
99	>>> client = MCastClient(grp_addr='224.0.0.1', src_port=59259)
100	>>> isinstance(client._source_port, int)
101	True
102	>>>
103
104	"""
105
106	__name__ = "tests.MulticastUDPClient" # skipcq: PYL-W0622	4✔
107
108	try:	4✔
109	import sys	4✔
110	if not hasattr(sys, 'modules') or not sys.modules: # pragma: no branch
111	raise ModuleNotFoundError(
112	"[CWE-440] OMG! sys.modules is not available or empty."
113	) from None
114	except ImportError as _cause: # pragma: no branch
115	raise ImportError("[CWE-440] Unable to import sys module.") from _cause
116
117	try:	4✔
118	import socket	4✔
119	import socketserver	4✔
120	except ImportError as _cause: # pragma: no branch
121	raise ImportError("[CWE-758] Test module failed completely.") from _cause
122
123	try:	4✔
124	import random	4✔
125	except ImportError as _cause: # pragma: no branch
126	raise ModuleNotFoundError("[CWE-758] Test module failed to randomize.") from _cause
127
128
129	class MCastClient(object): # skipcq: PYL-R0205	4✔
130	"""
131	For use as a test fixture.
132
133	A trivial implementation of a socket-based object with a function
134	named say. The say function of this class performs a send and recv on a given socket and
135	then prints out simple diognostics about the content sent and any response received.
136
137	Testing:
138
139	First some test fixtures:
140
141	>>> import socket as socket
142	>>> import random as random
143	>>>
144
145	Testcase 0: test the class MCastClient is.
146	A: Test that the MulticastUDPClient component is importable.
147	B: Test that the MCastClient class is importable.
148
149	>>> import tests.MulticastUDPClient
150	>>> from MulticastUDPClient import MCastClient as MCastClient
151	>>> MCastClient is not None
152	True
153	>>>
154
155	Testcase 1: Test the class MCastClient has a say function.
156	A: Test that the MulticastUDPClient component is importable.
157	B: Test that the MCastClient class is importable.
158	C: Test that the MCastClient class has the function named say.
159
160	>>> import tests.MulticastUDPClient
161	>>> from MulticastUDPClient import MCastClient as MCastClient
162	>>> MCastClient is not None
163	True
164	>>> MCastClient.say is not None
165	True
166	>>> type(MCastClient.say)
167	<class 'function'>
168	>>>
169
170
171	"""
172
173	__module__ = "tests.MulticastUDPClient.MCastClient"	4✔
174
175	_group_addr = None	4✔
176	"""The multicast group address."""	4✔
177
178	_source_port = None	4✔
179	"""The source port for the client."""	4✔
180
181	# skipcq: TCV-002
182	def __init__(self, args, *kwargs) -> None: # pragma: no cover
183	"""
184	Initialize a MCastClient object with optional group address and source port.
185
186	The client can be initialized with or without specifying a group address and source port.
187	If no source port is provided, a random port between 50000 and 59999 is generated.
188
189	Args:
190	*args: Variable length argument list (Unused).
191	**kwargs: Arbitrary keyword arguments.
192	- grp_addr (str): The multicast group address.
193	- src_port (int): The source port for the client.
194
195	Meta Testing:
196
197	First set up test fixtures by importing test context.
198
199	>>> import tests.MulticastUDPClient as MulticastUDPClient
200	>>> from MulticastUDPClient import MCastClient as MCastClient
201	>>>
202
203	Testcase 1: Initialization without any arguments.
204
205	>>> client = MCastClient()
206	>>> 50000 <= client._source_port <= 59999
207	True
208	>>> client._group_addr is None
209	True
210	>>>
211
212	Testcase 2: Initialization with only group address.
213
214	>>> tst_args = {}
215	>>> client = MCastClient(grp_addr="224.0.0.1")
216	>>> client._group_addr
217	'224.0.0.1'
218	>>> 50000 <= client._source_port <= 59999
219	True
220	>>>
221
222	Testcase 3: Initialization with only source port.
223
224	>>> client = MCastClient(src_port=55555)
225	>>> client._source_port
226	55555
227	>>> client._group_addr is None
228	True
229	>>>
230
231	Testcase 4: Initialization with both group address and source port.
232
233	>>> client = MCastClient(grp_addr="224.0.0.2", src_port=55556)
234	>>> client._group_addr
235	'224.0.0.2'
236	>>> client._source_port
237	55556
238	>>>
239
240
241	"""
242	# skipcq: TCV-002
243	if "grp_addr" in kwargs: # pragma: no branch
244	self._group_addr = kwargs.get("grp_addr", None) # skipcq: PTC-W0039 - ensure None
245	if "src_port" in kwargs: # pragma: no branch
246	self._source_port = kwargs.get("src_port", 0)
247	else: # pragma: no branch
248	self._source_port = int(
249	50000 + (
250	int(random.SystemRandom().randbytes(int(60000).__sizeof__()).hex(), 16) % 9999
251	)
252	)
253
254	# skipcq: TCV-002
255	@staticmethod
256	def say(address: str, port: int, sock: socket.socket, msg: str) -> None: # pragma: no cover
257	"""
258	Send a message to a specified multicast address and port, then receive and print it.
259
260	This function sends a UTF-8 encoded message to the specified multicast address and port
261	using the provided connection. It then waits for a response, decodes it, and prints both
262	the sent and received messages.
263
264	Args:
265	address (str): The multicast group address to send the message to.
266	port (int): The port number to send the message to.
267	sock (socket.socket): The socket connection to use for sending and receiving.
268	msg (str): The message to be sent.
269
270	Returns:
271	None
272
273	Prints:
274	The sent message and the received response.
275
276	Meta Testing:
277
278	First, set up test fixtures:
279
280	>>> import unittest.mock
281	>>> from MulticastUDPClient import MCastClient
282	>>>
283
284	Testcase 1: Test sending and receiving a message.
285
286	>>> mock_socket = unittest.mock.Mock()
287	>>> mock_socket.recv.return_value = b"Response received"
288	>>> client = MCastClient()
289	>>> client.say("224.0.0.1", 59991, mock_socket, "Test message")
290	Sent: Test message
291	Received: Response received
292	>>>
293
294	Testcase 2: Test sending a 'STOP' message.
295
296	>>> mock_socket.recv.return_value = b"Stopped"
297	>>> client.say("224.0.0.1", 59991, mock_socket, "STOP")
298	Sent: STOP
299	Received: Stopped
300	>>>
301
302
303	Note:
304	This function assumes that the connection is already properly configured
305	for multicast communication.
306
307	"""
308	# skipcq: TCV-002
309	sock.sendto(bytes(msg + "\n", "utf-8"), (address, port)) # pragma: no cover
310	received = str(sock.recv(1024), "utf-8") # pragma: no cover
311	sp = " " * 4 # pragma: no cover
312	if sys.stdout.isatty(): # pragma: no cover
313	print(f"Sent: {sp}{msg}") # skipcq: PYL-C0209 - must remain compatible
314	print(f"Received: {received}") # skipcq: PYL-C0209 - must remain compatible
315
316
317	class MyUDPHandler(socketserver.BaseRequestHandler):	4✔
318	"""
319	Subclasses socketserver.BaseRequestHandler to handle echo functionality.
320
321	Simplifies testing by echoing back the received string data in uppercase,
322	after printing the sender's IP address.
323
324	Meta Testing:
325
326	First set up test fixtures by importing test context.
327
328	>>> import tests.MulticastUDPClient as MulticastUDPClient
329	>>> from MulticastUDPClient import MyUDPHandler as MyUDPHandler
330	>>>
331
332	Testcase 1: MyUDPHandler should be automatically imported.
333
334	>>> MyUDPHandler.__name__ is not None
335	True
336	>>>
337
338	"""
339
340	__module__ = "tests.MulticastUDPClient.MyUDPHandler"	4✔
341
342	# skipcq: TCV-002
343	def handle(self) -> None: # pragma: no cover
344	"""
345	Handle incoming UDP requests.
346
347	This method overrides the `handle` method from `socketserver.BaseRequestHandler`
348	to process incoming UDP messages. It receives a message from a client, echoes it
349	back in uppercase, and prints diagnostic information.
350
351	Meta Testing:
352
353	First set up test fixtures by importing test context.
354
355	>>> import socket
356	>>> import threading
357	>>> from tests.MulticastUDPClient import MyUDPHandler
358	>>>
359
360	Testcase 1: Test handling a simple message.
361
362	>>> import socketserver
363	>>> server = socketserver.UDPServer(('localhost', 0), MyUDPHandler)
364	>>> threading.Thread(target=server.serve_forever, daemon=True).start()
365	>>> client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
366	>>> client_socket.sendto("hello world\n", server.server_address)
367	>>> data, _ = client_socket.recvfrom(1024)
368	>>> data
369	b'HELLO WORLD\n'
370	>>> server.shutdown()
371	>>>
372
373	Testcase 2: Test handling an empty message.
374
375	>>> import socketserver
376	>>> server = socketserver.UDPServer(('localhost', 0), MyUDPHandler)
377	>>> threading.Thread(target=server.serve_forever, daemon=True).start()
378	>>> client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
379	>>> client_socket.sendto("\n", server.server_address)
380	>>> data, _ = client_socket.recvfrom(1024)
381	>>> data
382	b'\n'
383	>>> server.shutdown()
384	>>>
385
386	Note:
387	This method assumes that the incoming request tuple contains a string and a socket,
388	as per `socketserver.BaseRequestHandler` for datagram services.
389
390	"""
391	# skipcq: TCV-002
392	data = self.request[0].strip() # pragma: no cover
393	sock = self.request[1] # pragma: no cover
394	print(f"{self.client_address[0]} wrote: ") # pragma: no cover
395	print(data) # pragma: no cover
396	sock.sendto(data.upper(), self.client_address) # pragma: no cover
397
398
399	# skipcq: TCV-002
400	def main() -> None: # pragma: no cover
401	"""
402	The main test operations.
403
404	Testing:
405
406	First some test fixtures:
407
408	>>> import socket as socket
409	>>> import random as random
410	>>>
411
412	Testcase 0: test the function main is.
413	A: Test that the MulticastUDPClient component is importable.
414	B: Test that the MulticastUDPClient has a main function.
415
416	>>> import tests.MulticastUDPClient
417	>>> tests.MulticastUDPClient.main is not None
418	True
419	>>> type(tests.MulticastUDPClient.main)
420	<class 'function'>
421	>>>
422
423
424	"""
425	# skipcq: TCV-002
426	HOST, PORT = "224.0.0.1", 59991 # pragma: no cover
427	data = "TEST This is a test" # pragma: no cover
428	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) # pragma: no cover
429	tsts_fxr = MCastClient() # pragma: no cover
430	print(str((HOST, PORT))) # pragma: no cover
431	tsts_fxr.say(HOST, PORT, sock, data) # pragma: no cover
432	tsts_fxr.say(HOST, PORT, sock, "STOP") # pragma: no cover
433
434
435	if __name__ == "__main__": # pragma: no branch
436	main() # skipcq: TCV-002
437	# skipcq: PYL-R1722
438	exit(0) # skipcq: PYL-R1722 -- intentionally allow overwriteing exit for testing.

reactive-firewall-org / multicast / 16584110377

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