test_connection.py 文件源码

def test_defer_connect(self):
        import socket
        for db in self.databases:
            d = db.copy()
            try:
                sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                sock.connect(d['unix_socket'])
            except KeyError:
                sock = socket.create_connection(
                                (d.get('host', 'localhost'), d.get('port', 3306)))
            for k in ['unix_socket', 'host', 'port']:
                try:
                    del d[k]
                except KeyError:
                    pass

            c = pymysql.connect(defer_connect=True, **d)
            self.assertFalse(c.open)
            c.connect(sock)
            c.close()
            sock.close()