Make `@context`-cancelled tests more pedantic

In order to match a very significant and coming-soon patch set to the
IPC `Context` and `Channel` cancellation semantics with significant but
subtle changes to the primitives and runtime logic:

- a new set of `Context` state pub meth APIs for checking exact
  inter-actor-linked-task outcomes such as `.outcome`, `.maybe_error`,
  and `.cancel_acked`.

- trying to move away from `Context.cancelled_caught` usage since the
  semantics from `trio` don't really map well (in terms of cancel
  requests and how they result in cancel-scope graceful closure) and
  `.cancel_acked: bool` is a better approach for IPC req-resp msging.
  - change test usage to access `._scope.cancelled_caught` directly.

- more pedantic ctxc-raising expects around the "type of self
  cancellation" and final outcome in ctxc cases:
  - `ContextCancelled` is raised by ctx (`Context.result()`) consumer
    methods when `Portal.cancel_actor()` is called (since it's an
    out-of-band request) despite `Channel._cancel_called` being set.
  - also raised by `.open_context().__aexit__()` on close.
  - `.outcome` is always `.maybe_error` is always one of
    `._local/remote_error`.

tests/test_cancellation.py

@@ -48,11 +48,13 @@ async def do_nuthin():
     ids=['no_args', 'unexpected_args'],
 )
 def test_remote_error(reg_addr, args_err):
-    """Verify an error raised in a subactor that is propagated
+    '''
+    Verify an error raised in a subactor that is propagated
     to the parent nursery, contains the underlying boxed builtin
     error type info and causes cancellation and reraising all the
     way up the stack.
-    """
+
+    '''
     args, errtype = args_err
 
     async def main():
@@ -65,7 +67,9 @@ def test_remote_error(reg_addr, args_err):
             # an exception group outside the nursery since the error
             # here and the far end task error are one in the same?
             portal = await nursery.run_in_actor(
-                assert_err, name='errorer', **args
+                assert_err,
+                name='errorer',
+                **args
             )
 
             # get result(s) from main task

tests/test_context_stream_semantics.py

@@ -5,7 +5,7 @@ Verify the we raise errors when streams are opened prior to
 sync-opening a ``tractor.Context`` beforehand.
 
 '''
-# from contextlib import asynccontextmanager as acm
+from contextlib import asynccontextmanager as acm
 from itertools import count
 import platform
 from pprint import pformat
