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.

2 files changed, 26 insertions, 2 deletions

diff --git a/lib/sql.py b/lib/sql.py
index e7d42e6..ffd9e27 100644
--- a/lib/sql.py
+++ b/lib/sql.py
@@ -1,7 +1,7 @@
 """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>
 #
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):