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

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

Implemented setting the username from python during authentication.

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