[apt-proxy-devel] r600 - people/halls/rework/apt_proxy/test
Chris Halls
halls at costa.debian.org
Tue Apr 4 19:59:33 UTC 2006
Author: halls
Date: Tue Apr 4 19:59:31 2006
New Revision: 600
Added:
people/halls/rework/apt_proxy/test/test_fetchers.py
Log:
Add fetchers test
Added: people/halls/rework/apt_proxy/test/test_fetchers.py
==============================================================================
--- (empty file)
+++ people/halls/rework/apt_proxy/test/test_fetchers.py Tue Apr 4 19:59:31 2006
@@ -0,0 +1,243 @@
+#
+# Copyright (C) 2006 Chris Halls <halls at debian.org>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of version 2.1 of the GNU Lesser General Public
+# License as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+"""This module tests the Fetcher classes"""
+
+from apt_proxy.apt_proxy_conf import apConfig
+from apt_proxy.test.test_apt_proxy import apTestHelper, FactoryTestHelper
+from apt_proxy.apt_proxy import Factory
+from apt_proxy.misc import log
+from apt_proxy.fetchers import HttpFetcher, FetcherHttpClient, FtpFetcher, Fetcher
+from twisted.internet import reactor, protocol, defer, error
+from twisted.protocols import ftp
+from twisted.cred import portal, checkers, credentials
+
+config1="""
+[DEFAULT]
+debug=all:9
+port=9999
+address=
+cleanup_freq=off
+max_versions=off
+
+[backend1]
+backends = http://localhost/nothing-really
+
+[ftp]
+backends = ftp://localhost/nothing-really
+"""
+
+class FetcherHttpTest(FactoryTestHelper):
+ def setUp(self):
+ """
+ Set up a factory using the additional config given
+ """
+ FactoryTestHelper.setUp(self, config1)
+
+ def testInit(self):
+ "Brief init test"
+ backend = self.factory.getBackend('backend1')
+ backendServer = backend.uris[0]
+ httpFetcher = HttpFetcher(backendServer)
+ httpFetcher.proxy = None # Would otherwise have been set by httpFetcher.connect
+ connection = FetcherHttpClient(httpFetcher)
+
+
+class FetcherFtpInitTest(FactoryTestHelper):
+ def setUp(self):
+ """
+ Set up a factory using the additional config given
+ """
+ FactoryTestHelper.setUp(self, config1)
+
+ def testInit(self):
+ "Brief init test"
+ backend = self.factory.getBackend('ftp')
+ 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):
+ """
+ Set up a factory using the additional config given
+ """
+ self.ftpserver = FtpServer()
+ port = self.ftpserver.start(FtpServerProtocolTest)
+ config = (config1 +
+ "[test_ftp]\n" +
+ "backends=http://127.0.0.1:" + str(port))
+ FactoryTestHelper.setUp(self, config)
+ self.backend = self.factory.getBackend('test_ftp')
+ self.backendServer = self.backend.uris[0]
+ self.ftpFetcher = FtpFetcher(self.backendServer)
+ self.ftpFetcher.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.ftpFetcher.disconnect()
+ self.ftpFetcher = None
+ self.ftpserver.stop()
+ self.ftpserver = None
+ FactoryTestHelper.tearDown(self)
+
+class FetcherFtpTest(FetcherFtpTestHelper):
+ def setUp(self):
+ FetcherFtpTestHelper.setUp(self, FtpServerProtocolTest)
+
+ def testConnect(self):
+ "Test connect"
+ return self.ftpFetcher.connect()
+ testConnect.timeout = 2
+
+ def testConnectFail(self):
+ "Test connect failure"
+ self.ftpserver.stop()
+ d = self.ftpFetcher.connect()
+ def callBack(result):
+ raise RuntimeError("Connect should have failed")
+ def errorBack(result):
+ result.trap(error.ConnectionRefusedError)
+ # Reverse meaning of deferred, ie errorback = as expected
+ d.addCallbacks(callBack, errorBack)
+ 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
+ self.error_code = None # Anticipated error
+ self.wait_for_mtime = False
+ self.wait_for_not_found = False
+ def download_failed(self, code, reason):
+ if self.error_code is not None and \
+ self.error_code == code:
+ self.deferred.callback()
+ else:
+ self.deferred.errback()
+ def server_mtime(self, time):
+ if self.wait_for_mtime == True:
+ self.deferred.callback(None)
+ def file_not_found(self):
+ if self.wait_for_not_found == True:
+ self.deferred.callback(None)
+
+class FetcherFtpProtocolTest(FetcherFtpTestHelper):
+ def setUp(self):
+ FetcherFtpTestHelper.setUp(self,FtpServerProtocolTest)
+ self.resultCallback = defer.Deferred()
+ self.fetcher = DummyFetcher(self.resultCallback)
+ self.fetcher.backendServer = self.backendServer
+
+ def tearDown(self):
+ FetcherFtpTestHelper.tearDown(self)
+
+ def testNotFound(self):
+ "Test for file not found"
+ d = self.ftpFetcher.connect()
+ d.addCallback(self.NotFoundConnectCallback)
+ return self.resultCallback
+ testNotFound.timeout = 1
+ def NotFoundConnectCallback(self,result):
+ self.fetcher.wait_for_not_found = True
+ self.ftpFetcher.download(self.fetcher, 'notHereFile', 0)
+
+ def MtimeConnectCallback(self,result):
+ log.debug("connection made", 'FetcherFtpProtocolTest')
+ self.fetcher.wait_for_mtime = True
+ self.ftpFetcher.download(self.fetcher, 'packages/Packages', 0)
+
+ def testMtime(self):
+ "Test mtime request"
+ def FetchSize():
+ pass
+ self.ftpFetcher.ftpFetchSize = FetchSize # We don't want to get size afterwards
+ d = self.ftpFetcher.connect()
+ d.addCallback(self.MtimeConnectCallback)
+ return self.resultCallback
+ testMtime.timeout = 1
+
+class FtpServer:
+ def start(self, tester):
+ """
+ Start FTP server, serving test data
+
+ @ret port number that server listens on
+
+ This routine was hacked from twisted/tap/ftp.py
+ """
+ f = ftp.FTPFactory()
+ r = ftp.FTPRealm()
+ r.tld = '../test_data'
+ f.tld = r.tld
+ p = portal.Portal(r)
+ p.registerChecker(checkers.AllowAnonymousAccess(), credentials.IAnonymous)
+
+ f.userAnonymous = 'anonymous'
+ f.portal = p
+ f.protocol = ftp.FTP
+
+ 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))
+ self.factory = f
+ return portnum
+
+ def stop(self):
+ #pass
+ self.port.stopListening()
+ self.factory.stopFactory()
More information about the apt-proxy-devel
mailing list