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

#!/usr/bin/env python
"""
Tests for the bbftp daemon when embedded in Python.

@author: Stephen Pascoe
@version: $Id$
"""

import unittest
import sys, os, signal, time, syslog
from glob import glob
import re, tempfile

HOME = os.path.abspath(os.getenv('NDG_DELIVERY_HOME', os.curdir))
BUILDDIR = glob('%s/build/lib.*' % HOME)[0]
BBFTP = glob('%s/src/bbftp-client*/bbftpc/bbftp' % HOME)[0]
DATADIR = '%s/test/data' % HOME
VERSION = open('%s/VERSION' % HOME).read()
NDG_MESSAGE_LEN = 256

NDG_HANDSHAKE = "NDG-Delivery-server %s" % VERSION

sys.path.append(BUILDDIR)
import bbftpd
        
class AuthzContext:
    def __init__(self, version_msg, user):
        self.version = version_msg
        self.username = user

class AuthContext:
    def authorise(self):
        # Read the auth version message
        msg = bbftpd.recv()
        # Trim to first '\0'
        x = msg.find('\0')
        if x:
            msg = msg[:x]

        syslog.syslog(syslog.LOG_DEBUG, 'AuthContext received Auth message: %s' % msg)

        # Send the response
        msg = NDG_HANDSHAKE + '\0' * (NDG_MESSAGE_LEN - len(NDG_HANDSHAKE))
        bbftpd.send(msg)

        # Get privatestr
        privatestr = bbftpd.recv()
        # Trim to first '\0'
        x = privatestr.find('\0')
        if x:
            privatestr = privatestr[:x]
        syslog.syslog(syslog.LOG_DEBUG, "AuthContext received privatestr: %s" % privatestr)

        return AuthzContext(msg, "TestCaseUser")


class FailingAuthContext(AuthContext):
    def authorise(self):
        az = super(FailingAuthContext).authorise()
        return None




class BaseFixture(unittest.TestCase):
    def setUp(self):
        # We want to mark the beginning of this test case in syslog
        syslog.openlog('test_embedded.py', 0, syslog.LOG_LOCAL0)
        syslog.syslog(syslog.LOG_DEBUG, 'Starting EmbeddedServerTestCase')

        self._startServer()

    def tearDown(self):
        self._stopServer()
        syslog.syslog(syslog.LOG_DEBUG, 'Ended EmbeddedServerTestCase')
        syslog.closelog()

    #----------------------------------------------------------------------------------


    def _startServer(self, authContext=None):
        if not authContext:
            authContext = AuthContext()
        # Start the server and store it's PID
        self.pid = bbftpd.run(authContext, ['-l', 'DEBUG'])

    def _stopServer(self):
        # Stop the server process
        os.kill(self.pid, signal.SIGTERM)
        os.waitpid(self.pid, 0)
        syslog.syslog(syslog.LOG_DEBUG, 'Stopping server')


    def _runClient(self, cmd, debug=False, user="testcase", privatestr=None):
        """Run the client.
        """

        if debug:
            f = "-d"
        else:
            f = "-m"

        if privatestr == None:
            p = ""
        else:
            p = "-P %s" % repr(privatestr)

        # This is ugly but I need to redirect stderr to stdout
        fh = os.popen('sh -c "%s %s %s -u %s -r 1 -e \'%s\' localhost" 2>&1 ' % (BBFTP, f, p, user, cmd))
        return fh

    def _readSyslog(self, logfile="/var/log/bbftpd/bbftpd.log"):
        """Get all bbftpd messages from syslog for this test case.

        @note: This requires read access to the logfile
        """

        # Read the logfile into a buffer
        log = open(logfile).readlines()

        # Find the latest testcase marker
        start_i = 0
        for i in xrange(len(log)-1, -1, -1):
            if re.search('test_embedded.py: Starting EmbeddedServerTestCase', log[i]):
                start_i = i+1
                break
        if not start_i:
            raise RuntimeError, "Can't find test case entry in syslog"

        filtered_log = []
        for line in log[start_i:]:
            if re.search(r'test_bbftpd.py:|bbftpd .*:', line):
                filtered_log.append(line)

        return ''.join(filtered_log)


    def _findLines(self, lines, string):
        """Look for lines matching each regular expression in lines.
        
        @return: True if all lines are found, else False
        """
        
        for line in lines:
            if not re.search('^%s$' % line, string, re.M):
                return False
            
        return True


    def assertLines(self, lines, string):
        """Assert that there is a line in string matching each regular expression in lines.
        """

        for line in lines:
            self.assert_(re.search('^%s$' % line, string, re.M))



