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

#!/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, stat
from glob import glob
import re, tempfile, getopt

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 delivery.server as server


class TestAuthHandler(server.BasicClientAuthHandler):
    
    def authorise(self):
        msg = self.recvCStr()

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

        self.send(NDG_HANDSHAKE)

        privatestr = self.recvCStr()
        syslog.syslog(syslog.LOG_DEBUG, "AuthContext received privatestr: %s" % privatestr)

        return self.makeAuthzHandler(msg, "TestCaseUser")

    def makeAuthzHandler(self, msg, user):
        return TestAuthzHandler(msg, "TestCaseUser")

class TestAuthzHandler(server.LiberalAuthzHandler):
    def __init__(self, version, username):
        super(TestAuthzHandler, self).__init__(username)
        self.version = version




class TestFailAuthHandler(server.AuthHandler):
    def authorise(self):
        raise server.AuthenticationFailure, "TestFailAuthHandler"


class TestPermAuthHandler(TestAuthHandler):
    def makeAuthzHandler(self, msg, user):
        return TestPermAuthzHandler(msg, "TestCaseUser")

class TestPermAuthzHandler(TestAuthzHandler):
    """Check's the path's other read permission and authorises accordingly.
    """

    def authzControl(self, msgcode, transferopt, path):
        # Ignore what is being authorised for now, just authorise if read allowed
        return self.authzPath(path)

    def authzRetr(self, path):
        return self.authzPath(path)

    def authzSend(self, path):
        return self.authzPath(path)

    def authzPath(self, path):
        mode = stat.S_IMODE(os.stat(path)[stat.ST_MODE])
        if (mode & 0004):
            syslog.syslog(syslog.LOG_DEBUG, 'TestPermAuthzHandler OK')
            return True
        else:
            syslog.syslog(syslog.LOG_DEBUG, 'TestPermAuthzHandler FAIL')
            raise server.AuthorisationFailure, "TestPermAuthzHandler: no read perms"


class BaseFixture(unittest.TestCase):

    authContext = TestAuthHandler()

    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 TestCase')

        self._startServer()

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

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


    def _startServer(self):
        # Start the server and store it's PID
        self.pid = server.start(self.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 TestCase', 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 AuthOK(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()

        print output

        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 AuthFail(BaseFixture):
    authContext = TestFailAuthHandler()

    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)


class Authz(BaseFixture):
    authContext = TestPermAuthHandler()

    def setUp(self):
        super(Authz, self).setUp()

        # Set modes on test files
        os.chmod('%s/bar' % (DATADIR), 0640)
        os.chmod('%s/foo' % (DATADIR), 0655)

    def testOK(self):
        """Test dir of a readable file.
        """

        fh = self._runClient("stat %s/foo" % (DATADIR), debug=True)
        output = fh.read()

        lines = self._readSyslog()


        self.assertLines(['Connection and authentication correct', 'stat /.*/foo OK'], output)
        self.assertLines(['.*Authz: MSG_STAT .*/foo', '.*TestPermAuthzHandler OK'], lines)

    def testFail(self):
        """Test dir of a non-readable file.
        """

        fh = self._runClient("stat %s/bar" % (DATADIR), debug=True)
        output = fh.read()

        lines = self._readSyslog()

        self.assertLines(['BBFTP-ERROR.* ndg_authz_control: AuthorisationFailure.* no read perms',
                          'stat /.*/bar FAILED'], output)
        self.assertLines(['.*Authz: MSG_STAT .*/bar', '.* TestPermAuthzHandler FAIL',
                          '.* ndg_authz_control: AuthorisationFailure.* no read perms'], lines)



import delivery.bbftpc

class PythonClientAuthOK(AuthOK):
    """Repeat AuthOK tests with a python-embedded client.
    """

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

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

        # Tell bbftp to dump output to a file which is then opened for reading.
        tmp = tempfile.mktemp('test_bbftpd')
        args = ['-e', cmd, f, '-u', user, '-o', tmp, '-f', tmp, '-r', '1']

        if privatestr != None:
            args += ['-P', privatestr]

        delivery.bbftpc.run(args + ['localhost'])

        fh.open(tmp)
        os.remove(tmp)

        return fh

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