Fix Connection class _copy_in private method

Closes: #555

Author: ABCDeath

asyncpg/connection.py

@@ -813,7 +813,7 @@ async def _copy_in(self, copy_stmt, source, timeout):
 
         if path is not None:
             # a path
-            f = await run_in_executor(None, open, path, 'wb')
+            f = await run_in_executor(None, open, path, 'rb')
             opened_by_us = True
         elif hasattr(source, 'read'):
             # file-like

tests/test_copy.py

@@ -8,6 +8,7 @@
 import asyncio
 import datetime
 import io
+import os
 import tempfile
 
 import asyncpg
@@ -582,6 +583,48 @@ async def __anext__(self):
         finally:
             await self.con.execute('DROP TABLE copytab')
 
+    async def test_copy_to_table_from_file_path(self):
+        await self.con.execute('''
+            CREATE TABLE copytab(a text, "b~" text, i int);
+        ''')
+
+        f = tempfile.NamedTemporaryFile(delete=False)
+        try:
+            f.write(
+                '\n'.join([
+                    'a1\tb1\t1',
+                    'a2\tb2\t2',
+                    'a3\tb3\t3',
+                    'a4\tb4\t4',
+                    'a5\tb5\t5',
+                    '*\t\\N\t\\N',
+                    ''
+                ]).encode('utf-8')
+            )
+            f.close()
+
+            res = await self.con.copy_to_table('copytab', source=f.name)
+            self.assertEqual(res, 'COPY 6')
+
+            output = await self.con.fetch("""
+                SELECT * FROM copytab ORDER BY a
+            """)
+            self.assertEqual(
+                output,
+                [
+                    ('*', None, None),
+                    ('a1', 'b1', 1),
+                    ('a2', 'b2', 2),
+                    ('a3', 'b3', 3),
+                    ('a4', 'b4', 4),
+                    ('a5', 'b5', 5),
+                ]
+            )
+
+        finally:
+            await self.con.execute('DROP TABLE public.copytab')
+            os.unlink(f.name)
+
     async def test_copy_records_to_table_1(self):
         await self.con.execute('''
             CREATE TABLE copytab(a text, b int, c timestamptz);