@@ -1207,6 +1207,185 @@ def test_ids_global(self):
1207 1207 1208 1208
self.assertEqual(cid2, int(cid1) + 1)
1209 1209 1210 +
def test_channel_list_interpreters_none(self):
1211 +
"""Test listing interpreters for a channel with no associations."""
1212 +
# Test for channel with no associated interpreters.
1213 +
cid = interpreters.channel_create()
1214 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1215 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1216 +
self.assertEqual(send_interps, [])
1217 +
self.assertEqual(recv_interps, [])
1218 + 1219 +
def test_channel_list_interpreters_basic(self):
1220 +
"""Test basic listing channel interpreters."""
1221 +
interp0 = interpreters.get_main()
1222 +
cid = interpreters.channel_create()
1223 +
interpreters.channel_send(cid, "send")
1224 +
# Test for a channel that has one end associated to an interpreter.
1225 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1226 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1227 +
self.assertEqual(send_interps, [interp0])
1228 +
self.assertEqual(recv_interps, [])
1229 + 1230 +
interp1 = interpreters.create()
1231 +
_run_output(interp1, dedent(f"""
1232 +
import _xxsubinterpreters as _interpreters
1233 +
obj = _interpreters.channel_recv({cid})
1234 +
"""))
1235 +
# Test for channel that has boths ends associated to an interpreter.
1236 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1237 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1238 +
self.assertEqual(send_interps, [interp0])
1239 +
self.assertEqual(recv_interps, [interp1])
1240 + 1241 +
def test_channel_list_interpreters_multiple(self):
1242 +
"""Test listing interpreters for a channel with many associations."""
1243 +
interp0 = interpreters.get_main()
1244 +
interp1 = interpreters.create()
1245 +
interp2 = interpreters.create()
1246 +
interp3 = interpreters.create()
1247 +
cid = interpreters.channel_create()
1248 + 1249 +
interpreters.channel_send(cid, "send")
1250 +
_run_output(interp1, dedent(f"""
1251 +
import _xxsubinterpreters as _interpreters
1252 +
_interpreters.channel_send({cid}, "send")
1253 +
"""))
1254 +
_run_output(interp2, dedent(f"""
1255 +
import _xxsubinterpreters as _interpreters
1256 +
obj = _interpreters.channel_recv({cid})
1257 +
"""))
1258 +
_run_output(interp3, dedent(f"""
1259 +
import _xxsubinterpreters as _interpreters
1260 +
obj = _interpreters.channel_recv({cid})
1261 +
"""))
1262 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1263 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1264 +
self.assertEqual(set(send_interps), {interp0, interp1})
1265 +
self.assertEqual(set(recv_interps), {interp2, interp3})
1266 + 1267 +
def test_channel_list_interpreters_destroyed(self):
1268 +
"""Test listing channel interpreters with a destroyed interpreter."""
1269 +
interp0 = interpreters.get_main()
1270 +
interp1 = interpreters.create()
1271 +
cid = interpreters.channel_create()
1272 +
interpreters.channel_send(cid, "send")
1273 +
_run_output(interp1, dedent(f"""
1274 +
import _xxsubinterpreters as _interpreters
1275 +
obj = _interpreters.channel_recv({cid})
1276 +
"""))
1277 +
# Should be one interpreter associated with each end.
1278 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1279 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1280 +
self.assertEqual(send_interps, [interp0])
1281 +
self.assertEqual(recv_interps, [interp1])
1282 + 1283 +
interpreters.destroy(interp1)
1284 +
# Destroyed interpreter should not be listed.
1285 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1286 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1287 +
self.assertEqual(send_interps, [interp0])
1288 +
self.assertEqual(recv_interps, [])
1289 + 1290 +
def test_channel_list_interpreters_released(self):
1291 +
"""Test listing channel interpreters with a released channel."""
1292 +
# Set up one channel with main interpreter on the send end and two
1293 +
# subinterpreters on the receive end.
1294 +
interp0 = interpreters.get_main()
1295 +
interp1 = interpreters.create()
1296 +
interp2 = interpreters.create()
1297 +
cid = interpreters.channel_create()
1298 +
interpreters.channel_send(cid, "data")
1299 +
_run_output(interp1, dedent(f"""
1300 +
import _xxsubinterpreters as _interpreters
1301 +
obj = _interpreters.channel_recv({cid})
1302 +
"""))
1303 +
interpreters.channel_send(cid, "data")
1304 +
_run_output(interp2, dedent(f"""
1305 +
import _xxsubinterpreters as _interpreters
1306 +
obj = _interpreters.channel_recv({cid})
1307 +
"""))
1308 +
# Check the setup.
1309 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1310 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1311 +
self.assertEqual(len(send_interps), 1)
1312 +
self.assertEqual(len(recv_interps), 2)
1313 + 1314 +
# Release the main interpreter from the send end.
1315 +
interpreters.channel_release(cid, send=True)
1316 +
# Send end should have no associated interpreters.
1317 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1318 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1319 +
self.assertEqual(len(send_interps), 0)
1320 +
self.assertEqual(len(recv_interps), 2)
1321 + 1322 +
# Release one of the subinterpreters from the receive end.
1323 +
_run_output(interp2, dedent(f"""
1324 +
import _xxsubinterpreters as _interpreters
1325 +
_interpreters.channel_release({cid})
1326 +
"""))
1327 +
# Receive end should have the released interpreter removed.
1328 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1329 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1330 +
self.assertEqual(len(send_interps), 0)
1331 +
self.assertEqual(recv_interps, [interp1])
1332 + 1333 +
def test_channel_list_interpreters_closed(self):
1334 +
"""Test listing channel interpreters with a closed channel."""
1335 +
interp0 = interpreters.get_main()
1336 +
interp1 = interpreters.create()
1337 +
cid = interpreters.channel_create()
1338 +
# Put something in the channel so that it's not empty.
1339 +
interpreters.channel_send(cid, "send")
1340 + 1341 +
# Check initial state.
1342 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1343 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1344 +
self.assertEqual(len(send_interps), 1)
1345 +
self.assertEqual(len(recv_interps), 0)
1346 + 1347 +
# Force close the channel.
1348 +
interpreters.channel_close(cid, force=True)
1349 +
# Both ends should raise an error.
1350 +
with self.assertRaises(interpreters.ChannelClosedError):
1351 +
interpreters.channel_list_interpreters(cid, send=True)
1352 +
with self.assertRaises(interpreters.ChannelClosedError):
1353 +
interpreters.channel_list_interpreters(cid, send=False)
1354 + 1355 +
def test_channel_list_interpreters_closed_send_end(self):
1356 +
"""Test listing channel interpreters with a channel's send end closed."""
1357 +
interp0 = interpreters.get_main()
1358 +
interp1 = interpreters.create()
1359 +
cid = interpreters.channel_create()
1360 +
# Put something in the channel so that it's not empty.
1361 +
interpreters.channel_send(cid, "send")
1362 + 1363 +
# Check initial state.
1364 +
send_interps = interpreters.channel_list_interpreters(cid, send=True)
1365 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1366 +
self.assertEqual(len(send_interps), 1)
1367 +
self.assertEqual(len(recv_interps), 0)
1368 + 1369 +
# Close the send end of the channel.
1370 +
interpreters.channel_close(cid, send=True)
1371 +
# Send end should raise an error.
1372 +
with self.assertRaises(interpreters.ChannelClosedError):
1373 +
interpreters.channel_list_interpreters(cid, send=True)
1374 +
# Receive end should not be closed (since channel is not empty).
1375 +
recv_interps = interpreters.channel_list_interpreters(cid, send=False)
1376 +
self.assertEqual(len(recv_interps), 0)
1377 + 1378 +
# Close the receive end of the channel from a subinterpreter.
1379 +
_run_output(interp1, dedent(f"""
1380 +
import _xxsubinterpreters as _interpreters
1381 +
_interpreters.channel_close({cid}, force=True)
1382 +
"""))
1383 +
# Both ends should raise an error.
1384 +
with self.assertRaises(interpreters.ChannelClosedError):
1385 +
interpreters.channel_list_interpreters(cid, send=True)
1386 +
with self.assertRaises(interpreters.ChannelClosedError):
1387 +
interpreters.channel_list_interpreters(cid, send=False)
1388 + 1210 1389
####################
1211 1390 1212 1391
def test_send_recv_main(self):
@@ -1540,6 +1719,23 @@ def test_close_used_multiple_times_by_single_user(self):
1540 1719
with self.assertRaises(interpreters.ChannelClosedError):
1541 1720
interpreters.channel_recv(cid)
1542 1721 1722 +
def test_channel_list_interpreters_invalid_channel(self):
1723 +
cid = interpreters.channel_create()
1724 +
# Test for invalid channel ID.
1725 +
with self.assertRaises(interpreters.ChannelNotFoundError):
1726 +
interpreters.channel_list_interpreters(1000, send=True)
1727 + 1728 +
interpreters.channel_close(cid)
1729 +
# Test for a channel that has been closed.
1730 +
with self.assertRaises(interpreters.ChannelClosedError):
1731 +
interpreters.channel_list_interpreters(cid, send=True)
1732 + 1733 +
def test_channel_list_interpreters_invalid_args(self):
1734 +
# Tests for invalid arguments passed to the API.
1735 +
cid = interpreters.channel_create()
1736 +
with self.assertRaises(TypeError):
1737 +
interpreters.channel_list_interpreters(cid)
1738 + 1543 1739 1544 1740
class ChannelReleaseTests(TestBase):
1545 1741 RetroSearch is an open source project built by @garambo | Open a GitHub Issue
Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo
HTML:
3.2
| Encoding:
UTF-8
| Version:
0.7.4