(self)
| 565 | pass |
| 566 | |
| 567 | def test_idle_iter(self): |
| 568 | client, _ = self._setup(IdleCmdHandler) |
| 569 | client.login('user', 'pass') |
| 570 | with client.idle() as idler: |
| 571 | # iteration should include response between 'IDLE' & '+ idling' |
| 572 | response = next(idler) |
| 573 | self.assertEqual(response, ('EXISTS', [b'0'])) |
| 574 | # iteration should produce responses |
| 575 | response = next(idler) |
| 576 | self.assertEqual(response, ('EXISTS', [b'2'])) |
| 577 | # fragmented response (with literal string) should arrive whole |
| 578 | expected_fetch_data = [ |
| 579 | (b'1 (BODY[HEADER.FIELDS (DATE)] {41}', |
| 580 | b'Date: Fri, 06 Dec 2024 06:00:00 +0000\r\n\r\n'), |
| 581 | b')'] |
| 582 | typ, data = next(idler) |
| 583 | self.assertEqual(typ, 'FETCH') |
| 584 | self.assertEqual(data, expected_fetch_data) |
| 585 | # response after a fragmented one should arrive separately |
| 586 | response = next(idler) |
| 587 | self.assertEqual(response, ('EXISTS', [b'3'])) |
| 588 | # iteration should have consumed untagged responses |
| 589 | _, data = client.response('EXISTS') |
| 590 | self.assertEqual(data, [None]) |
| 591 | # responses not iterated should be available after idle |
| 592 | _, data = client.response('RECENT') |
| 593 | self.assertEqual(data[0], b'1') |
| 594 | # responses received after 'DONE' should be available after idle |
| 595 | self.assertEqual(data[1], b'9') |
| 596 | |
| 597 | def test_idle_burst(self): |
| 598 | client, _ = self._setup(IdleCmdHandler) |
nothing calls this directly
no test coverage detected