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