source: TI05-delivery/trunk/test/test_embedded.py @ 1073

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI05-delivery/trunk/test/test_embedded.py@1073
Revision 1073, 4.9 KB checked in by spascoe, 13 years ago (diff)

The embedded bbftpd server now performs a callback to python to check
authentication. The callback works but it can't do anything at the moment.

  • Property svn:executable set to *
Line 
1#!/usr/bin/env python
2"""
3Tests for the bbftp daemon when embedded in Python.
4
5@author: Stephen Pascoe
6@version: $Id$
7"""
8
9import unittest
10import sys, os, signal, time, syslog
11from glob import glob
12import re, tempfile
13
14HOME = os.path.abspath(os.getenv('NDG_DELIVERY_HOME', os.curdir))
15BUILDDIR = glob('%s/build/lib.*' % HOME)[0]
16BBFTP = glob('%s/src/bbftp-client*/bbftpc/bbftp' % HOME)[0]
17DATADIR = '%s/test/data' % HOME
18sys.path.append(BUILDDIR)
19import bbftpd
20       
21class AuthContext(object):
22    def authorise(self):
23        # Return a dummy authzContext
24        return "authorised :-)"
25
26class EmbeddedServerTestCase(unittest.TestCase):
27    """Test the bbftpd module.
28    """
29   
30    def setUp(self):
31        # We want to mark the beginning of this test case in syslog
32        syslog.openlog('test_embedded.py', 0, syslog.LOG_LOCAL0)
33        syslog.syslog(syslog.LOG_DEBUG, 'Starting EmbeddedServerTestCase')
34
35        self._startServer()
36
37    def tearDown(self):
38        self._stopServer()
39        syslog.syslog(syslog.LOG_DEBUG, 'Ended EmbeddedServerTestCase')
40        syslog.closelog()
41
42    def testStartup(self):
43        lines = self._readSyslog()
44        self.assert_(self._findLines(['.*Starting bbftpd'], lines))
45
46    def testDir(self):
47        """Try connecting the client and listing a directory.
48        """
49
50
51        fh = self._runClient('-e "dir %s"' % DATADIR)
52        output = fh.read()
53
54
55        self.assert_(self._findLines([r'dir .*/data', r' d .*/\.', r' d .*/\.\.',
56                                      r' f .*/foo', r' f .*/bar', r' f .*/baz'], output))       
57
58        lines = self._readSyslog()
59        self.assert_(self._findLines(['.*Getting new bbftp connexion.*',
60                                      '.*Received auth message: NDG-bbFTP.*',
61                                      r'.*Authz: MSG_LIST_V2 .*/test/data/\*.*',
62                                      r'.*User  disconnected.*'], lines))
63
64
65    def testRetr(self):
66        """Try retrieving a file.
67        """
68
69        tmp = tempfile.mktemp('test_bbftpd')
70        fh = self._runClient('-e "get %s/foo %s"' % (DATADIR, tmp))
71
72        # Check the client output
73        output = fh.read()
74        self.assert_(self._findLines(['get.*nogzip'], output))
75
76        # Check retrieved file
77        self.assert_(os.system('diff --brief %s/foo %s' % (DATADIR, tmp)) == 0)
78        os.remove(tmp)
79
80        # Check syslog
81        lines = self._readSyslog()
82        self.assert_(self._findLines(['.*Authz: RETR .*/foo', '.*GET .*/foo.*'], lines))
83
84    def testStore(self):
85        """Try storing a file.
86        """
87
88
89        src = '%s/bar' % (DATADIR)
90        dest = '%s/new_bar' % (DATADIR)
91        os.system('cp %s %s' % (src, dest))
92        fh = self._runClient('-e "put %s %s"' % (src, dest))
93
94        # Check the client output
95        output = fh.read()
96        self.assert_(self._findLines(['put .* nogzip'], output))
97
98
99        # Check sent file
100        self.assert_(os.system('diff --brief %s %s' % (dest, src)) == 0)
101        os.remove(dest)
102
103        # Check syslog
104        lines = self._readSyslog()
105        self.assert_(self._findLines(['.*Authz: STORE .*/new_bar', '.*PUT .*/new_bar.*'], lines))
106
107
108    #----------------------------------------------------------------------------------
109
110
111    def _startServer(self):
112        # Start the server and store it's PID
113        self.pid = bbftpd.run(AuthContext(), ['-l', 'DEBUG'])
114
115    def _stopServer(self):
116        # Stop the server process
117        os.kill(self.pid, signal.SIGTERM)
118        os.waitpid(self.pid, 0)
119        syslog.syslog(syslog.LOG_DEBUG, 'Stopping server')
120
121    def _runClient(self, *args):
122        """Run the client with th -m option and return a file handle of the output.
123        """
124        fh = os.popen('%s -m -u testcase %s localhost' % (BBFTP, ' '.join(args)))
125        return fh
126
127    def _readSyslog(self, logfile="/var/log/bbftpd/bbftpd.log"):
128        """Get all bbftpd messages from syslog for this test case.
129
130        @note: This requires read access to the logfile
131        """
132
133        # Read the logfile into a buffer
134        log = open(logfile).readlines()
135
136        # Find the latest testcase marker
137        start_i = 0
138        for i in xrange(len(log)-1, -1, -1):
139            if re.search('test_embedded.py: Starting EmbeddedServerTestCase', log[i]):
140                start_i = i+1
141                break
142        if not start_i:
143            raise RuntimeError, "Can't find test case entry in syslog"
144
145        filtered_log = []
146        for line in log[start_i:]:
147            if re.search(r'test_bbftpd.py:|bbftpd .*:', line):
148                filtered_log.append(line)
149
150        return ''.join(filtered_log)
151
152
153    def _findLines(self, lines, string):
154        """Look for lines matching each regular expression in lines.
155       
156        @return: True if all lines are found, else False
157        """
158       
159        for line in lines:
160            if not re.search('^%s$' % line, string, re.M):
161                return False
162           
163        return True
164
165
166if __name__ == '__main__':
167    unittest.main()
Note: See TracBrowser for help on using the repository browser.