(self)
| 340 | self.assertIs(obj6, default) |
| 341 | |
| 342 | def test_send_buffer(self): |
| 343 | buf = bytearray(b'spamspamspam') |
| 344 | obj = None |
| 345 | rch, sch = channels.create() |
| 346 | |
| 347 | def f(): |
| 348 | nonlocal obj |
| 349 | while True: |
| 350 | try: |
| 351 | obj = rch.recv() |
| 352 | break |
| 353 | except channels.ChannelEmptyError: |
| 354 | time.sleep(0.1) |
| 355 | t = threading.Thread(target=f) |
| 356 | t.start() |
| 357 | |
| 358 | sch.send_buffer(buf) |
| 359 | t.join() |
| 360 | |
| 361 | self.assertIsNot(obj, buf) |
| 362 | self.assertIsInstance(obj, memoryview) |
| 363 | self.assertEqual(obj, buf) |
| 364 | |
| 365 | buf[4:8] = b'eggs' |
| 366 | self.assertEqual(obj, buf) |
| 367 | obj[4:8] = b'ham.' |
| 368 | self.assertEqual(obj, buf) |
| 369 | |
| 370 | def test_send_buffer_nowait(self): |
| 371 | buf = bytearray(b'spamspamspam') |
nothing calls this directly
no test coverage detected