[apt-proxy-devel] r602 - people/halls/rework/apt_proxy/test

[Chris Halls]

Author: halls
Date: Sat Apr  8 15:45:49 2006
New Revision: 602

Modified:
   people/halls/rework/apt_proxy/test/test_fetchers.py
Log:
Add rsync fetcher tests


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	Sat Apr  8 15:45:49 2006
@@ -20,10 +20,11 @@
 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 apt_proxy.fetchers import HttpFetcher, FetcherHttpClient, FtpFetcher, Fetcher, RsyncFetcher
 from twisted.internet import reactor, protocol, defer, error
 from twisted.protocols import ftp
 from twisted.cred import portal, checkers, credentials
+import time
 
 config1="""
 [DEFAULT]
@@ -241,3 +242,47 @@
         #pass
         self.port.stopListening()
         self.factory.stopFactory()
+
+class RsyncFetcherTest(FactoryTestHelper):
+    """
+    Set up a cache dir and a factory
+    """
+
+    rsync_config="""
+[DEFAULT]
+debug=all:9
+port=9999
+address=
+cleanup_freq=off
+max_versions=off
+
+[rsync]
+backends = rsync://127.0.0.1:0/test
+"""
+
+    class DummyFetcher:
+        def __init__(self, backend, backendServer):
+            self.backend = backend
+            self.backendServer = backendServer
+            self.CacheEntry = backend.get_cache_entry("testdir/testfile.deb")
+        def fetcher_internal_error(self, message):
+            log.debug('fetcher_internal_error: %s' % (message))
+
+    def setUp(self):
+        """
+        Set up a factory using the additional config given
+        """
+        FactoryTestHelper.setUp(self, self.rsync_config)
+        self.backend = self.factory.getBackend('rsync')
+        self.backendServer = self.backend.get_first_server()
+        self.f = RsyncFetcher(self.backendServer)
+    def tearDown(self):
+        self.f.disconnect()
+    def testRsyncInit(self):
+        self.assertEquals(self.f.backendServer, self.backendServer)
+    def testConnect(self):
+        return self.f.connect() # connect returns a deferred that fires
+    def testDownload(self):
+        self.f.connect()
+        dummyFetcher = self.DummyFetcher(self.backend, self.backendServer)
+        self.f.download(dummyFetcher, 'test', time.time())