diff options
author | Daniele Varrazzo <daniele.varrazzo@gmail.com> | 2017-03-16 00:24:47 +0000 |
---|---|---|
committer | Daniele Varrazzo <daniele.varrazzo@gmail.com> | 2017-03-16 00:55:20 +0000 |
commit | 3bfbd3a0a546b2b5bf30476dfa838cd05ddab442 (patch) | |
tree | 196e46db70b448e55a2cba0f156f14e27869644e /tests/test_sql.py | |
parent | 7187d6408a416fcb09aaa8b3039aba424320e9bd (diff) | |
download | psycopg2-3bfbd3a0a546b2b5bf30476dfa838cd05ddab442.tar.gz |
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.
Diffstat (limited to 'tests/test_sql.py')
-rwxr-xr-x | tests/test_sql.py | 26 |
1 files changed, 25 insertions, 1 deletions
diff --git a/tests/test_sql.py b/tests/test_sql.py index 16c4937..e35bf32 100755 --- a/tests/test_sql.py +++ b/tests/test_sql.py @@ -23,7 +23,9 @@ # License for more details. 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 from psycopg2 import sql @@ -149,6 +151,28 @@ class SqlFormatTests(ConnectingTestCase): self.assertEqual(cur.fetchall(), [(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): def test_class(self): |