16036635937

Committed 11 Jun 2025 04:56AM UTC coverage: 97.184% (+0.7%) from 96.484%

Build # 16036635937

Build Type

push

github

Committed by

reactive-firewall

Commit Message

[MERGE] Merged and resolved PR #439

Coverage Stats

127 of 133 branches covered (95.49%)

Branch coverage included in aggregate %.

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

18 existing lines in 2 files now uncovered.

2151 of 2211 relevant lines covered (97.29%)

5.02 hits per line

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

86.11

/tests/__init__.py

# -*- 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
# ..........................................
# 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.

"""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 _cause:  # pragma: no branch
        raise ModuleNotFoundError("[CWE-440] Module failed to import.") from _cause

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  # pragma: no cover
                                # 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 _root_cause:  # pragma: no branch
                raise ImportError("[CWE-909] Could Not Initialize Test Logger") from _root_cause
        # Setup logging for testing
        root = logging.getLogger()
        root.setLevel(logging.INFO)
        handler = ColoredStreamHandler()
        root.addHandler(handler)
except ImportError as _cause:  # pragma: no branch
        raise ModuleNotFoundError("[CWE-440] Logging failed to initialize.") from _cause

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 _cause:  # pragma: no branch
        raise ImportError("[CWE-440] multicast Failed to import.") from _cause

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 _cause:  # pragma: no branch
        raise ImportError("[CWE-440] multicast tests Failed to import.") from _cause

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
        # removed test_install_requires, in v2.0.9a3
        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_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 _root_cause:  # pragma: no branch
                        raise ImportError("[CWE-758] Test module failed completely.") from _root_cause
except Exception as _cause:  # pragma: no branch
        _LOGGER.debug(str(type(_cause)))
        _LOGGER.exception(str(_cause))
        _LOGGER.debug(str((_cause.args)))
        del _cause  # 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


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=False`: Disables verbose output for the test discovery. (changed in v2.0.9a3)
                - `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
                >>> 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=False, recurse=True, exclude_empty=True)
        doc_suite = unittest.TestSuite()
        try:
                doc_suite.addTests(doctest.DocTestSuite(module=module, test_finder=finder))
        except ValueError as _cause:
                # ValueError is raised when no tests are found
                _LOGGER.warning(
                        "No doctests found in %s: %s",  # lazy formatting to avoid PYL-W1203
                        module.__name__,
                        _cause,  # 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_build.BuildPEP621TestSuite,  # added in v2.0.9a3
                # removed test_install_requires.ParseRequirementsTestSuite in v2.0.9a3
        ],
        "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  # added in v2.0.7
        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:
        EXTRA_TESTS["coverage"].append(loadDocstringsFromModule(__module__))
        EXTRA_TESTS["coverage"].append(loadDocstringsFromModule(context))
        EXTRA_TESTS["coverage"].append(loadDocstringsFromModule("tests.MulticastUDPClient"))
except Exception:  # pragma: no branch
        _LOGGER.warning("Error loading optional doctests", exc_info=True)
        # reported, so now just continue with testing

try:
        from tests import test_extra  # added in v2.0.7
        depends.insert(11, test_extra)
        EXTRA_TESTS["security"].append(test_extra.ExtraDocsUtilsTestSuite)
        import docs.utils
        EXTRA_TESTS["security"].append(loadDocstringsFromModule(docs.utils))
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,
}


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:  # pragma: no branch
                raise ValueError(f"Unknown test group: {group}")
        selected_group = TEST_GROUPS[group]
        if category:  # pragma: no branch
                if category not in selected_group:  # pragma: no branch
                        raise ValueError(f"Unknown category '{category}' in group '{group}'")
                add_test_cases(selected_group[category])
        else:  # pragma: no branch
                # Load all categories in the group
                for category_tests in selected_group.values():
                        add_test_cases(category_tests)
        return suite


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()

1	# -- coding: utf-8 --
2
3	# Python Test Template
4	# ..................................
5	# Copyright (c) 2017-2025, Mr. Walls
6	# ..................................
7	# Licensed under MIT (the "License");
8	# you may not use this file except in compliance with the License.
9	# You may obtain a copy of the License at
10	# ..........................................
11	# https://github.com/reactive-firewall-org/multicast/tree/HEAD/LICENSE.md
12	# ..........................................
13	# Unless required by applicable law or agreed to in writing, software
14	# distributed under the License is distributed on an "AS IS" BASIS,
15	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16	# See the License for the specific language governing permissions and
17	# limitations under the License.
18
19	"""Multicast Testing Module.
20
21	Package containing test suites and utilities for the multicast module.
22
23	This package provides comprehensive testing coverage for various multicast functionalities
24	including server operations, data processing, cleanup routines, and exception handling.
25
26	For specific test cases, see:
27	- Server operations: test_hear_server.McastServerTestSuite
28	- Data processing: test_hear_data_processing.RecvDataProcessingTestSuite
29	- Cleanup routines: test_hear_cleanup.HearCleanupTestSuite
30	- Exception handling: test_exceptions.ExceptionsTestSuite
31
32	Robust imports: These statements import the entire "multicast" module,
33	allowing access to all its functionalities within the test environment.
34	This may be flagged as an intentional cyclic import by pylint.
35	See warning about cyclic-imports
36	[here](https://pylint.pycqa.org/en/latest/user_guide/messages/refactor/cyclic-import.html)
37
38	Testing:
39
40	Testcase 0: Load tests fixtures
41
42	>>> import tests as _tests
43	>>> _tests.__module__ is not None
44	True
45	>>> _tests.__package__ is not None
46	True
47	>>>
48
49
50	"""
51
52	__package__ = "tests" # skipcq: PYL-W0622
53
54	__module__ = "tests"
55
56	try:
57	import sys
58	import os
59	import unittest
60	from unittest import TestSuite as TestSuite
61	import types
62	from typing import Optional
63	except ImportError as _cause: # pragma: no branch
64	raise ModuleNotFoundError("[CWE-440] Module failed to import.") from _cause
65
66	try:
67	import logging
68	try:
69	class ANSIColors:	4✔
70	# Define ANSI color codes
71	BLACK = """\033[30m"""	4✔
72	RED = """\033[31m"""
73	GREEN = """\033[32m"""	4✔
74	YELLOW = """\033[33m"""	4✔
75	BLUE = """\033[34m"""	4✔
76	MAGENTA = """\033[35m"""	4✔
77	CYAN = """\033[36m"""	4✔
78	GREY = """\033[37m"""	4✔
79	AMBER = """\033[93m"""	4✔
80	REDBG = """\033[41m""" # Red background
81	ENDC = """\033[0m"""
82
83	logging_color = {	4✔
84	'debug': ANSIColors.BLUE, 'info': ANSIColors.GREY,	4✔
85	'warning': ANSIColors.AMBER,	4✔
86	'error': ANSIColors.RED,	4✔
87	'critical': str(str(ANSIColors.BLACK) + str(ANSIColors.REDBG)),
88	}	4✔
89		4✔
90	logging_level = {	4✔
91	'debug': logging.DEBUG,	4✔
92	'info': logging.INFO,	4✔
93	'warning': logging.WARNING,	4✔
94	'error': logging.ERROR,	4✔
95	'critical': logging.CRITICAL,	4✔
96	}	4✔
97		4✔
98	class ColoredStreamHandler(logging.StreamHandler):	4✔
99	def emit(self, record: logging.LogRecord) -> None:
100	# Get the log level as a string	4✔
101	loglevel = record.levelname.lower()
102	# Validate the log level
103	if not isinstance(loglevel, str) or loglevel not in logging_color:
104	raise ValueError("Invalid log level") from None # pragma: no cover
105	# Determine color based on whether the output is a terminal
106	if sys.stdout.isatty():
107	colorPrefix = logging_color[loglevel]	4✔
108	endColor = ANSIColors.ENDC
109	else:
110	colorPrefix = ""
111	endColor = ""
112	# Format the message
113	msg = self.format(record)
114	formatted_msg = f"{colorPrefix}{msg}{endColor}"
115	# Write the formatted message to the stream	4✔
116	self.stream.write(formatted_msg + self.terminator)	4✔
117	self.flush()
118		4✔
119	except Exception as _root_cause: # pragma: no branch
120	raise ImportError("[CWE-909] Could Not Initialize Test Logger") from _root_cause	4✔
121	# Setup logging for testing
122	root = logging.getLogger()
123	root.setLevel(logging.INFO)	4✔
UNCOV 124	handler = ColoredStreamHandler()	×
UNCOV 125	root.addHandler(handler)	×
126	except ImportError as _cause: # pragma: no branch
127	raise ModuleNotFoundError("[CWE-440] Logging failed to initialize.") from _cause	4✔
128		4✔
129	try:
130	if 'multicast' not in sys.modules:	4✔
131	import multicast # pylint: disable=cyclic-import - skipcq: PYL-R0401	4✔
132	else: # pragma: no branch
133	multicast = sys.modules["multicast"]	4✔
134	except Exception as _cause: # pragma: no branch	4✔
135	raise ImportError("[CWE-440] multicast Failed to import.") from _cause
136
137	try:
138	_LOGGER = logging.getLogger(__module__)
139	_LOGGER.debug("Initializing tests.")	4✔
140	_DIR_NAME = str(".")	4✔
141	_PARENT_DIR_NAME = str("..")	4✔
142	_BASE_NAME = os.path.dirname(__file__)	4✔
143	if 'multicast' in __file__:
144	sys.path.insert(0, os.path.abspath(os.path.join(_BASE_NAME, _PARENT_DIR_NAME)))
145	if 'tests' in __file__:
146	sys.path.insert(0, os.path.abspath(os.path.join(_BASE_NAME, _DIR_NAME)))	4✔
147	except ImportError as _cause: # pragma: no branch
148	raise ImportError("[CWE-440] multicast tests Failed to import.") from _cause
149
150	try:
151	from tests import profiling as profiling # skipcq: PYL-C0414
152	from tests import test_basic
153	from tests import test_exceptions
154	from tests import test_deps	4✔
155	# removed test_install_requires, in v2.0.9a3	4✔
156	from tests import test_manifest	4✔
157	from tests import test_build	4✔
158	from tests import test_usage	4✔
159	from tests import test_hear_server	4✔
160	from tests import test_hear_server_activate	4✔
161	from tests import test_hear_cleanup	4✔
162	from tests import test_hear_data_processing	4✔
163	from tests import test_hear_keyboard_interrupt	4✔
164
165	depends = [
166	profiling,
167	test_basic,	4✔
168	test_deps,	4✔
169	test_build,	4✔
170	test_manifest,	4✔
171	test_usage,	4✔
172	test_hear_server_activate,
173	test_hear_cleanup,	4✔
174	test_hear_data_processing,	4✔
175	test_exceptions,	4✔
176	test_hear_keyboard_interrupt,	4✔
177	test_hear_server	4✔
178	]	4✔
179		4✔
180	try:	4✔
181	from tests import test_fuzz
182	depends.insert(10, test_fuzz)	4✔
183	except Exception: # pragma: no branch
184	_LOGGER.exception("Error loading optional Fuzzing tests")
185
186	for unit_test in depends:
187	try:
188	if unit_test.__name__ is None: # pragma: no branch
189	raise ImportError(
190	f"Test module failed to import even the {str(unit_test)} tests."
191	) from None
192	except Exception as _root_cause: # pragma: no branch
193	raise ImportError("[CWE-758] Test module failed completely.") from _root_cause
194	except Exception as _cause: # pragma: no branch
195	_LOGGER.debug(str(type(_cause)))
196	_LOGGER.exception(str(_cause))
197	_LOGGER.debug(str((_cause.args)))	4✔
198	del _cause # skipcq - cleanup any error leaks early	4✔
199	exit(0) # skipcq: PYL-R1722 - intentionally allow overwriteing exit for testing	4✔
200
201	try:
202	if 'tests.context' not in sys.modules:
203	from tests import context	4✔
204	else: # pragma: no branch	4✔
205	context = sys.modules["tests.context"]	4✔
206	except ImportError as _cause: # pragma: no branch
207	raise ImportError("[CWE-440] context Failed to import.") from _cause
208
209
210	def loadDocstringsFromModule(module: types.ModuleType) -> TestSuite:
211	"""
212	Load and return a test suite containing doctests from the specified module.
213
214	This function attempts to import the `doctest` module and uses it to find
215	and load all doctests defined in the provided module. If the module is
216	valid and contains doctests, a `unittest.TestSuite` object is returned
217	containing those tests. If no doctests are found or if an error occurs
218	during the loading process, appropriate messages are printed to the	4✔
219	console.
220
221	Notes:
222	- The function checks if the `doctest` module is already imported to
223	avoid unnecessary imports.
224	- The `DocTestFinder` is configured with the following options:
225	- `verbose=False`: Disables verbose output for the test discovery. (changed in v2.0.9a3)
226	- `recurse=True`: Allows the finder to search for doctests in
227	nested functions and classes.	4✔
228	- `exclude_empty=True`: Excludes empty doctests from the results.
229	- If no doctests are found in the specified module, a message is printed
230	indicating that no tests were found.
231	- Any other exceptions encountered during the loading process are caught
232	and printed to the console.
233
234	See Also:
235	- get_test_suite: Function that uses `loadDocstringsFromModule` to build test suites
236	- load_tests: Function that loads both regular tests and doctests
237
238	Args:
239	module (module) -- The Python module from which to load doctests. This should be a
240	valid module object that has been imported. If the module is None,
241	the function will return None.
242
243	Returns:
244	(unittest.TestSuite or None) -- A `unittest.TestSuite` object containing the
245	doctests found in the specified module. If the module is None,
246	the function returns None.
247
248	Raises:
249	ImportError
250	If the `doctest` module fails to import, an ImportError is raised
251	with a message indicating the failure.
252
253	Meta-Testing:
254
255	>>> import multicast
256	>>> suite = loadDocstringsFromModule(multicast) #doctest: +ELLIPSIS
257	>>> if suite:
258	... print(f"Loaded {len(suite._tests)} doctests from "
259	... f"{multicast.__name__}") # doctest: +ELLIPSIS
260	Loaded ... doctests from ...
261	>>>
262
263	"""
264	if not module:
265	return None
266	try:
267	if 'doctest' not in sys.modules:
268	import doctest
269	else: # pragma: no branch
270	doctest = sys.modules["doctest"]
271	except Exception as _cause: # pragma: no branch
272	raise ImportError("[CWE-440] doctest Failed to import.") from _cause
273	finder = doctest.DocTestFinder(verbose=False, recurse=True, exclude_empty=True)
274	doc_suite = unittest.TestSuite()
275	try:
276	doc_suite.addTests(doctest.DocTestSuite(module=module, test_finder=finder))
277	except ValueError as _cause:
278	# ValueError is raised when no tests are found
279	_LOGGER.warning(
280	"No doctests found in %s: %s", # lazy formatting to avoid PYL-W1203
281	module.__name__,	4✔
UNCOV 282	_cause, # log as just warning level, instead of exception (error), but still detailed.	×
283	exc_info=True,	4✔
284	)
285	except Exception:
286	_LOGGER.exception(
287	"Error loading doctests from %s", # lazy formatting to avoid PYL-W1203
288	module.__name__,
289	)
290	return doc_suite	4✔
291		4✔
292		4✔
293	# === Test Suite Groups ===	4✔
UNCOV 294	MINIMUM_ACCEPTANCE_TESTS = {	×
295	"bootstrap": [
UNCOV 296	# Init/exceptions/env/skt tests	×
297	test_exceptions.ExceptionsTestSuite, # Also in basic, but crucial for bootstrap
298	],
299	"basic": [
300	test_basic.BasicTestSuite,
301	],
302	"build": [
303	# Build and packaging tests
304	test_build.BuildPEP517TestSuite,
305	test_manifest.ManifestInclusionTestSuite,
306	test_build.BuildPEP621TestSuite, # added in v2.0.9a3
307	# removed test_install_requires.ParseRequirementsTestSuite in v2.0.9a3	4✔
308	],
309	"doctests": [
310	# These will be loaded dynamically via DocTestSuite
311	loadDocstringsFromModule(multicast),	4✔
312	loadDocstringsFromModule(multicast.exceptions),
313	loadDocstringsFromModule(multicast.env),
314	loadDocstringsFromModule(multicast.skt),
315	loadDocstringsFromModule(multicast.recv),
316	loadDocstringsFromModule(multicast.send),
317	loadDocstringsFromModule(multicast.hear),
318	],
319	"say": [
320	# Tests focused on multicast/send.py
321	test_usage.MulticastTestSuite, # send-related tests
322	],
323	"hear": [
324	# Tests focused on multicast/recv.py and multicast/hear.py
325	test_hear_server.McastServerTestSuite,
326	test_hear_server.HearUDPHandlerTestSuite,
327	test_hear_server_activate.McastServerActivateTestSuite,
328	test_hear_data_processing.RecvDataProcessingTestSuite,
329	test_hear_data_processing.HearHandleNoneDataTestSuite,
330	test_hear_cleanup.HearCleanupTestSuite,
331	],
332	"usage": [
333	# Tests focused on multicast/__main__.py and API use cases
334	test_usage.BasicIntegrationTestSuite,
335	],
336	}
337
338
339	EXTRA_TESTS = {
340	"coverage": [
341	test_deps.BuildRequirementsTxtTestSuite,
342	test_hear_keyboard_interrupt.TestHearKeyboardInterrupt,
343	# Add other coverage-focused tests here
344	],
345	"linting": [], # To be implemented
346	"security": [], # To be implemented
347	}
348
349	try:
350	from tests import test_recv # added in v2.0.7
351	depends.insert(11, test_recv)
352	EXTRA_TESTS["coverage"].append(test_recv.McastRECVTestSuite)
353	except Exception: # pragma: no branch
354	_LOGGER.warning("Error loading optional debug tests", exc_info=True)
355
356	try:	4✔
357	EXTRA_TESTS["coverage"].append(loadDocstringsFromModule(__module__))
358	EXTRA_TESTS["coverage"].append(loadDocstringsFromModule(context))
359	EXTRA_TESTS["coverage"].append(loadDocstringsFromModule("tests.MulticastUDPClient"))
360	except Exception: # pragma: no branch
361	_LOGGER.warning("Error loading optional doctests", exc_info=True)
362	# reported, so now just continue with testing
363
364	try:
365	from tests import test_extra # added in v2.0.7
366	depends.insert(11, test_extra)	4✔
367	EXTRA_TESTS["security"].append(test_extra.ExtraDocsUtilsTestSuite)	4✔
368	import docs.utils	4✔
369	EXTRA_TESTS["security"].append(loadDocstringsFromModule(docs.utils))	4✔
370	except Exception: # pragma: no branch
371	_LOGGER.warning("Error loading optional extra tests", exc_info=True)
372
373	try:	4✔
374	FUZZING_TESTS = {	4✔
375	"slow": [	4✔
376	test_fuzz.HypothesisTestSuite, # Assuming this exists	4✔
377	],
378	# Future fuzzing test categories to be added
379	}
380	except Exception:
381	FUZZING_TESTS = {"slow": []}	4✔
382		4✔
383	PERFORMANCE_TESTS = {	4✔
384	"scalability": [], # Future implementation	4✔
385	"multi_sender": [], # Future implementation	4✔
386	"multi_receiver": [], # Future implementation	4✔
387	}
388
389	# Load specific group/category
390	TEST_GROUPS = {	4✔
391	"mat": MINIMUM_ACCEPTANCE_TESTS,	4✔
392	"extra": EXTRA_TESTS,
393	"fuzzing": FUZZING_TESTS,
394	"performance": PERFORMANCE_TESTS,
395	}
396
397
398	def get_test_suite(group: Optional[str] = None, category: Optional[str] = None) -> TestSuite:
399	"""Get a test suite based on group and category.
400		4✔
401	Args:
402	group (str): Test group ('mat', 'extra', 'fuzzing', 'performance')
403	category (str): Specific category within the group
404
405	Returns:
406	unittest.TestSuite: The configured test suite
407	"""	4✔
408	suite = unittest.TestSuite()
409	loader = unittest.TestLoader()
410
411	# Helper function to add test cases to suite
412	def add_test_cases(test_cases: list) -> None:
413	for test_case in test_cases:
414	if isinstance(test_case, unittest.TestSuite):
415	# Handle doctests	4✔
416	suite.addTests(test_case)
417	else:
418	suite.addTests(loader.loadTestsFromTestCase(test_case))
419
420	# Load all MATs if no group specified
421	if not group:
422	for category_tests in MINIMUM_ACCEPTANCE_TESTS.values():
423	add_test_cases(category_tests)
424	# and for coverage targets:
425	add_test_cases(EXTRA_TESTS["coverage"])	4✔
426	return suite	4✔
427	if group not in TEST_GROUPS: # pragma: no branch
428	raise ValueError(f"Unknown test group: {group}")
429	selected_group = TEST_GROUPS[group]	4✔
430	if category: # pragma: no branch	4✔
431	if category not in selected_group: # pragma: no branch	4✔
432	raise ValueError(f"Unknown category '{category}' in group '{group}'")
433	add_test_cases(selected_group[category])	4✔
434	else: # pragma: no branch
435	# Load all categories in the group	4✔
436	for category_tests in selected_group.values():
437	add_test_cases(category_tests)
438	return suite	4✔
439		4✔
440		4✔
441	def load_tests(loader, tests, pattern) -> unittest.TestSuite:
442	"""Will Load the tests from the project and then attempts to load the doctests too.	4✔
443		4✔
UNCOV 444	Meta Testing:	×
UNCOV 445		×
UNCOV 446	Testcase 0: Load test fixtures	×
UNCOV 447		×
UNCOV 448	>>> import tests as _tests	×
UNCOV 449	>>>	×
UNCOV 450		×
451	Testcase 1: Load test fixtures
452
UNCOV 453	>>> import tests as _tests	×
UNCOV 454	>>> _tests.load_tests is not None	×
UNCOV 455	True	×
456
457	"""
458	return get_test_suite()	4✔

reactive-firewall-org / multicast / 16036635937

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