We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
RemoteFileSystem uses wrong path separator
The RemoteFileSystem uses the client path  separators instead of the remote system path separators.

That means that when trying to write to a Linux RemoteTarget from a Windows Client, the path will be broken (backslashes instead of slashes as path separators, leading to wrong path names).


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `luigi/contrib/ssh.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 #
3 # Copyright 2012-2015 Spotify AB
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 # http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #
17 """
18 Light-weight remote execution library and utilities.
19 
20 There are some examples in the unittest, but
21 I added another more luigi-specific in the examples directory (examples/ssh_remote_execution.py
22 
23 :class:`RemoteContext` is meant to provide functionality similar to that of the standard library subprocess module,
24 but where the commands executed are run on a remote machine instead,
25 without the user having to think about prefixing everything with "ssh" and credentials etc.
26 
27 Using this mini library (which is just a convenience wrapper for subprocess),
28 :class:`RemoteTarget` is created to let you stream data from a remotely stored file using
29 the luigi :class:`~luigi.target.FileSystemTarget` semantics.
30 
31 As a bonus, :class:`RemoteContext` also provides a really cool feature that let's you
32 set up ssh tunnels super easily using a python context manager (there is an example
33 in the integration part of unittests).
34 
35 This can be super convenient when you want secure communication using a non-secure
36 protocol or circumvent firewalls (as long as they are open for ssh traffic).
37 """
38 
39 import contextlib
40 import os
41 import random
42 import subprocess
43 
44 import luigi
45 import luigi.format
46 import luigi.target
47 
48 
49 class RemoteContext(object):
50 
51     def __init__(self, host, username=None, key_file=None, connect_timeout=None, port=None):
52         self.host = host
53         self.username = username
54         self.key_file = key_file
55         self.connect_timeout = connect_timeout
56         self.port = port
57 
58     def __repr__(self):
59         return '%s(%r, %r, %r, %r, %r)' % (
60             type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout, self.port)
61 
62     def __eq__(self, other):
63         return repr(self) == repr(other)
64 
65     def __hash__(self):
66         return hash(repr(self))
67 
68     def _host_ref(self):
69         if self.username:
70             return "{0}@{1}".format(self.username, self.host)
71         else:
72             return self.host
73 
74     def _prepare_cmd(self, cmd):
75         connection_cmd = ["ssh", self._host_ref(),
76                           "-S", "none",  # disable ControlMaster since it causes all sorts of weird behaviour with subprocesses...
77                           "-o", "BatchMode=yes",  # no password prompts etc
78                           ]
79         if self.port:
80             connection_cmd.extend(["-p", self.port])
81 
82         if self.connect_timeout is not None:
83             connection_cmd += ['-o', 'ConnectTimeout=%d' % self.connect_timeout]
84 
85         if self.key_file:
86             connection_cmd.extend(["-i", self.key_file])
87         return connection_cmd + cmd
88 
89     def Popen(self, cmd, **kwargs):
90         """
91         Remote Popen.
92         """
93         prefixed_cmd = self._prepare_cmd(cmd)
94         return subprocess.Popen(prefixed_cmd, **kwargs)
95 
96     def check_output(self, cmd):
97         """
98         Execute a shell command remotely and return the output.
99 
100         Simplified version of Popen when you only want the output as a string and detect any errors.
101         """
102         p = self.Popen(cmd, stdout=subprocess.PIPE)
103         output, _ = p.communicate()
104         if p.returncode != 0:
105             raise subprocess.CalledProcessError(p.returncode, cmd)
106         return output
107 
108     @contextlib.contextmanager
109     def tunnel(self, local_port, remote_port=None, remote_host="localhost"):
110         """
111         Open a tunnel between localhost:local_port and remote_host:remote_port via the host specified by this context.
112 
113         Remember to close() the returned "tunnel" object in order to clean up
114         after yourself when you are done with the tunnel.
115         """
116         tunnel_host = "{0}:{1}:{2}".format(local_port, remote_host, remote_port)
117         proc = self.Popen(
118             # cat so we can shut down gracefully by closing stdin
119             ["-L", tunnel_host, "echo -n ready && cat"],
120             stdin=subprocess.PIPE,
121             stdout=subprocess.PIPE,
122         )
123         # make sure to get the data so we know the connection is established
124         ready = proc.stdout.read(5)
125         assert ready == b"ready", "Didn't get ready from remote echo"
126         yield  # user code executed here
127         proc.communicate()
128         assert proc.returncode == 0, "Tunnel process did an unclean exit (returncode %s)" % (proc.returncode,)
129 
130 
131 class RemoteFileSystem(luigi.target.FileSystem):
132 
133     def __init__(self, host, username=None, key_file=None, port=None):
134         self.remote_context = RemoteContext(host, username, key_file, port)
135 
136     def exists(self, path):
137         """
138         Return `True` if file or directory at `path` exist, False otherwise.
139         """
140         try:
141             self.remote_context.check_output(["test", "-e", path])
142         except subprocess.CalledProcessError as e:
143             if e.returncode == 1:
144                 return False
145             else:
146                 raise
147         return True
148 
149     def remove(self, path, recursive=True):
150         """
151         Remove file or directory at location `path`.
152         """
153         if recursive:
154             cmd = ["rm", "-r", path]
155         else:
156             cmd = ["rm", path]
157 
158         self.remote_context.check_output(cmd)
159 
160     def _scp(self, src, dest):
161         cmd = ["scp", "-q", "-B", "-C", "-o", "ControlMaster=no"]
162         if self.remote_context.key_file:
163             cmd.extend(["-i", self.remote_context.key_file])
164         if self.remote_context.port:
165             cmd.extend(["-P", self.remote_context.port])
166         if os.path.isdir(src):
167             cmd.extend(["-r"])
168         cmd.extend([src, dest])
169         p = subprocess.Popen(cmd)
170         output, _ = p.communicate()
171         if p.returncode != 0:
172             raise subprocess.CalledProcessError(p.returncode, cmd)
173 
174     def put(self, local_path, path):
175         # create parent folder if not exists
176         normpath = os.path.normpath(path)
177         folder = os.path.dirname(normpath)
178         if folder and not self.exists(folder):
179             self.remote_context.check_output(['mkdir', '-p', folder])
180 
181         tmp_path = path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)
182         self._scp(local_path, "%s:%s" % (self.remote_context._host_ref(), tmp_path))
183         self.remote_context.check_output(['mv', tmp_path, path])
184 
185     def get(self, path, local_path):
186         # Create folder if it does not exist
187         normpath = os.path.normpath(local_path)
188         folder = os.path.dirname(normpath)
189         if folder and not os.path.exists(folder):
190             os.makedirs(folder)
191 
192         tmp_local_path = local_path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)
193         self._scp("%s:%s" % (self.remote_context._host_ref(), path), tmp_local_path)
194         os.rename(tmp_local_path, local_path)
195 
196 
197 class AtomicRemoteFileWriter(luigi.format.OutputPipeProcessWrapper):
198 
199     def __init__(self, fs, path):
200         self._fs = fs
201         self.path = path
202 
203         # create parent folder if not exists
204         normpath = os.path.normpath(self.path)
205         folder = os.path.dirname(normpath)
206         if folder and not self.fs.exists(folder):
207             self.fs.remote_context.check_output(['mkdir', '-p', folder])
208 
209         self.__tmp_path = self.path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)
210         super(AtomicRemoteFileWriter, self).__init__(
211             self.fs.remote_context._prepare_cmd(['cat', '>', self.__tmp_path]))
212 
213     def __del__(self):
214         super(AtomicRemoteFileWriter, self).__del__()
215         if self.fs.exists(self.__tmp_path):
216             self.fs.remote_context.check_output(['rm', self.__tmp_path])
217 
218     def close(self):
219         super(AtomicRemoteFileWriter, self).close()
220         self.fs.remote_context.check_output(['mv', self.__tmp_path, self.path])
221 
222     @property
223     def tmp_path(self):
224         return self.__tmp_path
225 
226     @property
227     def fs(self):
228         return self._fs
229 
230 
231 class RemoteTarget(luigi.target.FileSystemTarget):
232     """
233     Target used for reading from remote files.
234 
235     The target is implemented using ssh commands streaming data over the network.
236     """
237 
238     def __init__(self, path, host, format=None, username=None, key_file=None, port=None):
239         super(RemoteTarget, self).__init__(path)
240         if format is None:
241             format = luigi.format.get_default_format()
242         self.format = format
243         self._fs = RemoteFileSystem(host, username, key_file, port)
244 
245     @property
246     def fs(self):
247         return self._fs
248 
249     def open(self, mode='r'):
250         if mode == 'w':
251             file_writer = AtomicRemoteFileWriter(self.fs, self.path)
252             if self.format:
253                 return self.format.pipe_writer(file_writer)
254             else:
255                 return file_writer
256         elif mode == 'r':
257             file_reader = luigi.format.InputPipeProcessWrapper(
258                 self.fs.remote_context._prepare_cmd(["cat", self.path]))
259             if self.format:
260                 return self.format.pipe_reader(file_reader)
261             else:
262                 return file_reader
263         else:
264             raise Exception("mode must be r/w")
265 
266     def put(self, local_path):
267         self.fs.put(local_path, self.path)
268 
269     def get(self, local_path):
270         self.fs.get(self.path, local_path)
271
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/luigi/contrib/ssh.py b/luigi/contrib/ssh.py
--- a/luigi/contrib/ssh.py
+++ b/luigi/contrib/ssh.py
@@ -40,6 +40,7 @@
 import os
 import random
 import subprocess
+import posixpath
 
 import luigi
 import luigi.format
@@ -173,7 +174,7 @@
 
     def put(self, local_path, path):
         # create parent folder if not exists
-        normpath = os.path.normpath(path)
+        normpath = posixpath.normpath(path)
         folder = os.path.dirname(normpath)
         if folder and not self.exists(folder):
             self.remote_context.check_output(['mkdir', '-p', folder])

{"golden_diff": "diff --git a/luigi/contrib/ssh.py b/luigi/contrib/ssh.py\n--- a/luigi/contrib/ssh.py\n+++ b/luigi/contrib/ssh.py\n@@ -40,6 +40,7 @@\n import os\n import random\n import subprocess\n+import posixpath\n \n import luigi\n import luigi.format\n@@ -173,7 +174,7 @@\n \n     def put(self, local_path, path):\n         # create parent folder if not exists\n-        normpath = os.path.normpath(path)\n+        normpath = posixpath.normpath(path)\n         folder = os.path.dirname(normpath)\n         if folder and not self.exists(folder):\n             self.remote_context.check_output(['mkdir', '-p', folder])\n", "issue": "RemoteFileSystem uses wrong path separator\nThe RemoteFileSystem uses the client path  separators instead of the remote system path separators.\n\nThat means that when trying to write to a Linux RemoteTarget from a Windows Client, the path will be broken (backslashes instead of slashes as path separators, leading to wrong path names).\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\"\"\"\nLight-weight remote execution library and utilities.\n\nThere are some examples in the unittest, but\nI added another more luigi-specific in the examples directory (examples/ssh_remote_execution.py\n\n:class:`RemoteContext` is meant to provide functionality similar to that of the standard library subprocess module,\nbut where the commands executed are run on a remote machine instead,\nwithout the user having to think about prefixing everything with \"ssh\" and credentials etc.\n\nUsing this mini library (which is just a convenience wrapper for subprocess),\n:class:`RemoteTarget` is created to let you stream data from a remotely stored file using\nthe luigi :class:`~luigi.target.FileSystemTarget` semantics.\n\nAs a bonus, :class:`RemoteContext` also provides a really cool feature that let's you\nset up ssh tunnels super easily using a python context manager (there is an example\nin the integration part of unittests).\n\nThis can be super convenient when you want secure communication using a non-secure\nprotocol or circumvent firewalls (as long as they are open for ssh traffic).\n\"\"\"\n\nimport contextlib\nimport os\nimport random\nimport subprocess\n\nimport luigi\nimport luigi.format\nimport luigi.target\n\n\nclass RemoteContext(object):\n\n    def __init__(self, host, username=None, key_file=None, connect_timeout=None, port=None):\n        self.host = host\n        self.username = username\n        self.key_file = key_file\n        self.connect_timeout = connect_timeout\n        self.port = port\n\n    def __repr__(self):\n        return '%s(%r, %r, %r, %r, %r)' % (\n            type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout, self.port)\n\n    def __eq__(self, other):\n        return repr(self) == repr(other)\n\n    def __hash__(self):\n        return hash(repr(self))\n\n    def _host_ref(self):\n        if self.username:\n            return \"{0}@{1}\".format(self.username, self.host)\n        else:\n            return self.host\n\n    def _prepare_cmd(self, cmd):\n        connection_cmd = [\"ssh\", self._host_ref(),\n                          \"-S\", \"none\",  # disable ControlMaster since it causes all sorts of weird behaviour with subprocesses...\n                          \"-o\", \"BatchMode=yes\",  # no password prompts etc\n                          ]\n        if self.port:\n            connection_cmd.extend([\"-p\", self.port])\n\n        if self.connect_timeout is not None:\n            connection_cmd += ['-o', 'ConnectTimeout=%d' % self.connect_timeout]\n\n        if self.key_file:\n            connection_cmd.extend([\"-i\", self.key_file])\n        return connection_cmd + cmd\n\n    def Popen(self, cmd, **kwargs):\n        \"\"\"\n        Remote Popen.\n        \"\"\"\n        prefixed_cmd = self._prepare_cmd(cmd)\n        return subprocess.Popen(prefixed_cmd, **kwargs)\n\n    def check_output(self, cmd):\n        \"\"\"\n        Execute a shell command remotely and return the output.\n\n        Simplified version of Popen when you only want the output as a string and detect any errors.\n        \"\"\"\n        p = self.Popen(cmd, stdout=subprocess.PIPE)\n        output, _ = p.communicate()\n        if p.returncode != 0:\n            raise subprocess.CalledProcessError(p.returncode, cmd)\n        return output\n\n    @contextlib.contextmanager\n    def tunnel(self, local_port, remote_port=None, remote_host=\"localhost\"):\n        \"\"\"\n        Open a tunnel between localhost:local_port and remote_host:remote_port via the host specified by this context.\n\n        Remember to close() the returned \"tunnel\" object in order to clean up\n        after yourself when you are done with the tunnel.\n        \"\"\"\n        tunnel_host = \"{0}:{1}:{2}\".format(local_port, remote_host, remote_port)\n        proc = self.Popen(\n            # cat so we can shut down gracefully by closing stdin\n            [\"-L\", tunnel_host, \"echo -n ready && cat\"],\n            stdin=subprocess.PIPE,\n            stdout=subprocess.PIPE,\n        )\n        # make sure to get the data so we know the connection is established\n        ready = proc.stdout.read(5)\n        assert ready == b\"ready\", \"Didn't get ready from remote echo\"\n        yield  # user code executed here\n        proc.communicate()\n        assert proc.returncode == 0, \"Tunnel process did an unclean exit (returncode %s)\" % (proc.returncode,)\n\n\nclass RemoteFileSystem(luigi.target.FileSystem):\n\n    def __init__(self, host, username=None, key_file=None, port=None):\n        self.remote_context = RemoteContext(host, username, key_file, port)\n\n    def exists(self, path):\n        \"\"\"\n        Return `True` if file or directory at `path` exist, False otherwise.\n        \"\"\"\n        try:\n            self.remote_context.check_output([\"test\", \"-e\", path])\n        except subprocess.CalledProcessError as e:\n            if e.returncode == 1:\n                return False\n            else:\n                raise\n        return True\n\n    def remove(self, path, recursive=True):\n        \"\"\"\n        Remove file or directory at location `path`.\n        \"\"\"\n        if recursive:\n            cmd = [\"rm\", \"-r\", path]\n        else:\n            cmd = [\"rm\", path]\n\n        self.remote_context.check_output(cmd)\n\n    def _scp(self, src, dest):\n        cmd = [\"scp\", \"-q\", \"-B\", \"-C\", \"-o\", \"ControlMaster=no\"]\n        if self.remote_context.key_file:\n            cmd.extend([\"-i\", self.remote_context.key_file])\n        if self.remote_context.port:\n            cmd.extend([\"-P\", self.remote_context.port])\n        if os.path.isdir(src):\n            cmd.extend([\"-r\"])\n        cmd.extend([src, dest])\n        p = subprocess.Popen(cmd)\n        output, _ = p.communicate()\n        if p.returncode != 0:\n            raise subprocess.CalledProcessError(p.returncode, cmd)\n\n    def put(self, local_path, path):\n        # create parent folder if not exists\n        normpath = os.path.normpath(path)\n        folder = os.path.dirname(normpath)\n        if folder and not self.exists(folder):\n            self.remote_context.check_output(['mkdir', '-p', folder])\n\n        tmp_path = path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        self._scp(local_path, \"%s:%s\" % (self.remote_context._host_ref(), tmp_path))\n        self.remote_context.check_output(['mv', tmp_path, path])\n\n    def get(self, path, local_path):\n        # Create folder if it does not exist\n        normpath = os.path.normpath(local_path)\n        folder = os.path.dirname(normpath)\n        if folder and not os.path.exists(folder):\n            os.makedirs(folder)\n\n        tmp_local_path = local_path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        self._scp(\"%s:%s\" % (self.remote_context._host_ref(), path), tmp_local_path)\n        os.rename(tmp_local_path, local_path)\n\n\nclass AtomicRemoteFileWriter(luigi.format.OutputPipeProcessWrapper):\n\n    def __init__(self, fs, path):\n        self._fs = fs\n        self.path = path\n\n        # create parent folder if not exists\n        normpath = os.path.normpath(self.path)\n        folder = os.path.dirname(normpath)\n        if folder and not self.fs.exists(folder):\n            self.fs.remote_context.check_output(['mkdir', '-p', folder])\n\n        self.__tmp_path = self.path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        super(AtomicRemoteFileWriter, self).__init__(\n            self.fs.remote_context._prepare_cmd(['cat', '>', self.__tmp_path]))\n\n    def __del__(self):\n        super(AtomicRemoteFileWriter, self).__del__()\n        if self.fs.exists(self.__tmp_path):\n            self.fs.remote_context.check_output(['rm', self.__tmp_path])\n\n    def close(self):\n        super(AtomicRemoteFileWriter, self).close()\n        self.fs.remote_context.check_output(['mv', self.__tmp_path, self.path])\n\n    @property\n    def tmp_path(self):\n        return self.__tmp_path\n\n    @property\n    def fs(self):\n        return self._fs\n\n\nclass RemoteTarget(luigi.target.FileSystemTarget):\n    \"\"\"\n    Target used for reading from remote files.\n\n    The target is implemented using ssh commands streaming data over the network.\n    \"\"\"\n\n    def __init__(self, path, host, format=None, username=None, key_file=None, port=None):\n        super(RemoteTarget, self).__init__(path)\n        if format is None:\n            format = luigi.format.get_default_format()\n        self.format = format\n        self._fs = RemoteFileSystem(host, username, key_file, port)\n\n    @property\n    def fs(self):\n        return self._fs\n\n    def open(self, mode='r'):\n        if mode == 'w':\n            file_writer = AtomicRemoteFileWriter(self.fs, self.path)\n            if self.format:\n                return self.format.pipe_writer(file_writer)\n            else:\n                return file_writer\n        elif mode == 'r':\n            file_reader = luigi.format.InputPipeProcessWrapper(\n                self.fs.remote_context._prepare_cmd([\"cat\", self.path]))\n            if self.format:\n                return self.format.pipe_reader(file_reader)\n            else:\n                return file_reader\n        else:\n            raise Exception(\"mode must be r/w\")\n\n    def put(self, local_path):\n        self.fs.put(local_path, self.path)\n\n    def get(self, local_path):\n        self.fs.get(self.path, local_path)\n", "path": "luigi/contrib/ssh.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\"\"\"\nLight-weight remote execution library and utilities.\n\nThere are some examples in the unittest, but\nI added another more luigi-specific in the examples directory (examples/ssh_remote_execution.py\n\n:class:`RemoteContext` is meant to provide functionality similar to that of the standard library subprocess module,\nbut where the commands executed are run on a remote machine instead,\nwithout the user having to think about prefixing everything with \"ssh\" and credentials etc.\n\nUsing this mini library (which is just a convenience wrapper for subprocess),\n:class:`RemoteTarget` is created to let you stream data from a remotely stored file using\nthe luigi :class:`~luigi.target.FileSystemTarget` semantics.\n\nAs a bonus, :class:`RemoteContext` also provides a really cool feature that let's you\nset up ssh tunnels super easily using a python context manager (there is an example\nin the integration part of unittests).\n\nThis can be super convenient when you want secure communication using a non-secure\nprotocol or circumvent firewalls (as long as they are open for ssh traffic).\n\"\"\"\n\nimport contextlib\nimport os\nimport random\nimport subprocess\nimport posixpath\n\nimport luigi\nimport luigi.format\nimport luigi.target\n\n\nclass RemoteContext(object):\n\n    def __init__(self, host, username=None, key_file=None, connect_timeout=None, port=None):\n        self.host = host\n        self.username = username\n        self.key_file = key_file\n        self.connect_timeout = connect_timeout\n        self.port = port\n\n    def __repr__(self):\n        return '%s(%r, %r, %r, %r, %r)' % (\n            type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout, self.port)\n\n    def __eq__(self, other):\n        return repr(self) == repr(other)\n\n    def __hash__(self):\n        return hash(repr(self))\n\n    def _host_ref(self):\n        if self.username:\n            return \"{0}@{1}\".format(self.username, self.host)\n        else:\n            return self.host\n\n    def _prepare_cmd(self, cmd):\n        connection_cmd = [\"ssh\", self._host_ref(),\n                          \"-S\", \"none\",  # disable ControlMaster since it causes all sorts of weird behaviour with subprocesses...\n                          \"-o\", \"BatchMode=yes\",  # no password prompts etc\n                          ]\n        if self.port:\n            connection_cmd.extend([\"-p\", self.port])\n\n        if self.connect_timeout is not None:\n            connection_cmd += ['-o', 'ConnectTimeout=%d' % self.connect_timeout]\n\n        if self.key_file:\n            connection_cmd.extend([\"-i\", self.key_file])\n        return connection_cmd + cmd\n\n    def Popen(self, cmd, **kwargs):\n        \"\"\"\n        Remote Popen.\n        \"\"\"\n        prefixed_cmd = self._prepare_cmd(cmd)\n        return subprocess.Popen(prefixed_cmd, **kwargs)\n\n    def check_output(self, cmd):\n        \"\"\"\n        Execute a shell command remotely and return the output.\n\n        Simplified version of Popen when you only want the output as a string and detect any errors.\n        \"\"\"\n        p = self.Popen(cmd, stdout=subprocess.PIPE)\n        output, _ = p.communicate()\n        if p.returncode != 0:\n            raise subprocess.CalledProcessError(p.returncode, cmd)\n        return output\n\n    @contextlib.contextmanager\n    def tunnel(self, local_port, remote_port=None, remote_host=\"localhost\"):\n        \"\"\"\n        Open a tunnel between localhost:local_port and remote_host:remote_port via the host specified by this context.\n\n        Remember to close() the returned \"tunnel\" object in order to clean up\n        after yourself when you are done with the tunnel.\n        \"\"\"\n        tunnel_host = \"{0}:{1}:{2}\".format(local_port, remote_host, remote_port)\n        proc = self.Popen(\n            # cat so we can shut down gracefully by closing stdin\n            [\"-L\", tunnel_host, \"echo -n ready && cat\"],\n            stdin=subprocess.PIPE,\n            stdout=subprocess.PIPE,\n        )\n        # make sure to get the data so we know the connection is established\n        ready = proc.stdout.read(5)\n        assert ready == b\"ready\", \"Didn't get ready from remote echo\"\n        yield  # user code executed here\n        proc.communicate()\n        assert proc.returncode == 0, \"Tunnel process did an unclean exit (returncode %s)\" % (proc.returncode,)\n\n\nclass RemoteFileSystem(luigi.target.FileSystem):\n\n    def __init__(self, host, username=None, key_file=None, port=None):\n        self.remote_context = RemoteContext(host, username, key_file, port)\n\n    def exists(self, path):\n        \"\"\"\n        Return `True` if file or directory at `path` exist, False otherwise.\n        \"\"\"\n        try:\n            self.remote_context.check_output([\"test\", \"-e\", path])\n        except subprocess.CalledProcessError as e:\n            if e.returncode == 1:\n                return False\n            else:\n                raise\n        return True\n\n    def remove(self, path, recursive=True):\n        \"\"\"\n        Remove file or directory at location `path`.\n        \"\"\"\n        if recursive:\n            cmd = [\"rm\", \"-r\", path]\n        else:\n            cmd = [\"rm\", path]\n\n        self.remote_context.check_output(cmd)\n\n    def _scp(self, src, dest):\n        cmd = [\"scp\", \"-q\", \"-B\", \"-C\", \"-o\", \"ControlMaster=no\"]\n        if self.remote_context.key_file:\n            cmd.extend([\"-i\", self.remote_context.key_file])\n        if self.remote_context.port:\n            cmd.extend([\"-P\", self.remote_context.port])\n        if os.path.isdir(src):\n            cmd.extend([\"-r\"])\n        cmd.extend([src, dest])\n        p = subprocess.Popen(cmd)\n        output, _ = p.communicate()\n        if p.returncode != 0:\n            raise subprocess.CalledProcessError(p.returncode, cmd)\n\n    def put(self, local_path, path):\n        # create parent folder if not exists\n        normpath = posixpath.normpath(path)\n        folder = os.path.dirname(normpath)\n        if folder and not self.exists(folder):\n            self.remote_context.check_output(['mkdir', '-p', folder])\n\n        tmp_path = path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        self._scp(local_path, \"%s:%s\" % (self.remote_context._host_ref(), tmp_path))\n        self.remote_context.check_output(['mv', tmp_path, path])\n\n    def get(self, path, local_path):\n        # Create folder if it does not exist\n        normpath = os.path.normpath(local_path)\n        folder = os.path.dirname(normpath)\n        if folder and not os.path.exists(folder):\n            os.makedirs(folder)\n\n        tmp_local_path = local_path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        self._scp(\"%s:%s\" % (self.remote_context._host_ref(), path), tmp_local_path)\n        os.rename(tmp_local_path, local_path)\n\n\nclass AtomicRemoteFileWriter(luigi.format.OutputPipeProcessWrapper):\n\n    def __init__(self, fs, path):\n        self._fs = fs\n        self.path = path\n\n        # create parent folder if not exists\n        normpath = os.path.normpath(self.path)\n        folder = os.path.dirname(normpath)\n        if folder and not self.fs.exists(folder):\n            self.fs.remote_context.check_output(['mkdir', '-p', folder])\n\n        self.__tmp_path = self.path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        super(AtomicRemoteFileWriter, self).__init__(\n            self.fs.remote_context._prepare_cmd(['cat', '>', self.__tmp_path]))\n\n    def __del__(self):\n        super(AtomicRemoteFileWriter, self).__del__()\n        if self.fs.exists(self.__tmp_path):\n            self.fs.remote_context.check_output(['rm', self.__tmp_path])\n\n    def close(self):\n        super(AtomicRemoteFileWriter, self).close()\n        self.fs.remote_context.check_output(['mv', self.__tmp_path, self.path])\n\n    @property\n    def tmp_path(self):\n        return self.__tmp_path\n\n    @property\n    def fs(self):\n        return self._fs\n\n\nclass RemoteTarget(luigi.target.FileSystemTarget):\n    \"\"\"\n    Target used for reading from remote files.\n\n    The target is implemented using ssh commands streaming data over the network.\n    \"\"\"\n\n    def __init__(self, path, host, format=None, username=None, key_file=None, port=None):\n        super(RemoteTarget, self).__init__(path)\n        if format is None:\n            format = luigi.format.get_default_format()\n        self.format = format\n        self._fs = RemoteFileSystem(host, username, key_file, port)\n\n    @property\n    def fs(self):\n        return self._fs\n\n    def open(self, mode='r'):\n        if mode == 'w':\n            file_writer = AtomicRemoteFileWriter(self.fs, self.path)\n            if self.format:\n                return self.format.pipe_writer(file_writer)\n            else:\n                return file_writer\n        elif mode == 'r':\n            file_reader = luigi.format.InputPipeProcessWrapper(\n                self.fs.remote_context._prepare_cmd([\"cat\", self.path]))\n            if self.format:\n                return self.format.pipe_reader(file_reader)\n            else:\n                return file_reader\n        else:\n            raise Exception(\"mode must be r/w\")\n\n    def put(self, local_path):\n        self.fs.put(local_path, self.path)\n\n    def get(self, local_path):\n        self.fs.get(self.path, local_path)\n", "path": "luigi/contrib/ssh.py"}]}