diff options
author | ben <ben@2b77aa54-bcbc-44c9-a7ec-4f6cf2b41109> | 2002-03-21 07:22:43 +0000 |
---|---|---|
committer | ben <ben@2b77aa54-bcbc-44c9-a7ec-4f6cf2b41109> | 2002-03-21 07:22:43 +0000 |
commit | 8c37a5bdfdd46d5cfad6e9d67925ddef9ca382bf (patch) | |
tree | 8f19be83962ef31d8ad58429d575c6f17d89c0ea /rdiff-backup/testing/connectiontest.py | |
parent | 8259a0d8a9ad1396a93cd6320943dc33446ac6ed (diff) | |
download | rdiff-backup-8c37a5bdfdd46d5cfad6e9d67925ddef9ca382bf.tar.gz |
First checkin
git-svn-id: http://svn.savannah.nongnu.org/svn/rdiff-backup/trunk@2 2b77aa54-bcbc-44c9-a7ec-4f6cf2b41109
Diffstat (limited to 'rdiff-backup/testing/connectiontest.py')
-rw-r--r-- | rdiff-backup/testing/connectiontest.py | 201 |
1 files changed, 201 insertions, 0 deletions
diff --git a/rdiff-backup/testing/connectiontest.py b/rdiff-backup/testing/connectiontest.py new file mode 100644 index 0000000..bcd98ea --- /dev/null +++ b/rdiff-backup/testing/connectiontest.py @@ -0,0 +1,201 @@ +import unittest, types, tempfile, os, sys +execfile("commontest.py") +rbexec("setconnections.py") + + +class LocalConnectionTest(unittest.TestCase): + """Test the dummy connection""" + lc = Globals.local_connection + + def testGetAttrs(self): + """Test getting of various attributes""" + assert type(self.lc.LocalConnection) is types.ClassType + try: self.lc.asotnuhaoseu + except NameError: pass + else: unittest.fail("NameError should be raised") + + def testSetattrs(self): + """Test setting of global attributes""" + self.lc.x = 5 + assert self.lc.x == 5 + self.lc.x = 7 + assert self.lc.x == 7 + + def testDelattrs(self): + """Testing deletion of attributes""" + self.lc.x = 5 + del self.lc.x + try: self.lc.x + except NameError: pass + else: unittest.fail("No exception raised") + + def testReval(self): + """Test string evaluation""" + assert self.lc.reval("pow", 2, 3) == 8 + + +class LowLevelPipeConnectionTest(unittest.TestCase): + """Test LLPC class""" + objs = ["Hello", ("Tuple", "of", "strings"), + [1, 2, 3, 4], 53.34235] + excts = [TypeError("te"), NameError("ne"), os.error("oe")] + filename = tempfile.mktemp() + + def testObjects(self): + """Try moving objects across connection""" + outpipe = open(self.filename, "w") + LLPC = LowLevelPipeConnection(None, outpipe) + for obj in self.objs: LLPC._putobj(obj, 3) + outpipe.close() + inpipe = open(self.filename, "r") + LLPC.inpipe = inpipe + for obj in self.objs: + gotten = LLPC._get() + assert gotten == (3, obj), gotten + inpipe.close + os.unlink(self.filename) + + def testBuf(self): + """Try moving a buffer""" + outpipe = open(self.filename, "w") + LLPC = LowLevelPipeConnection(None, outpipe) + inbuf = open("testfiles/various_file_types/regular_file", "r").read() + LLPC._putbuf(inbuf, 234) + outpipe.close() + inpipe = open(self.filename, "r") + LLPC.inpipe = inpipe + assert (234, inbuf) == LLPC._get() + inpipe.close() + os.unlink(self.filename) + + def testSendingExceptions(self): + """Exceptions should also be sent down pipe well""" + outpipe = open(self.filename, "w") + LLPC = LowLevelPipeConnection(None, outpipe) + for exception in self.excts: LLPC._putobj(exception, 0) + outpipe.close() + inpipe = open(self.filename, "r") + LLPC.inpipe = inpipe + for exception in self.excts: + incoming_exception = LLPC._get() + assert isinstance(incoming_exception[1], exception.__class__) + inpipe.close() + os.unlink(self.filename) + + +class PipeConnectionTest(unittest.TestCase): + """Test Pipe connection""" + regfilename = "testfiles/various_file_types/regular_file" + + def setUp(self): + """Must start a server for this""" + stdin, stdout = os.popen2("./server.py") + self.conn = PipeConnection(stdout, stdin) + #self.conn.Log.setverbosity(9) + #Log.setverbosity(9) + + def testBasic(self): + """Test some basic pipe functions""" + assert self.conn.ord("a") == 97 + assert self.conn.pow(2,3) == 8 + assert self.conn.reval("ord", "a") == 97 + + def testModules(self): + """Test module emulation""" + assert type(self.conn.tempfile.mktemp()) is types.StringType + assert self.conn.os.path.join("a", "b") == "a/b" + rp1 = RPath(self.conn, self.regfilename) + assert rp1.isreg() + + def testVirtualFiles(self): + """Testing virtual files""" + tempout = self.conn.open("testfiles/tempout", "w") + assert isinstance(tempout, VirtualFile) + regfilefp = open(self.regfilename, "r") + RPath.copyfileobj(regfilefp, tempout) + tempout.close() + regfilefp.close() + tempoutlocal = open("testfiles/tempout", "r") + regfilefp = open(self.regfilename, "r") + assert RPath.cmpfileobj(regfilefp, tempoutlocal) + tempoutlocal.close() + regfilefp.close() + os.unlink("testfiles/tempout") + + assert RPath.cmpfileobj(self.conn.open(self.regfilename, "r"), + open(self.regfilename, "r")) + + def testString(self): + """Test transmitting strings""" + assert "32" == self.conn.str(32) + assert 32 == self.conn.int("32") + + def testIterators(self): + """Test transmission of iterators""" + i = iter(map(RORPsubstitute, range(10))) + assert self.conn.hasattr(i, "next") + datastring = self.conn.reval("lambda i: i.next().data", i) + assert datastring == "Hello, there 0", datastring + + def testRPaths(self): + """Test transmission of rpaths""" + rp = RPath(self.conn, "testfiles/various_file_types/regular_file") + assert self.conn.reval("lambda rp: rp.data", rp) == rp.data + assert self.conn.reval("lambda rp: rp.conn is Globals.local_connection", rp) + + def testExceptions(self): + """Test exceptional results""" + self.assertRaises(os.error, self.conn.os.lstat, + "asoeut haosetnuhaoseu tn") + self.assertRaises(SyntaxError, self.conn.reval, + "aoetnsu aoehtnsu") + assert self.conn.pow(2,3) == 8 + + def tearDown(self): + """Bring down connection""" + self.conn.quit() + + +class RedirectedConnectionTest(unittest.TestCase): + """Test routing and redirection""" + def setUp(self): + """Must start two servers for this""" + #Log.setverbosity(9) + self.conna = SetConnections.init_connection("./server.py") + self.connb = SetConnections.init_connection("./server.py") + + def testBasic(self): + """Test basic operations with redirection""" + self.conna.Globals.set("tmp_connb", self.connb) + self.connb.Globals.set("tmp_conna", self.conna) + assert self.conna.Globals.get("tmp_connb") is self.connb + assert self.connb.Globals.get("tmp_conna") is self.conna + + self.conna.Test_SetConnGlobals(self.connb, "tmp_settest", 1) + assert self.connb.Globals.get("tmp_settest") + + assert self.conna.reval("Globals.get('tmp_connb').pow", 2, 3) == 8 + self.conna.reval("Globals.tmp_connb.reval", + "Globals.tmp_conna.Globals.set", "tmp_marker", 5) + assert self.conna.Globals.get("tmp_marker") == 5 + + def testRpaths(self): + """Test moving rpaths back and forth across connections""" + rp = RPath(self.conna, "foo") + self.connb.Globals.set("tmp_rpath", rp) + rp_returned = self.connb.Globals.get("tmp_rpath") + assert rp_returned.conn is rp.conn + assert rp_returned.path == rp.path + + def tearDown(self): + SetConnections.CloseConnections() + +class RORPsubstitute: + """Used in testIterators above to simulate a RORP""" + def __init__(self, i): + self.index = i + self.data = "Hello, there %d" % i + self.file = None + +if __name__ == "__main__": + unittest.main() |