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

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

Slight modifications to bbftpd module and a new test case for testing
the embedded server.

  • Property svn:executable set to *
RevLine 
[1070]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 EmbeddedServerTestCase(unittest.TestCase):
22    """Test the bbftpd module.
23    """
24   
25    def setUp(self):
26        # We want to mark the beginning of this test case in syslog
27        syslog.openlog('test_embedded.py', 0, syslog.LOG_LOCAL0)
28        syslog.syslog(syslog.LOG_DEBUG, 'Starting EmbeddedServerTestCase')
29
30        self._startServer()
31
32    def tearDown(self):
33        self._stopServer()
34        syslog.syslog(syslog.LOG_DEBUG, 'Ended EmbeddedServerTestCase')
35        syslog.closelog()
36
37    def testStartup(self):
38        lines = self._readSyslog()
39        self.assert_(self._findLines(['.*Starting bbftpd'], lines))
40
41    def testDir(self):
42        """Try connecting the client and listing a directory.
43        """
44
45
46        fh = self._runClient('-e "dir %s"' % DATADIR)
47        output = fh.read()
48
49
50        self.assert_(self._findLines([r'dir .*/data', r' d .*/\.', r' d .*/\.\.',
51                                      r' f .*/foo', r' f .*/bar', r' f .*/baz'], output))       
52
53        lines = self._readSyslog()
54        self.assert_(self._findLines(['.*Getting new bbftp connexion.*',
55                                      '.*Received auth message: NDG-bbFTP.*',
56                                      r'.*Authz: MSG_LIST_V2 .*/test/data/\*.*',
57                                      r'.*User  disconnected.*'], lines))
58
59
60    #----------------------------------------------------------------------------------
61
62
63    def _startServer(self):
64        # Start the server and store it's PID
65        self.pid = bbftpd.run(['-l', 'DEBUG'])
66
67    def _stopServer(self):
68        # Stop the server process
69        os.kill(self.pid, signal.SIGTERM)
70        os.waitpid(self.pid, 0)
71        syslog.syslog(syslog.LOG_DEBUG, 'Stopping server')
72
73    def _runClient(self, *args):
74        """Run the client with th -m option and return a file handle of the output.
75        """
76        fh = os.popen('%s -m -u testcase %s localhost' % (BBFTP, ' '.join(args)))
77        return fh
78
79    def _readSyslog(self, logfile="/var/log/bbftpd/bbftpd.log"):
80        """Get all bbftpd messages from syslog for this test case.
81
82        @note: This requires read access to the logfile
83        """
84
85        # Read the logfile into a buffer
86        log = open(logfile).readlines()
87
88        # Find the latest testcase marker
89        start_i = 0
90        for i in xrange(len(log)-1, -1, -1):
91            if re.search('test_embedded.py: Starting EmbeddedServerTestCase', log[i]):
92                start_i = i+1
93                break
94        if not start_i:
95            raise RuntimeError, "Can't find test case entry in syslog"
96
97        filtered_log = []
98        for line in log[start_i:]:
99            if re.search(r'test_bbftpd.py:|bbftpd .*:', line):
100                filtered_log.append(line)
101
102        return ''.join(filtered_log)
103
104
105    def _findLines(self, lines, string):
106        """Look for lines matching each regular expression in lines.
107       
108        @return: True if all lines are found, else False
109        """
110       
111        for line in lines:
112            if not re.search('^%s$' % line, string, re.M):
113                return False
114           
115        return True
116
117
118if __name__ == '__main__':
119    unittest.main()
Note: See TracBrowser for help on using the repository browser.