helpers contextmanager def runthreadedid source shared def run runinterpid source shared t threading threadtargetrun t start yield t join xxx run a func xxx expectchannelclosed is unnecessary once we improve exc propagation per init there aren t any others channel tests xxx check the following in the channel tests self assertisnotgot obj cid1 intout strip id2 interpreters create out runoutputid2 dedent import xxinterpchannels as channels cid channels create printcid test listing interpreters for a channel with no associations test for channel with no associated interpreters cid channels create sendinterps channels listinterpreterscid sendtrue recvinterps channels listinterpreterscid sendfalse self assertequalsendinterps self assertequalrecvinterps def testchannellistinterpretersbasicself test for a channel that has one end associated to an interpreter test for channel that has both ends associated to an interpreter sendinterps channels listinterpreterscid sendtrue recvinterps channels listinterpreterscid sendfalse self assertequalsendinterps interp0 self assertequalrecvinterps interp1 def testchannellistinterpretersmultipleself runoutputinterp2 dedentf import xxinterpchannels as channels obj channels recvcid sendinterps channels listinterpreterscid sendtrue recvinterps channels listinterpreterscid sendfalse self assertequalsetsendinterps interp0 interp1 self assertequalsetrecvinterps interp2 interp3 def testchannellistinterpretersdestroyedself should be one interpreter associated with each end sendinterps channels listinterpreterscid sendtrue recvinterps channels listinterpreterscid sendfalse self assertequalsendinterps interp0 self assertequalrecvinterps interp1 interpreters destroyinterp1 destroyed interpreter should not be listed sendinterps channels listinterpreterscid sendtrue recvinterps channels listinterpreterscid sendfalse self assertequalsendinterps interp0 self assertequalrecvinterps def testchannellistinterpretersreleasedself set up one channel with main interpreter on the send end and two subinterpreters on the receive end channels sendcid data blockingfalse runoutputinterp2 dedentf import xxinterpchannels as channels obj channels recvcid check the setup release the main interpreter from the send end send end should have no associated interpreters release one of the subinterpreters from the receive end receive end should have the released interpreter removed sendinterps channels listinterpreterscid sendtrue recvinterps channels listinterpreterscid sendfalse self assertequallensendinterps 0 self assertequalrecvinterps interp1 def testchannellistinterpretersclosedself put something in the channel so that it s not empty check initial state force close the channel both ends should raise an error test listing channel interpreters with a channel s send end closed interp0 interpreters getmain interp1 interpreters create cid channels create put something in the channel so that it s not empty channels sendcid send blockingfalse check initial state sendinterps channels listinterpreterscid sendtrue recvinterps channels listinterpreterscid sendfalse self assertequallensendinterps 1 self assertequallenrecvinterps 0 close the send end of the channel channels closecid sendtrue send end should raise an error with self assertraiseschannels channelclosederror channels listinterpreterscid sendtrue receive end should not be closed since channel is not empty recvinterps channels listinterpreterscid sendfalse self assertequallenrecvinterps 0 close the receive end of the channel from a subinterpreter runoutputinterp1 dedentf import xxinterpchannels as channels channels closecid forcetrue both ends should raise an error xxx check the following self assertisnotgot obj xxx what about between interpreters dictcidcid send obj channels recvcid self assertequalobj b spam self assertequalout strip send xxx for now there is no highlevel channel into which the sent channel id can be converted note this test caused crashes on some buildbots bpo33615 unittest skip disabled until highlevel channels exist def testrunstringargresolvedself cid channels create cid channels channelidcid resolvetrue interp interpreters create out runoutputinterp dedent import xxinterpchannels as channels printchan id end channels sendchan id b spam blockingfalse sendrecv def testsendrecvdifferentinterpretersself cid channels create id1 interpreters create out runoutputid1 dedentf import xxinterpchannels as channels channels sendcid b spam blockingfalse t threading threadtargetf t start channels sendcid b spam obj recvwaitcid t join self assertequalobj b eggs def testsendnotfoundself with self assertraiseschannels channelnotfounderror channels send10 b spam def testrecvnotfoundself with self assertraiseschannels channelnotfounderror channels recv10 def testrecvemptyself cid channels create with self assertraiseschannels channelemptyerror channels recvcid def testrecvdefaultself default object cid channels create obj1 channels recvcid default channels sendcid none blockingfalse channels sendcid 1 blockingfalse channels sendcid b spam blockingfalse channels sendcid b eggs blockingfalse obj2 channels recvcid default obj3 channels recvcid default obj4 channels recvcid obj5 channels recvcid default obj6 channels recvcid default self assertisobj1 default self assertisobj2 none self assertequalobj3 1 self assertequalobj4 b spam self assertequalobj5 b eggs self assertisobj6 default def testrecvsendinginterpdestroyedself with self subtest closed cid1 channels create interp interpreters create interpreters runstringinterp dedentf import xxinterpchannels as channels channels sendcid1 b spam blockingfalse channels sendcid2 b eggs blockingfalse interpreters destroyinterp channels recvcid2 with self assertraisesregexruntimeerror f channel cid2 is empty channels recvcid2 del cid2 sendbuffer def testsendbufferself buf bytearrayb spamspamspam cid channels create channels sendbuffercid buf blockingfalse obj channels recvcid self assertisnotobj buf self assertisinstanceobj memoryview self assertequalobj buf buf4 8 b eggs self assertequalobj buf obj4 8 b ham self assertequalobj buf send with waiting def buildsendwaiterself obj bufferfalse we want a long enough sleep that send actually has to wait if buffer send channels sendbuffer else send channels send cid channels create try started time monotonic sendcid obj blockingfalse stopped time monotonic channels recvcid finally channels destroycid delay stopped started seconds delay 3 def wait time sleepdelay return wait def testsendblockingwaitingself received none obj b spam wait self buildsendwaiterobj cid channels create def f nonlocal received wait received recvwaitcid t threading threadtargetf t start channels sendcid obj blockingtrue t join self assertequalreceived obj def testsendbufferblockingwaitingself received none obj bytearrayb spam wait self buildsendwaiterobj buffertrue cid channels create def f nonlocal received wait received recvwaitcid t threading threadtargetf t start channels sendbuffercid obj blockingtrue t join self assertequalreceived obj def testsendblockingnowaitself received none obj b spam cid channels create def f nonlocal received received recvwaitcid t threading threadtargetf t start channels sendcid obj blockingtrue t join self assertequalreceived obj def testsendbufferblockingnowaitself received none obj bytearrayb spam cid channels create def f nonlocal received received recvwaitcid t threading threadtargetf t start channels sendbuffercid obj blockingtrue t join self assertequalreceived obj def testsendtimeoutself obj b spam with self subtest nonblocking with timeout cid channels create with self assertraisesvalueerror channels sendcid obj blockingfalse timeout0 1 with self subtest timeout hit cid channels create with self assertraisestimeouterror channels sendcid obj blockingtrue timeout0 1 with self assertraiseschannels channelemptyerror received channels recvcid printreprreceived with self subtest timeout not hit cid channels create def f recvwaitcid t threading threadtargetf t start channels sendcid obj blockingtrue timeout10 t join def testsendbuffertimeoutself try self hasrunoncetimeout except attributeerror at the moment this test leaks a few references it looks like the leak originates with the addition of channels sendbuffer gh110246 whereas the tests were added afterward we want this test even if the refleak isn t fixed yet so we skip here raise unittest skiptest temporarily skipped due to refleaks else self hasrunoncetimeout true obj bytearrayb spam with self subtest nonblocking with timeout cid channels create with self assertraisesvalueerror channels sendbuffercid obj blockingfalse timeout0 1 with self subtest timeout hit cid channels create with self assertraisestimeouterror channels sendbuffercid obj blockingtrue timeout0 1 with self assertraiseschannels channelemptyerror received channels recvcid printreprreceived with self subtest timeout not hit cid channels create def f recvwaitcid t threading threadtargetf t start channels sendbuffercid obj blockingtrue timeout10 t join def testsendclosedwhilewaitingself obj b spam wait self buildsendwaiterobj with self subtest without timeout cid channels create def f wait channels closecid forcetrue t threading threadtargetf t start with self assertraiseschannels channelclosederror channels sendcid obj blockingtrue t join with self subtest with timeout cid channels create def f wait channels closecid forcetrue t threading threadtargetf t start with self assertraiseschannels channelclosederror channels sendcid obj blockingtrue timeout30 t join def testsendbufferclosedwhilewaitingself try self hasrunonceclosed except attributeerror at the moment this test leaks a few references it looks like the leak originates with the addition of channels sendbuffer gh110246 whereas the tests were added afterward we want this test even if the refleak isn t fixed yet so we skip here raise unittest skiptest temporarily skipped due to refleaks else self hasrunonceclosed true obj bytearrayb spam wait self buildsendwaiterobj buffertrue with self subtest without timeout cid channels create def f wait channels closecid forcetrue t threading threadtargetf t start with self assertraiseschannels channelclosederror channels sendbuffercid obj blockingtrue t join with self subtest with timeout cid channels create def f wait channels closecid forcetrue t threading threadtargetf t start with self assertraiseschannels channelclosederror channels sendbuffercid obj blockingtrue timeout30 t join close def testclosesingleuserself cid channels create channels sendcid b spam blockingfalse channels recvcid channels closecid with self assertraiseschannels channelclosederror channels sendcid b eggs with self assertraiseschannels channelclosederror channels recvcid def testclosemultipleusersself cid channels create id1 interpreters create id2 interpreters create interpreters runstringid1 dedentf import xxinterpchannels as channels channels sendcid b spam blockingfalse channels closecid excsnap interpreters runstringid1 dedentf channels sendcid b spam self assertequalexcsnap type name channelclosederror def testclosemultipletimesself cid channels create channels sendcid b spam blockingfalse channels recvcid channels closecid with self assertraiseschannels channelclosederror channels closecid def testcloseemptyself tests false false true false false true true true for send recv in tests with self subtestsend recv cid channels create channels sendcid b spam blockingfalse channels recvcid channels closecid sendsend recvrecv with self assertraiseschannels channelclosederror channels sendcid b eggs with self assertraiseschannels channelclosederror channels recvcid def testclosedefaultswithunuseditemsself cid channels create channels sendcid b spam blockingfalse channels sendcid b ham blockingfalse with self assertraiseschannels channelnotemptyerror channels closecid channels recvcid channels sendcid b eggs blockingfalse def testcloserecvwithunuseditemsunforcedself cid channels create channels sendcid b spam blockingfalse channels sendcid b ham blockingfalse with self assertraiseschannels channelnotemptyerror channels closecid recvtrue channels recvcid channels sendcid b eggs blockingfalse channels recvcid channels recvcid channels closecid recvtrue def testclosesendwithunuseditemsunforcedself cid channels create channels sendcid b spam blockingfalse channels sendcid b ham blockingfalse channels closecid sendtrue with self assertraiseschannels channelclosederror channels sendcid b eggs channels recvcid channels recvcid with self assertraiseschannels channelclosederror channels recvcid def testclosebothwithunuseditemsunforcedself cid channels create channels sendcid b spam blockingfalse channels sendcid b ham blockingfalse with self assertraiseschannels channelnotemptyerror channels closecid recvtrue sendtrue channels recvcid channels sendcid b eggs blockingfalse channels recvcid channels recvcid channels closecid recvtrue def testcloserecvwithunuseditemsforcedself cid channels create channels sendcid b spam blockingfalse channels sendcid b ham blockingfalse channels closecid recvtrue forcetrue with self assertraiseschannels channelclosederror channels sendcid b eggs with self assertraiseschannels channelclosederror channels recvcid def testclosesendwithunuseditemsforcedself cid channels create channels sendcid b spam blockingfalse channels sendcid b ham blockingfalse channels closecid sendtrue forcetrue with self assertraiseschannels channelclosederror channels sendcid b eggs with self assertraiseschannels channelclosederror channels recvcid def testclosebothwithunuseditemsforcedself cid channels create channels sendcid b spam blockingfalse channels sendcid b ham blockingfalse channels closecid sendtrue recvtrue forcetrue with self assertraiseschannels channelclosederror channels sendcid b eggs with self assertraiseschannels channelclosederror channels recvcid def testcloseneverusedself cid channels create channels closecid with self assertraiseschannels channelclosederror channels sendcid b spam with self assertraiseschannels channelclosederror channels recvcid def testclosebyunassociatedinterpself cid channels create channels sendcid b spam blockingfalse interp interpreters create interpreters runstringinterp dedentf import xxinterpchannels as channels channels closecid forcetrue test for invalid channel id test for a channel that has been closed tests for invalid arguments passed to the api xxx add more test coverage a la the tests for close main interp other run in current thread new thread other thread different threads end opposite force no force used not used associated not associated empty emptied never emptied partly emptied closed not closed released not released creator interp other associated interpreter not running associated interpreter destroyed use prerelease release after check release in main interp1 creator same other incl interp2 use none send recv sendrecv in none same otherincl interp2 sameotherincl interp2 all prerelease none send recv both in none same otherincl interp2 sameotherincl interp2 all prerelease forced none send recv both in none same otherincl interp2 sameotherincl interp2 all release same release forced same use after none send recv sendrecv in none same otherincl interp2 sameotherincl interp2 all release after none send recv sendrecv in none same otherincl interp2 sameotherincl interp2 all check released sendrecv for sameotherincl interp2 check closed sendrecv for sameotherincl interp2 out runoutputid2 dedentf import xxinterpchannels as channels obj channels recvcid channels releasecid printreprobj self assertequalout strip b spam def testnokwargsself cid channels create channels sendcid b spam blockingfalse channels recvcid channels releasecid with self assertraiseschannels channelclosederror channels sendcid b eggs with self assertraiseschannels channelclosederror channels recvcid def testmultipletimesself cid channels create channels sendcid b spam blockingfalse channels recvcid channels releasecid sendtrue recvtrue with self assertraiseschannels channelclosederror channels releasecid sendtrue recvtrue def testwithunuseditemsself cid channels create channels sendcid b spam blockingfalse channels sendcid b ham blockingfalse channels releasecid sendtrue recvtrue with self assertraiseschannels channelclosederror channels recvcid def testneverusedself cid channels create channels releasecid with self assertraiseschannels channelclosederror channels sendcid b spam with self assertraiseschannels channelclosederror channels recvcid def testbyunassociatedinterpself cid channels create channels sendcid b spam blockingfalse interp interpreters create interpreters runstringinterp dedentf import xxinterpchannels as channels channels releasecid xxx something s not right with this test with self assertraiseschannels channelclosederror channels recvcid def testpartiallyself xxx is partial close too weirdconfusing cid channels create channels sendcid none blockingfalse channels recvcid channels sendcid b spam blockingfalse channels releasecid sendtrue obj channels recvcid self assertequalobj b spam def testusedmultipletimesbysingleuserself cid channels create channels sendcid b spam blockingfalse channels sendcid b spam blockingfalse channels sendcid b spam blockingfalse channels recvcid channels releasecid sendtrue recvtrue with self assertraiseschannels channelclosederror channels sendcid b eggs with self assertraiseschannels channelclosederror channels recvcid class channelclosefixturenamedtuple channelclosefixture end interp other extra creator set this to true to avoid creating interpreters e g when scanning through test permutations without running them quick false def newcls end interp other extra creator assert end in send recv if cls quick known else interp interpreter fromrawinterp other interpreter fromrawother extra interpreter fromrawextra known interp name interp other name other extra name extra if not creator creator same self super newcls end interp other extra creator self prepped set self state channelstate self known known return self property def stateself return self state property def cidself try return self cid except attributeerror creator self getinterpreterself creator self cid self newchannelcreator return self cid def getinterpreterself interp interp self getinterpreterinterp self prepinterpreterinterp return interp def expectclosederrorself endnone if end is none end self end if end recv and self state closed send return false return boolself state closed def prepinterpreterself interp self prepinterpreterinterp def recordactionself action result self state result def cleanupself cleanupinterpreters cleanupchannels internal methods def newchannelself creator if creator name main return channels create else ch channels create runinterpcreator id f import xxsubinterpreters cid xxsubchannels create we purposefully send back an int to avoid tying the channel to the other interpreter xxsubchannels sendch intcid blockingfalse del xxsubinterpreters unittest skip these tests take several hours to run class exhaustivechannelteststestbase def iteractionsetsself used not used associated not associated empty emptied never emptied partly emptied closed not closed released not released never used yield only preclosed and possible used after for closeactions in self itercloseactionsets same other yield closeactions for postactions in self iterpostcloseactionsets yield closeactions postactions for closeactions in self itercloseactionsets other extra yield closeactions for postactions in self iterpostcloseactionsets yield closeactions postactions used for useactions in self iteruseactionsets same other yield useactions for closeactions in self itercloseactionsets same other actions useactions closeactions yield actions for postactions in self iterpostcloseactionsets yield actions postactions for closeactions in self itercloseactionsets other extra actions useactions closeactions yield actions for postactions in self iterpostcloseactionsets yield actions postactions for useactions in self iteruseactionsets other extra yield useactions for closeactions in self itercloseactionsets same other actions useactions closeactions yield actions for postactions in self iterpostcloseactionsets yield actions postactions for closeactions in self itercloseactionsets other extra actions useactions closeactions yield actions for postactions in self iterpostcloseactionsets yield actions postactions def iteruseactionsetsself interp1 interp2 interps interp1 interp2 only recv end used yield channelaction use recv interp1 yield channelaction use recv interp2 yield channelaction use recv interp1 channelaction use recv interp2 never emptied yield channelaction use send interp1 yield channelaction use send interp2 yield channelaction use send interp1 channelaction use send interp2 partially emptied for interp1 in interps for interp2 in interps for interp3 in interps yield channelaction use send interp1 channelaction use send interp2 channelaction use recv interp3 fully emptied for interp1 in interps for interp2 in interps for interp3 in interps for interp4 in interps yield channelaction use send interp1 channelaction use send interp2 channelaction use recv interp3 channelaction use recv interp4 def itercloseactionsetsself interp1 interp2 ends recv send interps interp1 interp2 for force in true false op forceclose if force else close for interp in interps for end in ends yield channelactionop end interp for recvop in close forceclose for sendop in close forceclose for recv in interps for send in interps yield channelactionrecvop recv recv channelactionsendop send send def iterpostcloseactionsetsself for interp in same extra other yield channelaction use recv interp yield channelaction use send interp def runactionsself fix actions for action in actions self runactionfix action def runactionself fix action hideclosedtrue end action resolveendfix end interp action resolveinterpfix interp fix other fix extra fix prepinterpreterinterp if interp name main result runaction fix cid action action end fix state hideclosedhideclosed fix recordactionaction result else cid channels create runinterpinterp id f result helpers runaction fix cid repraction action reprend reprfix state hideclosedhideclosed channels sendcid result pending tobytes1 little blockingfalse channels sendcid b x if result closed else b blockingfalse xxx threads runinterpinterp id with helpers expectchannelclosed channels sendcid b spam blockingfalse runinterpinterp id with helpers expectchannelclosed channels closecid forcetrue this is useful for scanning through the possible tests xxx things slow down if we have too many interpreters xxx things slow down if we have too many interpreters helpers contextmanager def run_threaded id source shared def run run_interp id source shared t threading thread target run t start yield t join xxx run a func xxx expect_channel_closed is unnecessary once we improve exc propagation per __init__ there aren t any others xxx already destroyed channel tests unforced xxx check the following in the channel tests self assertisnot got obj import _xxinterpchannels as _channels cid _channels create print cid import _xxinterpchannels as _channels cid _channels create print cid test listing interpreters for a channel with no associations test for channel with no associated interpreters test basic listing channel interpreters test for a channel that has one end associated to an interpreter import _xxinterpchannels as _channels obj _channels recv cid test for channel that has both ends associated to an interpreter test listing interpreters for a channel with many associations import _xxinterpchannels as _channels _channels send cid send blocking false import _xxinterpchannels as _channels obj _channels recv cid import _xxinterpchannels as _channels obj _channels recv cid test listing channel interpreters with a destroyed interpreter import _xxinterpchannels as _channels obj _channels recv cid should be one interpreter associated with each end destroyed interpreter should not be listed test listing channel interpreters with a released channel set up one channel with main interpreter on the send end and two subinterpreters on the receive end import _xxinterpchannels as _channels obj _channels recv cid import _xxinterpchannels as _channels obj _channels recv cid check the setup release the main interpreter from the send end send end should have no associated interpreters release one of the subinterpreters from the receive end import _xxinterpchannels as _channels _channels release cid receive end should have the released interpreter removed test listing channel interpreters with a closed channel put something in the channel so that it s not empty check initial state force close the channel both ends should raise an error test listing channel interpreters with a channel s send end closed put something in the channel so that it s not empty check initial state close the send end of the channel send end should raise an error receive end should not be closed since channel is not empty close the receive end of the channel from a subinterpreter import _xxinterpchannels as _channels _channels close cid force true both ends should raise an error xxx check the following self assertisnot got obj xxx what about between interpreters import _xxinterpchannels as _channels print cid end _channels send cid b spam blocking false xxx for now there is no high level channel into which the sent channel id can be converted note this test caused crashes on some buildbots bpo 33615 import _xxinterpchannels as _channels print chan id end _channels send chan id b spam blocking false send recv import _xxinterpchannels as _channels cid _channels create orig b spam _channels send cid orig blocking false obj _channels recv cid assert obj is not orig assert obj orig import _xxinterpchannels as _channels _channels send cid b spam blocking false import time import _xxinterpchannels as _channels while true try obj _channels recv cid break except _channels channelemptyerror time sleep 0 1 assert obj b spam _channels send cid b eggs import _xxinterpchannels as _channels _channels send cid1 b spam blocking false import _xxinterpchannels as _channels _channels send cid2 b spam blocking false send_buffer send with waiting we want a long enough sleep that send actually has to wait seconds at the moment this test leaks a few references it looks like the leak originates with the addition of _channels send_buffer gh 110246 whereas the tests were added afterward we want this test even if the refleak isn t fixed yet so we skip here at the moment this test leaks a few references it looks like the leak originates with the addition of _channels send_buffer gh 110246 whereas the tests were added afterward we want this test even if the refleak isn t fixed yet so we skip here close import _xxinterpchannels as _channels _channels send cid b spam blocking false import _xxinterpchannels as _channels _channels recv cid _channels send cid b spam _channels send cid b spam import _xxinterpchannels as _channels _channels close cid force true test for invalid channel id test for a channel that has been closed tests for invalid arguments passed to the api xxx add more test coverage a la the tests for close main interp other run in current thread new thread other thread different threads end opposite force no force used not used associated not associated empty emptied never emptied partly emptied closed not closed released not released creator interp other associated interpreter not running associated interpreter destroyed use pre release release after check release in main interp1 creator same other incl interp2 use none send recv send recv in none same other incl interp2 same other incl interp2 all pre release none send recv both in none same other incl interp2 same other incl interp2 all pre release forced none send recv both in none same other incl interp2 same other incl interp2 all release same release forced same use after none send recv send recv in none same other incl interp2 same other incl interp2 all release after none send recv send recv in none same other incl interp2 same other incl interp2 all check released send recv for same other incl interp2 check closed send recv for same other incl interp2 import _xxinterpchannels as _channels _channels send cid b spam blocking false import _xxinterpchannels as _channels obj _channels recv cid _channels release cid print repr obj _channels release cid import _xxinterpchannels as _channels _channels release cid xxx something s not right with this test import _xxinterpchannels as _channels obj _channels send cid b spam blocking false _channels release cid xxx is partial close too weird confusing set this to true to avoid creating interpreters e g when scanning through test permutations without running them internal methods import _xxsubinterpreters cid _xxsubchannels create we purposefully send back an int to avoid tying the channel to the other interpreter _xxsubchannels send ch int cid blocking false del _xxsubinterpreters import _xxinterpchannels as channels import test test__xxinterpchannels as helpers channelstate helpers channelstate try cid except nameerror cid channels _channel_id self cid main interp other run in current thread new thread other thread different threads end opposite force no force used not used associated not associated empty emptied never emptied partly emptied closed not closed released not released creator interp other associated interpreter not running associated interpreter destroyed close after unbound use pre close close after check close in main interp1 creator same other extra use none send recv send recv in none same other same other all pre close none send recv in none same other same other all pre close forced none send recv in none same other same other all close same close forced same use after none send recv send recv in none same other extra same other all close after none send recv send recv in none same other extra same other all check closed send recv for same other incl interp2 used not used associated not associated empty emptied never emptied partly emptied closed not closed released not released never used only pre closed and possible used after used only recv end used never emptied partially emptied fully emptied 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 xxx threads with helpers expect_channel_closed channels recv cid with helpers expect_channel_closed channels send cid b spam blocking false with helpers expect_channel_closed channels close cid with helpers expect_channel_closed channels close cid force true this is useful for scanning through the possible tests xxx things slow down if we have too many interpreters xxx things slow down if we have too many interpreters
from collections import namedtuple
import contextlib
import sys
from textwrap import dedent
import threading
import time
import unittest

