Changeset 1333
- Timestamp:
- 27/07/06 11:56:03 (15 years ago)
- Location:
- TI05-delivery/trunk
- Files:
-
- 2 edited
Legend:
- Unmodified
- Added
- Removed
-
TI05-delivery/trunk/lib/python/pybbftp/server.py
r1289 r1333 56 56 57 57 58 # Import all LOG_* levels from syslog. 59 import syslog 60 for x in dir(syslog): 61 if x[:4] == 'LOG_': 62 globals()[x] = getattr(syslog, x) 63 58 64 #-------------------------------------------------------------------------------------------------------------- 59 65 … … 135 141 raise NotImplementedError 136 142 143 def log(self, priority, msg): 144 """Send a message to the bbftpd logger. 145 146 @param priority one of server.LOG_* 147 @param msg the message to log 148 """ 149 bbftpd.log(priority, msg) 150 137 151 138 152 class AuthzHandler(object): … … 172 186 173 187 raise NotImplementedError 174 188 189 def log(self, priority, msg): 190 """Send a message to the bbftpd logger. 191 192 @param priority one of server.LOG_* 193 @param msg the message to log 194 """ 195 bbftpd.log(priority, msg) 196 175 197 176 198 #-------------------------------------------------------------------------------------------------------------- -
TI05-delivery/trunk/test/test_embedded.py
r1332 r1333 8 8 9 9 import unittest 10 import sys, os, signal, time, s yslog, stat10 import sys, os, signal, time, stat 11 11 from glob import glob 12 12 import re, tempfile, getopt … … 37 37 38 38 msg = self.recv() 39 self.log Message('AuthContext received Auth message: %s' % msg)39 self.log(server.LOG_DEBUG, 'AuthContext received Auth message: %s' % msg) 40 40 41 41 self.send(NDG_HANDSHAKE) 42 42 43 43 privatestr = self.recv() 44 self.log Message("AuthContext received privatestr: %s" % privatestr)44 self.log(server.LOG_DEBUG, "AuthContext received privatestr: %s" % privatestr) 45 45 46 46 return self.makeAuthzHandler(msg, "TestCaseUser") … … 49 49 return TestAuthzHandler(msg, "TestCaseUser") 50 50 51 def logMessage(self, msg):52 server.bbftpd.log(syslog.LOG_DEBUG, msg)53 51 54 52 class TestAuthzHandler(server.LiberalAuthzHandler): … … 74 72 75 73 def authzControl(self, msgcode, transferopt, path): 76 se rver.bbftpd.log(syslog.LOG_DEBUG, 'TestPermAuthzHandler.authzControl: msgcode = %s' % hex(msgcode))74 self.log(server.LOG_DEBUG, 'TestPermAuthzHandler.authzControl: msgcode = %s' % hex(msgcode)) 77 75 return self.authzPath(path) 78 76 … … 86 84 mode = stat.S_IMODE(os.stat(path)[stat.ST_MODE]) 87 85 if (mode & 0004): 88 se rver.bbftpd.log(syslog.LOG_DEBUG, 'TestPermAuthzHandler OK')86 self.log(server.LOG_DEBUG, 'TestPermAuthzHandler OK') 89 87 return True 90 88 else: 91 se rver.bbftpd.log(syslog.LOG_DEBUG, 'TestPermAuthzHandler FAIL')89 self.log(server.LOG_DEBUG, 'TestPermAuthzHandler FAIL') 92 90 raise server.AuthorisationFailure, "TestPermAuthzHandler: no read perms" 93 91 … … 98 96 99 97 def setUp(self): 100 # We want to mark the beginning of this test case in syslog101 #syslog.openlog('test_embedded.py', 0, syslog.LOG_LOCAL0)102 #syslog.syslog(syslog.LOG_DEBUG, 'Starting TestCase')103 104 98 self._clearLog() 105 99 self._startServer() … … 107 101 def tearDown(self): 108 102 self._stopServer() 109 #syslog.syslog(syslog.LOG_DEBUG, 'Ended TestCase')110 #syslog.closelog()111 103 112 104 #---------------------------------------------------------------------------------- … … 121 113 os.kill(self.pid, signal.SIGTERM) 122 114 os.waitpid(self.pid, 0) 123 #syslog.syslog(syslog.LOG_DEBUG, 'Stopping server')124 115 125 116 … … 148 139 return open(logfile).read() 149 140 150 151 def _readSyslog(self, logfile="/var/log/bbftpd/bbftpd.log"):152 """Get all bbftpd messages from syslog for this test case.153 154 @note: This requires read access to the logfile155 """156 157 # Read the logfile into a buffer158 log = open(logfile).readlines()159 160 # Find the latest testcase marker161 start_i = 0162 for i in xrange(len(log)-1, -1, -1):163 if re.search('test_embedded.py: Starting TestCase', log[i]):164 start_i = i+1165 break166 if not start_i:167 raise RuntimeError, "Can't find test case entry in syslog"168 169 filtered_log = []170 for line in log[start_i:]:171 if re.search(r'test_bbftpd.py:|bbftpd .*:', line):172 filtered_log.append(line)173 174 return ''.join(filtered_log)175 141 176 142 … … 203 169 def testStartup(self): 204 170 lines = self._readLog() 205 # Give syslog time to flush it's logs.206 time.sleep(1)207 171 self.assertLines(['.*Starting bbftpd'], lines) 208 172 … … 264 228 os.remove(tmp) 265 229 266 # Check syslog230 # Check log 267 231 lines = self._readLog() 268 232 self.assertLines(['.*Authz: RETR .*/foo', '.*GET TestCaseUser .*/foo.*'], lines) … … 287 251 os.remove(dest) 288 252 289 # Check syslog253 # Check log 290 254 lines = self._readLog() 291 255 self.assertLines(['.*Authz: STORE .*/new_bar', '.*PUT TestCaseUser .*/new_bar.*'], lines)
Note: See TracChangeset
for help on using the changeset viewer.