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

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

Logging to syslog has been changed to use LOG_DEBUG. The tests have
been changed to assume messages are logged to /var/log/bbftpd/bbftpd.log.

  • 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
14class ExtendedAPITestCase(unittest.TestCase):
15    """Test the basic bbftp private authentication API with stub extensions.
16    """
17
18    def setUp(self):
19        self._getPaths()
20        # Check bbftpd isn't running
21        if self._getServerPids():
22            raise RuntimeError, 'A bbftpd process is already running'
23
24        # We want to mark the beginning of this test case in syslog
25        syslog.openlog('test_bbftpd.py', 0, syslog.LOG_LOCAL0)
26        syslog.syslog(syslog.LOG_DEBUG, 'Starting ExtendedAPITestCase')
27       
28    def tearDown(self):
29        syslog.syslog(syslog.LOG_DEBUG, 'Ended ExtendedAPITestCase')
30        syslog.closelog()
31
32    def testStartup(self):
33        """Start and stop the server.
34        """
35        try:
36            self._startServer()
37            pid = self._getServerPid()
38            self._stopServer(pid)
39        except Exception, e:
40            self.fail(e)
41
42    def testStartupWithSyslog(self):
43        """Start and stop the server, check syslog for INFO entries.
44        """
45        try:
46            self._startServer()
47            pid = self._getServerPid()
48            self._stopServer(pid)
49
50            lines = self._readSyslog()
51        except Exception, e:
52            self.fail(e)
53
54        self.assert_(self._findLines(['.*initialising private auth.*', '.*finalising private auth.*'], lines))
55
56    def testDir(self):
57        """Try connecting the client and listing a directory.
58        """
59
60        self._startServer()
61        pid = self._getServerPid()
62
63        fh = self._runClient('-e "dir %s"' % self.DATADIR)
64        output = fh.read()
65
66        self._stopServer(pid)
67
68        self.assert_(self._findLines([r'dir .*/data', r' d .*/\.', r' d .*/\.\.',
69                                      r' f .*/foo', r' f .*/bar', r' f .*/baz'], output))       
70
71        lines = self._readSyslog()
72        self.assert_(self._findLines(['.*Getting new bbftp connexion.*',
73                                      '.*Received auth message: NDG-bbFTP.*',
74                                      r'.*Authz: MSG_LIST_V2 .*/test/data/\*.*',
75                                      r'.*User  disconnected.*'], lines))
76       
77    def testRetr(self):
78        """Try retrieving a file.
79        """
80
81        self._startServer()
82        pid = self._getServerPid()
83
84        tmp = tempfile.mktemp('test_bbftpd')
85        fh = self._runClient('-e "get %s/foo %s"' % (self.DATADIR, tmp))
86
87        # Check the client output
88        output = fh.read()
89        self.assert_(self._findLines(['get.*nogzip'], output))
90
91        self._stopServer(pid)
92
93        # Check retrieved file
94        self.assert_(os.system('diff --brief %s/foo %s' % (self.DATADIR, tmp)) == 0)
95        os.remove(tmp)
96
97        # Check syslog
98        lines = self._readSyslog()
99        self.assert_(self._findLines(['.*Authz: RETR .*/foo', '.*GET .*/foo.*'], lines))
100
101    def testStore(self):
102        """Try storing a file.
103        """
104
105        self._startServer()
106        pid = self._getServerPid()
107
108        src = '%s/bar' % (self.DATADIR)
109        dest = '%s/new_bar' % (self.DATADIR)
110        os.system('cp %s %s' % (src, dest))
111        fh = self._runClient('-e "put %s %s"' % (src, dest))
112
113        # Check the client output
114        output = fh.read()
115        self.assert_(self._findLines(['put .* nogzip'], output))
116
117        self._stopServer(pid)
118
119        # Check sent file
120        self.assert_(os.system('diff --brief %s %s' % (dest, src)) == 0)
121        os.remove(dest)
122
123        # Check syslog
124        lines = self._readSyslog()
125        self.assert_(self._findLines(['.*Authz: STORE .*/new_bar', '.*PUT .*/new_bar.*'], lines))
126
127    #------------------------------------------------------------------------------
128           
129    def _getPaths(self):
130        """Find the server executable and any other important files.
131        """
132        self.HOME = os.path.abspath(os.getenv('NDG_DELIVERY_HOME', os.curdir))
133        try:
134            self.BBFTPD = glob('%s/src/bbftp-server*/bbftpd/bbftpd' % self.HOME)[0]
135        except IndexError:
136            raise RuntimeError, 'Cannot find bbftpd executable'
137        try:
138            self.BBFTP = glob('%s/src/bbftp-client*/bbftpc/bbftp' % self.HOME)[0]
139        except:
140            raise RuntimeError, 'Cannot find bbftp executable'
141        dd = '%s/test/data' % self.HOME
142        if os.path.exists(dd):
143            self.DATADIR = dd
144        else:
145            raise RuntimeError, 'Cannot find test data directory'
146
147    def _startServer(self):
148        os.system('%s -b -l DEBUG' % self.BBFTPD)
149
150    def _getServerPids(self):
151        pids = os.popen('ps -C bbftpd -o %p --no-headers').readlines()
152        return [int(x.strip()) for x in pids]
153
154    def _getServerPid(self):
155        return self._getServerPids()[0]
156
157    def _stopServer(self, pid):
158        os.kill(pid, signal.SIGTERM)
159        # Wait upto 5 seconds for the server to stop
160        for x in range(10):
161            if pid not in self._getServerPids():
162                return
163            time.sleep(0.5)
164        raise RuntimeError, 'Failed to stop server'
165
166    def _readSyslog(self, logfile="/var/log/bbftpd/bbftpd.log"):
167        """Get all bbftpd messages from syslog for this test case.
168
169        @note: This requires read access to the logfile
170        """
171
172        # Read the logfile into a buffer
173        log = open(logfile).readlines()
174
175        # Find the latest testcase marker
176        start_i = 0
177        for i in xrange(len(log)-1, -1, -1):
178            if re.search('test_bbftpd.py: Starting ExtendedAPITestCase', log[i]):
179                start_i = i+1
180                break
181        if not start_i:
182            raise RuntimeError, "Can't find test case entry in syslog"
183
184        filtered_log = []
185        for line in log[start_i:]:
186            if re.search(r'test_bbftpd.py:|bbftpd .*:', line):
187                filtered_log.append(line)
188
189        return ''.join(filtered_log)
190
191    def _runClient(self, *args):
192        """Run the client with th -m option and return a file handle of the output.
193        """
194        fh = os.popen('%s -m -u testcase %s localhost' % (self.BBFTP, ' '.join(args)))
195        return fh
196
197    def _findLines(self, lines, string):
198        """Look for lines matching each regular expression in lines.
199       
200        @return: True if all lines are found, else False
201        """
202       
203        for line in lines:
204            if not re.search('^%s$' % line, string, re.M):
205                return False
206           
207        return True
208
209if __name__ == '__main__':
210    unittest.main()
Note: See TracBrowser for help on using the repository browser.