1import unittest
2import tempfile
3import os
4import sys
5import subprocess
6from commontest import old_test_dir, abs_test_dir
7from rdiff_backup.connection import LowLevelPipeConnection, PipeConnection, \
8    VirtualFile, SetConnections
9from rdiff_backup import Globals, rpath, FilenameMapping  # , log
10
11SourceDir = 'rdiff_backup'
12regfilename = os.path.join(old_test_dir, b"various_file_types",
13                           b"regular_file")
14
15
16class LocalConnectionTest(unittest.TestCase):
17    """Test the dummy connection"""
18    lc = Globals.local_connection
19
20    def testGetAttrs(self):
21        """Test getting of various attributes"""
22        assert type(self.lc.LocalConnection) is type
23        try:
24            self.lc.asotnuhaoseu
25        except (NameError, KeyError):
26            pass
27        else:
28            unittest.fail("NameError or KeyError should be raised")
29
30    def testSetattrs(self):
31        """Test setting of global attributes"""
32        self.lc.x = 5
33        assert self.lc.x == 5
34        self.lc.x = 7
35        assert self.lc.x == 7
36
37    def testDelattrs(self):
38        """Testing deletion of attributes"""
39        self.lc.x = 5
40        del self.lc.x
41        try:
42            self.lc.x
43        except (NameError, KeyError):
44            pass
45        else:
46            unittest.fail("No exception raised")
47
48    def testReval(self):
49        """Test string evaluation"""
50        assert self.lc.reval("pow", 2, 3) == 8
51
52
53class LowLevelPipeConnectionTest(unittest.TestCase):
54    """Test LLPC class"""
55    objs = ["Hello", ("Tuple", "of", "strings"), [1, 2, 3, 4], 53.34235]
56    excts = [TypeError("te"), NameError("ne"), os.error("oe")]
57    filename = tempfile.mktemp()
58
59    def testObjects(self):
60        """Try moving objects across connection"""
61        with open(self.filename, "wb") as outpipe:
62            LLPC = LowLevelPipeConnection(None, outpipe)
63            for obj in self.objs:
64                LLPC._putobj(obj, 3)
65        with open(self.filename, "rb") as inpipe:
66            LLPC.inpipe = inpipe
67            for obj in self.objs:
68                gotten = LLPC._get()
69                assert gotten == (3, obj), gotten
70        os.unlink(self.filename)
71
72    def testBuf(self):
73        """Try moving a buffer"""
74        with open(self.filename, "wb") as outpipe:
75            LLPC = LowLevelPipeConnection(None, outpipe)
76            with open(regfilename, "rb") as inpipe:
77                inbuf = inpipe.read()
78            LLPC._putbuf(inbuf, 234)
79        with open(self.filename, "rb") as inpipe:
80            LLPC.inpipe = inpipe
81            assert (234, inbuf) == LLPC._get()
82        os.unlink(self.filename)
83
84    def testSendingExceptions(self):
85        """Exceptions should also be sent down pipe well"""
86        with open(self.filename, "wb") as outpipe:
87            LLPC = LowLevelPipeConnection(None, outpipe)
88            for exception in self.excts:
89                LLPC._putobj(exception, 0)
90        with open(self.filename, "rb") as inpipe:
91            LLPC.inpipe = inpipe
92            for exception in self.excts:
93                incoming_exception = LLPC._get()
94                assert isinstance(incoming_exception[1], exception.__class__)
95        os.unlink(self.filename)
96
97
98class PipeConnectionTest(unittest.TestCase):
99    """Test Pipe connection"""
100
101    def setUp(self):
102        """Must start a server for this"""
103        pipe_cmd = "%s testing/server.py %s" \
104            % (sys.executable, SourceDir)
105        self.p = subprocess.Popen(pipe_cmd,
106                                  shell=True,
107                                  stdin=subprocess.PIPE,
108                                  stdout=subprocess.PIPE,
109                                  close_fds=True)
110        (stdin, stdout) = (self.p.stdin, self.p.stdout)
111        self.conn = PipeConnection(stdout, stdin)
112        Globals.security_level = "override"
113
114    def testBasic(self):
115        """Test some basic pipe functions"""
116        assert self.conn.ord("a") == 97
117        assert self.conn.pow(2, 3) == 8
118        assert self.conn.reval("ord", "a") == 97
119
120    def testModules(self):
121        """Test module emulation"""
122        assert type(self.conn.tempfile.mktemp()) is str
123        assert self.conn.os.path.join(b"a", b"b") == b"a/b"
124        rp1 = rpath.RPath(self.conn, regfilename)
125        assert rp1.isreg()
126
127    def testVirtualFiles(self):
128        """Testing virtual files"""
129        # generate file name for temporary file
130        temp_file = os.path.join(abs_test_dir, b"tempout")
131
132        tempout = self.conn.open(temp_file, "wb")
133        assert isinstance(tempout, VirtualFile)
134        regfilefp = open(regfilename, "rb")
135        rpath.copyfileobj(regfilefp, tempout)
136        tempout.close()
137        regfilefp.close()
138        tempoutlocal = open(temp_file, "rb")
139        regfilefp = open(regfilename, "rb")
140        assert rpath.cmpfileobj(regfilefp, tempoutlocal)
141        tempoutlocal.close()
142        regfilefp.close()
143        os.unlink(temp_file)
144
145        with open(regfilename, "rb") as localfh:
146            assert rpath.cmpfileobj(self.conn.open(regfilename, "rb"), localfh)
147
148    def testString(self):
149        """Test transmitting strings"""
150        assert "32" == self.conn.str(32)
151        assert 32 == self.conn.int("32")
152
153    def testIterators(self):
154        """Test transmission of iterators"""
155        i = iter([5, 10, 15] * 100)
156        assert self.conn.hasattr(i, "__next__") and self.conn.hasattr(
157            i, "__iter__")
158        ret_val = self.conn.reval("lambda i: next(i)*next(i)", i)
159        assert ret_val == 50, ret_val
160
161    def testRPaths(self):
162        """Test transmission of rpaths"""
163        rp = rpath.RPath(self.conn, regfilename)
164        assert self.conn.reval("lambda rp: rp.data", rp) == rp.data
165        assert self.conn.reval(
166            "lambda rp: rp.conn is Globals.local_connection", rp)
167
168    def testQuotedRPaths(self):
169        """Test transmission of quoted rpaths"""
170        qrp = FilenameMapping.QuotedRPath(self.conn, regfilename)
171        assert self.conn.reval("lambda qrp: qrp.data", qrp) == qrp.data
172        assert qrp.isreg(), qrp
173        qrp_class_str = self.conn.reval("lambda qrp: str(qrp.__class__)", qrp)
174        assert qrp_class_str.find("QuotedRPath") > -1, qrp_class_str
175
176    def testExceptions(self):
177        """Test exceptional results"""
178        self.assertRaises(os.error, self.conn.os.lstat,
179                          "asoeut haosetnuhaoseu tn")
180        self.assertRaises(SyntaxError, self.conn.reval, "aoetnsu aoehtnsu")
181        assert self.conn.pow(2, 3) == 8
182
183    def tearDown(self):
184        """Bring down connection"""
185        self.conn.quit()
186        if (self.p.poll() is None):
187            self.p.terminate()
188
189
190class RedirectedConnectionTest(unittest.TestCase):
191    """Test routing and redirection"""
192
193    def setUp(self):
194        """Must start two servers for this"""
195        self.conna = SetConnections.init_connection(
196            "%s testing/server.py %s" % (sys.executable, SourceDir))
197        self.connb = SetConnections.init_connection(
198            "%s testing/server.py %s" % (sys.executable, SourceDir))
199
200    def testBasic(self):
201        """Test basic operations with redirection"""
202        self.conna.Globals.set("tmp_val", 1)
203        self.connb.Globals.set("tmp_val", 2)
204        assert self.conna.Globals.get("tmp_val") == 1
205        assert self.connb.Globals.get("tmp_val") == 2
206
207        self.conna.Globals.set("tmp_connb", self.connb)
208        self.connb.Globals.set("tmp_conna", self.conna)
209        assert self.conna.Globals.get("tmp_connb") is self.connb
210        assert self.connb.Globals.get("tmp_conna") is self.conna
211
212        val = self.conna.reval("Globals.get('tmp_connb').Globals.get",
213                               "tmp_val")
214        assert val == 2, val
215        val = self.connb.reval("Globals.get('tmp_conna').Globals.get",
216                               "tmp_val")
217        assert val == 1, val
218
219        assert self.conna.reval("Globals.get('tmp_connb').pow", 2, 3) == 8
220        self.conna.reval("Globals.tmp_connb.reval",
221                         "Globals.tmp_conna.Globals.set", "tmp_marker", 5)
222        assert self.conna.Globals.get("tmp_marker") == 5
223
224    def testRpaths(self):
225        """Test moving rpaths back and forth across connections"""
226        rp = rpath.RPath(self.conna, "foo")
227        self.connb.Globals.set("tmp_rpath", rp)
228        rp_returned = self.connb.Globals.get("tmp_rpath")
229        assert rp_returned.conn is rp.conn
230        assert rp_returned.path == rp.path
231
232    def tearDown(self):
233        SetConnections.CloseConnections()
234
235
236if __name__ == "__main__":
237    unittest.main()
238