mirror of
https://github.com/psycopg/psycopg2.git
synced 2024-11-14 04:56:33 +03:00
Added test to verify sql objects work with copy_expert()
I'll be honest: I lucked out, I didn't think about this combination. But maybe sheer luck, maybe using common code paths, it just works. Let's make it stays so.
This commit is contained in:
parent
7187d6408a
commit
3bfbd3a0a5
|
@ -1,7 +1,7 @@
|
||||||
"""SQL composition utility module
|
"""SQL composition utility module
|
||||||
"""
|
"""
|
||||||
|
|
||||||
# psycopg/sql.py - Implementation of the JSON adaptation objects
|
# psycopg/sql.py - SQL composition utility module
|
||||||
#
|
#
|
||||||
# Copyright (C) 2016 Daniele Varrazzo <daniele.varrazzo@gmail.com>
|
# Copyright (C) 2016 Daniele Varrazzo <daniele.varrazzo@gmail.com>
|
||||||
#
|
#
|
||||||
|
|
|
@ -23,7 +23,9 @@
|
||||||
# License for more details.
|
# License for more details.
|
||||||
|
|
||||||
import datetime as dt
|
import datetime as dt
|
||||||
from testutils import unittest, ConnectingTestCase, skip_before_python
|
from cStringIO import StringIO
|
||||||
|
from testutils import (unittest, ConnectingTestCase,
|
||||||
|
skip_before_postgres, skip_before_python, skip_copy_if_green)
|
||||||
|
|
||||||
import psycopg2
|
import psycopg2
|
||||||
from psycopg2 import sql
|
from psycopg2 import sql
|
||||||
|
@ -149,6 +151,28 @@ class SqlFormatTests(ConnectingTestCase):
|
||||||
self.assertEqual(cur.fetchall(),
|
self.assertEqual(cur.fetchall(),
|
||||||
[(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])
|
[(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])
|
||||||
|
|
||||||
|
@skip_copy_if_green
|
||||||
|
@skip_before_postgres(8, 2)
|
||||||
|
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')
|
||||||
|
|
||||||
|
|
||||||
class IdentifierTests(ConnectingTestCase):
|
class IdentifierTests(ConnectingTestCase):
|
||||||
def test_class(self):
|
def test_class(self):
|
||||||
|
|
Loading…
Reference in New Issue
Block a user