Wrap `send_chan_aclose_masks_beg.py` as test suite

Call it `test_trioisms::test_unmask_aclose_as_checkpoint_on_aexit` and
parametrize all script-mod`.main()` toggles including `.xfails()` for
the `raise_unmasked=False` cases.
cancelled_masking_guards
Tyler Goodlet 2025-09-05 18:43:58 -04:00
parent 759174729c
commit 04c3d5e239
1 changed files with 60 additions and 6 deletions

View File

@ -6,11 +6,18 @@ want to see changed.
from contextlib import ( from contextlib import (
asynccontextmanager as acm, asynccontextmanager as acm,
) )
from types import ModuleType
from functools import partial
import pytest import pytest
from _pytest import pathlib
from tractor.trionics import collapse_eg from tractor.trionics import collapse_eg
import trio import trio
from trio import TaskStatus from trio import TaskStatus
from tractor._testing import (
examples_dir,
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -106,8 +113,9 @@ def test_acm_embedded_nursery_propagates_enter_err(
debug_mode: bool, debug_mode: bool,
): ):
''' '''
Demo how a masking `trio.Cancelled` could be handled by unmasking from the Demo how a masking `trio.Cancelled` could be handled by unmasking
`.__context__` field when a user (by accident) re-raises from a `finally:`. from the `.__context__` field when a user (by accident) re-raises
from a `finally:`.
''' '''
import tractor import tractor
@ -158,13 +166,13 @@ def test_acm_embedded_nursery_propagates_enter_err(
assert len(assert_eg.exceptions) == 1 assert len(assert_eg.exceptions) == 1
def test_gatherctxs_with_memchan_breaks_multicancelled( def test_gatherctxs_with_memchan_breaks_multicancelled(
debug_mode: bool, debug_mode: bool,
): ):
''' '''
Demo how a using an `async with sndchan` inside a `.trionics.gather_contexts()` task Demo how a using an `async with sndchan` inside
will break a strict-eg-tn's multi-cancelled absorption.. a `.trionics.gather_contexts()` task will break a strict-eg-tn's
multi-cancelled absorption..
''' '''
from tractor import ( from tractor import (
@ -190,7 +198,6 @@ def test_gatherctxs_with_memchan_breaks_multicancelled(
f'Closed {task!r}\n' f'Closed {task!r}\n'
) )
async def main(): async def main():
async with ( async with (
# XXX should ensure ONLY the KBI # XXX should ensure ONLY the KBI
@ -211,3 +218,50 @@ def test_gatherctxs_with_memchan_breaks_multicancelled(
with pytest.raises(KeyboardInterrupt): with pytest.raises(KeyboardInterrupt):
trio.run(main) trio.run(main)
@pytest.mark.parametrize(
'raise_unmasked', [
True,
pytest.param(
False,
marks=pytest.mark.xfail(
reason="see examples/trio/send_chan_aclose_masks.py"
)
),
]
)
@pytest.mark.parametrize(
'child_errors_mid_stream',
[True, False],
)
def test_unmask_aclose_as_checkpoint_on_aexit(
raise_unmasked: bool,
child_errors_mid_stream: bool,
debug_mode: bool,
):
'''
Verify that our unmasker util works over the common case where
a mem-chan's `.aclose()` is included in an `@acm` stack
and it being currently a checkpoint, can `trio.Cancelled`-mask an embedded
exception from user code resulting in a silent failure which
appears like graceful cancellation.
This test suite is mostly implemented as an example script so it
could more easily be shared with `trio`-core peeps as `tractor`-less
minimum reproducing example.
'''
mod: ModuleType = pathlib.import_path(
examples_dir()
/ 'trio'
/ 'send_chan_aclose_masks_beg.py',
root=examples_dir(),
consider_namespace_packages=False,
)
with pytest.raises(RuntimeError):
trio.run(partial(
mod.main,
raise_unmasked=raise_unmasked,
child_errors_mid_stream=child_errors_mid_stream,
))