(self)
| 748 | socket.create_connection((trusted_host, port), timeout=TIMEOUT).close() |
| 749 | |
| 750 | def test_with_statement(self): |
| 751 | self.client.quit() |
| 752 | |
| 753 | def is_client_connected(): |
| 754 | if self.client.sock is None: |
| 755 | return False |
| 756 | try: |
| 757 | self.client.sendcmd('noop') |
| 758 | except (OSError, EOFError): |
| 759 | return False |
| 760 | return True |
| 761 | |
| 762 | # base test |
| 763 | with ftplib.FTP(timeout=TIMEOUT) as self.client: |
| 764 | self.client.connect(self.server.host, self.server.port) |
| 765 | self.client.sendcmd('noop') |
| 766 | self.assertTrue(is_client_connected()) |
| 767 | self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit') |
| 768 | self.assertFalse(is_client_connected()) |
| 769 | |
| 770 | # QUIT sent inside the with block |
| 771 | with ftplib.FTP(timeout=TIMEOUT) as self.client: |
| 772 | self.client.connect(self.server.host, self.server.port) |
| 773 | self.client.sendcmd('noop') |
| 774 | self.client.quit() |
| 775 | self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit') |
| 776 | self.assertFalse(is_client_connected()) |
| 777 | |
| 778 | # force a wrong response code to be sent on QUIT: error_perm |
| 779 | # is expected and the connection is supposed to be closed |
| 780 | try: |
| 781 | with ftplib.FTP(timeout=TIMEOUT) as self.client: |
| 782 | self.client.connect(self.server.host, self.server.port) |
| 783 | self.client.sendcmd('noop') |
| 784 | self.server.handler_instance.next_response = '550 error on quit' |
| 785 | except ftplib.error_perm as err: |
| 786 | self.assertEqual(str(err), '550 error on quit') |
| 787 | else: |
| 788 | self.fail('Exception not raised') |
| 789 | # needed to give the threaded server some time to set the attribute |
| 790 | # which otherwise would still be == 'noop' |
| 791 | time.sleep(0.1) |
| 792 | self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit') |
| 793 | self.assertFalse(is_client_connected()) |
| 794 | |
| 795 | def test_source_address(self): |
| 796 | self.client.quit() |
nothing calls this directly
no test coverage detected