[AIRFLOW-2525] Fix PostgresHook.copy_expert to work with "COPY FROM"

airflow/hooks/postgres_hook.py

@@ -64,10 +64,11 @@ def copy_expert(self, sql, filename, open=open):
         Executes SQL using psycopg2 copy_expert method
         Necessary to execute COPY command without access to a superuser
         """
-        f = open(filename, 'w')
-        with closing(self.get_conn()) as conn:
-            with closing(conn.cursor()) as cur:
-                cur.copy_expert(sql, f)
+        with open(filename, 'w+') as f:
+            with closing(self.get_conn()) as conn:
+                with closing(conn.cursor()) as cur:
+                    cur.copy_expert(sql, f)
+                    conn.commit()
 
     @staticmethod
     def _serialize_cell(cell, conn):

tests/hooks/test_postgres_hook.py

@@ -43,17 +43,16 @@ def get_conn(self):
 
     def test_copy_expert(self):
         m = mock.mock_open(read_data='{"some": "json"}')
-        with mock.patch('airflow.hooks.postgres_hook.open', m, create=True) as m:
+        with mock.patch('airflow.hooks.postgres_hook.open', m):
             statement = "SQL"
             filename = "filename"
 
             self.cur.fetchall.return_value = None
-            f = m(filename, 'w')
-            def test_open(filename, mode):
-                return f
 
-            self.assertEqual(None, self.db_hook.copy_expert(statement, filename, open=test_open))
+            self.assertEqual(None, self.db_hook.copy_expert(statement, filename, open=m))
 
             self.conn.close.assert_called_once()
             self.cur.close.assert_called_once()
-            self.cur.copy_expert.assert_called_once_with(statement, f)
+            self.conn.commit.assert_called_once()
+            self.cur.copy_expert.assert_called_once_with(statement, m.return_value)
+            m.assert_called_once_with(filename, "w+")