source: TI05-delivery/trunk/test/test_bbftpd.py @ 1369

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

Corrections to bbftp.h.in required rerunning autoconf.

  • Property svn:executable set to *
Line 
1#!/usr/bin/env python
2"""
3Tests for the bbftp daemon.
4
5@author: Stephen Pascoe
6@version: $Id$
7"""
8
9import unittest
10import os, signal, time, syslog
11from glob import glob
12import re, tempfile
13
14
15
16class ExtendedAPITestCase(unittest.TestCase):
17    """Test the basic bbftp private authentication API with stub extensions.
18
19    This test case runs bbftpd without python embedding and checks the auth API
20    is being called correctly.
21    """
22
23    BBFTP_LOG = './bbftpc.log'
24
25    def setUp(self):
26        self._getPaths()
27        # Check bbftpd isn't running
28        if self._getServerPids():
29            raise RuntimeError, 'A bbftpd process is already running'
30
31        # We want to mark the beginning of this test case in syslog
32        syslog.openlog('test_bbftpd.py', 0, syslog.LOG_LOCAL0)
33        syslog.syslog(syslog.LOG_DEBUG, 'Starting ExtendedAPITestCase')
34       
35    def tearDown(self):
36        syslog.syslog(syslog.LOG_DEBUG, 'Ended ExtendedAPITestCase')
37        syslog.closelog()
38
39    def testStartup(self):
40        """Start and stop the server.
41        """
42        try:
43            self._startServer()
44            pid = self._getServerPid()
45            self._stopServer(pid)
46        except Exception, e:
47            self.fail(e)
48
49    def testStartupWithSyslog(self):
50        """Start and stop the server, check syslog for INFO entries.
51        """
52        try:
53            self._startServer()
54            pid = self._getServerPid()
55            self._stopServer(pid)
56
57            lines = self._readLog()
58        except Exception, e:
59            self.fail(e)
60
61        self.assert_(self._findLines(['.*Starting bbftpd in background mode'], lines))
62
63    def testHandshake(self):
64        """Connect with client in verbose mode and test Auth handhsake is received.
65        """
66
67        self._startServer()
68        pid = self._getServerPid()
69
70        fh = self._runClient('dir .', debug=True)
71        output = fh.read()
72
73        self._stopServer(pid)
74
75        self.assert_(self._findLines(['Received Auth handshake: NDG-Delivery-server %s.*' % self.VERSION], output))
76
77        lines = self._readLog()
78        self.assert_(self._findLines(['.*Received auth message: NDG-Delivery-client %s.*' % self.VERSION], lines))
79
80    def testPrivateStr(self):
81        """Connect with a private string.  Check syslog for string value.
82        """
83
84        self._startServer()
85        pid = self._getServerPid()
86
87        fh = self._runClient('dir .', privatestr="Tester calling")
88        output = fh.read()
89
90        self._stopServer(pid)
91
92        lines = self._readLog()
93        self.assert_(self._findLines(['.*Private string: Tester calling.*'], lines))
94
95    def testDir(self):
96        """Try connecting the client and listing a directory.
97        """
98
99        self._startServer()
100        pid = self._getServerPid()
101
102        fh = self._runClient('dir %s' % self.DATADIR)
103        output = fh.read()
104
105        self._stopServer(pid)
106
107        self.assert_(self._findLines([r'dir .*/data', r' d .*/\.', r' d .*/\.\.',
108                                      r' f .*/foo', r' f .*/bar', r' f .*/baz'], output))       
109
110        lines = self._readLog()
111        self.assert_(self._findLines(['.*Getting new bbftp connexion.*',
112                                      r'.*Authz: MSG_LIST_V2 .*/test/data/\*.*',
113                                      r'.*User  disconnected.*'], lines))
114       
115    def testRetr(self):
116        """Try retrieving a file.
117        """
118
119        self._startServer()
120        pid = self._getServerPid()
121
122        tmp = tempfile.mktemp('test_bbftpd')
123        fh = self._runClient('get %s/foo %s' % (self.DATADIR, tmp))
124
125        # Check the client output
126        output = fh.read()
127        self.assert_(self._findLines(['get.*nogzip'], output))
128
129        self._stopServer(pid)
130
131        # Check retrieved file
132        self.assert_(os.system('diff --brief %s/foo %s' % (self.DATADIR, tmp)) == 0)
133        os.remove(tmp)
134
135        # Check syslog
136        lines = self._readLog()
137        self.assert_(self._findLines(['.*Authz: RETR .*/foo', '.*GET .*/foo.*'], lines))
138
139    def testStore(self):
140        """Try storing a file.
141        """
142
143        self._startServer()
144        pid = self._getServerPid()
145
146        src = '%s/bar' % (self.DATADIR)
147        dest = '%s/new_bar' % (self.DATADIR)
148        os.system('cp %s %s' % (src, dest))
149        fh = self._runClient('put %s %s' % (src, dest))
150
151        # Check the client output
152        output = fh.read()
153        self.assert_(self._findLines(['put .* nogzip'], output))
154
155        self._stopServer(pid)
156
157        # Check sent file
158        self.assert_(os.system('diff --brief %s %s' % (dest, src)) == 0)
159        os.remove(dest)
160
161        # Check syslog
162        lines = self._readLog()
163        self.assert_(self._findLines(['.*Authz: STORE .*/new_bar', '.*PUT .*/new_bar.*'], lines))
164
165    #------------------------------------------------------------------------------
166           
167    def _getPaths(self):
168        """Find the server executable and any other important files.
169        """
170        self.HOME = os.path.abspath(os.getenv('NDG_DELIVERY_HOME', os.curdir))
171        try:
172            self.BBFTPD = glob('%s/src/bbftp-server*/bbftpd/bbftpd' % self.HOME)[0]
173        except IndexError:
174            raise RuntimeError, 'Cannot find bbftpd executable'
175        try:
176            self.BBFTP = glob('%s/src/bbftp-client*/bbftpc/bbftp' % self.HOME)[0]
177        except:
178            raise RuntimeError, 'Cannot find bbftp executable'
179        dd = '%s/test/data' % self.HOME
180        if os.path.exists(dd):
181            self.DATADIR = dd
182        else:
183            raise RuntimeError, 'Cannot find test data directory'
184        self.VERSION = open('%s/VERSION' % self.HOME).read()
185
186
187    def _startServer(self):
188        #os.system('valgrind --trace-children=yes --log-file=bbftpd.vg %s -b -l DEBUG' % self.BBFTPD)
189        os.system('%s -b -l DEBUG' % self.BBFTPD)
190
191    def _getServerPids(self):
192        pids = os.popen('ps -C bbftpd -o %p --no-headers').readlines()
193        return [int(x.strip()) for x in pids]
194
195    def _getServerPid(self):
196        return self._getServerPids()[0]
197
198    def _stopServer(self, pid):
199        os.kill(pid, signal.SIGTERM)
200        # Wait upto 5 seconds for the server to stop
201        for x in range(10):
202            if pid not in self._getServerPids():
203                return
204            time.sleep(0.5)
205        raise RuntimeError, 'Failed to stop server'
206
207    def _readLog(self, logfile='./bbftpd.log'):
208        return open(logfile).read()
209
210    def _runClient(self, cmd, debug=False, user="testcase", privatestr=None):
211        """Run the client.
212        """
213
214        #print "Waiting for gdb to connect to server"
215        #os.system('ps -u spascoe')
216        #time.sleep(15)
217
218        if debug:
219            f = "-d"
220        else:
221            f = "-m"
222
223        if privatestr:
224            p = '-P %s' % repr(privatestr)
225        else:
226            p = ''
227       
228        #fh = os.popen('%s %s -u %s %s -e %s localhost' % (self.BBFTP, f, user, p, repr(cmd)))
229        os.system('%s %s -u %s %s -e %s localhost >%s' % (self.BBFTP, f, user, p, repr(cmd), self.BBFTP_LOG))
230        fh = open(self.BBFTP_LOG)
231        return fh
232
233    def _findLines(self, lines, string):
234        """Look for lines matching each regular expression in lines.
235       
236        @return: True if all lines are found, else False
237        """
238       
239        for line in lines:
240            if not re.search('^%s$' % line, string, re.M):
241                return False
242           
243        return True
244
245
246if __name__ == '__main__':
247    unittest.main()
Note: See TracBrowser for help on using the repository browser.