from test.support import import_helper

from test.test__xxsubinterpreters import (
    interpreters,
    _run_output,
    clean_up_interpreters,
)


channels = import_helper.import_module('_xxinterpchannels')





def recv_wait(cid):
    while True:
        try:
            return channels.recv(cid)
        except channels.ChannelEmptyError:
            time.sleep(0.1)











def run_interp(id, source, **shared):
    _run_interp(id, source, shared)


def _run_interp(id, source, shared, _mainns={}):
    source = dedent(source)
    main = interpreters.get_main()
    if main == id:
        if interpreters.get_current() != main:
            raise RuntimeError
        
        exec(source, _mainns)
    else:
        interpreters.run_string(id, source, shared)


class Interpreter(namedtuple('Interpreter', 'name id')):

    @classmethod
    def from_raw(cls, raw):
        if isinstance(raw, cls):
            return raw
        elif isinstance(raw, str):
            return cls(raw)
        else:
            raise NotImplementedError

    def __new__(cls, name=None, id=None):
        main = interpreters.get_main()
        if id == main:
            if not name:
                name = 'main'
            elif name != 'main':
                raise ValueError(
                    'name mismatch (expected "main", got "{}")'.format(name))
            id = main
        elif id is not None:
            if not name:
                name = 'interp'
            elif name == 'main':
                raise ValueError('name mismatch (unexpected "main")')
            if not isinstance(id, interpreters.InterpreterID):
                id = interpreters.InterpreterID(id)
        elif not name or name == 'main':
            name = 'main'
            id = main
        else:
            id = interpreters.create()
        self = super().__new__(cls, name, id)
        return self




