(self)
| 1296 | _channels.recv(cid) |
| 1297 | |
| 1298 | def test_multiple_users(self): |
| 1299 | cid = _channels.create(REPLACE) |
| 1300 | id1 = _interpreters.create() |
| 1301 | id2 = _interpreters.create() |
| 1302 | _interpreters.run_string(id1, dedent(f""" |
| 1303 | import _interpchannels as _channels |
| 1304 | _channels.send({cid}, b'spam', blocking=False) |
| 1305 | """)) |
| 1306 | out = _run_output(id2, dedent(f""" |
| 1307 | import _interpchannels as _channels |
| 1308 | obj, _ = _channels.recv({cid}) |
| 1309 | _channels.release({cid}) |
| 1310 | print(repr(obj)) |
| 1311 | """)) |
| 1312 | _interpreters.run_string(id1, dedent(f""" |
| 1313 | _channels.release({cid}) |
| 1314 | """)) |
| 1315 | |
| 1316 | self.assertEqual(out.strip(), "b'spam'") |
| 1317 | |
| 1318 | def test_no_kwargs(self): |
| 1319 | cid = _channels.create(REPLACE) |
nothing calls this directly
no test coverage detected