test_sql.py 文件源码

def test_copy(self):
        cur = self.conn.cursor()
        cur.execute("""
            create table test_compose (
                id serial primary key,
                foo text, bar text, "ba'z" text)
            """)

        s = StringIO("10\ta\tb\tc\n20\td\te\tf\n")
        cur.copy_expert(
            sql.SQL("copy {t} (id, foo, bar, {f}) from stdin").format(
                t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s)

        s1 = StringIO()
        cur.copy_expert(
            sql.SQL("copy (select {f} from {t} order by id) to stdout").format(
                t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s1)
        s1.seek(0)
        self.assertEqual(s1.read(), 'c\nf\n')