test_connection.py 文件源码

def test_replication_not_supported(self):
        conn = self.repl_connect()
        if conn is None: return
        cur = conn.cursor()
        f = StringIO()
        self.assertRaises(psycopg2.NotSupportedError,
            cur.copy_expert, "START_REPLICATION 0/0", f)