[apt-proxy-devel] r605 - people/halls/rework/apt_proxy/test
Chris Halls
halls at costa.debian.org
Tue Apr 25 09:01:21 UTC 2006
Author: halls
Date: Tue Apr 25 09:01:20 2006
New Revision: 605
Modified:
people/halls/rework/apt_proxy/test/test_fetchers.py
Log:
* Remove obsolete ftp test classes (now using twisted's ftp server to test)
* Add rync test classes
Modified: people/halls/rework/apt_proxy/test/test_fetchers.py
==============================================================================
--- people/halls/rework/apt_proxy/test/test_fetchers.py (original)
+++ people/halls/rework/apt_proxy/test/test_fetchers.py Tue Apr 25 09:01:20 2006
@@ -24,7 +24,8 @@
from twisted.internet import reactor, protocol, defer, error
from twisted.protocols import ftp
from twisted.cred import portal, checkers, credentials
-import time
+from twisted.python import failure
+import time, os, socket, signal, string
config1="""
[DEFAULT]
@@ -70,39 +71,13 @@
backendServer = backend.uris[0]
ftpFetcher = FtpFetcher(backendServer)
-class FtpServerProtocolTest(protocol.Protocol):
- def connectionMade(self):
- """
- Ftp connection made
- """
- log.debug("connection made to test ftp server.", 'FtpServerProtocolTest')
- self.transport.loseConnection()
-
-class FtpServer2:
- def start(self, tester):
- """
- Start web server, serving test data
-
- @ret port number that server listens on
- """
- f = protocol.Factory()
- f.protocol = tester
- self.port = reactor.listenTCP(0, f, interface="127.0.0.1")
- portnum = self.port.getHost().port
- log.debug("Ftp server listening on port %s" %(portnum))
-
- return portnum
-
- def stop(self):
- self.port.stopListening()
-
class FetcherFtpTestHelper(FactoryTestHelper):
- def setUp(self, serverProtocol):
+ def setUp(self):
"""
Set up a factory using the additional config given
"""
self.ftpserver = FtpServer()
- port = self.ftpserver.start(FtpServerProtocolTest)
+ port = self.ftpserver.start()
config = (config1 +
"[test_ftp]\n" +
"backends=http://127.0.0.1:" + str(port))
@@ -124,7 +99,7 @@
class FetcherFtpTest(FetcherFtpTestHelper):
def setUp(self):
- FetcherFtpTestHelper.setUp(self, FtpServerProtocolTest)
+ FetcherFtpTestHelper.setUp(self)
def testConnect(self):
"Test connect"
@@ -144,20 +119,6 @@
return d
testConnectFail.timeout = 2
-class FtpServerProtocolTest(protocol.Protocol):
- """
- Helper class for FetcherFtpProtocolTest
- """
- def connectionMade(self):
- """
- Ftp connection made
- """
- log.debug("connection made to test ftp server.", 'FtpServerProtocolTest')
- self.sendCommand("220 (apt-proxy test suite)")
-
- def sendCommand(self, command):
- self.transport.write("%s\n" % (command))
-
class DummyFetcher:
def __init__(self, deferred):
self.deferred = deferred
@@ -179,7 +140,7 @@
class FetcherFtpProtocolTest(FetcherFtpTestHelper):
def setUp(self):
- FetcherFtpTestHelper.setUp(self,FtpServerProtocolTest)
+ FetcherFtpTestHelper.setUp(self)
self.resultCallback = defer.Deferred()
self.fetcher = DummyFetcher(self.resultCallback)
self.fetcher.backendServer = self.backendServer
@@ -213,7 +174,7 @@
testMtime.timeout = 1
class FtpServer:
- def start(self, tester):
+ def start(self):
"""
Start FTP server, serving test data
@@ -221,10 +182,10 @@
This routine was hacked from twisted/tap/ftp.py
"""
+ root = '../test_data'
f = ftp.FTPFactory()
- r = ftp.FTPRealm()
- r.tld = '../test_data'
- f.tld = r.tld
+ r = ftp.FTPRealm(root)
+ f.tld = root
p = portal.Portal(r)
p.registerChecker(checkers.AllowAnonymousAccess(), credentials.IAnonymous)
@@ -286,3 +247,134 @@
self.f.connect()
dummyFetcher = self.DummyFetcher(self.backend, self.backendServer)
self.f.download(dummyFetcher, 'test', time.time())
+
+class RsyncServer(protocol.ProcessProtocol):
+ """
+ Starts an rsync daemon on localhost for testing
+ """
+ rsyncCommand = '/usr/bin/rsync'
+
+ def start(self):
+ """
+ Start rsync server, serving test data
+
+ @ret port number that server listens on
+ """
+ self.rsync_dir = '../test_data'
+
+ # Find a port number for the rsync server process:
+ # Start listening on a random port, then close it
+ s = socket.socket()
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ s.listen(1)
+ self.rsync_port = s.getsockname()[1]
+ s.close()
+
+ self.rsync_confpath = self.rsync_dir + os.sep + 'testrsync.conf'
+ self.write_rsyncconf()
+
+ args = (self.rsyncCommand, '--daemon', '--config=' + self.rsync_confpath, '--verbose', '--no-detach')
+ self.rsyncProcess = reactor.spawnProcess(self, self.rsyncCommand, args, None,self.rsync_dir)
+
+ # wait for server to start
+ s2 = socket.socket()
+ while s2:
+ try:
+ s2.connect(('127.0.0.1', self.rsync_port))
+ break
+ except:
+ pass
+ reactor.iterate(0.1)
+ s2.close()
+
+ log.debug("rsync server listening on port %s" %(self.rsync_port))
+ return self.rsync_port
+
+ def stop(self):
+ if self.rsyncProcess and self.rsyncProcess.pid:
+ log.debug("killing rsync child pid " +
+ str(self.rsyncProcess.pid), 'RsyncServer')
+ self.rsyncProcess.loseConnection()
+ os.kill(self.rsyncProcess.pid, signal.SIGTERM)
+
+ def write_rsyncconf(self):
+ f = open(self.rsync_confpath, 'w')
+ f.write("address = 127.0.0.1\n")
+ f.write("port = %s\n" % (self.rsync_port))
+ f.write("log file = %s\n" %(self.rsync_dir+os.sep+'testrsync.log'))
+ f.write("[apt-proxy]\n")
+ f.write("path = %s\n" %(self.rsync_dir))
+ f.write("use chroot = false\n") # Can't chroot becuase daemon isn't root
+ f.close()
+
+ def outReceived(self, data):
+ "Data received from rsync process to stdout"
+ for s in string.split(data, '\n'):
+ if len(s):
+ log.debug('rsync: ' + s, 'RsyncServer')
+
+ def errReceived(self, data):
+ "Data received from rsync process to stderr"
+ for s in string.split(data, '\n'):
+ if len(s):
+ log.err('rsync error: ' + s, 'RsyncServer')
+
+ def processEnded(self, status_object):
+ if isinstance(status_object, failure.Failure):
+ log.debug("rsync failure: %s" %(status_object)
+ ,'RsyncServer')
+ else:
+ log.debug("Status: %d" %(status_object.value.exitCode)
+ ,'RsyncServer')
+
+ # Success?
+ exitcode = status_object.value.exitCode
+
+
+class FetcherRsyncTestHelper(FactoryTestHelper):
+ def setUp(self):
+ """
+ Set up a factory using the additional config given
+ """
+ self.rsyncserver = RsyncServer()
+ port = self.rsyncserver.start()
+ config = (config1 +
+ "[test_rsync]\n" +
+ "backends=http://127.0.0.1:" + str(port) + '/apt-proxy')
+ FactoryTestHelper.setUp(self, config)
+ self.backend = self.factory.getBackend('test_rsync')
+ self.backendServer = self.backend.uris[0]
+ self.rsyncFetcher = RsyncFetcher(self.backendServer)
+ self.rsyncFetcher.debug = 1
+ def tearDown(self):
+ # We don't care about deferreds left over e.g. pending connection
+ #delayeds = reactor.getDelayedCalls()
+ #for d in delayeds:
+ # d.cancel()
+ self.rsyncFetcher.disconnect()
+ self.rsyncFetcher = None
+ self.rsyncserver.stop()
+ self.rsyncserver = None
+ FactoryTestHelper.tearDown(self)
+
+class FetcherRsyncProtocolTest(FetcherRsyncTestHelper):
+ def setUp(self):
+ FetcherRsyncTestHelper.setUp(self)
+ self.resultCallback = defer.Deferred()
+ self.fetcher = DummyFetcher(self.resultCallback)
+ self.fetcher.backendServer = self.backendServer
+
+ def tearDown(self):
+ FetcherRsyncTestHelper.tearDown(self)
+
+ def testNotFound(self):
+ "Test for file not found"
+ d = self.rsyncFetcher.connect()
+ d.addCallback(self.NotFound2)
+ return self.resultCallback
+ testNotFound.timeout = 1
+ def NotFound2(self,result):
+ self.fetcher.wait_for_not_found = True
+ fileName = 'notHereFile'
+ self.fetcher.cacheEntry = self.backend.get_cache_entry(fileName)
+ self.rsyncFetcher.download(self.fetcher, fileName, 0)
More information about the apt-proxy-devel
mailing list