Changeset 1104 for TI05-delivery/trunk

Timestamp:

06/06/06 16:35:10 (15 years ago)

Author:

spascoe

Message:

Enhanced test cases.

Location:

TI05-delivery/trunk

Files:

• src/bbftp-server-3.2.0/bbftpd/bbftpd_private.c (modified) (1 diff)
• test/test_embedded.py (modified) (4 diffs)

TI05-delivery/trunk/src/bbftp-server-3.2.0/bbftpd/bbftpd_private.c

@@ -55 +55 @@
 #include <version.h>
 
-#ifdef NDG_AUTH
-#include <signal.h>
-#endif // NDG_AUTH
-
 extern  int     incontrolsock ;
 extern  int     outcontrolsock ;

TI05-delivery/trunk/test/test_embedded.py

@@ -54 +54 @@
         return AuthzContext(msg, "TestCaseUser")
 
-class EmbeddedServerTestCase(unittest.TestCase):
-    """Test the bbftpd module.
-    """
-    
+
+class FailingAuthContext(AuthContext):
+    def authorise(self):
+        az = super(FailingAuthContext).authorise()
+        return None
+
+
+
+
+class BaseFixture(unittest.TestCase):
     def setUp(self):
         # We want to mark the beginning of this test case in syslog
@@ -70 +76 @@
         syslog.closelog()
 
-    def testStartup(self):
-        lines = self._readSyslog()
-        # Give syslog time to flush it's logs.
-        time.sleep(1)
-        self.assert_(self._findLines(['.*Starting bbftpd'], lines))
-
-    def testDir(self):
-        """Try connecting the client and listing a directory.
-        """
-
-
-        fh = self._runClient("dir %s" % DATADIR)
-        output = fh.read()
-
-
-        self.assert_(self._findLines([r'dir .*/data', r' d .*/\.', r' d .*/\.\.',
-                                      r' f .*/foo', r' f .*/bar', r' f .*/baz'], output))       
-
-        lines = self._readSyslog()
-        self.assert_(self._findLines(['.*Getting new bbftp connexion.*',
-                                      r'.*Authz: MSG_LIST_V2 .*/test/data/\*.*',
-                                      r'.*User TestCaseUser disconnected.*'], lines))
-
-
-    def testHandshake(self):
-        """Verify handshake messages are exchanged.
-        """
-
-        fh = self._runClient("dir .", debug=True)
-        output = fh.read()
-
-        self.assert_(self._findLines(['Received Auth handshake: NDG-Delivery-server %s' % VERSION], output))
-
-        lines = self._readSyslog()
-        self.assert_(self._findLines(['.*AuthContext received Auth message: NDG-Delivery-client %s' % VERSION], lines))
-
-    def testPrivateStr(self):
-        """Verify the private string is sent to server.
-        """
-
-        fh = self._runClient("dir .", privatestr="testPrivateStr")
-        output = fh.read()
-
-        lines = self._readSyslog()
-        self.assert_(self._findLines(['.*AuthContext received privatestr: testPrivateStr'], lines))
-
-    def testRetr(self):
-        """Try retrieving a file.
-        """
-
-        tmp = tempfile.mktemp('test_bbftpd')
-        fh = self._runClient("get %s/foo %s" % (DATADIR, tmp))
-
-        # Check the client output
-        output = fh.read()
-        self.assert_(self._findLines(['get.*nogzip'], output))
-
-        # Check retrieved file
-        self.assert_(os.system('diff --brief %s/foo %s' % (DATADIR, tmp)) == 0)
-        os.remove(tmp)
-
-        # Check syslog
-        lines = self._readSyslog()
-        self.assert_(self._findLines(['.*Authz: RETR .*/foo', '.*GET TestCaseUser .*/foo.*'], lines))
-
-    def testStore(self):
-        """Try storing a file.
-        """
-
-
-        src = '%s/bar' % (DATADIR)
-        dest = '%s/new_bar' % (DATADIR)
-        os.system('cp %s %s' % (src, dest))
-        fh = self._runClient("put %s %s" % (src, dest))
-
-        # Check the client output
-        output = fh.read()
-        self.assert_(self._findLines(['put .* nogzip'], output))
-
-
-        # Check sent file
-        self.assert_(os.system('diff --brief %s %s' % (dest, src)) == 0)
-        os.remove(dest)
-
-        # Check syslog
-        lines = self._readSyslog()
-        self.assert_(self._findLines(['.*Authz: STORE .*/new_bar', '.*PUT TestCaseUser .*/new_bar.*'], lines))
-
-
     #----------------------------------------------------------------------------------
 
 
-    def _startServer(self):
+    def _startServer(self, authContext=None):
+        if not authContext:
+            authContext = AuthContext()
         # Start the server and store it's PID
-        self.authContext = AuthContext()
-        self.pid = bbftpd.run(self.authContext, ['-l', 'DEBUG'])
+        self.pid = bbftpd.run(authContext, ['-l', 'DEBUG'])
 
     def _stopServer(self):