class EmbeddedServerTestCase(BaseFixture):
    """Test the bbftpd module.
    """
    
    def testStartup(self):
        lines = self._readSyslog()
        # Give syslog time to flush it's logs.
        time.sleep(1)
        self.assertLines(['.*Starting bbftpd'], lines)

    def testDir(self):
        """Try connecting the client and listing a directory.
        """


        fh = self._runClient("dir %s" % DATADIR)
        output = fh.read()


        self.assertLines([r'dir .*/data', r' d .*/\.', r' d .*/\.\.',
                        r' f .*/foo', r' f .*/bar', r' f .*/baz'], output)

        lines = self._readSyslog()
        self.assertLines(['.*Getting new bbftp connexion.*',
                          r'.*Authz: MSG_LIST_V2 .*/test/data/\*.*',
                          r'.*User TestCaseUser disconnected.*'], lines)


    def testHandshake(self):
        """Verify handshake messages are exchanged.
        """

        fh = self._runClient("dir .", debug=True)
        output = fh.read()

        self.assertLines(['Received Auth handshake: NDG-Delivery-server %s' % VERSION], output)

        lines = self._readSyslog()
        self.assertLines(['.*AuthContext received Auth message: NDG-Delivery-client %s' % VERSION], lines)

    def testPrivateStr(self):
        """Verify the private string is sent to server.
        """

        fh = self._runClient("dir .", privatestr="testPrivateStr")
        output = fh.read()

        lines = self._readSyslog()
        self.assertLines(['.*AuthContext received privatestr: testPrivateStr'], lines)

    def testRetr(self):
        """Try retrieving a file.
        """

        tmp = tempfile.mktemp('test_bbftpd')
        fh = self._runClient("get %s/foo %s" % (DATADIR, tmp))

        # Check the client output
        output = fh.read()
        self.assertLines(['get.*nogzip'], output)

        # Check retrieved file
        self.assert_(os.system('diff --brief %s/foo %s' % (DATADIR, tmp)) == 0)
        os.remove(tmp)

        # Check syslog
        lines = self._readSyslog()
        self.assertLines(['.*Authz: RETR .*/foo', '.*GET TestCaseUser .*/foo.*'], lines)

    def testStore(self):
        """Try storing a file.
        """


        src = '%s/bar' % (DATADIR)
        dest = '%s/new_bar' % (DATADIR)
        os.system('cp %s %s' % (src, dest))
        fh = self._runClient("put %s %s" % (src, dest))

        # Check the client output
        output = fh.read()
        self.assertLines(['put .* nogzip'], output)


        # Check sent file
        self.assert_(os.system('diff --brief %s %s' % (dest, src)) == 0)
        os.remove(dest)

        # Check syslog
        lines = self._readSyslog()
        self.assertLines(['.*Authz: STORE .*/new_bar', '.*PUT TestCaseUser .*/new_bar.*'], lines)



class AuthFailure(BaseFixture):
    def setUp(self):
        # We want to mark the beginning of this test case in syslog
        syslog.openlog('test_embedded.py', 0, syslog.LOG_LOCAL0)
        syslog.syslog(syslog.LOG_DEBUG, 'Starting EmbeddedServerTestCase')

        self._startServer(authContext=FailingAuthContext())

    def test(self):
        """Fail authorisation
        """

        fh = self._runClient("dir .", debug=True)
        output = fh.read()

        lines = self._readSyslog()

        self.assertLines(['.*Error while private authentication.*'], output)
        self.assertLines(['.*bbftpd_private_auth failed.*'], lines)

if __name__ == '__main__':
    unittest.main()