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

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

There are various changes here.

The server now calls python for authorisation. This is untested, although the test suite passes. More flesh on delivery.server and it's now used in the
test suite. The server now logs python exceptions to syslog from within C code.

  • 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
18VERSION = open('%s/VERSION' % HOME).read()
19NDG_MESSAGE_LEN = 256
20
21NDG_HANDSHAKE = "NDG-Delivery-server %s" % VERSION
22
23sys.path.append(BUILDDIR)
24
25import delivery.server as server
26
27
28class TestAuthHandler(server.BasicClientAuthHandler):
29    def authorise(self):
30        msg = self.recvCStr()
31
32        syslog.syslog(syslog.LOG_DEBUG, 'AuthContext received Auth message: %s' % msg)
33
34        self.send(NDG_HANDSHAKE)
35
36        privatestr = self.recvCStr()
37        syslog.syslog(syslog.LOG_DEBUG, "AuthContext received privatestr: %s" % privatestr)
38
39        return TestAuthzHandler(msg, "TestCaseUser")
40
41class TestAuthzHandler(server.LiberalAuthzHandler):
42    def __init__(self, version, username):
43        super(TestAuthzHandler, self).__init__(username)
44        self.version = version
45
46
47class TestFailAuthHandler(server.AuthHandler):
48    def authorise(self):
49        return None
50
51class BaseFixture(unittest.TestCase):
52    def setUp(self):
53        # We want to mark the beginning of this test case in syslog
54        syslog.openlog('test_embedded.py', 0, syslog.LOG_LOCAL0)
55        syslog.syslog(syslog.LOG_DEBUG, 'Starting EmbeddedServerTestCase')
56
57        self._startServer()
58
59    def tearDown(self):
60        self._stopServer()
61        syslog.syslog(syslog.LOG_DEBUG, 'Ended EmbeddedServerTestCase')
62        syslog.closelog()
63
64    #----------------------------------------------------------------------------------
65
66
67    def _startServer(self, authContext=None):
68        if not authContext:
69            authContext = TestAuthHandler()
70        # Start the server and store it's PID
71        self.pid = server.start(authContext, ['-l', 'DEBUG'])
72
73    def _stopServer(self):
74        # Stop the server process
75        os.kill(self.pid, signal.SIGTERM)
76        os.waitpid(self.pid, 0)
77        syslog.syslog(syslog.LOG_DEBUG, 'Stopping server')
78
79
80    def _runClient(self, cmd, debug=False, user="testcase", privatestr=None):
81        """Run the client.
82        """
83
84        if debug:
85            f = "-d"
86        else:
87            f = "-m"
88
89        if privatestr == None:
90            p = ""
91        else:
92            p = "-P %s" % repr(privatestr)
93
94        # This is ugly but I need to redirect stderr to stdout
95        fh = os.popen('sh -c "%s %s %s -u %s -r 1 -e \'%s\' localhost" 2>&1 ' % (BBFTP, f, p, user, cmd))
96        return fh
97
98    def _readSyslog(self, logfile="/var/log/bbftpd/bbftpd.log"):
99        """Get all bbftpd messages from syslog for this test case.
100
101        @note: This requires read access to the logfile
102        """
103
104        # Read the logfile into a buffer
105        log = open(logfile).readlines()
106
107        # Find the latest testcase marker
108        start_i = 0
109        for i in xrange(len(log)-1, -1, -1):
110            if re.search('test_embedded.py: Starting EmbeddedServerTestCase', log[i]):
111                start_i = i+1
112                break
113        if not start_i:
114            raise RuntimeError, "Can't find test case entry in syslog"
115
116        filtered_log = []
117        for line in log[start_i:]:
118            if re.search(r'test_bbftpd.py:|bbftpd .*:', line):
119                filtered_log.append(line)
120
121        return ''.join(filtered_log)
122
123
124    def _findLines(self, lines, string):
125        """Look for lines matching each regular expression in lines.
126       
127        @return: True if all lines are found, else False
128        """
129       
130        for line in lines:
131            if not re.search('^%s$' % line, string, re.M):
132                return False
133           
134        return True
135
136
137    def assertLines(self, lines, string):
138        """Assert that there is a line in string matching each regular expression in lines.
139        """
140
141        for line in lines:
142            self.assert_(re.search('^%s$' % line, string, re.M))
143
144
145
146class EmbeddedServerTestCase(BaseFixture):
147    """Test the bbftpd module.
148    """
149   
150    def testStartup(self):
151        lines = self._readSyslog()
152        # Give syslog time to flush it's logs.
153        time.sleep(1)
154        self.assertLines(['.*Starting bbftpd'], lines)
155
156    def testDir(self):
157        """Try connecting the client and listing a directory.
158        """
159
160
161        fh = self._runClient("dir %s" % DATADIR)
162        output = fh.read()
163
164
165        self.assertLines([r'dir .*/data', r' d .*/\.', r' d .*/\.\.',
166                        r' f .*/foo', r' f .*/bar', r' f .*/baz'], output)
167
168        lines = self._readSyslog()
169        self.assertLines(['.*Getting new bbftp connexion.*',
170                          r'.*Authz: MSG_LIST_V2 .*/test/data/\*.*',
171                          r'.*User TestCaseUser disconnected.*'], lines)
172
173
174    def testHandshake(self):
175        """Verify handshake messages are exchanged.
176        """
177
178        fh = self._runClient("dir .", debug=True)
179        output = fh.read()
180
181        self.assertLines(['Received Auth handshake: NDG-Delivery-server %s' % VERSION], output)
182
183        lines = self._readSyslog()
184        self.assertLines(['.*AuthContext received Auth message: NDG-Delivery-client %s' % VERSION], lines)
185
186    def testPrivateStr(self):
187        """Verify the private string is sent to server.
188        """
189
190        fh = self._runClient("dir .", privatestr="testPrivateStr")
191        output = fh.read()
192
193        lines = self._readSyslog()
194        self.assertLines(['.*AuthContext received privatestr: testPrivateStr'], lines)
195
196    def testRetr(self):
197        """Try retrieving a file.
198        """
199
200        tmp = tempfile.mktemp('test_bbftpd')
201        fh = self._runClient("get %s/foo %s" % (DATADIR, tmp))
202
203        # Check the client output
204        output = fh.read()
205
206        self.assertLines(['get.*nogzip'], output)
207
208        # Check retrieved file
209        self.assert_(os.system('diff --brief %s/foo %s' % (DATADIR, tmp)) == 0)
210        os.remove(tmp)
211
212        # Check syslog
213        lines = self._readSyslog()
214        self.assertLines(['.*Authz: RETR .*/foo', '.*GET TestCaseUser .*/foo.*'], lines)
215
216    def testStore(self):
217        """Try storing a file.
218        """
219
220
221        src = '%s/bar' % (DATADIR)
222        dest = '%s/new_bar' % (DATADIR)
223        os.system('cp %s %s' % (src, dest))
224        fh = self._runClient("put %s %s" % (src, dest))
225
226        # Check the client output
227        output = fh.read()
228        self.assertLines(['put .* nogzip'], output)
229
230
231        # Check sent file
232        self.assert_(os.system('diff --brief %s %s' % (dest, src)) == 0)
233        os.remove(dest)
234
235        # Check syslog
236        lines = self._readSyslog()
237        self.assertLines(['.*Authz: STORE .*/new_bar', '.*PUT TestCaseUser .*/new_bar.*'], lines)
238
239
240
241class AuthFailure(BaseFixture):
242    def setUp(self):
243        # We want to mark the beginning of this test case in syslog
244        syslog.openlog('test_embedded.py', 0, syslog.LOG_LOCAL0)
245        syslog.syslog(syslog.LOG_DEBUG, 'Starting EmbeddedServerTestCase')
246
247        self._startServer(authContext=TestFailAuthHandler())
248
249    def test(self):
250        """Fail authorisation
251        """
252
253        fh = self._runClient("dir .", debug=True)
254        output = fh.read()
255
256        lines = self._readSyslog()
257
258        self.assertLines(['.*Error while private authentication.*'], output)
259        self.assertLines(['.*bbftpd_private_auth failed.*'], lines)
260
261if __name__ == '__main__':
262    unittest.main()
Note: See TracBrowser for help on using the repository browser.