[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