[apt-proxy-devel] r650 - trunk/apt_proxy/test
Chris Halls
halls at alioth.debian.org
Tue Apr 17 21:54:50 UTC 2007
Author: halls
Date: Tue Apr 17 21:54:50 2007
New Revision: 650
Modified:
trunk/apt_proxy/test/test_fetchers.py
Log:
Test http proxy and ftp username/password
Modified: trunk/apt_proxy/test/test_fetchers.py
==============================================================================
--- trunk/apt_proxy/test/test_fetchers.py (original)
+++ trunk/apt_proxy/test/test_fetchers.py Tue Apr 17 21:54:50 2007
@@ -42,12 +42,19 @@
max_versions=off
[backend1]
-backends = http://localhost/nothing-really
+backends = http://host1/nothing-really
[backend2]
-backends = http://localhost/nothing-really
+backends = http://host2:82/nothing-really
+
+[backend3]
+backends = http://host3/nothing-really
http_proxy = user1:password at test:1234
+[backend4]
+backends = http://host4:84/foo
+http_proxy = user2:password2 at test2:4321
+
[ftp]
backends = ftp://localhost/nothing-really
"""
@@ -58,23 +65,45 @@
Set up a factory using the additional config given
"""
FactoryTestHelper.setUp(self, config1)
- backend = self.factory.getBackend('backend1')
+
+ def getConnection(self, backendName):
+ """
+ Get a connection for given backend name
+ """
+ backend = self.factory.getBackend(backendName)
backendServer = backend.uris[0]
httpFetcher = HttpFetcher(backendServer)
- httpFetcher.proxy = None # Would otherwise have been set by httpFetcher.connect
- self.connection = FetcherHttpClient(httpFetcher)
- self.connection.transport = StringIO()
+ httpFetcher.proxy = backend.config.http_proxy # Would otherwise have been set by httpFetcher.connect
+ connection = FetcherHttpClient(httpFetcher)
+ connection.transport = StringIO()
def endHeaders():
pass
- self.connection.endHeaders = endHeaders
+ connection.endHeaders = endHeaders
+ return connection
def testProxyHeader(self):
- self.connection.proxy = self.factory.getBackend('backend2').config.http_proxy
- self.connection.download(None, '/a/b/c', None)
- fetcherHeaders = self.connection.transport.getvalue()
+ connection = self.getConnection('backend3')
+ connection.download(None, 'a/b/c', None)
+ fetcherHeaders = connection.transport.getvalue()
authheader = "Proxy-Authorization: Basic " + base64.encodestring('user1:password')[:-1]
self.assertNotEquals(re.search(r'(?:\n|^)'+authheader+r'\r\n', fetcherHeaders), None)
+ def HttpCommandTest(self, backendName, expectedResult):
+ connection = self.getConnection(backendName)
+ connection.download(None, 'a/b/c', None)
+ fetcherCommand = connection.transport.getvalue().split('\r\n')[0]
+ self.assertEquals(fetcherCommand, expectedResult + ' HTTP/1.0')
+
+ def testGetWithoutProxy(self):
+ self.HttpCommandTest('backend1', 'GET /nothing-really/a/b/c')
+
+ def testHostWithProxy(self):
+ self.HttpCommandTest('backend3', 'GET http://host3/nothing-really/a/b/c')
+
+ def testHostPortWithProxy(self):
+ self.HttpCommandTest('backend4', 'GET http://host4:84/foo/a/b/c')
+
+
class FetcherFtpInitTest(FactoryTestHelper):
def setUp(self):
"""
@@ -94,22 +123,28 @@
Set up a factory using the additional config given
"""
self.ftpserver = FtpServer()
- port = self.ftpserver.start()
+ port = self.ftpserver.start('username', 'password')
config = (config1 +
"[test_ftp]\n" +
- "backends=http://127.0.0.1:" + str(port))
+ "backends=http://127.0.0.1:" + str(port) + "\n" +
+ "[test_ftp_pass]\n" +
+ "backends=http://username:password@127.0.0.1:" + str(port) + "\n")
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
+ log.debug("Connected... %s" % (self.ftpFetcher))
+
def tearDown(self):
# We don't care about deferreds left over e.g. pending connection
#delayeds = reactor.getDelayedCalls()
#for d in delayeds:
# d.cancel()
reactor.iterate(0.1) # Process FTP callbacks before disconnecting
+ log.debug("Disconnecting... %s client=%s" % (self.ftpFetcher, self.ftpFetcher.ftpclient))
self.ftpFetcher.disconnect()
+ reactor.runUntilCurrent()
self.ftpFetcher = None
self.ftpserver.stop()
self.ftpserver = None
@@ -128,9 +163,24 @@
return self.ftpFetcher.connect()
testConnect.timeout = 2
+ def testFtpPass(self):
+ "Test connect with username + password"
+ backend = self.factory.getBackend('test_ftp_pass')
+ backendServer = backend.uris[0]
+ ftpFetcher = FtpFetcher(backendServer)
+ ftpFetcher.debug = 1
+ d = ftpFetcher.connect()
+ d.addBoth(lambda x: ftpFetcher.disconnect())
+ return d
+ testFtpPass.timeout = 2
+
def testConnectFail(self):
"Test connect failure"
self.ftpserver.stop()
+ reactor.runUntilCurrent()
+ reactor.iterate(0.1)
+ reactor.iterate(0.1)
+ reactor.iterate(0.1)
d = self.ftpFetcher.connect()
def callBack(result):
raise RuntimeError("Connect should have failed")
@@ -150,7 +200,7 @@
def download_failed(self, code, reason):
if self.error_code is not None and \
self.error_code == code:
- self.deferred.callback()
+ self.deferred.callback(None)
else:
self.deferred.errback(None)
def server_mtime(self, time):
@@ -196,7 +246,7 @@
testMtime.timeout = 1
class FtpServer:
- def start(self):
+ def start(self, username = None, password = None):
"""
Start FTP server, serving test data
@@ -212,6 +262,12 @@
p.registerChecker(checkers.AllowAnonymousAccess(), credentials.IAnonymous)
f.userAnonymous = 'anonymous'
+
+ if username:
+ checker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
+ checker.addUser(username, password)
+ p.registerChecker(checker)
+
f.portal = p
f.protocol = ftp.FTP
@@ -222,9 +278,8 @@
return portnum
def stop(self):
- #pass
- self.port.stopListening()
self.factory.stopFactory()
+ self.port.stopListening()
class RsyncFetcherTest(FactoryTestHelper):
"""
More information about the apt-proxy-devel
mailing list