HAL9000
2609787d35
Introduce a reference-counted shared stream wrapper manager so concurrent ValidationPipeline.run() calls correctly restore the true sys.stdout/stderr after all pipelines finish. - Add _install_thread_local_streams() / _release_thread_local_streams() protected by threading.Lock; first caller saves originals and installs wrappers, last caller restores them - Remove io.TextIOBase inheritance from _ThreadLocalStream to avoid Python 3.13 read-only slot conflict; use cast(TextIO, ...) at assignment sites - encoding property returns str | None matching io.TextIOBase signature - Replace assert guards with explicit RuntimeError fail-fast checks - Add Behave scenario: Concurrent pipelines restore global streams after execution - Split validation_pipeline_steps.py (547→428 lines) by extracting MockValidationExecutor to _validation_pipeline_mock.py and edge-case steps to validation_pipeline_edge_steps.py - Update CHANGELOG.md and CONTRIBUTORS.md ISSUES CLOSED: #7623
43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
"""Step definitions for validation pipeline edge-case executor scenarios.
|
|
|
|
Covers non-standard executor return values (non-bool passed, non-string
|
|
message, non-dict data) and stdout/stderr printing executors.
|
|
|
|
These steps rely on ``context.vp_executor`` being a ``MockValidationExecutor``
|
|
instance, which is set up by the ``a validation pipeline test environment``
|
|
Given step in ``validation_pipeline_steps.py``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from behave import given, use_step_matcher
|
|
|
|
use_step_matcher("re")
|
|
|
|
|
|
@given(r'the vp executor returns non-bool passed for "(?P<name>[^"]+)"')
|
|
def step_given_executor_non_bool_passed(context: Any, name: str) -> None:
|
|
context.vp_executor.set_non_bool_passed(name)
|
|
|
|
|
|
@given(r'the vp executor returns non-string message for "(?P<name>[^"]+)"')
|
|
def step_given_executor_non_string_message(context: Any, name: str) -> None:
|
|
context.vp_executor.set_non_string_message(name)
|
|
|
|
|
|
@given(r'the vp executor returns non-dict data for "(?P<name>[^"]+)"')
|
|
def step_given_executor_non_dict_data(context: Any, name: str) -> None:
|
|
context.vp_executor.set_non_dict_data(name)
|
|
|
|
|
|
@given(r'the vp executor prints to stdout for "(?P<name>[^"]+)"')
|
|
def step_given_executor_prints_stdout(context: Any, name: str) -> None:
|
|
context.vp_executor.set_prints_stdout(name)
|
|
|
|
|
|
@given(r'the vp executor prints to stderr for "(?P<name>[^"]+)"')
|
|
def step_given_executor_prints_stderr(context: Any, name: str) -> None:
|
|
context.vp_executor.set_prints_stderr(name)
|