@@ -250,6 +250,17 @@ def test_simple_context(
         trio.run(main)
 
 
+@acm
+async def expect_ctxc(yay: bool) -> None:
+    if yay:
+        try:
+            yield
+        except ContextCancelled:
+            return
+    else:
+        yield
+
+
 @pytest.mark.parametrize(
     'callee_returns_early',
     [True, False],
@@ -280,23 +291,60 @@ def test_caller_cancels(
     async def check_canceller(
         ctx: Context,
     ) -> None:
-        # should not raise yet return the remote
+        actor: Actor = current_actor()
-        # context cancelled error.
+        uid: tuple = actor.uid
-        res = await ctx.result()
 
+        if (
+            cancel_method == 'portal'
+            and not callee_returns_early
+        ):
+            try:
+                res = await ctx.result()
+                assert 0, 'Portal cancel should raise!'
+
+            except ContextCancelled as ctxc:
+                assert ctx.chan._cancel_called
+                assert ctxc.canceller == uid
+                assert ctxc is ctx.maybe_error
+
+        # NOTE: should not ever raise even in the `ctx`
+        # case since self-cancellation should swallow the ctxc
+        # silently!
+        else:
+            res = await ctx.result()
+
+        # we actually get a result
         if callee_returns_early:
             assert res == 'yo'
+            assert ctx.outcome is res
+            assert ctx.maybe_error is None
 
         else:
-            err = res
+            err: Exception = ctx.outcome
             assert isinstance(err, ContextCancelled)
             assert (
                 tuple(err.canceller)
                 ==
-                current_actor().uid
+                uid
             )
+            assert (
+                err
+                is ctx.maybe_error
+                is ctx._remote_error
+            )
+            if le := ctx._local_error:
+                assert err is le
+
+            # else:
+                # TODO: what should this be then?
+                # not defined until block closes right?
+                #
+                # await tractor.pause()
+                # assert ctx._local_error is None
+
 
     async def main():
+
         async with tractor.open_nursery(
             debug_mode=debug_mode,
         ) as an:
@@ -306,11 +354,16 @@ def test_caller_cancels(
             )
             timeout = 0.5 if not callee_returns_early else 2
             with trio.fail_after(timeout):
-                async with portal.open_context(
+                async with (
-                    simple_setup_teardown,
+
-                    data=10,
+                    expect_ctxc(yay=cancel_method == 'portal'),
-                    block_forever=not callee_returns_early,
+
-                ) as (ctx, sent):
+                    portal.open_context(
+                        simple_setup_teardown,
+                        data=10,
+                        block_forever=not callee_returns_early,
+                    ) as (ctx, sent),
+                ):
 
                     if callee_returns_early:
                         # ensure we block long enough before sending
@@ -332,6 +385,16 @@ def test_caller_cancels(
             if cancel_method != 'portal':
                 await portal.cancel_actor()
 
+            # since the `.cancel_actor()` call just above
+            # will cause the `.open_context().__aexit__()` raise
+            # a ctxc which should in turn cause `ctx._scope` to
+            # catch any cancellation?
+            if (
+                not callee_returns_early
+                and cancel_method == 'portal'
+            ):
+                assert ctx._scope.cancelled_caught
+
     trio.run(main)
 
 
@@ -434,7 +497,6 @@ async def test_callee_closes_ctx_after_stream_open(
 
 @tractor.context
 async def expect_cancelled(
-
     ctx: Context,
 
 ) -> None:
@@ -454,7 +516,7 @@ async def expect_cancelled(
         raise
 
     else:
-        assert 0, "Wasn't cancelled!?"
+        assert 0, "callee wasn't cancelled !?"
 
 
 @pytest.mark.parametrize(
@@ -473,8 +535,8 @@ async def test_caller_closes_ctx_after_callee_opens_stream(
     async with tractor.open_nursery(
         debug_mode=debug_mode,
     ) as an:
-        root: Actor = current_actor()
 
+        root: Actor = current_actor()
         portal = await an.start_actor(
             'ctx_cancelled',
             enable_modules=[__name__],
@@ -487,11 +549,13 @@ async def test_caller_closes_ctx_after_callee_opens_stream(
 
             await portal.run(assert_state, value=True)
 
-            # call cancel explicitly
+            # call `ctx.cancel()` explicitly
             if use_ctx_cancel_method:
-
                 await ctx.cancel()
 
+                # NOTE: means the local side `ctx._scope` will
+                # have been cancelled by an ctxc ack and thus
+                # `._scope.cancelled_caught` should be set.
                 try:
                     async with ctx.open_stream() as stream:
                         async for msg in stream:
@@ -520,20 +584,35 @@ async def test_caller_closes_ctx_after_callee_opens_stream(
                 assert portal.channel.connected()
 
                 # ctx is closed here
-                await portal.run(assert_state, value=False)
+                await portal.run(
+                    assert_state,
+                    value=False,
+                )
 
             else:
                 try:
                     with trio.fail_after(0.2):
                         await ctx.result()
                         assert 0, "Callee should have blocked!?"
+
                 except trio.TooSlowError:
                     # NO-OP -> since already called above
                     await ctx.cancel()
 
-        # local scope should have absorbed the cancellation
+        # NOTE: local scope should have absorbed the cancellation since
-        assert ctx.cancelled_caught
+        # in this case we call `ctx.cancel()` and the local
-        assert ctx._remote_error is ctx._local_error
+        # `._scope` gets `.cancel_called` on the ctxc ack.
+        if use_ctx_cancel_method:
+            assert ctx._scope.cancelled_caught
+
+        # rxed ctxc response from far end
+        assert ctx.cancel_acked
+        assert (
+            ctx._remote_error
+            is ctx._local_error
+            is ctx.maybe_error
+            is ctx.outcome
+        )
 
         try:
             async with ctx.open_stream() as stream: