| import contextlib |
| import os |
| import pickle |
| from textwrap import dedent, indent |
| import threading |
| import time |
| import unittest |
| |
| from test import support |
| from test.support import script_helper |
| |
| interpreters = support.import_module('_xxsubinterpreters') |
| |
| |
| def _captured_script(script): |
| r, w = os.pipe() |
| indented = script.replace('\n', '\n ') |
| wrapped = dedent(f""" |
| import contextlib |
| with open({w}, 'w') as chan: |
| with contextlib.redirect_stdout(chan): |
| {indented} |
| """) |
| return wrapped, open(r) |
| |
| |
| def _run_output(interp, request, shared=None): |
| script, chan = _captured_script(request) |
| with chan: |
| interpreters.run_string(interp, script, shared) |
| return chan.read() |
| |
| |
| @contextlib.contextmanager |
| def _running(interp): |
| r, w = os.pipe() |
| def run(): |
| interpreters.run_string(interp, dedent(f""" |
| # wait for "signal" |
| with open({r}) as chan: |
| chan.read() |
| """)) |
| |
| t = threading.Thread(target=run) |
| t.start() |
| |
| yield |
| |
| with open(w, 'w') as chan: |
| chan.write('done') |
| t.join() |
| |
| |
| class IsShareableTests(unittest.TestCase): |
| |
| def test_default_shareables(self): |
| shareables = [ |
| # singletons |
| None, |
| # builtin objects |
| b'spam', |
| ] |
| for obj in shareables: |
| with self.subTest(obj): |
| self.assertTrue( |
| interpreters.is_shareable(obj)) |
| |
| 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 |
| True, |
| False, |
| NotImplemented, |
| ..., |
| # builtin types and objects |
| type, |
| object, |
| object(), |
| Exception(), |
| 42, |
| 100.0, |
| 'spam', |
| # user-defined types and objects |
| Cheese, |
| Cheese('Wensleydale'), |
| SubBytes(b'spam'), |
| ] |
| for obj in not_shareables: |
| with self.subTest(obj): |
| self.assertFalse( |
| interpreters.is_shareable(obj)) |
| |
| |
| class TestBase(unittest.TestCase): |
| |
| def tearDown(self): |
| for id in interpreters.list_all(): |
| if id == 0: # main |
| continue |
| try: |
| interpreters.destroy(id) |
| except RuntimeError: |
| pass # already destroyed |
| |
| for cid in interpreters.channel_list_all(): |
| try: |
| interpreters.channel_destroy(cid) |
| except interpreters.ChannelNotFoundError: |
| pass # already destroyed |
| |
| |
| class ListAllTests(TestBase): |
| |
| def test_initial(self): |
| main = interpreters.get_main() |
| ids = interpreters.list_all() |
| self.assertEqual(ids, [main]) |
| |
| def test_after_creating(self): |
| main = interpreters.get_main() |
| first = interpreters.create() |
| second = interpreters.create() |
| ids = interpreters.list_all() |
| self.assertEqual(ids, [main, first, second]) |
| |
| def test_after_destroying(self): |
| main = interpreters.get_main() |
| first = interpreters.create() |
| second = interpreters.create() |
| interpreters.destroy(first) |
| ids = interpreters.list_all() |
| self.assertEqual(ids, [main, second]) |
| |
| |
| class GetCurrentTests(TestBase): |
| |
| def test_main(self): |
| main = interpreters.get_main() |
| cur = interpreters.get_current() |
| self.assertEqual(cur, main) |
| |
| def test_subinterpreter(self): |
| main = interpreters.get_main() |
| interp = interpreters.create() |
| out = _run_output(interp, dedent(""" |
| import _xxsubinterpreters as _interpreters |
| print(int(_interpreters.get_current())) |
| """)) |
| cur = int(out.strip()) |
| _, expected = interpreters.list_all() |
| self.assertEqual(cur, expected) |
| self.assertNotEqual(cur, main) |
| |
| |
| class GetMainTests(TestBase): |
| |
| def test_from_main(self): |
| [expected] = interpreters.list_all() |
| main = interpreters.get_main() |
| self.assertEqual(main, expected) |
| |
| def test_from_subinterpreter(self): |
| [expected] = interpreters.list_all() |
| interp = interpreters.create() |
| out = _run_output(interp, dedent(""" |
| import _xxsubinterpreters as _interpreters |
| print(int(_interpreters.get_main())) |
| """)) |
| main = int(out.strip()) |
| self.assertEqual(main, expected) |
| |
| |
| class IsRunningTests(TestBase): |
| |
| def test_main(self): |
| main = interpreters.get_main() |
| self.assertTrue(interpreters.is_running(main)) |
| |
| def test_subinterpreter(self): |
| interp = interpreters.create() |
| self.assertFalse(interpreters.is_running(interp)) |
| |
| with _running(interp): |
| self.assertTrue(interpreters.is_running(interp)) |
| self.assertFalse(interpreters.is_running(interp)) |
| |
| def test_from_subinterpreter(self): |
| interp = interpreters.create() |
| out = _run_output(interp, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| if _interpreters.is_running({int(interp)}): |
| print(True) |
| else: |
| print(False) |
| """)) |
| self.assertEqual(out.strip(), 'True') |
| |
| def test_already_destroyed(self): |
| interp = interpreters.create() |
| interpreters.destroy(interp) |
| with self.assertRaises(RuntimeError): |
| interpreters.is_running(interp) |
| |
| def test_does_not_exist(self): |
| with self.assertRaises(RuntimeError): |
| interpreters.is_running(1_000_000) |
| |
| def test_bad_id(self): |
| with self.assertRaises(RuntimeError): |
| interpreters.is_running(-1) |
| |
| |
| class InterpreterIDTests(TestBase): |
| |
| def test_with_int(self): |
| id = interpreters.InterpreterID(10, force=True) |
| |
| self.assertEqual(int(id), 10) |
| |
| def test_coerce_id(self): |
| id = interpreters.InterpreterID('10', force=True) |
| self.assertEqual(int(id), 10) |
| |
| id = interpreters.InterpreterID(10.0, force=True) |
| self.assertEqual(int(id), 10) |
| |
| class Int(str): |
| def __init__(self, value): |
| self._value = value |
| def __int__(self): |
| return self._value |
| |
| id = interpreters.InterpreterID(Int(10), force=True) |
| self.assertEqual(int(id), 10) |
| |
| def test_bad_id(self): |
| for id in [-1, 'spam']: |
| with self.subTest(id): |
| with self.assertRaises(ValueError): |
| interpreters.InterpreterID(id) |
| with self.assertRaises(OverflowError): |
| interpreters.InterpreterID(2**64) |
| with self.assertRaises(TypeError): |
| interpreters.InterpreterID(object()) |
| |
| def test_does_not_exist(self): |
| id = interpreters.channel_create() |
| with self.assertRaises(RuntimeError): |
| interpreters.InterpreterID(int(id) + 1) # unforced |
| |
| def test_repr(self): |
| id = interpreters.InterpreterID(10, force=True) |
| self.assertEqual(repr(id), 'InterpreterID(10)') |
| |
| def test_equality(self): |
| id1 = interpreters.create() |
| id2 = interpreters.InterpreterID(int(id1)) |
| id3 = interpreters.create() |
| |
| self.assertTrue(id1 == id1) |
| self.assertTrue(id1 == id2) |
| self.assertTrue(id1 == int(id1)) |
| self.assertFalse(id1 == id3) |
| |
| self.assertFalse(id1 != id1) |
| self.assertFalse(id1 != id2) |
| self.assertTrue(id1 != id3) |
| |
| |
| class CreateTests(TestBase): |
| |
| def test_in_main(self): |
| id = interpreters.create() |
| |
| self.assertIn(id, interpreters.list_all()) |
| |
| @unittest.skip('enable this test when working on pystate.c') |
| def test_unique_id(self): |
| seen = set() |
| for _ in range(100): |
| id = interpreters.create() |
| interpreters.destroy(id) |
| seen.add(id) |
| |
| self.assertEqual(len(seen), 100) |
| |
| def test_in_thread(self): |
| lock = threading.Lock() |
| id = None |
| def f(): |
| nonlocal id |
| id = interpreters.create() |
| lock.acquire() |
| lock.release() |
| |
| t = threading.Thread(target=f) |
| with lock: |
| t.start() |
| t.join() |
| self.assertIn(id, interpreters.list_all()) |
| |
| def test_in_subinterpreter(self): |
| main, = interpreters.list_all() |
| id1 = interpreters.create() |
| out = _run_output(id1, dedent(""" |
| import _xxsubinterpreters as _interpreters |
| id = _interpreters.create() |
| print(int(id)) |
| """)) |
| id2 = int(out.strip()) |
| |
| self.assertEqual(set(interpreters.list_all()), {main, id1, id2}) |
| |
| def test_in_threaded_subinterpreter(self): |
| main, = interpreters.list_all() |
| id1 = interpreters.create() |
| id2 = None |
| def f(): |
| nonlocal id2 |
| out = _run_output(id1, dedent(""" |
| import _xxsubinterpreters as _interpreters |
| id = _interpreters.create() |
| print(int(id)) |
| """)) |
| id2 = int(out.strip()) |
| |
| t = threading.Thread(target=f) |
| t.start() |
| t.join() |
| |
| self.assertEqual(set(interpreters.list_all()), {main, id1, id2}) |
| |
| def test_after_destroy_all(self): |
| before = set(interpreters.list_all()) |
| # Create 3 subinterpreters. |
| ids = [] |
| for _ in range(3): |
| id = interpreters.create() |
| ids.append(id) |
| # Now destroy them. |
| for id in ids: |
| interpreters.destroy(id) |
| # Finally, create another. |
| id = interpreters.create() |
| self.assertEqual(set(interpreters.list_all()), before | {id}) |
| |
| def test_after_destroy_some(self): |
| before = set(interpreters.list_all()) |
| # Create 3 subinterpreters. |
| id1 = interpreters.create() |
| id2 = interpreters.create() |
| id3 = interpreters.create() |
| # Now destroy 2 of them. |
| interpreters.destroy(id1) |
| interpreters.destroy(id3) |
| # Finally, create another. |
| id = interpreters.create() |
| self.assertEqual(set(interpreters.list_all()), before | {id, id2}) |
| |
| |
| class DestroyTests(TestBase): |
| |
| def test_one(self): |
| id1 = interpreters.create() |
| id2 = interpreters.create() |
| id3 = interpreters.create() |
| self.assertIn(id2, interpreters.list_all()) |
| interpreters.destroy(id2) |
| self.assertNotIn(id2, interpreters.list_all()) |
| self.assertIn(id1, interpreters.list_all()) |
| self.assertIn(id3, interpreters.list_all()) |
| |
| def test_all(self): |
| before = set(interpreters.list_all()) |
| ids = set() |
| for _ in range(3): |
| id = interpreters.create() |
| ids.add(id) |
| self.assertEqual(set(interpreters.list_all()), before | ids) |
| for id in ids: |
| interpreters.destroy(id) |
| self.assertEqual(set(interpreters.list_all()), before) |
| |
| def test_main(self): |
| main, = interpreters.list_all() |
| with self.assertRaises(RuntimeError): |
| interpreters.destroy(main) |
| |
| def f(): |
| with self.assertRaises(RuntimeError): |
| interpreters.destroy(main) |
| |
| t = threading.Thread(target=f) |
| t.start() |
| t.join() |
| |
| def test_already_destroyed(self): |
| id = interpreters.create() |
| interpreters.destroy(id) |
| with self.assertRaises(RuntimeError): |
| interpreters.destroy(id) |
| |
| def test_does_not_exist(self): |
| with self.assertRaises(RuntimeError): |
| interpreters.destroy(1_000_000) |
| |
| def test_bad_id(self): |
| with self.assertRaises(RuntimeError): |
| interpreters.destroy(-1) |
| |
| def test_from_current(self): |
| main, = interpreters.list_all() |
| id = interpreters.create() |
| script = dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| try: |
| _interpreters.destroy({int(id)}) |
| except RuntimeError: |
| pass |
| """) |
| |
| interpreters.run_string(id, script) |
| self.assertEqual(set(interpreters.list_all()), {main, id}) |
| |
| def test_from_sibling(self): |
| main, = interpreters.list_all() |
| id1 = interpreters.create() |
| id2 = interpreters.create() |
| script = dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| _interpreters.destroy({int(id2)}) |
| """) |
| interpreters.run_string(id1, script) |
| |
| self.assertEqual(set(interpreters.list_all()), {main, id1}) |
| |
| def test_from_other_thread(self): |
| id = interpreters.create() |
| def f(): |
| interpreters.destroy(id) |
| |
| t = threading.Thread(target=f) |
| t.start() |
| t.join() |
| |
| def test_still_running(self): |
| main, = interpreters.list_all() |
| interp = interpreters.create() |
| with _running(interp): |
| with self.assertRaises(RuntimeError): |
| interpreters.destroy(interp) |
| self.assertTrue(interpreters.is_running(interp)) |
| |
| |
| class RunStringTests(TestBase): |
| |
| SCRIPT = dedent(""" |
| with open('{}', 'w') as out: |
| out.write('{}') |
| """) |
| FILENAME = 'spam' |
| |
| def setUp(self): |
| super().setUp() |
| self.id = interpreters.create() |
| self._fs = None |
| |
| def tearDown(self): |
| if self._fs is not None: |
| self._fs.close() |
| super().tearDown() |
| |
| @property |
| def fs(self): |
| if self._fs is None: |
| self._fs = FSFixture(self) |
| return self._fs |
| |
| def test_success(self): |
| script, file = _captured_script('print("it worked!", end="")') |
| with file: |
| interpreters.run_string(self.id, script) |
| out = file.read() |
| |
| self.assertEqual(out, 'it worked!') |
| |
| def test_in_thread(self): |
| script, file = _captured_script('print("it worked!", end="")') |
| with file: |
| def f(): |
| interpreters.run_string(self.id, script) |
| |
| t = threading.Thread(target=f) |
| t.start() |
| t.join() |
| out = file.read() |
| |
| self.assertEqual(out, 'it worked!') |
| |
| def test_create_thread(self): |
| script, file = _captured_script(""" |
| import threading |
| def f(): |
| print('it worked!', end='') |
| |
| t = threading.Thread(target=f) |
| t.start() |
| t.join() |
| """) |
| with file: |
| interpreters.run_string(self.id, script) |
| out = file.read() |
| |
| self.assertEqual(out, 'it worked!') |
| |
| @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") |
| def test_fork(self): |
| import tempfile |
| with tempfile.NamedTemporaryFile('w+') as file: |
| file.write('') |
| file.flush() |
| |
| expected = 'spam spam spam spam spam' |
| script = dedent(f""" |
| # (inspired by Lib/test/test_fork.py) |
| import os |
| pid = os.fork() |
| if pid == 0: # child |
| with open('{file.name}', 'w') as out: |
| out.write('{expected}') |
| # Kill the unittest runner in the child process. |
| os._exit(1) |
| else: |
| SHORT_SLEEP = 0.1 |
| import time |
| for _ in range(10): |
| spid, status = os.waitpid(pid, os.WNOHANG) |
| if spid == pid: |
| break |
| time.sleep(SHORT_SLEEP) |
| assert(spid == pid) |
| """) |
| interpreters.run_string(self.id, script) |
| |
| file.seek(0) |
| content = file.read() |
| self.assertEqual(content, expected) |
| |
| def test_already_running(self): |
| with _running(self.id): |
| with self.assertRaises(RuntimeError): |
| interpreters.run_string(self.id, 'print("spam")') |
| |
| def test_does_not_exist(self): |
| id = 0 |
| while id in interpreters.list_all(): |
| id += 1 |
| with self.assertRaises(RuntimeError): |
| interpreters.run_string(id, 'print("spam")') |
| |
| def test_error_id(self): |
| with self.assertRaises(RuntimeError): |
| interpreters.run_string(-1, 'print("spam")') |
| |
| def test_bad_id(self): |
| with self.assertRaises(TypeError): |
| interpreters.run_string('spam', 'print("spam")') |
| |
| def test_bad_script(self): |
| with self.assertRaises(TypeError): |
| interpreters.run_string(self.id, 10) |
| |
| def test_bytes_for_script(self): |
| with self.assertRaises(TypeError): |
| interpreters.run_string(self.id, b'print("spam")') |
| |
| @contextlib.contextmanager |
| def assert_run_failed(self, exctype, msg=None): |
| with self.assertRaises(interpreters.RunFailedError) as caught: |
| yield |
| if msg is None: |
| self.assertEqual(str(caught.exception).split(':')[0], |
| str(exctype)) |
| else: |
| self.assertEqual(str(caught.exception), |
| "{}: {}".format(exctype, msg)) |
| |
| def test_invalid_syntax(self): |
| with self.assert_run_failed(SyntaxError): |
| # missing close paren |
| interpreters.run_string(self.id, 'print("spam"') |
| |
| def test_failure(self): |
| with self.assert_run_failed(Exception, 'spam'): |
| interpreters.run_string(self.id, 'raise Exception("spam")') |
| |
| def test_SystemExit(self): |
| with self.assert_run_failed(SystemExit, '42'): |
| interpreters.run_string(self.id, 'raise SystemExit(42)') |
| |
| def test_sys_exit(self): |
| with self.assert_run_failed(SystemExit): |
| interpreters.run_string(self.id, dedent(""" |
| import sys |
| sys.exit() |
| """)) |
| |
| with self.assert_run_failed(SystemExit, '42'): |
| interpreters.run_string(self.id, dedent(""" |
| import sys |
| sys.exit(42) |
| """)) |
| |
| def test_with_shared(self): |
| r, w = os.pipe() |
| |
| shared = { |
| 'spam': b'ham', |
| 'eggs': b'-1', |
| 'cheddar': None, |
| } |
| script = dedent(f""" |
| eggs = int(eggs) |
| spam = 42 |
| result = spam + eggs |
| |
| ns = dict(vars()) |
| del ns['__builtins__'] |
| import pickle |
| with open({w}, 'wb') as chan: |
| pickle.dump(ns, chan) |
| """) |
| interpreters.run_string(self.id, script, shared) |
| with open(r, 'rb') as chan: |
| ns = pickle.load(chan) |
| |
| self.assertEqual(ns['spam'], 42) |
| self.assertEqual(ns['eggs'], -1) |
| self.assertEqual(ns['result'], 41) |
| self.assertIsNone(ns['cheddar']) |
| |
| def test_shared_overwrites(self): |
| interpreters.run_string(self.id, dedent(""" |
| spam = 'eggs' |
| ns1 = dict(vars()) |
| del ns1['__builtins__'] |
| """)) |
| |
| shared = {'spam': b'ham'} |
| script = dedent(f""" |
| ns2 = dict(vars()) |
| del ns2['__builtins__'] |
| """) |
| interpreters.run_string(self.id, script, shared) |
| |
| r, w = os.pipe() |
| script = dedent(f""" |
| ns = dict(vars()) |
| del ns['__builtins__'] |
| import pickle |
| with open({w}, 'wb') as chan: |
| pickle.dump(ns, chan) |
| """) |
| interpreters.run_string(self.id, script) |
| with open(r, 'rb') as chan: |
| ns = pickle.load(chan) |
| |
| self.assertEqual(ns['ns1']['spam'], 'eggs') |
| self.assertEqual(ns['ns2']['spam'], b'ham') |
| self.assertEqual(ns['spam'], b'ham') |
| |
| def test_shared_overwrites_default_vars(self): |
| r, w = os.pipe() |
| |
| shared = {'__name__': b'not __main__'} |
| script = dedent(f""" |
| spam = 42 |
| |
| ns = dict(vars()) |
| del ns['__builtins__'] |
| import pickle |
| with open({w}, 'wb') as chan: |
| pickle.dump(ns, chan) |
| """) |
| interpreters.run_string(self.id, script, shared) |
| with open(r, 'rb') as chan: |
| ns = pickle.load(chan) |
| |
| self.assertEqual(ns['__name__'], b'not __main__') |
| |
| def test_main_reused(self): |
| r, w = os.pipe() |
| interpreters.run_string(self.id, dedent(f""" |
| spam = True |
| |
| ns = dict(vars()) |
| del ns['__builtins__'] |
| import pickle |
| with open({w}, 'wb') as chan: |
| pickle.dump(ns, chan) |
| del ns, pickle, chan |
| """)) |
| with open(r, 'rb') as chan: |
| ns1 = pickle.load(chan) |
| |
| r, w = os.pipe() |
| interpreters.run_string(self.id, dedent(f""" |
| eggs = False |
| |
| ns = dict(vars()) |
| del ns['__builtins__'] |
| import pickle |
| with open({w}, 'wb') as chan: |
| pickle.dump(ns, chan) |
| """)) |
| with open(r, 'rb') as chan: |
| ns2 = pickle.load(chan) |
| |
| self.assertIn('spam', ns1) |
| self.assertNotIn('eggs', ns1) |
| self.assertIn('eggs', ns2) |
| self.assertIn('spam', ns2) |
| |
| def test_execution_namespace_is_main(self): |
| r, w = os.pipe() |
| |
| script = dedent(f""" |
| spam = 42 |
| |
| ns = dict(vars()) |
| ns['__builtins__'] = str(ns['__builtins__']) |
| import pickle |
| with open({w}, 'wb') as chan: |
| pickle.dump(ns, chan) |
| """) |
| interpreters.run_string(self.id, script) |
| with open(r, 'rb') as chan: |
| ns = pickle.load(chan) |
| |
| ns.pop('__builtins__') |
| ns.pop('__loader__') |
| self.assertEqual(ns, { |
| '__name__': '__main__', |
| '__annotations__': {}, |
| '__doc__': None, |
| '__package__': None, |
| '__spec__': None, |
| 'spam': 42, |
| }) |
| |
| # XXX Fix this test! |
| @unittest.skip('blocking forever') |
| def test_still_running_at_exit(self): |
| script = dedent(f""" |
| from textwrap import dedent |
| import threading |
| import _xxsubinterpreters as _interpreters |
| id = _interpreters.create() |
| def f(): |
| _interpreters.run_string(id, dedent(''' |
| import time |
| # Give plenty of time for the main interpreter to finish. |
| time.sleep(1_000_000) |
| ''')) |
| |
| t = threading.Thread(target=f) |
| t.start() |
| """) |
| with support.temp_dir() as dirname: |
| filename = script_helper.make_script(dirname, 'interp', script) |
| with script_helper.spawn_python(filename) as proc: |
| retcode = proc.wait() |
| |
| self.assertEqual(retcode, 0) |
| |
| |
| class ChannelIDTests(TestBase): |
| |
| def test_default_kwargs(self): |
| cid = interpreters._channel_id(10, force=True) |
| |
| self.assertEqual(int(cid), 10) |
| self.assertEqual(cid.end, 'both') |
| |
| def test_with_kwargs(self): |
| cid = interpreters._channel_id(10, send=True, force=True) |
| self.assertEqual(cid.end, 'send') |
| |
| cid = interpreters._channel_id(10, send=True, recv=False, force=True) |
| self.assertEqual(cid.end, 'send') |
| |
| cid = interpreters._channel_id(10, recv=True, force=True) |
| self.assertEqual(cid.end, 'recv') |
| |
| cid = interpreters._channel_id(10, recv=True, send=False, force=True) |
| self.assertEqual(cid.end, 'recv') |
| |
| cid = interpreters._channel_id(10, send=True, recv=True, force=True) |
| self.assertEqual(cid.end, 'both') |
| |
| def test_coerce_id(self): |
| cid = interpreters._channel_id('10', force=True) |
| self.assertEqual(int(cid), 10) |
| |
| cid = interpreters._channel_id(10.0, force=True) |
| self.assertEqual(int(cid), 10) |
| |
| class Int(str): |
| def __init__(self, value): |
| self._value = value |
| def __int__(self): |
| return self._value |
| |
| cid = interpreters._channel_id(Int(10), force=True) |
| self.assertEqual(int(cid), 10) |
| |
| def test_bad_id(self): |
| for cid in [-1, 'spam']: |
| with self.subTest(cid): |
| with self.assertRaises(ValueError): |
| interpreters._channel_id(cid) |
| with self.assertRaises(OverflowError): |
| interpreters._channel_id(2**64) |
| with self.assertRaises(TypeError): |
| interpreters._channel_id(object()) |
| |
| def test_bad_kwargs(self): |
| with self.assertRaises(ValueError): |
| interpreters._channel_id(10, send=False, recv=False) |
| |
| def test_does_not_exist(self): |
| cid = interpreters.channel_create() |
| with self.assertRaises(interpreters.ChannelNotFoundError): |
| interpreters._channel_id(int(cid) + 1) # unforced |
| |
| def test_repr(self): |
| cid = interpreters._channel_id(10, force=True) |
| self.assertEqual(repr(cid), 'ChannelID(10)') |
| |
| cid = interpreters._channel_id(10, send=True, force=True) |
| self.assertEqual(repr(cid), 'ChannelID(10, send=True)') |
| |
| cid = interpreters._channel_id(10, recv=True, force=True) |
| self.assertEqual(repr(cid), 'ChannelID(10, recv=True)') |
| |
| cid = interpreters._channel_id(10, send=True, recv=True, force=True) |
| self.assertEqual(repr(cid), 'ChannelID(10)') |
| |
| def test_equality(self): |
| cid1 = interpreters.channel_create() |
| cid2 = interpreters._channel_id(int(cid1)) |
| cid3 = interpreters.channel_create() |
| |
| self.assertTrue(cid1 == cid1) |
| self.assertTrue(cid1 == cid2) |
| self.assertTrue(cid1 == int(cid1)) |
| self.assertFalse(cid1 == cid3) |
| |
| self.assertFalse(cid1 != cid1) |
| self.assertFalse(cid1 != cid2) |
| self.assertTrue(cid1 != cid3) |
| |
| |
| class ChannelTests(TestBase): |
| |
| def test_sequential_ids(self): |
| before = interpreters.channel_list_all() |
| id1 = interpreters.channel_create() |
| id2 = interpreters.channel_create() |
| id3 = interpreters.channel_create() |
| after = interpreters.channel_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(""" |
| import _xxsubinterpreters as _interpreters |
| cid = _interpreters.channel_create() |
| print(int(cid)) |
| """)) |
| cid1 = int(out.strip()) |
| |
| id2 = interpreters.create() |
| out = _run_output(id2, dedent(""" |
| import _xxsubinterpreters as _interpreters |
| cid = _interpreters.channel_create() |
| print(int(cid)) |
| """)) |
| cid2 = int(out.strip()) |
| |
| self.assertEqual(cid2, int(cid1) + 1) |
| |
| #################### |
| |
| def test_drop_single_user(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_recv(cid) |
| interpreters.channel_drop_interpreter(cid, send=True, recv=True) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_send(cid, b'eggs') |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| def test_drop_multiple_users(self): |
| cid = interpreters.channel_create() |
| id1 = interpreters.create() |
| id2 = interpreters.create() |
| interpreters.run_string(id1, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| _interpreters.channel_send({int(cid)}, b'spam') |
| """)) |
| out = _run_output(id2, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| obj = _interpreters.channel_recv({int(cid)}) |
| _interpreters.channel_drop_interpreter({int(cid)}) |
| print(repr(obj)) |
| """)) |
| interpreters.run_string(id1, dedent(f""" |
| _interpreters.channel_drop_interpreter({int(cid)}) |
| """)) |
| |
| self.assertEqual(out.strip(), "b'spam'") |
| |
| def test_drop_no_kwargs(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_recv(cid) |
| interpreters.channel_drop_interpreter(cid) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_send(cid, b'eggs') |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| def test_drop_multiple_times(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_recv(cid) |
| interpreters.channel_drop_interpreter(cid, send=True, recv=True) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_drop_interpreter(cid, send=True, recv=True) |
| |
| def test_drop_with_unused_items(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_send(cid, b'ham') |
| interpreters.channel_drop_interpreter(cid, send=True, recv=True) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| def test_drop_never_used(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_drop_interpreter(cid) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_send(cid, b'spam') |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| def test_drop_by_unassociated_interp(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interp = interpreters.create() |
| interpreters.run_string(interp, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| _interpreters.channel_drop_interpreter({int(cid)}) |
| """)) |
| obj = interpreters.channel_recv(cid) |
| interpreters.channel_drop_interpreter(cid) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_send(cid, b'eggs') |
| self.assertEqual(obj, b'spam') |
| |
| def test_drop_close_if_unassociated(self): |
| cid = interpreters.channel_create() |
| interp = interpreters.create() |
| interpreters.run_string(interp, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| obj = _interpreters.channel_send({int(cid)}, b'spam') |
| _interpreters.channel_drop_interpreter({int(cid)}) |
| """)) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| def test_drop_partially(self): |
| # XXX Is partial close too wierd/confusing? |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, None) |
| interpreters.channel_recv(cid) |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_drop_interpreter(cid, send=True) |
| obj = interpreters.channel_recv(cid) |
| |
| self.assertEqual(obj, b'spam') |
| |
| def test_drop_used_multiple_times_by_single_user(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_recv(cid) |
| interpreters.channel_drop_interpreter(cid, send=True, recv=True) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_send(cid, b'eggs') |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| #################### |
| |
| def test_close_single_user(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_recv(cid) |
| interpreters.channel_close(cid) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_send(cid, b'eggs') |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| def test_close_multiple_users(self): |
| cid = interpreters.channel_create() |
| id1 = interpreters.create() |
| id2 = interpreters.create() |
| interpreters.run_string(id1, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| _interpreters.channel_send({int(cid)}, b'spam') |
| """)) |
| interpreters.run_string(id2, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| _interpreters.channel_recv({int(cid)}) |
| """)) |
| interpreters.channel_close(cid) |
| with self.assertRaises(interpreters.RunFailedError) as cm: |
| interpreters.run_string(id1, dedent(f""" |
| _interpreters.channel_send({int(cid)}, b'spam') |
| """)) |
| self.assertIn('ChannelClosedError', str(cm.exception)) |
| with self.assertRaises(interpreters.RunFailedError) as cm: |
| interpreters.run_string(id2, dedent(f""" |
| _interpreters.channel_send({int(cid)}, b'spam') |
| """)) |
| self.assertIn('ChannelClosedError', str(cm.exception)) |
| |
| def test_close_multiple_times(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_recv(cid) |
| interpreters.channel_close(cid) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_close(cid) |
| |
| def test_close_with_unused_items(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_send(cid, b'ham') |
| interpreters.channel_close(cid) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| def test_close_never_used(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_close(cid) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_send(cid, b'spam') |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| def test_close_by_unassociated_interp(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interp = interpreters.create() |
| interpreters.run_string(interp, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| _interpreters.channel_close({int(cid)}) |
| """)) |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_close(cid) |
| |
| def test_close_used_multiple_times_by_single_user(self): |
| cid = interpreters.channel_create() |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_send(cid, b'spam') |
| interpreters.channel_recv(cid) |
| interpreters.channel_close(cid) |
| |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_send(cid, b'eggs') |
| with self.assertRaises(interpreters.ChannelClosedError): |
| interpreters.channel_recv(cid) |
| |
| #################### |
| |
| def test_send_recv_main(self): |
| cid = interpreters.channel_create() |
| orig = b'spam' |
| interpreters.channel_send(cid, orig) |
| obj = interpreters.channel_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(""" |
| import _xxsubinterpreters as _interpreters |
| cid = _interpreters.channel_create() |
| orig = b'spam' |
| _interpreters.channel_send(cid, orig) |
| obj = _interpreters.channel_recv(cid) |
| assert obj is not orig |
| assert obj == orig |
| """)) |
| |
| def test_send_recv_different_interpreters(self): |
| cid = interpreters.channel_create() |
| id1 = interpreters.create() |
| out = _run_output(id1, dedent(f""" |
| import _xxsubinterpreters as _interpreters |
| _interpreters.channel_send({int(cid)}, b'spam') |
| """)) |
| obj = interpreters.channel_recv(cid) |
| |
| self.assertEqual(obj, b'spam') |
| |
| def test_send_recv_different_threads(self): |
| cid = interpreters.channel_create() |
| |
| def f(): |
| while True: |
| try: |
| obj = interpreters.channel_recv(cid) |
| break |
| except interpreters.ChannelEmptyError: |
| time.sleep(0.1) |
| interpreters.channel_send(cid, obj) |
| t = threading.Thread(target=f) |
| t.start() |
| |
| interpreters.channel_send(cid, b'spam') |
| t.join() |
| obj = interpreters.channel_recv(cid) |
| |
| self.assertEqual(obj, b'spam') |
| |
| def test_send_recv_different_interpreters_and_threads(self): |
| cid = interpreters.channel_create() |
| id1 = interpreters.create() |
| out = None |
| |
| def f(): |
| nonlocal out |
| out = _run_output(id1, dedent(f""" |
| import time |
| import _xxsubinterpreters as _interpreters |
| while True: |
| try: |
| obj = _interpreters.channel_recv({int(cid)}) |
| break |
| except _interpreters.ChannelEmptyError: |
| time.sleep(0.1) |
| assert(obj == b'spam') |
| _interpreters.channel_send({int(cid)}, b'eggs') |
| """)) |
| t = threading.Thread(target=f) |
| t.start() |
| |
| interpreters.channel_send(cid, b'spam') |
| t.join() |
| obj = interpreters.channel_recv(cid) |
| |
| self.assertEqual(obj, b'eggs') |
| |
| def test_send_not_found(self): |
| with self.assertRaises(interpreters.ChannelNotFoundError): |
| interpreters.channel_send(10, b'spam') |
| |
| def test_recv_not_found(self): |
| with self.assertRaises(interpreters.ChannelNotFoundError): |
| interpreters.channel_recv(10) |
| |
| def test_recv_empty(self): |
| cid = interpreters.channel_create() |
| with self.assertRaises(interpreters.ChannelEmptyError): |
| interpreters.channel_recv(cid) |
| |
| def test_run_string_arg(self): |
| cid = interpreters.channel_create() |
| interp = interpreters.create() |
| |
| out = _run_output(interp, dedent(""" |
| import _xxsubinterpreters as _interpreters |
| print(cid.end) |
| _interpreters.channel_send(cid, b'spam') |
| """), |
| dict(cid=cid.send)) |
| obj = interpreters.channel_recv(cid) |
| |
| self.assertEqual(obj, b'spam') |
| self.assertEqual(out.strip(), 'send') |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |