test_copy.py 文件源码

def test_copy_rowcount_error(self):
        curs = self.conn.cursor()

        curs.execute("insert into tcopy (data) values ('fff')")
        self.assertEqual(curs.rowcount, 1)

        self.assertRaises(psycopg2.DataError,
            curs.copy_from, StringIO('aaa\nbbb\nccc\n'), 'tcopy')
        self.assertEqual(curs.rowcount, -1)