(self)
| 122 | queues.create('1') |
| 123 | |
| 124 | def test_shareable(self): |
| 125 | queue1 = queues.create() |
| 126 | |
| 127 | interp = interpreters.create() |
| 128 | interp.exec(dedent(f""" |
| 129 | from concurrent.interpreters import _queues as queues |
| 130 | queue1 = queues.Queue({queue1.id}) |
| 131 | """)); |
| 132 | |
| 133 | with self.subTest('same interpreter'): |
| 134 | queue2 = queues.create() |
| 135 | queue1.put(queue2) |
| 136 | queue3 = queue1.get() |
| 137 | self.assertIs(queue3, queue2) |
| 138 | |
| 139 | with self.subTest('from current interpreter'): |
| 140 | queue4 = queues.create() |
| 141 | queue1.put(queue4) |
| 142 | out = _run_output(interp, dedent(""" |
| 143 | queue4 = queue1.get() |
| 144 | print(queue4.id) |
| 145 | """)) |
| 146 | qid = int(out) |
| 147 | self.assertEqual(qid, queue4.id) |
| 148 | |
| 149 | with self.subTest('from subinterpreter'): |
| 150 | out = _run_output(interp, dedent(""" |
| 151 | queue5 = queues.create() |
| 152 | queue1.put(queue5) |
| 153 | print(queue5.id) |
| 154 | """)) |
| 155 | qid = int(out) |
| 156 | queue5 = queue1.get() |
| 157 | self.assertEqual(queue5.id, qid) |
| 158 | |
| 159 | def test_id_type(self): |
| 160 | queue = queues.create() |
nothing calls this directly
no test coverage detected