#! /usr/bin/env python3
# -*- coding: utf-8 -*-
# Multicast Python Module (Testing) (Extra Fuzzing Group)
# ..................................
# Copyright (c) 2017-2026, 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:
"""Handle imports with CWE-758 mitigation.
This implementation uses a nested try-except pattern to:
1. Attempt direct context import
2. Fallback to relative import
3. Validate context module integrity
4. Import required dependencies
References:
- CWE-758: Reliance on Undefined, Unspecified, or Implementation-Defined Behavior
"""
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 context import BasicUsageTestSuite
import io
from unittest.mock import patch
from context import subprocess
from context import Process
from context import string
except Exception as baton:
raise ImportError(f"[CWE-758] {__file__} Failed to import test context") from baton
_has_hypothesis: bool = False
"""
Hypothesis is not compatible with this project's license, however for users that accept the
relevant terms and conditions of the Hypothesis module and have installed it, the following
optional tests extend coverage to leverage hypothesis.
Otherwise this file is effectively omitted from testing.
Disclaimer:
For clarity this file (as python sourcecode) is Licensed under MIT (the "License");
The resulting python tests (as software) are only provided "AS IS" and WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. The dependencies of the resulting
python tests (as software), are NOT covered by the same Licensed. Assembly may be required.
"""
if not _has_hypothesis:
try:
from hypothesis import given
from hypothesis import settings
from hypothesis import strategies as st
_has_hypothesis = True
except ImportError: # pragma: no branch
_has_hypothesis = False
[docs]
def onlyIfHasHypothesis(has_hypothesis: bool) -> callable:
"""
Conditionally enable a test suite class based on the availability of the hypothesis library.
If the provided flag is False, returns a dummy class with a placeholder method that does nothing,
allowing tests dependent on hypothesis to be safely bypassed. If the provided flag is True,
the original class is returned unchanged.
Arguments:
has_hypothesis (bool): Flag indicating whether the hypothesis library is available.
Returns:
callable: A decorator function that returns either the original class or a dummy class
with a placeholder method, depending on the has_hypothesis flag.
"""
def decorator(cls: callable) -> callable: # skipcq: PY-D0003 -- decorator ok
if not has_hypothesis:
# Create an empty class with a method that returns None
return type(cls.__name__, (object,), {
'method': lambda self: None
})
return cls
return decorator
[docs]
@context.markWithMetaTag("fuzzing", "slow")
@onlyIfHasHypothesis(_has_hypothesis)
class HypothesisTestSuite(BasicUsageTestSuite):
"""
A test suite that uses Hypothesis to perform fuzz testing on the multicast sender and receiver.
This class extends `context.BasicUsageTestSuite` and provides a test case
to verify the robustness of the multicast implementation when handling
random binary data of varying sizes.
Methods:
- `test_multicast_sender_with_random_data(self, data)`: Tests sending and
receiving random binary data using the multicast sender and receiver.
Note:
This test ensures that the multicast sender and receiver can handle binary
data up to 1472 bytes, which is the typical maximum size for UDP packets
without fragmentation.
"""
__module__ = "tests.test_fuzz"
__name__ = "tests.test_fuzz.HypothesisTestSuite"
[docs]
@given(st.binary(min_size=1, max_size=1472))
@settings(deadline=None)
def test_multicast_sender_with_random_data(self, data: any) -> None:
"""
Tests the multicast sender and receiver with random binary data.
This test uses the `hypothesis` library to generate random binary data
between 1 and 1472 bytes, sends it using the multicast sender, and verifies
that the multicast receiver successfully receives the data.
Args:
data (bytes): Random binary data generated by Hypothesis.
Notes:
- A random port number is used for each test run to prevent port collisions.
- The test sets up a receiver process and sends the data multiple times.
- If the receiver process encounters an error, the test is skipped.
"""
theResult: bool = False
fail_fixture: str = "SAY --> HEAR == error"
_fixture_port_num: int = self._always_generate_random_port_WHEN_called()
try:
self.assertIsNotNone(_fixture_port_num)
self.assertIsInstance(_fixture_port_num, int)
_fixture_SAY_args = [
"SAY",
"--port",
str(_fixture_port_num),
"--group",
"'224.0.0.1'",
"--message",
f"'{data}'"
]
_fixture_HEAR_args = [
"HEAR",
"--port",
str(_fixture_port_num),
"--groups",
"'224.0.0.1'",
"--group",
"'224.0.0.1'"
]
p = Process(
target=multicast.__main__.McastDispatch().doStep,
name="HEAR",
args=[_fixture_HEAR_args]
)
p.start()
try:
self.assertIsNotNone(multicast.__main__.McastDispatch().doStep(_fixture_SAY_args))
self.assertIsNotNone(
multicast.__main__.McastDispatch().doStep(_fixture_SAY_args)
) # preemptive extra retry
except Exception as _root_cause:
p.join()
raise unittest.SkipTest(fail_fixture) from _root_cause
p.join()
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.skipTest(fail_fixture)
self.assertTrue(theResult, fail_fixture)
if __name__ == '__main__':
unittest.main()