@contextlib.contextmanager
def expect_channel_closed():
    try:
        yield
    except channels.ChannelClosedError:
        pass
    else:
        assert False, 'channel not closed'


class ChannelAction(namedtuple('ChannelAction', 'action end interp')):

    def __new__(cls, action, end=None, interp=None):
        if not end:
            end = 'both'
        if not interp:
            interp = 'main'
        self = super().__new__(cls, action, end, interp)
        return self

    def __init__(self, *args, **kwargs):
        if self.action == 'use':
            if self.end not in ('same', 'opposite', 'send', 'recv'):
                raise ValueError(self.end)
        elif self.action in ('close', 'force-close'):
            if self.end not in ('both', 'same', 'opposite', 'send', 'recv'):
                raise ValueError(self.end)
        else:
            raise ValueError(self.action)
        if self.interp not in ('main', 'same', 'other', 'extra'):
            raise ValueError(self.interp)

    def resolve_end(self, end):
        if self.end == 'same':
            return end
        elif self.end == 'opposite':
            return 'recv' if end == 'send' else 'send'
        else:
            return self.end

    def resolve_interp(self, interp, other, extra):
        if self.interp == 'same':
            return interp
        elif self.interp == 'other':
            if other is None:
                raise RuntimeError
            return other
        elif self.interp == 'extra':
            if extra is None:
                raise RuntimeError
            return extra
        elif self.interp == 'main':
            if interp.name == 'main':
                return interp
            elif other and other.name == 'main':
                return other
            else:
                raise RuntimeError
        


