#! /usr/bin/env python
# -*- coding: utf-8 -*-
# Python Test Repo Template
# ..................................
# Copyright (c) 2017-2024, 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
# ..........................................
# http://www.github.com/reactive-firewall/python-repo/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."""
__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 err:
raise ImportError("[CWE-440] Unable to import sys module.") from err
try:
if 'os' not in sys.modules:
import os
else: # pragma: no branch
os = sys.modules["""os"""]
except Exception as badErr: # pragma: no branch
baton = ModuleNotFoundError(badErr, str("[CWE-758] Test module failed completely."))
baton.module = __module__
baton.path = __file__
baton.__cause__ = badErr
raise baton from badErr
try:
if 'functools' not in sys.modules:
import functools
else: # pragma: no branch
functools = sys.modules["""functools"""]
except Exception as badErr: # pragma: no branch
baton = ModuleNotFoundError(badErr, str("[CWE-758] Test module failed completely."))
baton.module = __module__
baton.path = __file__
baton.__cause__ = badErr
raise baton from badErr
try:
if 'socket' not in sys.modules:
import socket
else: # pragma: no branch
socket = sys.modules["""socket"""]
except Exception as badErr: # pragma: no branch
baton = ModuleNotFoundError(badErr, str("[CWE-758] Test module failed completely."))
baton.module = __module__
baton.path = __file__
baton.__cause__ = badErr
raise baton from badErr
try:
if 'socketserver' not in sys.modules:
import socketserver
else: # pragma: no branch
socketserver = sys.modules["""socketserver"""]
except Exception as badErr: # pragma: no branch
baton = ModuleNotFoundError(badErr, str("[CWE-758] Test module failed completely."))
baton.module = __module__
baton.path = __file__
baton.__cause__ = badErr
raise baton from badErr
try:
if 'random' not in sys.modules:
import random
else: # pragma: no branch
random = sys.modules["""random"""]
except Exception as badErr: # pragma: no branch
baton = ModuleNotFoundError(badErr, str("[CWE-758] Test module failed completely."))
baton.module = __module__
baton.path = __file__
baton.__cause__ = badErr
raise baton from badErr
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."""
def __init__(self, *args, **kwargs):
"""
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
>>>
"""
if str("""grp_addr""") in kwargs:
self._group_addr = kwargs.get("""grp_addr""", None) # skipcq: PTC-W0039 - ensure None
if str("""src_port""") in kwargs:
self._source_port = kwargs.get("""src_port""", 0)
else:
self._source_port = int(
50000 + (
int(
random.SystemRandom().randbytes(
int(60000).__sizeof__()
).hex(),
16
) % 9999
)
)
[docs]
@staticmethod
def say(address, port, sock, msg):
"""
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.
"""
sock.sendto(bytes(msg + "\n", "utf-8"), (address, port))
received = str(sock.recv(1024), "utf-8")
print(str("Sent: {}").format(msg)) # skipcq: PYL-C0209 - must remain compatible
print(str("Received: {}").format(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"""
[docs]
def handle(self):
"""
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.
"""
data = self.request[0].strip()
sock = self.request[1]
print(str("{} wrote:").format(self.client_address[0]))
print(data)
sock.sendto(data.upper(), self.client_address)
[docs]
def main():
"""
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'>
>>>
"""
HOST, PORT = "224.0.0.1", 59991
data = "TEST This is a test"
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
tsts_fxr = MCastClient()
print(str((HOST, PORT)))
tsts_fxr.say(HOST, PORT, sock, data)
tsts_fxr.say(HOST, PORT, sock, str("""STOP"""))
if __name__ == "__main__":
main()
exit(0) # skipcq: PYL-R1722 - intentionally allow overwriteing exit for testing