# -*- coding: utf-8 -*-
# Python Test Template
# ..................................
# 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
# ..........................................
# http://www.github.com/reactive-firewall/python-reop/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.
"""Multicast Testing Module.
Package containing test suites and utilities for the multicast module.
This package provides comprehensive testing coverage for various multicast functionalities
including server operations, data processing, cleanup routines, and exception handling.
For specific test cases, see:
- Server operations: test_hear_server.McastServerTestSuite
- Data processing: test_hear_data_processing.RecvDataProcessingTestSuite
- Cleanup routines: test_hear_cleanup.HearCleanupTestSuite
- Exception handling: test_exceptions.ExceptionsTestSuite
Robust imports: These statements import the entire "multicast" module,
allowing access to all its functionalities within the test environment.
This may be flagged as an intentional cyclic import by pylint.
See warning about cyclic-imports
[here](https://pylint.pycqa.org/en/latest/user_guide/messages/refactor/cyclic-import.html)
Testing:
Testcase 0: Load tests fixtures
>>> import tests as _tests
>>> _tests.__module__ is not None
True
>>> _tests.__package__ is not None
True
>>>
"""
__package__ = "tests" # skipcq: PYL-W0622
__module__ = "tests"
try:
import sys
import os
import unittest
from unittest import TestSuite as TestSuite
import types
from typing import Optional
except ImportError as baton: # pragma: no branch
raise ModuleNotFoundError("[CWE-440] Module failed to import.") from baton
try:
import logging
try:
class ANSIColors:
# Define ANSI color codes
BLACK = """\033[30m"""
RED = """\033[31m"""
GREEN = """\033[32m"""
YELLOW = """\033[33m"""
BLUE = """\033[34m"""
MAGENTA = """\033[35m"""
CYAN = """\033[36m"""
GREY = """\033[37m"""
AMBER = """\033[93m"""
REDBG = """\033[41m""" # Red background
ENDC = """\033[0m"""
logging_color = {
'debug': ANSIColors.BLUE, 'info': ANSIColors.GREY,
'warning': ANSIColors.AMBER,
'error': ANSIColors.RED,
'critical': str(str(ANSIColors.BLACK) + str(ANSIColors.REDBG)),
}
logging_level = {
'debug': logging.DEBUG,
'info': logging.INFO,
'warning': logging.WARNING,
'error': logging.ERROR,
'critical': logging.CRITICAL,
}
class ColoredStreamHandler(logging.StreamHandler):
def emit(self, record: logging.LogRecord) -> None:
# Get the log level as a string
loglevel = record.levelname.lower()
# Validate the log level
if not isinstance(loglevel, str) or loglevel not in logging_color:
raise ValueError("Invalid log level") from None
# Determine color based on whether the output is a terminal
if sys.stdout.isatty():
colorPrefix = logging_color[loglevel]
endColor = ANSIColors.ENDC
else:
colorPrefix = ""
endColor = ""
# Format the message
msg = self.format(record)
formatted_msg = f"{colorPrefix}{msg}{endColor}"
# Write the formatted message to the stream
self.stream.write(formatted_msg + self.terminator)
self.flush()
except Exception as _cause:
raise ImportError("[CWE-909] Could Not Initialize Test Logger") from _cause
# Setup logging for testing
root = logging.getLogger()
root.setLevel(logging.INFO)
handler = ColoredStreamHandler()
root.addHandler(handler)
except ImportError as baton: # pragma: no branch
raise ModuleNotFoundError("[CWE-440] Logging failed to initialize.") from baton
try:
if 'multicast' not in sys.modules:
import multicast # pylint: disable=cyclic-import - skipcq: PYL-R0401
else: # pragma: no branch
multicast = sys.modules["multicast"]
except Exception as err: # pragma: no branch
raise ImportError("[CWE-440] multicast Failed to import.") from err
try:
_LOGGER = logging.getLogger(__module__)
_LOGGER.debug("Initializing tests.")
_DIR_NAME = str(".")
_PARENT_DIR_NAME = str("..")
_BASE_NAME = os.path.dirname(__file__)
if 'multicast' in __file__:
sys.path.insert(0, os.path.abspath(os.path.join(_BASE_NAME, _PARENT_DIR_NAME)))
if 'tests' in __file__:
sys.path.insert(0, os.path.abspath(os.path.join(_BASE_NAME, _DIR_NAME)))
except ImportError as err: # pragma: no branch
raise ImportError("[CWE-440] multicast tests Failed to import.") from err
try:
from tests import profiling as profiling # skipcq: PYL-C0414
from tests import test_basic
from tests import test_exceptions
from tests import test_deps
from tests import test_install_requires
from tests import test_manifest
from tests import test_build
from tests import test_usage
from tests import test_hear_server
from tests import test_hear_server_activate
from tests import test_hear_cleanup
from tests import test_hear_data_processing
from tests import test_hear_keyboard_interrupt
depends = [
profiling,
test_basic,
test_deps,
test_install_requires,
test_build,
test_manifest,
test_usage,
test_hear_server_activate,
test_hear_cleanup,
test_hear_data_processing,
test_exceptions,
test_hear_keyboard_interrupt,
test_hear_server
]
try:
from tests import test_fuzz
depends.insert(10, test_fuzz)
except Exception: # pragma: no branch
_LOGGER.exception("Error loading optional Fuzzing tests")
for unit_test in depends:
try:
if unit_test.__name__ is None: # pragma: no branch
raise ImportError(
f"Test module failed to import even the {str(unit_test)} tests."
) from None
except Exception as impErr: # pragma: no branch
raise ImportError(str("[CWE-758] Test module failed completely.")) from impErr
except Exception as badErr: # pragma: no branch
_LOGGER.debug(str(type(badErr)))
_LOGGER.exception(str(badErr))
_LOGGER.debug(str((badErr.args)))
badErr = None
del badErr # skipcq - cleanup any error leaks early
exit(0) # skipcq: PYL-R1722 - intentionally allow overwriteing exit for testing
try:
if 'tests.context' not in sys.modules:
from tests import context
else: # pragma: no branch
context = sys.modules["tests.context"]
except ImportError as _cause: # pragma: no branch
raise ImportError("[CWE-440] context Failed to import.") from _cause
[docs]
def loadDocstringsFromModule(module: types.ModuleType) -> TestSuite:
"""
Load and return a test suite containing doctests from the specified module.
This function attempts to import the `doctest` module and uses it to find
and load all doctests defined in the provided module. If the module is
valid and contains doctests, a `unittest.TestSuite` object is returned
containing those tests. If no doctests are found or if an error occurs
during the loading process, appropriate messages are printed to the
console.
Notes:
- The function checks if the `doctest` module is already imported to
avoid unnecessary imports.
- The `DocTestFinder` is configured with the following options:
- `verbose=True`: Enables verbose output for the test discovery.
- `recurse=True`: Allows the finder to search for doctests in
nested functions and classes.
- `exclude_empty=True`: Excludes empty doctests from the results.
- If no doctests are found in the specified module, a message is printed
indicating that no tests were found.
- Any other exceptions encountered during the loading process are caught
and printed to the console.
See Also:
- get_test_suite: Function that uses `loadDocstringsFromModule` to build test suites
- load_tests: Function that loads both regular tests and doctests
Args:
module (module) -- The Python module from which to load doctests. This should be a
valid module object that has been imported. If the module is None,
the function will return None.
Returns:
(unittest.TestSuite or None) -- A `unittest.TestSuite` object containing the
doctests found in the specified module. If the module is None,
the function returns None.
Raises:
ImportError
If the `doctest` module fails to import, an ImportError is raised
with a message indicating the failure.
Meta-Testing:
>>> import multicast
>>> suite = loadDocstringsFromModule(multicast) #doctest: +ELLIPSIS
Finding tests in multicast...
>>> if suite:
... print(f"Loaded {len(suite._tests)} doctests from "
... f"{multicast.__name__}") # doctest: +ELLIPSIS
Loaded ... doctests from ...
>>>
"""
if not module:
return None
try:
if 'doctest' not in sys.modules:
import doctest
else: # pragma: no branch
doctest = sys.modules["doctest"]
except Exception as _cause: # pragma: no branch
raise ImportError("[CWE-440] doctest Failed to import.") from _cause
finder = doctest.DocTestFinder(verbose=True, recurse=True, exclude_empty=True)
doc_suite = unittest.TestSuite()
try:
doc_suite.addTests(doctest.DocTestSuite(module=module, test_finder=finder))
except ValueError as e:
# ValueError is raised when no tests are found
_LOGGER.warning(
"No doctests found in %s: %s", # lazy formatting to avoid PYL-W1203
module.__name__,
e, # log as just warning level, instead of exception (error), but still detailed.
exc_info=True,
)
except Exception:
_LOGGER.exception(
"Error loading doctests from %s", # lazy formatting to avoid PYL-W1203
module.__name__,
)
return doc_suite
# === Test Suite Groups ===
MINIMUM_ACCEPTANCE_TESTS = {
"bootstrap": [
# Init/exceptions/env/skt tests
test_exceptions.ExceptionsTestSuite, # Also in basic, but crucial for bootstrap
],
"basic": [
test_basic.BasicTestSuite,
],
"build": [
# Build and packaging tests
test_build.BuildPEP517TestSuite,
test_manifest.ManifestInclusionTestSuite,
test_install_requires.ParseRequirementsTestSuite,
],
"doctests": [
# These will be loaded dynamically via DocTestSuite
loadDocstringsFromModule(multicast),
loadDocstringsFromModule(multicast.exceptions),
loadDocstringsFromModule(multicast.env),
loadDocstringsFromModule(multicast.skt),
loadDocstringsFromModule(multicast.recv),
loadDocstringsFromModule(multicast.send),
loadDocstringsFromModule(multicast.hear),
],
"say": [
# Tests focused on multicast/send.py
test_usage.MulticastTestSuite, # send-related tests
],
"hear": [
# Tests focused on multicast/recv.py and multicast/hear.py
test_hear_server.McastServerTestSuite,
test_hear_server.HearUDPHandlerTestSuite,
test_hear_server_activate.McastServerActivateTestSuite,
test_hear_data_processing.RecvDataProcessingTestSuite,
test_hear_data_processing.HearHandleNoneDataTestSuite,
test_hear_cleanup.HearCleanupTestSuite,
],
"usage": [
# Tests focused on multicast/__main__.py and API use cases
test_usage.BasicIntegrationTestSuite,
],
}
EXTRA_TESTS = {
"coverage": [
test_deps.BuildRequirementsTxtTestSuite,
test_hear_keyboard_interrupt.TestHearKeyboardInterrupt,
# Add other coverage-focused tests here
],
"linting": [], # To be implemented
"security": [], # To be implemented
}
try:
from tests import test_recv
depends.insert(11, test_recv)
EXTRA_TESTS["coverage"].append(test_recv.McastRECVTestSuite)
except Exception: # pragma: no branch
_LOGGER.warning("Error loading optional debug tests", exc_info=True)
try:
from tests import test_extra
depends.insert(11, test_extra)
EXTRA_TESTS["security"].append(test_extra.ExtraDocsUtilsTestSuite)
except Exception: # pragma: no branch
_LOGGER.warning("Error loading optional extra tests", exc_info=True)
try:
FUZZING_TESTS = {
"slow": [
test_fuzz.HypothesisTestSuite, # Assuming this exists
],
# Future fuzzing test categories to be added
}
except Exception:
FUZZING_TESTS = {"slow": []}
PERFORMANCE_TESTS = {
"scalability": [], # Future implementation
"multi_sender": [], # Future implementation
"multi_receiver": [], # Future implementation
}
# Load specific group/category
TEST_GROUPS = {
"mat": MINIMUM_ACCEPTANCE_TESTS,
"extra": EXTRA_TESTS,
"fuzzing": FUZZING_TESTS,
"performance": PERFORMANCE_TESTS,
}
[docs]
def get_test_suite(group: Optional[str] = None, category: Optional[str] = None) -> TestSuite:
"""Get a test suite based on group and category.
Args:
group (str): Test group ('mat', 'extra', 'fuzzing', 'performance')
category (str): Specific category within the group
Returns:
unittest.TestSuite: The configured test suite
"""
suite = unittest.TestSuite()
loader = unittest.TestLoader()
# Helper function to add test cases to suite
def add_test_cases(test_cases: list) -> None:
for test_case in test_cases:
if isinstance(test_case, unittest.TestSuite):
# Handle doctests
suite.addTests(test_case)
else:
suite.addTests(loader.loadTestsFromTestCase(test_case))
# Load all MATs if no group specified
if not group:
for category_tests in MINIMUM_ACCEPTANCE_TESTS.values():
add_test_cases(category_tests)
# and for coverage targets:
add_test_cases(EXTRA_TESTS["coverage"])
return suite
if group not in TEST_GROUPS:
raise ValueError(f"Unknown test group: {group}")
selected_group = TEST_GROUPS[group]
if category:
if category not in selected_group:
raise ValueError(f"Unknown category '{category}' in group '{group}'")
add_test_cases(selected_group[category])
else:
# Load all categories in the group
for category_tests in selected_group.values():
add_test_cases(category_tests)
return suite
[docs]
def load_tests(loader, tests, pattern) -> unittest.TestSuite:
"""Will Load the tests from the project and then attempts to load the doctests too.
Meta Testing:
Testcase 0: Load test fixtures
>>> import tests as _tests
>>>
Testcase 1: Load test fixtures
>>> import tests as _tests
>>> _tests.load_tests is not None
True
"""
return get_test_suite()