class ChannelState(namedtuple('ChannelState', 'pending closed')):

    def __new__(cls, pending=0, *, closed=False):
        self = super().__new__(cls, pending, closed)
        return self

    def incr(self):
        return type(self)(self.pending + 1, closed=self.closed)

    def decr(self):
        return type(self)(self.pending - 1, closed=self.closed)

    def close(self, *, force=True):
        if self.closed:
            if not force or self.pending == 0:
                return self
        return type(self)(0 if force else self.pending, closed=True)


def run_action(cid, action, end, state, *, hideclosed=True):
    if state.closed:
        if action == 'use' and end == 'recv' and state.pending:
            expectfail = False
        else:
            expectfail = True
    else:
        expectfail = False

    try:
        result = _run_action(cid, action, end, state)
    except channels.ChannelClosedError:
        if not hideclosed and not expectfail:
            raise
        result = state.close()
    else:
        if expectfail:
            raise ...  
    return result


def _run_action(cid, action, end, state):
    if action == 'use':
        if end == 'send':
            channels.send(cid, b'spam', blocking=False)
            return state.incr()
        elif end == 'recv':
            if not state.pending:
                try:
                    channels.recv(cid)
                except channels.ChannelEmptyError:
                    return state
                else:
                    raise Exception('expected ChannelEmptyError')
            else:
                channels.recv(cid)
                return state.decr()
        else:
            raise ValueError(end)
    elif action == 'close':
        kwargs = {}
        if end in ('recv', 'send'):
            kwargs[end] = True
        channels.close(cid, **kwargs)
        return state.close()
    elif action == 'force-close':
        kwargs = {
            'force': True,
            }
        if end in ('recv', 'send'):
            kwargs[end] = True
        channels.close(cid, **kwargs)
        return state.close(force=True)
    else:
        raise ValueError(action)


def clean_up_channels():
    for cid in channels.list_all():
        try:
            channels.destroy(cid)
        except channels.ChannelNotFoundError:
            pass  


class TestBase(unittest.TestCase):

    def tearDown(self):
        clean_up_channels()
        clean_up_interpreters()





