import contextlib import json import os import os.path import sys import threading from textwrap import dedent import unittest import time from test import support from test.support import import_helper from test.support import threading_helper from test.support import os_helper _interpreters = import_helper.import_module('_xxsubinterpreters') _channels = import_helper.import_module('_xxinterpchannels') from test.support import interpreters def _captured_script(script): r, w = os.pipe() indented = script.replace('\n', '\n ') wrapped = dedent(f""" import contextlib with open({w}, 'w', encoding='utf-8') as spipe: with contextlib.redirect_stdout(spipe): {indented} """) return wrapped, open(r, encoding='utf-8') def clean_up_interpreters(): for interp in interpreters.list_all(): if interp.id == 0: # main continue try: interp.close() except RuntimeError: pass # already destroyed def _run_output(interp, request, channels=None): script, rpipe = _captured_script(request) with rpipe: interp.run(script, channels=channels) return rpipe.read() @contextlib.contextmanager def _running(interp): r, w = os.pipe() def run(): interp.run(dedent(f""" # wait for "signal" with open({r}) as rpipe: rpipe.read() """)) t = threading.Thread(target=run) t.start() yield with open(w, 'w') as spipe: spipe.write('done') t.join() class TestBase(unittest.TestCase): def pipe(self): def ensure_closed(fd): try: os.close(fd) except OSError: pass r, w = os.pipe() self.addCleanup(lambda: ensure_closed(r)) self.addCleanup(lambda: ensure_closed(w)) return r, w def tearDown(self): clean_up_interpreters() class CreateTests(TestBase): def test_in_main(self): interp = interpreters.create() self.assertIsInstance(interp, interpreters.Interpreter) self.assertIn(interp, interpreters.list_all()) def test_in_thread(self): lock = threading.Lock() interp = None def f(): nonlocal interp interp = interpreters.create() lock.acquire() lock.release() t = threading.Thread(target=f) with lock: t.start() t.join() self.assertIn(interp, interpreters.list_all()) def test_in_subinterpreter(self): main, = interpreters.list_all() interp = interpreters.create() out = _run_output(interp, dedent(""" from test.support import interpreters interp = interpreters.create() print(interp.id) """)) interp2 = interpreters.Interpreter(int(out)) self.assertEqual(interpreters.list_all(), [main, interp, interp2]) def test_after_destroy_all(self): before = set(interpreters.list_all()) # Create 3 subinterpreters. interp_lst = [] for _ in range(3): interps = interpreters.create() interp_lst.append(interps) # Now destroy them. for interp in interp_lst: interp.close() # Finally, create another. interp = interpreters.create() self.assertEqual(set(interpreters.list_all()), before | {interp}) def test_after_destroy_some(self): before = set(interpreters.list_all()) # Create 3 subinterpreters. interp1 = interpreters.create() interp2 = interpreters.create() interp3 = interpreters.create() # Now destroy 2 of them. interp1.close() interp2.close() # Finally, create another. interp = interpreters.create() self.assertEqual(set(interpreters.list_all()), before | {interp3, interp}) class GetCurrentTests(TestBase): def test_main(self): main = interpreters.get_main() current = interpreters.get_current() self.assertEqual(current, main) def test_subinterpreter(self): main = _interpreters.get_main() interp = interpreters.create() out = _run_output(interp, dedent(""" from test.support import interpreters cur = interpreters.get_current() print(cur.id) """)) current = interpreters.Interpreter(int(out)) self.assertNotEqual(current, main) class ListAllTests(TestBase): def test_initial(self): interps = interpreters.list_all() self.assertEqual(1, len(interps)) def test_after_creating(self): main = interpreters.get_current() first = interpreters.create() second = interpreters.create() ids = [] for interp in interpreters.list_all(): ids.append(interp.id) self.assertEqual(ids, [main.id, first.id, second.id]) def test_after_destroying(self): main = interpreters.get_current() first = interpreters.create() second = interpreters.create() first.close() ids = [] for interp in interpreters.list_all(): ids.append(interp.id) self.assertEqual(ids, [main.id, second.id]) class TestInterpreterAttrs(TestBase): def test_id_type(self): main = interpreters.get_main() current = interpreters.get_current() interp = interpreters.create() self.assertIsInstance(main.id, _interpreters.InterpreterID) self.assertIsInstance(current.id, _interpreters.InterpreterID) self.assertIsInstance(interp.id, _interpreters.InterpreterID) def test_main_id(self): main = interpreters.get_main() self.assertEqual(main.id, 0) def test_custom_id(self): interp = interpreters.Interpreter(1) self.assertEqual(interp.id, 1) with self.assertRaises(TypeError): interpreters.Interpreter('1') def test_id_readonly(self): interp = interpreters.Interpreter(1) with self.assertRaises(AttributeError): interp.id = 2 @unittest.skip('not ready yet (see bpo-32604)') def test_main_isolated(self): main = interpreters.get_main() self.assertFalse(main.isolated) @unittest.skip('not ready yet (see bpo-32604)') def test_subinterpreter_isolated_default(self): interp = interpreters.create() self.assertFalse(interp.isolated) def test_subinterpreter_isolated_explicit(self): interp1 = interpreters.create(isolated=True) interp2 = interpreters.create(isolated=False) self.assertTrue(interp1.isolated) self.assertFalse(interp2.isolated) @unittest.skip('not ready yet (see bpo-32604)') def test_custom_isolated_default(self): interp = interpreters.Interpreter(1) self.assertFalse(interp.isolated) def test_custom_isolated_explicit(self): interp1 = interpreters.Interpreter(1, isolated=True) interp2 = interpreters.Interpreter(1, isolated=False) self.assertTrue(interp1.isolated) self.assertFalse(interp2.isolated) def test_isolated_readonly(self): interp = interpreters.Interpreter(1) with self.assertRaises(AttributeError): interp.isolated = True def test_equality(self): interp1 = interpreters.create() interp2 = interpreters.create() self.assertEqual(interp1, interp1) self.assertNotEqual(interp1, interp2) class TestInterpreterIsRunning(TestBase): def test_main(self): main = interpreters.get_main() self.assertTrue(main.is_running()) @unittest.skip('Fails on FreeBSD') def test_subinterpreter(self): interp = interpreters.create() self.assertFalse(interp.is_running()) with _running(interp): self.assertTrue(interp.is_running()) self.assertFalse(interp.is_running()) def test_finished(self): r, w = self.pipe() interp = interpreters.create() interp.run(f"""if True: import os os.write({w}, b'x') """) self.assertFalse(interp.is_running()) self.assertEqual(os.read(r, 1), b'x') def test_from_subinterpreter(self): interp = interpreters.create() out = _run_output(interp, dedent(f""" import _xxsubinterpreters as _interpreters if _interpreters.is_running({interp.id}): print(True) else: print(False) """)) self.assertEqual(out.strip(), 'True') def test_already_destroyed(self): interp = interpreters.create() interp.close() with self.assertRaises(RuntimeError): interp.is_running() def test_does_not_exist(self): interp = interpreters.Interpreter(1_000_000) with self.assertRaises(RuntimeError): interp.is_running() def test_bad_id(self): interp = interpreters.Interpreter(-1) with self.assertRaises(ValueError): interp.is_running() def test_with_only_background_threads(self): r_interp, w_interp = self.pipe() r_thread, w_thread = self.pipe() DONE = b'D' FINISHED = b'F' interp = interpreters.create() interp.run(f"""if True: import os import threading def task(): v = os.read({r_thread}, 1) assert v == {DONE!r} os.write({w_interp}, {FINISHED!r}) t = threading.Thread(target=task) t.start() """) self.assertFalse(interp.is_running()) os.write(w_thread, DONE) interp.run('t.join()') self.assertEqual(os.read(r_interp, 1), FINISHED) class TestInterpreterClose(TestBase): def test_basic(self): main = interpreters.get_main() interp1 = interpreters.create() interp2 = interpreters.create() interp3 = interpreters.create() self.assertEqual(set(interpreters.list_all()), {main, interp1, interp2, interp3}) interp2.close() self.assertEqual(set(interpreters.list_all()), {main, interp1, interp3}) def test_all(self): before = set(interpreters.list_all()) interps = set() for _ in range(3): interp = interpreters.create() interps.add(interp) self.assertEqual(set(interpreters.list_all()), before | interps) for interp in interps: interp.close() self.assertEqual(set(interpreters.list_all()), before) def test_main(self): main, = interpreters.list_all() with self.assertRaises(RuntimeError): main.close() def f(): with self.assertRaises(RuntimeError): main.close() t = threading.Thread(target=f) t.start() t.join() def test_already_destroyed(self): interp = interpreters.create() interp.close() with self.assertRaises(RuntimeError): interp.close() def test_does_not_exist(self): interp = interpreters.Interpreter(1_000_000) with self.assertRaises(RuntimeError): interp.close() def test_bad_id(self): interp = interpreters.Interpreter(-1) with self.assertRaises(ValueError): interp.close() def test_from_current(self): main, = interpreters.list_all() interp = interpreters.create() out = _run_output(interp, dedent(f""" from test.support import interpreters interp = interpreters.Interpreter({int(interp.id)}) try: interp.close() except RuntimeError: print('failed') """)) self.assertEqual(out.strip(), 'failed') self.assertEqual(set(interpreters.list_all()), {main, interp}) def test_from_sibling(self): main, = interpreters.list_all() interp1 = interpreters.create() interp2 = interpreters.create() self.assertEqual(set(interpreters.list_all()), {main, interp1, interp2}) interp1.run(dedent(f""" from test.support import interpreters interp2 = interpreters.Interpreter(int({interp2.id})) interp2.close() interp3 = interpreters.create() interp3.close() """)) self.assertEqual(set(interpreters.list_all()), {main, interp1}) def test_from_other_thread(self): interp = interpreters.create() def f(): interp.close() t = threading.Thread(target=f) t.start() t.join() @unittest.skip('Fails on FreeBSD') def test_still_running(self): main, = interpreters.list_all() interp = interpreters.create() with _running(interp): with self.assertRaises(RuntimeError): interp.close() self.assertTrue(interp.is_running()) def test_subthreads_still_running(self): r_interp, w_interp = self.pipe() r_thread, w_thread = self.pipe() FINISHED = b'F' interp = interpreters.create() interp.run(f"""if True: import os import threading import time done = False def notify_fini(): global done done = True t.join() threading._register_atexit(notify_fini) def task(): while not done: time.sleep(0.1) os.write({w_interp}, {FINISHED!r}) t = threading.Thread(target=task) t.start() """) interp.close() self.assertEqual(os.read(r_interp, 1), FINISHED) class TestInterpreterRun(TestBase): def test_success(self): interp = interpreters.create() script, file = _captured_script('print("it worked!", end="")') with file: interp.run(script) out = file.read() self.assertEqual(out, 'it worked!') def test_failure(self): interp = interpreters.create() with self.assertRaises(interpreters.RunFailedError): interp.run('raise Exception') def test_in_thread(self): interp = interpreters.create() script, file = _captured_script('print("it worked!", end="")') with file: def f(): interp.run(script) t = threading.Thread(target=f) t.start() t.join() out = file.read() self.assertEqual(out, 'it worked!') @support.requires_fork() def test_fork(self): interp = interpreters.create() import tempfile with tempfile.NamedTemporaryFile('w+', encoding='utf-8') as file: file.write('') file.flush() expected = 'spam spam spam spam spam' script = dedent(f""" import os try: os.fork() except RuntimeError: with open('{file.name}', 'w', encoding='utf-8') as out: out.write('{expected}') """) interp.run(script) file.seek(0) content = file.read() self.assertEqual(content, expected) @unittest.skip('Fails on FreeBSD') def test_already_running(self): interp = interpreters.create() with _running(interp): with self.assertRaises(RuntimeError): interp.run('print("spam")') def test_does_not_exist(self): interp = interpreters.Interpreter(1_000_000) with self.assertRaises(RuntimeError): interp.run('print("spam")') def test_bad_id(self): interp = interpreters.Interpreter(-1) with self.assertRaises(ValueError): interp.run('print("spam")') def test_bad_script(self): interp = interpreters.create() with self.assertRaises(TypeError): interp.run(10) def test_bytes_for_script(self): interp = interpreters.create() with self.assertRaises(TypeError): interp.run(b'print("spam")') def test_with_background_threads_still_running(self): r_interp, w_interp = self.pipe() r_thread, w_thread = self.pipe() RAN = b'R' DONE = b'D' FINISHED = b'F' interp = interpreters.create() interp.run(f"""if True: import os import threading def task(): v = os.read({r_thread}, 1) assert v == {DONE!r} os.write({w_interp}, {FINISHED!r}) t = threading.Thread(target=task) t.start() os.write({w_interp}, {RAN!r}) """) interp.run(f"""if True: os.write({w_interp}, {RAN!r}) """) os.write(w_thread, DONE) interp.run('t.join()') self.assertEqual(os.read(r_interp, 1), RAN) self.assertEqual(os.read(r_interp, 1), RAN) self.assertEqual(os.read(r_interp, 1), FINISHED) # test_xxsubinterpreters covers the remaining Interpreter.run() behavior. class StressTests(TestBase): # In these tests we generally want a lot of interpreters, # but not so many that any test takes too long. @support.requires_resource('cpu') def test_create_many_sequential(self): alive = [] for _ in range(100): interp = interpreters.create() alive.append(interp) @support.requires_resource('cpu') def test_create_many_threaded(self): alive = [] def task(): interp = interpreters.create() alive.append(interp) threads = (threading.Thread(target=task) for _ in range(200)) with threading_helper.start_threads(threads): pass class StartupTests(TestBase): # We want to ensure the initial state of subinterpreters # matches expectations. _subtest_count = 0 @contextlib.contextmanager def subTest(self, *args): with super().subTest(*args) as ctx: self._subtest_count += 1 try: yield ctx finally: if self._debugged_in_subtest: if self._subtest_count == 1: # The first subtest adds a leading newline, so we # compensate here by not printing a trailing newline. print('### end subtest debug ###', end='') else: print('### end subtest debug ###') self._debugged_in_subtest = False def debug(self, msg, *, header=None): if header: self._debug(f'--- {header} ---') if msg: if msg.endswith(os.linesep): self._debug(msg[:-len(os.linesep)]) else: self._debug(msg) self._debug('') self._debug('------') else: self._debug(msg) _debugged = False _debugged_in_subtest = False def _debug(self, msg): if not self._debugged: print() self._debugged = True if self._subtest is not None: if True: if not self._debugged_in_subtest: self._debugged_in_subtest = True print('### start subtest debug ###') print(msg) else: print(msg) def create_temp_dir(self): import tempfile tmp = tempfile.mkdtemp(prefix='test_interpreters_') tmp = os.path.realpath(tmp) self.addCleanup(os_helper.rmtree, tmp) return tmp def write_script(self, *path, text): filename = os.path.join(*path) dirname = os.path.dirname(filename) if dirname: os.makedirs(dirname, exist_ok=True) with open(filename, 'w', encoding='utf-8') as outfile: outfile.write(dedent(text)) return filename @support.requires_subprocess() def run_python(self, argv, *, cwd=None): # This method is inspired by # EmbeddingTestsMixin.run_embedded_interpreter() in test_embed.py. import shlex import subprocess if isinstance(argv, str): argv = shlex.split(argv) argv = [sys.executable, *argv] try: proc = subprocess.run( argv, cwd=cwd, capture_output=True, text=True, ) except Exception as exc: self.debug(f'# cmd: {shlex.join(argv)}') if isinstance(exc, FileNotFoundError) and not exc.filename: if os.path.exists(argv[0]): exists = 'exists' else: exists = 'does not exist' self.debug(f'{argv[0]} {exists}') raise # re-raise assert proc.stderr == '' or proc.returncode != 0, proc.stderr if proc.returncode != 0 and support.verbose: self.debug(f'# python3 {shlex.join(argv[1:])} failed:') self.debug(proc.stdout, header='stdout') self.debug(proc.stderr, header='stderr') self.assertEqual(proc.returncode, 0) self.assertEqual(proc.stderr, '') return proc.stdout def test_sys_path_0(self): # The main interpreter's sys.path[0] should be used by subinterpreters. script = ''' import sys from test.support import interpreters orig = sys.path[0] interp = interpreters.create() interp.run(f"""if True: import json import sys print(json.dumps({{ 'main': {orig!r}, 'sub': sys.path[0], }}, indent=4), flush=True) """) ''' # / # pkg/ # __init__.py # __main__.py # script.py # script.py cwd = self.create_temp_dir() self.write_script(cwd, 'pkg', '__init__.py', text='') self.write_script(cwd, 'pkg', '__main__.py', text=script) self.write_script(cwd, 'pkg', 'script.py', text=script) self.write_script(cwd, 'script.py', text=script) cases = [ ('script.py', cwd), ('-m script', cwd), ('-m pkg', cwd), ('-m pkg.script', cwd), ('-c "import script"', ''), ] for argv, expected in cases: with self.subTest(f'python3 {argv}'): out = self.run_python(argv, cwd=cwd) data = json.loads(out) sp0_main, sp0_sub = data['main'], data['sub'] self.assertEqual(sp0_sub, sp0_main) self.assertEqual(sp0_sub, expected) # XXX Also check them all with the -P cmdline flag? class FinalizationTests(TestBase): def test_gh_109793(self): import subprocess argv = [sys.executable, '-c', '''if True: import _xxsubinterpreters as _interpreters interpid = _interpreters.create() raise Exception '''] proc = subprocess.run(argv, capture_output=True, text=True) self.assertIn('Traceback', proc.stderr) if proc.returncode == 0 and support.verbose: print() print("--- cmd unexpected succeeded ---") print(f"stdout:\n{proc.stdout}") print(f"stderr:\n{proc.stderr}") print("------") self.assertEqual(proc.returncode, 1) class TestIsShareable(TestBase): def test_default_shareables(self): shareables = [ # singletons None, # builtin objects b'spam', 'spam', 10, -10, True, False, 100.0, (), (1, ('spam', 'eggs'), True), ] for obj in shareables: with self.subTest(obj): shareable = interpreters.is_shareable(obj) self.assertTrue(shareable) def test_not_shareable(self): class Cheese: def __init__(self, name): self.name = name def __str__(self): return self.name class SubBytes(bytes): """A subclass of a shareable type.""" not_shareables = [ # singletons NotImplemented, ..., # builtin types and objects type, object, object(), Exception(), # user-defined types and objects Cheese, Cheese('Wensleydale'), SubBytes(b'spam'), ] for obj in not_shareables: with self.subTest(repr(obj)): self.assertFalse( interpreters.is_shareable(obj)) class TestChannels(TestBase): def test_create(self): r, s = interpreters.create_channel() self.assertIsInstance(r, interpreters.RecvChannel) self.assertIsInstance(s, interpreters.SendChannel) def test_list_all(self): self.assertEqual(interpreters.list_all_channels(), []) created = set() for _ in range(3): ch = interpreters.create_channel() created.add(ch) after = set(interpreters.list_all_channels()) self.assertEqual(after, created) def test_shareable(self): rch, sch = interpreters.create_channel() self.assertTrue( interpreters.is_shareable(rch)) self.assertTrue( interpreters.is_shareable(sch)) sch.send_nowait(rch) sch.send_nowait(sch) rch2 = rch.recv() sch2 = rch.recv() self.assertEqual(rch2, rch) self.assertEqual(sch2, sch) def test_is_closed(self): rch, sch = interpreters.create_channel() rbefore = rch.is_closed sbefore = sch.is_closed rch.close() rafter = rch.is_closed safter = sch.is_closed self.assertFalse(rbefore) self.assertFalse(sbefore) self.assertTrue(rafter) self.assertTrue(safter) class TestRecvChannelAttrs(TestBase): def test_id_type(self): rch, _ = interpreters.create_channel() self.assertIsInstance(rch.id, _channels.ChannelID) def test_custom_id(self): rch = interpreters.RecvChannel(1) self.assertEqual(rch.id, 1) with self.assertRaises(TypeError): interpreters.RecvChannel('1') def test_id_readonly(self): rch = interpreters.RecvChannel(1) with self.assertRaises(AttributeError): rch.id = 2 def test_equality(self): ch1, _ = interpreters.create_channel() ch2, _ = interpreters.create_channel() self.assertEqual(ch1, ch1) self.assertNotEqual(ch1, ch2) class TestSendChannelAttrs(TestBase): def test_id_type(self): _, sch = interpreters.create_channel() self.assertIsInstance(sch.id, _channels.ChannelID) def test_custom_id(self): sch = interpreters.SendChannel(1) self.assertEqual(sch.id, 1) with self.assertRaises(TypeError): interpreters.SendChannel('1') def test_id_readonly(self): sch = interpreters.SendChannel(1) with self.assertRaises(AttributeError): sch.id = 2 def test_equality(self): _, ch1 = interpreters.create_channel() _, ch2 = interpreters.create_channel() self.assertEqual(ch1, ch1) self.assertNotEqual(ch1, ch2) class TestSendRecv(TestBase): def test_send_recv_main(self): r, s = interpreters.create_channel() orig = b'spam' s.send_nowait(orig) obj = r.recv() self.assertEqual(obj, orig) self.assertIsNot(obj, orig) def test_send_recv_same_interpreter(self): interp = interpreters.create() interp.run(dedent(""" from test.support import interpreters r, s = interpreters.create_channel() orig = b'spam' s.send_nowait(orig) obj = r.recv() assert obj == orig, 'expected: obj == orig' assert obj is not orig, 'expected: obj is not orig' """)) @unittest.skip('broken (see BPO-...)') def test_send_recv_different_interpreters(self): r1, s1 = interpreters.create_channel() r2, s2 = interpreters.create_channel() orig1 = b'spam' s1.send_nowait(orig1) out = _run_output( interpreters.create(), dedent(f""" obj1 = r.recv() assert obj1 == b'spam', 'expected: obj1 == orig1' # When going to another interpreter we get a copy. assert id(obj1) != {id(orig1)}, 'expected: obj1 is not orig1' orig2 = b'eggs' print(id(orig2)) s.send_nowait(orig2) """), channels=dict(r=r1, s=s2), ) obj2 = r2.recv() self.assertEqual(obj2, b'eggs') self.assertNotEqual(id(obj2), int(out)) def test_send_recv_different_threads(self): r, s = interpreters.create_channel() def f(): while True: try: obj = r.recv() break except interpreters.ChannelEmptyError: time.sleep(0.1) s.send(obj) t = threading.Thread(target=f) t.start() orig = b'spam' s.send(orig) obj = r.recv() t.join() self.assertEqual(obj, orig) self.assertIsNot(obj, orig) def test_send_recv_nowait_main(self): r, s = interpreters.create_channel() orig = b'spam' s.send_nowait(orig) obj = r.recv_nowait() self.assertEqual(obj, orig) self.assertIsNot(obj, orig) def test_send_recv_nowait_main_with_default(self): r, _ = interpreters.create_channel() obj = r.recv_nowait(None) self.assertIsNone(obj) def test_send_recv_nowait_same_interpreter(self): interp = interpreters.create() interp.run(dedent(""" from test.support import interpreters r, s = interpreters.create_channel() orig = b'spam' s.send_nowait(orig) obj = r.recv_nowait() assert obj == orig, 'expected: obj == orig' # When going back to the same interpreter we get the same object. assert obj is not orig, 'expected: obj is not orig' """)) @unittest.skip('broken (see BPO-...)') def test_send_recv_nowait_different_interpreters(self): r1, s1 = interpreters.create_channel() r2, s2 = interpreters.create_channel() orig1 = b'spam' s1.send_nowait(orig1) out = _run_output( interpreters.create(), dedent(f""" obj1 = r.recv_nowait() assert obj1 == b'spam', 'expected: obj1 == orig1' # When going to another interpreter we get a copy. assert id(obj1) != {id(orig1)}, 'expected: obj1 is not orig1' orig2 = b'eggs' print(id(orig2)) s.send_nowait(orig2) """), channels=dict(r=r1, s=s2), ) obj2 = r2.recv_nowait() self.assertEqual(obj2, b'eggs') self.assertNotEqual(id(obj2), int(out)) def test_recv_timeout(self): r, _ = interpreters.create_channel() with self.assertRaises(TimeoutError): r.recv(timeout=1) def test_recv_channel_does_not_exist(self): ch = interpreters.RecvChannel(1_000_000) with self.assertRaises(interpreters.ChannelNotFoundError): ch.recv() def test_send_channel_does_not_exist(self): ch = interpreters.SendChannel(1_000_000) with self.assertRaises(interpreters.ChannelNotFoundError): ch.send(b'spam') def test_recv_nowait_channel_does_not_exist(self): ch = interpreters.RecvChannel(1_000_000) with self.assertRaises(interpreters.ChannelNotFoundError): ch.recv_nowait() def test_send_nowait_channel_does_not_exist(self): ch = interpreters.SendChannel(1_000_000) with self.assertRaises(interpreters.ChannelNotFoundError): ch.send_nowait(b'spam') def test_recv_nowait_empty(self): ch, _ = interpreters.create_channel() with self.assertRaises(interpreters.ChannelEmptyError): ch.recv_nowait() def test_recv_nowait_default(self): default = object() rch, sch = interpreters.create_channel() obj1 = rch.recv_nowait(default) sch.send_nowait(None) sch.send_nowait(1) sch.send_nowait(b'spam') sch.send_nowait(b'eggs') obj2 = rch.recv_nowait(default) obj3 = rch.recv_nowait(default) obj4 = rch.recv_nowait() obj5 = rch.recv_nowait(default) obj6 = rch.recv_nowait(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_send_buffer(self): buf = bytearray(b'spamspamspam') obj = None rch, sch = interpreters.create_channel() def f(): nonlocal obj while True: try: obj = rch.recv() break except interpreters.ChannelEmptyError: time.sleep(0.1) t = threading.Thread(target=f) t.start() sch.send_buffer(buf) t.join() 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 test_send_buffer_nowait(self): buf = bytearray(b'spamspamspam') rch, sch = interpreters.create_channel() sch.send_buffer_nowait(buf) obj = rch.recv() 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)