diff --git a/src/pynguin/analyses/module.py b/src/pynguin/analyses/module.py index 4be96e199..f974b1edf 100644 --- a/src/pynguin/analyses/module.py +++ b/src/pynguin/analyses/module.py @@ -416,6 +416,21 @@ def _is_blacklisted(element: Any) -> bool: )) +def public_top_level_module(module_name: str) -> str: + """Return the top-level module of a module name without leading underscores. + + C implementations of standard-library modules often report a private module, + e.g., ``io.TextIOWrapper.__module__`` is ``"_io"``. + + Args: + module_name: The (possibly dotted) module name. + + Returns: + The top-level module name without leading underscores. + """ + return module_name.split(".", maxsplit=1)[0].lstrip("_") + + def _c_is_whitelisted(element: ModuleType | str) -> bool: """Checks if the given element belongs to the C module whitelist. @@ -429,9 +444,7 @@ def _c_is_whitelisted(element: ModuleType | str) -> bool: try: module_name = element if isinstance(element, str) else element.__name__ - top_level = module_name.split(".")[0] - public_top_level = top_level.lstrip("_") - return public_top_level in c_module_whitelist + return public_top_level_module(module_name) in c_module_whitelist except Exception: # noqa: BLE001 LOGGER.warning( "Could not check if %s is whitelisted. Assuming it is not.", element, exc_info=True diff --git a/src/pynguin/assertion/assertiontraceobserver.py b/src/pynguin/assertion/assertiontraceobserver.py index 66e008c86..fb55d7715 100644 --- a/src/pynguin/assertion/assertiontraceobserver.py +++ b/src/pynguin/assertion/assertiontraceobserver.py @@ -26,7 +26,7 @@ import pynguin.configuration as config import pynguin.testcase.execution as ex import pynguin.utils.typetracing as tt -from pynguin.analyses.module import MODULE_BLACKLIST +from pynguin.analyses.module import MODULE_BLACKLIST, public_top_level_module from pynguin.assertion.assertion_to_ast import assertion_to_cst from pynguin.utils.exceptions import TracingAbortedException from pynguin.utils.naming import get_module_alias @@ -105,7 +105,13 @@ def _is_blacklisted_module(module_name: str | None) -> bool: return False blacklist = set(MODULE_BLACKLIST).union(config.configuration.ignore_modules) blacklist.discard("builtins") - return module_name in blacklist or module_name.split(".")[0] in blacklist + # Also match C implementations, e.g., "_io" for "io", next to private entries + # of the blacklist itself, e.g., "_thread". + return ( + module_name in blacklist + or module_name.split(".")[0] in blacklist + or public_top_level_module(module_name) in blacklist + ) def _is_blacklisted_value(value: Any) -> bool: diff --git a/tests/analyses/test_native_libraries.py b/tests/analyses/test_native_libraries.py index cbd0a8e16..28a58be26 100644 --- a/tests/analyses/test_native_libraries.py +++ b/tests/analyses/test_native_libraries.py @@ -48,6 +48,19 @@ def test_c_is_whitelisted(): assert not pm._c_is_whitelisted(fake_mod) +@pytest.mark.parametrize( + "module_name,expected", + [ + ("_io", "io"), + ("_json.decoder", "json"), + ("json", "json"), + ("numpy.linalg", "numpy"), + ], +) +def test_public_top_level_module(module_name, expected): + assert pm.public_top_level_module(module_name) == expected + + def test_c_is_whitelisted_exception(): """Verify that non-module or broken objects return False without crashing.""" diff --git a/tests/assertion/test_assertion_generation_integration.py b/tests/assertion/test_assertion_generation_integration.py index 749f33a3e..0fad5e24f 100644 --- a/tests/assertion/test_assertion_generation_integration.py +++ b/tests/assertion/test_assertion_generation_integration.py @@ -1170,3 +1170,31 @@ def test_mutation_analysis_integration_package_with_relative_import( assert num_created > 0 assert metrics.num_created_mutants == num_created assert metrics.num_killed_mutants > 0 + + +def _tc_std_streams(alias: str) -> tc.TestCase: + return make_test_case( + call_stmt("stream_holder_0", f"{alias}.StreamHolder()"), + call_stmt("shlex_0", f"{alias}.create_lexer()"), + ) + + +def test_no_assertions_on_std_streams(subject_properties: SubjectProperties): + # Stream objects such as sys.stdin and sys.stdin.buffer are io types from the "_io" + # module, and pytest's capture replaces them, so assertions on them do not hold there. + module_name = "tests.fixtures.examples.std_streams" + config.configuration.module_name = module_name + alias = get_module_alias(module_name) + with install_import_hook(module_name, subject_properties): + with subject_properties.instrumentation_tracer: + module = importlib.import_module(module_name) + importlib.reload(module) + + test_case = _tc_std_streams(alias) + _suite(test_case).accept(ag.AssertionGenerator(TestCaseExecutor(subject_properties))) + + source = _render(test_case) + assert "assert stream_holder_0.name == 'stdin'" in source + assert "_io." not in source + for stream_field in ("stream_holder_0.text", "stream_holder_0.binary", "shlex_0.instream"): + assert stream_field not in source diff --git a/tests/assertion/test_assertiontraceobserver.py b/tests/assertion/test_assertiontraceobserver.py index 5d2e40b12..45d1ee758 100644 --- a/tests/assertion/test_assertiontraceobserver.py +++ b/tests/assertion/test_assertiontraceobserver.py @@ -19,6 +19,7 @@ from __future__ import annotations import __future__ +import io import logging import os import sys @@ -708,6 +709,37 @@ def test_is_unstable_runtime_value(): assert ato._is_unstable_runtime_value(list(sys.path)) is False +@pytest.mark.parametrize( + "module_name,blacklisted", + [ + ("_io", True), + ("_asyncio", True), + ("_csv", True), + ("_thread", True), + ("_iox", False), + ("_pytest", False), + ], +) +def test_is_blacklisted_module_for_private_modules(module_name, blacklisted): + # C implementations report private modules, e.g., "_io" for the blacklisted "io". + assert ato._is_blacklisted_module(module_name) is blacklisted + + +@pytest.mark.parametrize( + "stream", + [ + io.TextIOWrapper(io.BytesIO(), encoding="utf-8"), + io.BufferedReader(io.BytesIO()), + io.StringIO(), + io.BytesIO(), + ], +) +def test_is_blacklisted_value_for_io_streams(stream): + # The stream types report "_io" as their module, while MODULE_BLACKLIST lists "io". + assert type(stream).__module__ == "_io" + assert ato._is_blacklisted_value(stream) is True + + # --- RemoteAssertionTraceObserver: static class-field assertions -------------- diff --git a/tests/fixtures/examples/std_streams.py b/tests/fixtures/examples/std_streams.py new file mode 100644 index 000000000..8b1dafc26 --- /dev/null +++ b/tests/fixtures/examples/std_streams.py @@ -0,0 +1,29 @@ +# This file is part of Pynguin. +# +# SPDX-FileCopyrightText: 2019–2026 Pynguin Contributors +# +# SPDX-License-Identifier: MIT +# +"""Fixture whose objects keep references to the standard input streams.""" + +import shlex +import sys + + +class StreamHolder: + """Keeps the text and binary standard input streams.""" + + def __init__(self) -> None: + """Store the currently bound stdin and its binary buffer.""" + self.text = sys.stdin + self.binary = sys.stdin.buffer + self.name = "stdin" + + +def create_lexer() -> shlex.shlex: + """Create a lexer that reads from the standard input. + + Returns: + A lexer whose ``instream`` is ``sys.stdin``. + """ + return shlex.shlex()