@@ -187 +105 @@
         else:
             p = "-P %s" % repr(privatestr)
-        
-        fh = os.popen('%s %s %s -u %s -e %s localhost' % (BBFTP, f, p, user, repr(cmd)))
+
+        # This is ugly but I need to redirect stderr to stdout
+        fh = os.popen('sh -c "%s %s %s -u %s -r 1 -e \'%s\' localhost" 2>&1 ' % (BBFTP, f, p, user, cmd))
         return fh
-
 
     def _readSyslog(self, logfile="/var/log/bbftpd/bbftpd.log"):
@@ -231 +149 @@
 
 
+    def assertLines(self, lines, string):
+        """Assert that there is a line in string matching each regular expression in lines.
+        """
+
+        for line in lines:
+            self.assert_(re.search('^%s$' % line, string, re.M))
+
+
+
+class EmbeddedServerTestCase(BaseFixture):
+    """Test the bbftpd module.
+    """
+    
+    def testStartup(self):
+        lines = self._readSyslog()
+        # Give syslog time to flush it's logs.
+        time.sleep(1)
+        self.assertLines(['.*Starting bbftpd'], lines)
+
+    def testDir(self):
+        """Try connecting the client and listing a directory.
+        """
+
+
+        fh = self._runClient("dir %s" % DATADIR)
+        output = fh.read()
+
+
+        self.assertLines([r'dir .*/data', r' d .*/\.', r' d .*/\.\.',
+                        r' f .*/foo', r' f .*/bar', r' f .*/baz'], output)
+
+        lines = self._readSyslog()
+        self.assertLines(['.*Getting new bbftp connexion.*',
+                          r'.*Authz: MSG_LIST_V2 .*/test/data/\*.*',
+                          r'.*User TestCaseUser disconnected.*'], lines)
+
+
+    def testHandshake(self):
+        """Verify handshake messages are exchanged.
+        """
+
+        fh = self._runClient("dir .", debug=True)
+        output = fh.read()
+
+        self.assertLines(['Received Auth handshake: NDG-Delivery-server %s' % VERSION], output)
+
+        lines = self._readSyslog()
+        self.assertLines(['.*AuthContext received Auth message: NDG-Delivery-client %s' % VERSION], lines)
+
+    def testPrivateStr(self):
+        """Verify the private string is sent to server.
+        """
+
+        fh = self._runClient("dir .", privatestr="testPrivateStr")
+        output = fh.read()
+
+        lines = self._readSyslog()
+        self.assertLines(['.*AuthContext received privatestr: testPrivateStr'], lines)
+
+    def testRetr(self):
+        """Try retrieving a file.
+        """
+
+        tmp = tempfile.mktemp('test_bbftpd')
+        fh = self._runClient("get %s/foo %s" % (DATADIR, tmp))
+
+        # Check the client output
+        output = fh.read()
+        self.assertLines(['get.*nogzip'], output)
+
+        # Check retrieved file
+        self.assert_(os.system('diff --brief %s/foo %s' % (DATADIR, tmp)) == 0)
+        os.remove(tmp)
+
+        # Check syslog
+        lines = self._readSyslog()
+        self.assertLines(['.*Authz: RETR .*/foo', '.*GET TestCaseUser .*/foo.*'], lines)
+
+    def testStore(self):
+        """Try storing a file.
+        """
+
+
+        src = '%s/bar' % (DATADIR)
+        dest = '%s/new_bar' % (DATADIR)
+        os.system('cp %s %s' % (src, dest))
+        fh = self._runClient("put %s %s" % (src, dest))
+
+        # Check the client output
+        output = fh.read()
+        self.assertLines(['put .* nogzip'], output)
+
+
+        # Check sent file
+        self.assert_(os.system('diff --brief %s %s' % (dest, src)) == 0)
+        os.remove(dest)
+
+        # Check syslog
+        lines = self._readSyslog()
+        self.assertLines(['.*Authz: STORE .*/new_bar', '.*PUT TestCaseUser .*/new_bar.*'], lines)
+
+
+
+class AuthFailure(BaseFixture):
+    def setUp(self):
+        # We want to mark the beginning of this test case in syslog
+        syslog.openlog('test_embedded.py', 0, syslog.LOG_LOCAL0)
+        syslog.syslog(syslog.LOG_DEBUG, 'Starting EmbeddedServerTestCase')
+
+        self._startServer(authContext=FailingAuthContext())
+
+    def test(self):
+        """Fail authorisation
+        """
+
+        fh = self._runClient("dir .", debug=True)
+        output = fh.read()
+
+        lines = self._readSyslog()
+
+        self.assertLines(['.*Error while private authentication.*'], output)
+        self.assertLines(['.*bbftpd_private_auth failed.*'], lines)
+
 if __name__ == '__main__':
     unittest.main()