From ab1224341621ee432eb89ffec6bc9891c51effb2 Mon Sep 17 00:00:00 2001 From: Esmail Date: Mon, 26 Oct 2026 12:58:48 +0100 Subject: [PATCH 1/3] Treat the currently bound standard streams as unstable runtime values The identity blacklist of process-dependent values is captured when the module is imported in the master process. multiprocess then rebinds sys.stdin to os.devnull in the worker, so a SUT that stores sys.stdin (e.g., shlex.shlex()) got type assertions on the worker's stream, which fail under pytest's default capture. Also compare against the currently bound sys.stdin, sys.stdout and sys.stderr at check time (closes #156). --- .../assertion/assertiontraceobserver.py | 15 ++++++++++++--- tests/assertion/test_assertiontraceobserver.py | 18 ++++++++++++++++++ 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/src/pynguin/assertion/assertiontraceobserver.py b/src/pynguin/assertion/assertiontraceobserver.py index 66e008c86..8d009bb49 100644 --- a/src/pynguin/assertion/assertiontraceobserver.py +++ b/src/pynguin/assertion/assertiontraceobserver.py @@ -74,6 +74,10 @@ if val is not None ) +# Standard streams that can be rebound after the values above were captured, e.g., +# multiprocess rebinds sys.stdin to os.devnull in every worker process. +_REBINDABLE_STD_STREAMS = ("stdin", "stdout", "stderr") + def _is_unstable_runtime_value(value: Any) -> bool: """Check whether a value is a known volatile runtime/environment singleton. @@ -83,10 +87,15 @@ def _is_unstable_runtime_value(value: Any) -> bool: Returns: True, if ``value`` is (by identity) one of the known - process-/environment-dependent singletons, regardless of the name it - is currently bound to. + process-/environment-dependent singletons or the currently bound + standard stream, regardless of the name it is currently bound to. """ - return any(value is unstable for unstable in _UNSTABLE_RUNTIME_VALUES) + if any(value is unstable for unstable in _UNSTABLE_RUNTIME_VALUES): + return True + return any( + (stream := getattr(sys, name, None)) is not None and value is stream + for name in _REBINDABLE_STD_STREAMS + ) def _is_blacklisted_module(module_name: str | None) -> bool: diff --git a/tests/assertion/test_assertiontraceobserver.py b/tests/assertion/test_assertiontraceobserver.py index 5d2e40b12..b2bc3f6e5 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,23 @@ def test_is_unstable_runtime_value(): assert ato._is_unstable_runtime_value(list(sys.path)) is False +@pytest.mark.parametrize("stream", ["stdin", "stdout", "stderr"]) +def test_is_unstable_runtime_value_for_rebound_std_stream(monkeypatch, stream): + # multiprocess rebinds sys.stdin to os.devnull in each worker process, i.e., after + # the original streams were captured at import time. + replacement = io.TextIOWrapper(io.BytesIO(), encoding="utf-8") + monkeypatch.setattr(sys, stream, replacement) + assert ato._is_unstable_runtime_value(replacement) is True + assert ato._is_blacklisted_value(replacement) is True + other_stream = io.TextIOWrapper(io.BytesIO(), encoding="utf-8") + assert ato._is_unstable_runtime_value(other_stream) is False + + +def test_is_unstable_runtime_value_without_stdin(monkeypatch): + monkeypatch.setattr(sys, "stdin", None) + assert ato._is_unstable_runtime_value(None) is False + + # --- RemoteAssertionTraceObserver: static class-field assertions -------------- From f2fa5dc7e5decf9c8dc0860853463d814da04401 Mon Sep 17 00:00:00 2001 From: Esmail Date: Mon, 26 Oct 2026 19:48:50 +0100 Subject: [PATCH 2/3] Blacklist io stream types by mapping the _io module to io The stream types of io report "_io" as their module, while MODULE_BLACKLIST lists "io", so file objects were never blacklisted. Matching the live sys.stdin only covered the stream that OutputSuppressionContext binds during execution, but not, e.g., sys.stdin.buffer. Drop the live check and treat "_io" as "io" in _is_blacklisted_module instead. Add a regression test that runs a shlex.shlex() and a holder of sys.stdin/sys.stdin.buffer through the real executor. --- .../assertion/assertiontraceobserver.py | 24 +++++++------- .../test_assertion_generation_integration.py | 28 ++++++++++++++++ .../assertion/test_assertiontraceobserver.py | 33 ++++++++++--------- tests/fixtures/examples/std_streams.py | 29 ++++++++++++++++ 4 files changed, 87 insertions(+), 27 deletions(-) create mode 100644 tests/fixtures/examples/std_streams.py diff --git a/src/pynguin/assertion/assertiontraceobserver.py b/src/pynguin/assertion/assertiontraceobserver.py index 8d009bb49..d085698d6 100644 --- a/src/pynguin/assertion/assertiontraceobserver.py +++ b/src/pynguin/assertion/assertiontraceobserver.py @@ -74,9 +74,9 @@ if val is not None ) -# Standard streams that can be rebound after the values above were captured, e.g., -# multiprocess rebinds sys.stdin to os.devnull in every worker process. -_REBINDABLE_STD_STREAMS = ("stdin", "stdout", "stderr") +# C implementations of blacklisted modules report their private module name, e.g., +# the stream types of ``io`` (``io.TextIOWrapper.__module__ == "_io"``). +_PRIVATE_MODULE_ALIASES = {"_io": "io"} def _is_unstable_runtime_value(value: Any) -> bool: @@ -87,15 +87,10 @@ def _is_unstable_runtime_value(value: Any) -> bool: Returns: True, if ``value`` is (by identity) one of the known - process-/environment-dependent singletons or the currently bound - standard stream, regardless of the name it is currently bound to. + process-/environment-dependent singletons, regardless of the name it + is currently bound to. """ - if any(value is unstable for unstable in _UNSTABLE_RUNTIME_VALUES): - return True - return any( - (stream := getattr(sys, name, None)) is not None and value is stream - for name in _REBINDABLE_STD_STREAMS - ) + return any(value is unstable for unstable in _UNSTABLE_RUNTIME_VALUES) def _is_blacklisted_module(module_name: str | None) -> bool: @@ -114,7 +109,12 @@ 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 + top_level = module_name.split(".")[0] + return ( + module_name in blacklist + or top_level in blacklist + or _PRIVATE_MODULE_ALIASES.get(top_level) in blacklist + ) def _is_blacklisted_value(value: Any) -> bool: 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 b2bc3f6e5..05e6bd3fe 100644 --- a/tests/assertion/test_assertiontraceobserver.py +++ b/tests/assertion/test_assertiontraceobserver.py @@ -709,21 +709,24 @@ def test_is_unstable_runtime_value(): assert ato._is_unstable_runtime_value(list(sys.path)) is False -@pytest.mark.parametrize("stream", ["stdin", "stdout", "stderr"]) -def test_is_unstable_runtime_value_for_rebound_std_stream(monkeypatch, stream): - # multiprocess rebinds sys.stdin to os.devnull in each worker process, i.e., after - # the original streams were captured at import time. - replacement = io.TextIOWrapper(io.BytesIO(), encoding="utf-8") - monkeypatch.setattr(sys, stream, replacement) - assert ato._is_unstable_runtime_value(replacement) is True - assert ato._is_blacklisted_value(replacement) is True - other_stream = io.TextIOWrapper(io.BytesIO(), encoding="utf-8") - assert ato._is_unstable_runtime_value(other_stream) is False - - -def test_is_unstable_runtime_value_without_stdin(monkeypatch): - monkeypatch.setattr(sys, "stdin", None) - assert ato._is_unstable_runtime_value(None) is False +def test_is_blacklisted_module_maps_private_io_module(): + assert ato._is_blacklisted_module("_io") is True + assert ato._is_blacklisted_module("_iox") is False + + +@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() From 4d15c0c43ecd46a91d8d52210baa7f6956e4d9be Mon Sep 17 00:00:00 2001 From: Esmail Date: Tue, 27 Oct 2026 10:46:07 +0100 Subject: [PATCH 3/3] Share the lstrip("_") module normalisation with the C whitelist Replace the hard-coded {"_io": "io"} mapping with public_top_level_module, the top_level.lstrip("_") normalisation that _c_is_whitelisted already used. _is_blacklisted_module now also matches other C implementations of blacklisted modules, e.g., _asyncio, _csv, _pickle, _queue or _socket. --- src/pynguin/analyses/module.py | 19 ++++++++++++++++--- .../assertion/assertiontraceobserver.py | 13 +++++-------- tests/analyses/test_native_libraries.py | 13 +++++++++++++ .../assertion/test_assertiontraceobserver.py | 17 ++++++++++++++--- 4 files changed, 48 insertions(+), 14 deletions(-) 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 d085698d6..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 @@ -74,10 +74,6 @@ if val is not None ) -# C implementations of blacklisted modules report their private module name, e.g., -# the stream types of ``io`` (``io.TextIOWrapper.__module__ == "_io"``). -_PRIVATE_MODULE_ALIASES = {"_io": "io"} - def _is_unstable_runtime_value(value: Any) -> bool: """Check whether a value is a known volatile runtime/environment singleton. @@ -109,11 +105,12 @@ def _is_blacklisted_module(module_name: str | None) -> bool: return False blacklist = set(MODULE_BLACKLIST).union(config.configuration.ignore_modules) blacklist.discard("builtins") - top_level = module_name.split(".")[0] + # 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 top_level in blacklist - or _PRIVATE_MODULE_ALIASES.get(top_level) in blacklist + or module_name.split(".")[0] in blacklist + or public_top_level_module(module_name) in blacklist ) 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_assertiontraceobserver.py b/tests/assertion/test_assertiontraceobserver.py index 05e6bd3fe..45d1ee758 100644 --- a/tests/assertion/test_assertiontraceobserver.py +++ b/tests/assertion/test_assertiontraceobserver.py @@ -709,9 +709,20 @@ def test_is_unstable_runtime_value(): assert ato._is_unstable_runtime_value(list(sys.path)) is False -def test_is_blacklisted_module_maps_private_io_module(): - assert ato._is_blacklisted_module("_io") is True - assert ato._is_blacklisted_module("_iox") 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(