test_postgres_hook.py 文件源码

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:
            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.conn.close.assert_called_once()
            self.cur.close.assert_called_once()
            self.cur.copy_expert.assert_called_once_with(statement, f)