Method run_action

Lib/test/test__interpchannels.py:1691–1721

(self, fix, action, *, hideclosed=True)

            self.run_action(fix, action)

    def run_action(self, fix, action, *, hideclosed=True):
        end = action.resolve_end(fix.end)
        interp = action.resolve_interp(fix.interp, fix.other, fix.extra)
        fix.prep_interpreter(interp)
        if interp.name == 'main':
            result = run_action(
                fix.cid,
                action.action,
                end,
                fix.state,
                hideclosed=hideclosed,
                )
            fix.record_action(action, result)
        else:
            _cid = _channels.create(REPLACE)
            run_interp(interp.id, f"""
                result = helpers.run_action(
                    {fix.cid},
                    {repr(action.action)},
                    {repr(end)},
                    {repr(fix.state)},
                    hideclosed={hideclosed},
                    )
                _channels.send({_cid}, result.pending.to_bytes(1, 'little'), blocking=False)
                _channels.send({_cid}, b'X' if result.closed else b'', blocking=False)
                """)
            result = ChannelState(
                pending=int.from_bytes(recv_nowait(_cid), 'little'),
                closed=bool(recv_nowait(_cid)),
                )
            fix.record_action(action, result)

    def iter_fixtures(self):
        # XXX threads?