(self)
| 1055 | _channels.recv(cid) |
| 1056 | |
| 1057 | def test_close_multiple_users(self): |
| 1058 | cid = _channels.create(REPLACE) |
| 1059 | id1 = _interpreters.create() |
| 1060 | id2 = _interpreters.create() |
| 1061 | _interpreters.run_string(id1, dedent(f""" |
| 1062 | import _interpchannels as _channels |
| 1063 | _channels.send({cid}, b'spam', blocking=False) |
| 1064 | """)) |
| 1065 | _interpreters.run_string(id2, dedent(f""" |
| 1066 | import _interpchannels as _channels |
| 1067 | _channels.recv({cid}) |
| 1068 | """)) |
| 1069 | _channels.close(cid) |
| 1070 | |
| 1071 | excsnap = _interpreters.run_string(id1, dedent(f""" |
| 1072 | _channels.send({cid}, b'spam') |
| 1073 | """)) |
| 1074 | self.assertEqual(excsnap.type.__name__, 'ChannelClosedError') |
| 1075 | |
| 1076 | excsnap = _interpreters.run_string(id2, dedent(f""" |
| 1077 | _channels.send({cid}, b'spam') |
| 1078 | """)) |
| 1079 | self.assertEqual(excsnap.type.__name__, 'ChannelClosedError') |
| 1080 | |
| 1081 | def test_close_multiple_times(self): |
| 1082 | cid = _channels.create(REPLACE) |
nothing calls this directly
no test coverage detected