class ChannelIDTests(TestBase):

    def test_default_kwargs(self):
        cid = channels._channel_id(10, force=True)

        self.assertEqual(int(cid), 10)
        self.assertEqual(cid.end, 'both')

    def test_with_kwargs(self):
        cid = channels._channel_id(10, send=True, force=True)
        self.assertEqual(cid.end, 'send')

        cid = channels._channel_id(10, send=True, recv=False, force=True)
        self.assertEqual(cid.end, 'send')

        cid = channels._channel_id(10, recv=True, force=True)
        self.assertEqual(cid.end, 'recv')

        cid = channels._channel_id(10, recv=True, send=False, force=True)
        self.assertEqual(cid.end, 'recv')

        cid = channels._channel_id(10, send=True, recv=True, force=True)
        self.assertEqual(cid.end, 'both')

    def test_coerce_id(self):
        class Int(str):
            def __index__(self):
                return 10

        cid = channels._channel_id(Int(), force=True)
        self.assertEqual(int(cid), 10)

    def test_bad_id(self):
        self.assertRaises(TypeError, channels._channel_id, object())
        self.assertRaises(TypeError, channels._channel_id, 10.0)
        self.assertRaises(TypeError, channels._channel_id, '10')
        self.assertRaises(TypeError, channels._channel_id, b'10')
        self.assertRaises(ValueError, channels._channel_id, -1)
        self.assertRaises(OverflowError, channels._channel_id, 2**64)

    def test_bad_kwargs(self):
        with self.assertRaises(ValueError):
            channels._channel_id(10, send=False, recv=False)

    def test_does_not_exist(self):
        cid = channels.create()
        with self.assertRaises(channels.ChannelNotFoundError):
            channels._channel_id(int(cid) + 1)  

    def test_str(self):
        cid = channels._channel_id(10, force=True)
        self.assertEqual(str(cid), '10')

    def test_repr(self):
        cid = channels._channel_id(10, force=True)
        self.assertEqual(repr(cid), 'ChannelID(10)')

        cid = channels._channel_id(10, send=True, force=True)
        self.assertEqual(repr(cid), 'ChannelID(10, send=True)')

        cid = channels._channel_id(10, recv=True, force=True)
        self.assertEqual(repr(cid), 'ChannelID(10, recv=True)')

        cid = channels._channel_id(10, send=True, recv=True, force=True)
        self.assertEqual(repr(cid), 'ChannelID(10)')

    def test_equality(self):
        cid1 = channels.create()
        cid2 = channels._channel_id(int(cid1))
        cid3 = channels.create()

        self.assertTrue(cid1 == cid1)
        self.assertTrue(cid1 == cid2)
        self.assertTrue(cid1 == int(cid1))
        self.assertTrue(int(cid1) == cid1)
        self.assertTrue(cid1 == float(int(cid1)))
        self.assertTrue(float(int(cid1)) == cid1)
        self.assertFalse(cid1 == float(int(cid1)) + 0.1)
        self.assertFalse(cid1 == str(int(cid1)))
        self.assertFalse(cid1 == 2**1000)
        self.assertFalse(cid1 == float('inf'))
        self.assertFalse(cid1 == 'spam')
        self.assertFalse(cid1 == cid3)

        self.assertFalse(cid1 != cid1)
        self.assertFalse(cid1 != cid2)
        self.assertTrue(cid1 != cid3)

    def test_shareable(self):
        chan = channels.create()

        obj = channels.create()
        channels.send(chan, obj, blocking=False)
        got = channels.recv(chan)

        self.assertEqual(got, obj)
        self.assertIs(type(got), type(obj))
        
        


