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

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

Python code can read an authorisation message from the client.

Test cases confirm message passing client->server->python is working.

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