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

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

Implemented full client-server handshake. A version message is now
exchanged client->server then server->client. These messages are logged
in debug mode. Message passing is implemented in Python on the server (at the
moment this is only implemented within test/test_embedded.py).

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