class ChannelTests(TestBase):

    def test_create_cid(self):
        cid = channels.create()
        self.assertIsInstance(cid, channels.ChannelID)

    def test_sequential_ids(self):
        before = channels.list_all()
        id1 = channels.create()
        id2 = channels.create()
        id3 = channels.create()
        after = channels.list_all()

        self.assertEqual(id2, int(id1) + 1)
        self.assertEqual(id3, int(id2) + 1)
        self.assertEqual(set(after) - set(before), {id1, id2, id3})

    def test_ids_global(self):
        id1 = interpreters.create()
        out = _run_output(id1, dedent())
        cid1 = int(out.strip())

        id2 = interpreters.create()
        out = _run_output(id2, dedent())
        cid2 = int(out.strip())

        self.assertEqual(cid2, int(cid1) + 1)

    def test_channel_list_interpreters_none(self):
        
        
        cid = channels.create()
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(send_interps, [])
        self.assertEqual(recv_interps, [])

    def test_channel_list_interpreters_basic(self):
        
        interp0 = interpreters.get_main()
        cid = channels.create()
        channels.send(cid, "send", blocking=False)
        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(send_interps, [interp0])
        self.assertEqual(recv_interps, [])

        interp1 = interpreters.create()
        _run_output(interp1, dedent(f))
        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(send_interps, [interp0])
        self.assertEqual(recv_interps, [interp1])

    def test_channel_list_interpreters_multiple(self):
        
        interp0 = interpreters.get_main()
        interp1 = interpreters.create()
        interp2 = interpreters.create()
        interp3 = interpreters.create()
        cid = channels.create()

        channels.send(cid, "send", blocking=False)
        _run_output(interp1, dedent(f))
        _run_output(interp2, dedent(f))
        _run_output(interp3, dedent(f))
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(set(send_interps), {interp0, interp1})
        self.assertEqual(set(recv_interps), {interp2, interp3})

    def test_channel_list_interpreters_destroyed(self):
        
        interp0 = interpreters.get_main()
        interp1 = interpreters.create()
        cid = channels.create()
        channels.send(cid, "send", blocking=False)
        _run_output(interp1, dedent(f))
        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(send_interps, [interp0])
        self.assertEqual(recv_interps, [interp1])

        interpreters.destroy(interp1)
        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(send_interps, [interp0])
        self.assertEqual(recv_interps, [])

    def test_channel_list_interpreters_released(self):
        
        
        
        interp0 = interpreters.get_main()
        interp1 = interpreters.create()
        interp2 = interpreters.create()
        cid = channels.create()
        channels.send(cid, "data", blocking=False)
        _run_output(interp1, dedent(f))
        channels.send(cid, "data", blocking=False)
        _run_output(interp2, dedent(f))
        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(len(send_interps), 1)
        self.assertEqual(len(recv_interps), 2)

        
        channels.release(cid, send=True)
        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(len(send_interps), 0)
        self.assertEqual(len(recv_interps), 2)

        
        _run_output(interp2, dedent(f))
        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(len(send_interps), 0)
        self.assertEqual(recv_interps, [interp1])

    def test_channel_list_interpreters_closed(self):
        
        interp0 = interpreters.get_main()
        interp1 = interpreters.create()
        cid = channels.create()
        
        channels.send(cid, "send", blocking=False)

        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(len(send_interps), 1)
        self.assertEqual(len(recv_interps), 0)

        
        channels.close(cid, force=True)
        
        with self.assertRaises(channels.ChannelClosedError):
            channels.list_interpreters(cid, send=True)
        with self.assertRaises(channels.ChannelClosedError):
            channels.list_interpreters(cid, send=False)

    def test_channel_list_interpreters_closed_send_end(self):
        
        interp0 = interpreters.get_main()
        interp1 = interpreters.create()
        cid = channels.create()
        
        channels.send(cid, "send", blocking=False)

        
        send_interps = channels.list_interpreters(cid, send=True)
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(len(send_interps), 1)
        self.assertEqual(len(recv_interps), 0)

        
        channels.close(cid, send=True)
        
        with self.assertRaises(channels.ChannelClosedError):
            channels.list_interpreters(cid, send=True)
        
        recv_interps = channels.list_interpreters(cid, send=False)
        self.assertEqual(len(recv_interps), 0)

        
        _run_output(interp1, dedent(f))
        return
        
        with self.assertRaises(channels.ChannelClosedError):
            channels.list_interpreters(cid, send=True)
        with self.assertRaises(channels.ChannelClosedError):
            channels.list_interpreters(cid, send=False)

    def test_allowed_types(self):
        cid = channels.create()
        objects = [
            None,
            'spam',
            b'spam',
            42,
        ]
        for obj in objects:
            with self.subTest(obj):
                channels.send(cid, obj, blocking=False)
                got = channels.recv(cid)

                self.assertEqual(got, obj)
                self.assertIs(type(got), type(obj))
                
                
                

    def test_run_string_arg_unresolved(self):
        cid = channels.create()
        interp = interpreters.create()

        out = _run_output(interp, dedent(),
            dict(cid=cid.send))
        obj = channels.recv(cid)

        self.assertEqual(obj, b'spam')
        self.assertEqual(out.strip(), 'send')

    
    
    
    @unittest.skip('disabled until high-level channels exist')
    def test_run_string_arg_resolved(self):
        cid = channels.create()
        cid = channels._channel_id(cid, _resolve=True)
        interp = interpreters.create()

        out = _run_output(interp, dedent(),
            dict(chan=cid.send))
        obj = channels.recv(cid)

        self.assertEqual(obj, b'spam')
        self.assertEqual(out.strip(), 'send')

    
    

    def test_send_recv_main(self):
        cid = channels.create()
        orig = b'spam'
        channels.send(cid, orig, blocking=False)
        obj = channels.recv(cid)

        self.assertEqual(obj, orig)
        self.assertIsNot(obj, orig)

    def test_send_recv_same_interpreter(self):
        id1 = interpreters.create()
        out = _run_output(id1, dedent())

    def test_send_recv_different_interpreters(self):
        cid = channels.create()
        id1 = interpreters.create()
        out = _run_output(id1, dedent(f))
        obj = channels.recv(cid)

        self.assertEqual(obj, b'spam')

    def test_send_recv_different_threads(self):
        cid = channels.create()

        def f():
            obj = recv_wait(cid)
            channels.send(cid, obj)
        t = threading.Thread(target=f)
        t.start()

        channels.send(cid, b'spam')
        obj = recv_wait(cid)
        t.join()

        self.assertEqual(obj, b'spam')

    def test_send_recv_different_interpreters_and_threads(self):
        cid = channels.create()
        id1 = interpreters.create()
        out = None

        def f():
            nonlocal out
            out = _run_output(id1, dedent(f))
        t = threading.Thread(target=f)
        t.start()

        channels.send(cid, b'spam')
        obj = recv_wait(cid)
        t.join()

        self.assertEqual(obj, b'eggs')

    def test_send_not_found(self):
        with self.assertRaises(channels.ChannelNotFoundError):
            channels.send(10, b'spam')

    def test_recv_not_found(self):
        with self.assertRaises(channels.ChannelNotFoundError):
            channels.recv(10)

    def test_recv_empty(self):
        cid = channels.create()
        with self.assertRaises(channels.ChannelEmptyError):
            channels.recv(cid)

    def test_recv_default(self):
        default = object()
        cid = channels.create()
        obj1 = channels.recv(cid, default)
        channels.send(cid, None, blocking=False)
        channels.send(cid, 1, blocking=False)
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'eggs', blocking=False)
        obj2 = channels.recv(cid, default)
        obj3 = channels.recv(cid, default)
        obj4 = channels.recv(cid)
        obj5 = channels.recv(cid, default)
        obj6 = channels.recv(cid, default)

        self.assertIs(obj1, default)
        self.assertIs(obj2, None)
        self.assertEqual(obj3, 1)
        self.assertEqual(obj4, b'spam')
        self.assertEqual(obj5, b'eggs')
        self.assertIs(obj6, default)

    def test_recv_sending_interp_destroyed(self):
        with self.subTest('closed'):
            cid1 = channels.create()
            interp = interpreters.create()
            interpreters.run_string(interp, dedent(f))
            interpreters.destroy(interp)

            with self.assertRaisesRegex(RuntimeError,
                                        f'channel {cid1} is closed'):
                channels.recv(cid1)
            del cid1
        with self.subTest('still open'):
            cid2 = channels.create()
            interp = interpreters.create()
            interpreters.run_string(interp, dedent(f))
            channels.send(cid2, b'eggs', blocking=False)
            interpreters.destroy(interp)

            channels.recv(cid2)
            with self.assertRaisesRegex(RuntimeError,
                                        f'channel {cid2} is empty'):
                channels.recv(cid2)
            del cid2

    
    

    def test_send_buffer(self):
        buf = bytearray(b'spamspamspam')
        cid = channels.create()
        channels.send_buffer(cid, buf, blocking=False)
        obj = channels.recv(cid)

        self.assertIsNot(obj, buf)
        self.assertIsInstance(obj, memoryview)
        self.assertEqual(obj, buf)

        buf[4:8] = b'eggs'
        self.assertEqual(obj, buf)
        obj[4:8] = b'ham.'
        self.assertEqual(obj, buf)

    
    

    def build_send_waiter(self, obj, *, buffer=False):
        

        if buffer:
            send = channels.send_buffer
        else:
            send = channels.send

        cid = channels.create()
        try:
            started = time.monotonic()
            send(cid, obj, blocking=False)
            stopped = time.monotonic()
            channels.recv(cid)
        finally:
            channels.destroy(cid)
        delay = stopped - started  
        delay *= 3

        def wait():
            time.sleep(delay)
        return wait

    def test_send_blocking_waiting(self):
        received = None
        obj = b'spam'
        wait = self.build_send_waiter(obj)
        cid = channels.create()
        def f():
            nonlocal received
            wait()
            received = recv_wait(cid)
        t = threading.Thread(target=f)
        t.start()
        channels.send(cid, obj, blocking=True)
        t.join()

        self.assertEqual(received, obj)

    def test_send_buffer_blocking_waiting(self):
        received = None
        obj = bytearray(b'spam')
        wait = self.build_send_waiter(obj, buffer=True)
        cid = channels.create()
        def f():
            nonlocal received
            wait()
            received = recv_wait(cid)
        t = threading.Thread(target=f)
        t.start()
        channels.send_buffer(cid, obj, blocking=True)
        t.join()

        self.assertEqual(received, obj)

    def test_send_blocking_no_wait(self):
        received = None
        obj = b'spam'
        cid = channels.create()
        def f():
            nonlocal received
            received = recv_wait(cid)
        t = threading.Thread(target=f)
        t.start()
        channels.send(cid, obj, blocking=True)
        t.join()

        self.assertEqual(received, obj)

    def test_send_buffer_blocking_no_wait(self):
        received = None
        obj = bytearray(b'spam')
        cid = channels.create()
        def f():
            nonlocal received
            received = recv_wait(cid)
        t = threading.Thread(target=f)
        t.start()
        channels.send_buffer(cid, obj, blocking=True)
        t.join()

        self.assertEqual(received, obj)

    def test_send_timeout(self):
        obj = b'spam'

        with self.subTest('non-blocking with timeout'):
            cid = channels.create()
            with self.assertRaises(ValueError):
                channels.send(cid, obj, blocking=False, timeout=0.1)

        with self.subTest('timeout hit'):
            cid = channels.create()
            with self.assertRaises(TimeoutError):
                channels.send(cid, obj, blocking=True, timeout=0.1)
            with self.assertRaises(channels.ChannelEmptyError):
                received = channels.recv(cid)
                print(repr(received))

        with self.subTest('timeout not hit'):
            cid = channels.create()
            def f():
                recv_wait(cid)
            t = threading.Thread(target=f)
            t.start()
            channels.send(cid, obj, blocking=True, timeout=10)
            t.join()

    def test_send_buffer_timeout(self):
        try:
            self._has_run_once_timeout
        except AttributeError:
            
            
            
            
            
            raise unittest.SkipTest('temporarily skipped due to refleaks')
        else:
            self._has_run_once_timeout = True

        obj = bytearray(b'spam')

        with self.subTest('non-blocking with timeout'):
            cid = channels.create()
            with self.assertRaises(ValueError):
                channels.send_buffer(cid, obj, blocking=False, timeout=0.1)

        with self.subTest('timeout hit'):
            cid = channels.create()
            with self.assertRaises(TimeoutError):
                channels.send_buffer(cid, obj, blocking=True, timeout=0.1)
            with self.assertRaises(channels.ChannelEmptyError):
                received = channels.recv(cid)
                print(repr(received))

        with self.subTest('timeout not hit'):
            cid = channels.create()
            def f():
                recv_wait(cid)
            t = threading.Thread(target=f)
            t.start()
            channels.send_buffer(cid, obj, blocking=True, timeout=10)
            t.join()

    def test_send_closed_while_waiting(self):
        obj = b'spam'
        wait = self.build_send_waiter(obj)

        with self.subTest('without timeout'):
            cid = channels.create()
            def f():
                wait()
                channels.close(cid, force=True)
            t = threading.Thread(target=f)
            t.start()
            with self.assertRaises(channels.ChannelClosedError):
                channels.send(cid, obj, blocking=True)
            t.join()

        with self.subTest('with timeout'):
            cid = channels.create()
            def f():
                wait()
                channels.close(cid, force=True)
            t = threading.Thread(target=f)
            t.start()
            with self.assertRaises(channels.ChannelClosedError):
                channels.send(cid, obj, blocking=True, timeout=30)
            t.join()

    def test_send_buffer_closed_while_waiting(self):
        try:
            self._has_run_once_closed
        except AttributeError:
            
            
            
            
            
            raise unittest.SkipTest('temporarily skipped due to refleaks')
        else:
            self._has_run_once_closed = True

        obj = bytearray(b'spam')
        wait = self.build_send_waiter(obj, buffer=True)

        with self.subTest('without timeout'):
            cid = channels.create()
            def f():
                wait()
                channels.close(cid, force=True)
            t = threading.Thread(target=f)
            t.start()
            with self.assertRaises(channels.ChannelClosedError):
                channels.send_buffer(cid, obj, blocking=True)
            t.join()

        with self.subTest('with timeout'):
            cid = channels.create()
            def f():
                wait()
                channels.close(cid, force=True)
            t = threading.Thread(target=f)
            t.start()
            with self.assertRaises(channels.ChannelClosedError):
                channels.send_buffer(cid, obj, blocking=True, timeout=30)
            t.join()

    
    

    def test_close_single_user(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.recv(cid)
        channels.close(cid)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_close_multiple_users(self):
        cid = channels.create()
        id1 = interpreters.create()
        id2 = interpreters.create()
        interpreters.run_string(id1, dedent(f))
        interpreters.run_string(id2, dedent(f))
        channels.close(cid)

        excsnap = interpreters.run_string(id1, dedent(f))
        self.assertEqual(excsnap.type.__name__, 'ChannelClosedError')

        excsnap = interpreters.run_string(id2, dedent(f))
        self.assertEqual(excsnap.type.__name__, 'ChannelClosedError')

    def test_close_multiple_times(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.recv(cid)
        channels.close(cid)

        with self.assertRaises(channels.ChannelClosedError):
            channels.close(cid)

    def test_close_empty(self):
        tests = [
            (False, False),
            (True, False),
            (False, True),
            (True, True),
            ]
        for send, recv in tests:
            with self.subTest((send, recv)):
                cid = channels.create()
                channels.send(cid, b'spam', blocking=False)
                channels.recv(cid)
                channels.close(cid, send=send, recv=recv)

                with self.assertRaises(channels.ChannelClosedError):
                    channels.send(cid, b'eggs')
                with self.assertRaises(channels.ChannelClosedError):
                    channels.recv(cid)

    def test_close_defaults_with_unused_items(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'ham', blocking=False)

        with self.assertRaises(channels.ChannelNotEmptyError):
            channels.close(cid)
        channels.recv(cid)
        channels.send(cid, b'eggs', blocking=False)

    def test_close_recv_with_unused_items_unforced(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'ham', blocking=False)

        with self.assertRaises(channels.ChannelNotEmptyError):
            channels.close(cid, recv=True)
        channels.recv(cid)
        channels.send(cid, b'eggs', blocking=False)
        channels.recv(cid)
        channels.recv(cid)
        channels.close(cid, recv=True)

    def test_close_send_with_unused_items_unforced(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'ham', blocking=False)
        channels.close(cid, send=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        channels.recv(cid)
        channels.recv(cid)
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_close_both_with_unused_items_unforced(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'ham', blocking=False)

        with self.assertRaises(channels.ChannelNotEmptyError):
            channels.close(cid, recv=True, send=True)
        channels.recv(cid)
        channels.send(cid, b'eggs', blocking=False)
        channels.recv(cid)
        channels.recv(cid)
        channels.close(cid, recv=True)

    def test_close_recv_with_unused_items_forced(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'ham', blocking=False)
        channels.close(cid, recv=True, force=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_close_send_with_unused_items_forced(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'ham', blocking=False)
        channels.close(cid, send=True, force=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_close_both_with_unused_items_forced(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'ham', blocking=False)
        channels.close(cid, send=True, recv=True, force=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_close_never_used(self):
        cid = channels.create()
        channels.close(cid)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'spam')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_close_by_unassociated_interp(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        interp = interpreters.create()
        interpreters.run_string(interp, dedent(f))
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)
        with self.assertRaises(channels.ChannelClosedError):
            channels.close(cid)

    def test_close_used_multiple_times_by_single_user(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'spam', blocking=False)
        channels.recv(cid)
        channels.close(cid, force=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_channel_list_interpreters_invalid_channel(self):
        cid = channels.create()
        
        with self.assertRaises(channels.ChannelNotFoundError):
            channels.list_interpreters(1000, send=True)

        channels.close(cid)
        
        with self.assertRaises(channels.ChannelClosedError):
            channels.list_interpreters(cid, send=True)

    def test_channel_list_interpreters_invalid_args(self):
        
        cid = channels.create()
        with self.assertRaises(TypeError):
            channels.list_interpreters(cid)


class ChannelReleaseTests(TestBase):

    

    

    

    

    def test_single_user(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.recv(cid)
        channels.release(cid, send=True, recv=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_multiple_users(self):
        cid = channels.create()
        id1 = interpreters.create()
        id2 = interpreters.create()
        interpreters.run_string(id1, dedent(f))
        out = _run_output(id2, dedent(f))
        interpreters.run_string(id1, dedent(f))

        self.assertEqual(out.strip(), "b'spam'")

    def test_no_kwargs(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.recv(cid)
        channels.release(cid)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_multiple_times(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.recv(cid)
        channels.release(cid, send=True, recv=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.release(cid, send=True, recv=True)

    def test_with_unused_items(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'ham', blocking=False)
        channels.release(cid, send=True, recv=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_never_used(self):
        cid = channels.create()
        channels.release(cid)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'spam')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_by_unassociated_interp(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        interp = interpreters.create()
        interpreters.run_string(interp, dedent(f))
        obj = channels.recv(cid)
        channels.release(cid)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        self.assertEqual(obj, b'spam')

    def test_close_if_unassociated(self):
        
        cid = channels.create()
        interp = interpreters.create()
        interpreters.run_string(interp, dedent(f))

        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)

    def test_partially(self):
        
        cid = channels.create()
        channels.send(cid, None, blocking=False)
        channels.recv(cid)
        channels.send(cid, b'spam', blocking=False)
        channels.release(cid, send=True)
        obj = channels.recv(cid)

        self.assertEqual(obj, b'spam')

    def test_used_multiple_times_by_single_user(self):
        cid = channels.create()
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'spam', blocking=False)
        channels.send(cid, b'spam', blocking=False)
        channels.recv(cid)
        channels.release(cid, send=True, recv=True)

        with self.assertRaises(channels.ChannelClosedError):
            channels.send(cid, b'eggs')
        with self.assertRaises(channels.ChannelClosedError):
            channels.recv(cid)


class ChannelCloseFixture(namedtuple('ChannelCloseFixture',
                                     'end interp other extra creator')):

    
    
    QUICK = False

    def __new__(cls, end, interp, other, extra, creator):
        assert end in ('send', 'recv')
        if cls.QUICK:
            known = {}
        else:
            interp = Interpreter.from_raw(interp)
            other = Interpreter.from_raw(other)
            extra = Interpreter.from_raw(extra)
            known = {
                interp.name: interp,
                other.name: other,
                extra.name: extra,
                }
        if not creator:
            creator = 'same'
        self = super().__new__(cls, end, interp, other, extra, creator)
        self._prepped = set()
        self._state = ChannelState()
        self._known = known
        return self

    @property
    def state(self):
        return self._state

    @property
    def cid(self):
        try:
            return self._cid
        except AttributeError:
            creator = self._get_interpreter(self.creator)
            self._cid = self._new_channel(creator)
            return self._cid

    def get_interpreter(self, interp):
        interp = self._get_interpreter(interp)
        self._prep_interpreter(interp)
        return interp

    def expect_closed_error(self, end=None):
        if end is None:
            end = self.end
        if end == 'recv' and self.state.closed == 'send':
            return False
        return bool(self.state.closed)

    def prep_interpreter(self, interp):
        self._prep_interpreter(interp)

    def record_action(self, action, result):
        self._state = result

    def clean_up(self):
        clean_up_interpreters()
        clean_up_channels()

    

    def _new_channel(self, creator):
        if creator.name == 'main':
            return channels.create()
        else:
            ch = channels.create()
            run_interp(creator.id, f)
            self._cid = channels.recv(ch)
        return self._cid

    def _get_interpreter(self, interp):
        if interp in ('same', 'interp'):
            return self.interp
        elif interp == 'other':
            return self.other
        elif interp == 'extra':
            return self.extra
        else:
            name = interp
            try:
                interp = self._known[name]
            except KeyError:
                interp = self._known[name] = Interpreter(name)
            return interp

    def _prep_interpreter(self, interp):
        if interp.id in self._prepped:
            return
        self._prepped.add(interp.id)
        if interp.name == 'main':
            return
        run_interp(interp.id, f)


@unittest.skip('these tests take several hours to run')
class ExhaustiveChannelTests(TestBase):

    

    

    

    def iter_action_sets(self):
        
        
        
        

        
        yield []

        
        for closeactions in self._iter_close_action_sets('same', 'other'):
            yield closeactions
            for postactions in self._iter_post_close_action_sets():
                yield closeactions + postactions
        for closeactions in self._iter_close_action_sets('other', 'extra'):
            yield closeactions
            for postactions in self._iter_post_close_action_sets():
                yield closeactions + postactions

        
        for useactions in self._iter_use_action_sets('same', 'other'):
            yield useactions
            for closeactions in self._iter_close_action_sets('same', 'other'):
                actions = useactions + closeactions
                yield actions
                for postactions in self._iter_post_close_action_sets():
                    yield actions + postactions
            for closeactions in self._iter_close_action_sets('other', 'extra'):
                actions = useactions + closeactions
                yield actions
                for postactions in self._iter_post_close_action_sets():
                    yield actions + postactions
        for useactions in self._iter_use_action_sets('other', 'extra'):
            yield useactions
            for closeactions in self._iter_close_action_sets('same', 'other'):
                actions = useactions + closeactions
                yield actions
                for postactions in self._iter_post_close_action_sets():
                    yield actions + postactions
            for closeactions in self._iter_close_action_sets('other', 'extra'):
                actions = useactions + closeactions
                yield actions
                for postactions in self._iter_post_close_action_sets():
                    yield actions + postactions

    def _iter_use_action_sets(self, interp1, interp2):
        interps = (interp1, interp2)

        
        yield [
            ChannelAction('use', 'recv', interp1),
            ]
        yield [
            ChannelAction('use', 'recv', interp2),
            ]
        yield [
            ChannelAction('use', 'recv', interp1),
            ChannelAction('use', 'recv', interp2),
            ]

        
        yield [
            ChannelAction('use', 'send', interp1),
            ]
        yield [
            ChannelAction('use', 'send', interp2),
            ]
        yield [
            ChannelAction('use', 'send', interp1),
            ChannelAction('use', 'send', interp2),
            ]

        
        for interp1 in interps:
            for interp2 in interps:
                for interp3 in interps:
                    yield [
                        ChannelAction('use', 'send', interp1),
                        ChannelAction('use', 'send', interp2),
                        ChannelAction('use', 'recv', interp3),
                        ]

        
        for interp1 in interps:
            for interp2 in interps:
                for interp3 in interps:
                    for interp4 in interps:
                        yield [
                            ChannelAction('use', 'send', interp1),
                            ChannelAction('use', 'send', interp2),
                            ChannelAction('use', 'recv', interp3),
                            ChannelAction('use', 'recv', interp4),
                            ]

    def _iter_close_action_sets(self, interp1, interp2):
        ends = ('recv', 'send')
        interps = (interp1, interp2)
        for force in (True, False):
            op = 'force-close' if force else 'close'
            for interp in interps:
                for end in ends:
                    yield [
                        ChannelAction(op, end, interp),
                        ]
        for recvop in ('close', 'force-close'):
            for sendop in ('close', 'force-close'):
                for recv in interps:
                    for send in interps:
                        yield [
                            ChannelAction(recvop, 'recv', recv),
                            ChannelAction(sendop, 'send', send),
                            ]

    def _iter_post_close_action_sets(self):
        for interp in ('same', 'extra', 'other'):
            yield [
                ChannelAction('use', 'recv', interp),
                ]
            yield [
                ChannelAction('use', 'send', interp),
                ]

    def run_actions(self, fix, actions):
        for action in actions:
            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()
            run_interp(interp.id, f)
            result = ChannelState(
                pending=int.from_bytes(channels.recv(_cid), 'little'),
                closed=bool(channels.recv(_cid)),
                )
            fix.record_action(action, result)

    def iter_fixtures(self):
        
        interpreters = [
            ('main', 'interp', 'extra'),
            ('interp', 'main', 'extra'),
            ('interp1', 'interp2', 'extra'),
            ('interp1', 'interp2', 'main'),
        ]
        for interp, other, extra in interpreters:
            for creator in ('same', 'other', 'creator'):
                for end in ('send', 'recv'):
                    yield ChannelCloseFixture(end, interp, other, extra, creator)

    def _close(self, fix, *, force):
        op = 'force-close' if force else 'close'
        close = ChannelAction(op, fix.end, 'same')
        if not fix.expect_closed_error():
            self.run_action(fix, close, hideclosed=False)
        else:
            with self.assertRaises(channels.ChannelClosedError):
                self.run_action(fix, close, hideclosed=False)

    def _assert_closed_in_interp(self, fix, interp=None):
        if interp is None or interp.name == 'main':
            with self.assertRaises(channels.ChannelClosedError):
                channels.recv(fix.cid)
            with self.assertRaises(channels.ChannelClosedError):
                channels.send(fix.cid, b'spam')
            with self.assertRaises(channels.ChannelClosedError):
                channels.close(fix.cid)
            with self.assertRaises(channels.ChannelClosedError):
                channels.close(fix.cid, force=True)
        else:
            run_interp(interp.id, )
            run_interp(interp.id, )
            run_interp(interp.id, )
            run_interp(interp.id, )

    def _assert_closed(self, fix):
        self.assertTrue(fix.state.closed)

        for _ in range(fix.state.pending):
            channels.recv(fix.cid)
        self._assert_closed_in_interp(fix)

        for interp in ('same', 'other'):
            interp = fix.get_interpreter(interp)
            if interp.name == 'main':
                continue
            self._assert_closed_in_interp(fix, interp)

        interp = fix.get_interpreter('fresh')
        self._assert_closed_in_interp(fix, interp)

    def _iter_close_tests(self, verbose=False):
        i = 0
        for actions in self.iter_action_sets():
            print()
            for fix in self.iter_fixtures():
                i += 1
                if i > 1000:
                    return
                if verbose:
                    if (i - 1) % 6 == 0:
                        print()
                    print(i, fix, '({} actions)'.format(len(actions)))
                else:
                    if (i - 1) % 6 == 0:
                        print(' ', end='')
                    print('.', end=''); sys.stdout.flush()
                yield i, fix, actions
            if verbose:
                print('---')
        print()

    
    def _skim_close_tests(self):
        ChannelCloseFixture.QUICK = True
        for i, fix, actions in self._iter_close_tests():
            pass

    def test_close(self):
        for i, fix, actions in self._iter_close_tests():
            with self.subTest('{} {}  {}'.format(i, fix, actions)):
                fix.prep_interpreter(fix.interp)
                self.run_actions(fix, actions)

                self._close(fix, force=False)

                self._assert_closed(fix)
            
            fix.clean_up()

    def test_force_close(self):
        for i, fix, actions in self._iter_close_tests():
            with self.subTest('{} {}  {}'.format(i, fix, actions)):
                fix.prep_interpreter(fix.interp)
                self.run_actions(fix, actions)

                self._close(fix, force=True)

                self._assert_closed(fix)
            
            fix.clean_up()


if __name__ == '__main__':
    unittest.main()