1import os
2import stat
3from io import BytesIO
4
5from py import path
6
7from fabric import Connection
8
9
10def _support(*parts):
11    return os.path.join(os.path.dirname(__file__), "_support", *parts)
12
13
14class Transfer_:
15    class get:
16        def setup(self):
17            self.c = Connection("localhost")
18            self.remote = _support("file.txt")
19
20        def base_case(self, tmpdir):
21            # Copy file from support to tempdir
22            with tmpdir.as_cwd():
23                result = self.c.get(self.remote)
24
25            # Make sure it arrived
26            local = tmpdir.join("file.txt")
27            assert local.check()
28            assert local.read() == "yup\n"
29            # Sanity check result object
30            assert result.remote == self.remote
31            assert result.orig_remote == self.remote
32            assert result.local == str(local)
33            assert result.orig_local is None
34
35        def file_like_objects(self):
36            fd = BytesIO()
37            result = self.c.get(remote=self.remote, local=fd)
38            assert fd.getvalue() == b"yup\n"
39            assert result.remote == self.remote
40            assert result.local is fd
41
42        def mode_preservation(self, tmpdir):
43            # Use a dummy file which is given an unusual, highly unlikely to be
44            # default umask, set of permissions (oct 641, aka -rw-r----x)
45            local = tmpdir.join("funky-local.txt")
46            remote = tmpdir.join("funky-remote.txt")
47            remote.write("whatever")
48            remote.chmod(0o641)
49            self.c.get(remote=str(remote), local=str(local))
50            assert stat.S_IMODE(local.stat().mode) == 0o641
51
52    class put:
53        def setup(self):
54            self.c = Connection("localhost")
55            self.remote = path.local.mkdtemp().join("file.txt").realpath()
56
57        def base_case(self):
58            # Copy file from 'local' (support dir) to 'remote' (tempdir)
59            local_dir = _support()
60            with path.local(local_dir).as_cwd():
61                tmpdir = self.remote.dirpath()
62                # TODO: wrap chdir at the Connection level
63                self.c.sftp().chdir(str(tmpdir))
64                result = self.c.put("file.txt")
65            # Make sure it arrived
66            assert self.remote.check()
67            assert self.remote.read() == "yup\n"
68            # Sanity check result object
69            assert result.remote == self.remote
70            assert result.orig_remote is None
71            assert result.local == _support("file.txt")
72            assert result.orig_local == "file.txt"
73
74        def file_like_objects(self):
75            fd = BytesIO()
76            fd.write(b"yup\n")
77            remote_str = str(self.remote)
78            result = self.c.put(local=fd, remote=remote_str)
79            assert self.remote.read() == "yup\n"
80            assert result.remote == remote_str
81            assert result.local is fd
82
83        def mode_preservation(self, tmpdir):
84            # Use a dummy file which is given an unusual, highly unlikely to be
85            # default umask, set of permissions (oct 641, aka -rw-r----x)
86            local = tmpdir.join("funky-local.txt")
87            local.write("whatever")
88            local.chmod(0o641)
89            remote = tmpdir.join("funky-remote.txt")
90            self.c.put(remote=str(remote), local=str(local))
91            assert stat.S_IMODE(remote.stat().mode) == 0o641
92