#!/usr/bin/python '''executes a function in a Python module Copyright (c) 2011 jc.unternet.net licensed under GPL for pytest to work, the module under test must end with something like this: if __name__ == '__main__': command = os.path.splitext(os.path.basename(sys.argv[0]))[0] print eval(command)(*sys.argv[1:]) what happens: when you run ./pytest.py, this searches the current directory for .py files, creates a subdirectory of the same name for each, and mounts a virtual filesystem under the directory. for example, the function bleah in blarg.py can be tested as: ./blarg/bleah arg1 arg2 after you fusermount -u blarg, the blarg folder is automatically deleted. ''' import os, sys, re, stat, errno, syslog, fcntl, fuse from fuse import Fuse fuse.fuse_python_api = (0, 2) DEBUGGING = bool(os.getenv('DEBUGGING')) class TestStat(fuse.Stat): 'subclass of fuse.Stat applicable to this application' def __init__(self, filename): attributes = ['st_mode', 'st_ino', 'st_dev', 'st_nlink', 'st_uid', 'st_gid', 'st_size', 'st_atime', 'st_mtime', 'st_ctime'] log('stat %s' % filename) try: source_stat = os.stat(filename) except: log('error in stat: %s' % repr(sys.exc_info()), syslog.LOG_ERR) log('cwd: %s' % os.getcwd()) log('dir: %s' % repr(os.listdir('.'))) log('source_stat: %s' % repr(source_stat)) for attribute in attributes: try: setattr(self, attribute, getattr(source_stat, attribute)) except: log('failed to set %s' % attribute, syslog.LOG_ERR) self.st_dev = 0 # not the same device as that of actual file log('initial stat: %s' % repr(self)) class TestFS(Fuse): 'filesystem for testing functions in Python scripts' def __init__(self, **kwargs): self.mountpoint = sys.argv[1] self.modulename = os.path.basename(self.mountpoint) self.mounted = __import__(self.modulename) self.filename = sys.argv[1] + '.py' self.contents = readfile(self.filename) self.functions = filter( lambda f: hasattr(getattr(self.mounted, f), '__call__'), dir(self.mounted)) self.opened = {} super(TestFS, self).__init__(**kwargs) def fsdestroy(self, data = None): 'called by FUSE after shutdown of filesystem' log('destroy %s: %s' % (self.mountpoint, data)) os.rmdir(self.mountpoint) def getattr(self, path): 'return attributes of virtual file or directory' log('getattr %s' % path) st = TestStat(self.filename) log('getattr got stat of %s' % st) if path == '/': st.st_mode = stat.S_IFDIR | 0755 st.st_nlink = 2 elif path[1:] in self.functions: st.st_nlink = 1 st.st_size = len(self.contents) else: st = -errno.ENOENT log('getattr returning %s' % repr(st)) return st def readdir(self, path, offset): 'return virtual directory contents' log('readdir %s' % path) if path == '/': try: ret = ['.', '..'] + self.functions except: log('error in readdir /', syslog.LOG_ERR) ret = ['.', '..'] else: ret = [path[1:]] log(repr(ret)) for r in ret: log('yielding %s' % r) yield fuse.Direntry(r) def open(self, path, flags): 'open a virtual file' log('open %s %s' % (path, flags)) if not path[1:] in self.functions: return -errno.ENOENT else: self.opened[path] = flags return 0 def read(self, path, size, offset): 'read bytes from the file' log('read %d bytes, offset %d, of %s' % (size, offset, path)) if path in self.opened: return self.contents[offset:offset + size] else: return -errno.ENOSYS def write(self, path, buffer, offset): 'write to file not permitted' log('attempted write to %s' % path) return -errno.EPERM def release(self, path, descriptor): 'what happens on file.close()' log('release %d for %s' % (descriptor, path)) flags = self.opened.pop(path) log('closing flags: %s' % flags) def sync(self): 'since writes not permitted, sync not necessary' return -errno.ENOSYS def readfile(filename): 'return contents of a file, closing it properly' input = open(filename) data = input.read() input.close() return data def main(sources): 'start up the filesystem for each Python script except for this one' script = os.path.basename(sys.argv[0]) if not len(sources): sources = filter(lambda filename: ( filename.endswith('.py') and filename != script), os.listdir('.')) sys.argv.append('') # Fuse expects mountpoint in sys.argv else: sys.argv[2:] = [] for sourcefile in map(os.path.abspath, sources): sys.argv[1] = mountpoint = os.path.splitext(sourcefile)[0] name = os.path.basename(mountpoint) if not os.path.exists(mountpoint): os.mkdir(mountpoint, 0700) syslog.openlog(name, syslog.LOG_PID, syslog.LOG_KERN) log('starting FUSE "%s"' % name) server = TestFS(version = "%prog " + fuse.__version__, usage = name + '\n\n' + Fuse.fusage, dash_s_do = 'setsingle') server.parse(values = server, errex = 1) server.main() def debug(message): 'print messages to sys.stderr, not useful after FUSE takes over' if DEBUGGING: print >>sys.stderr, message def log(message, level = syslog.LOG_DEBUG): 'log messages to syslog, found in /var/log/user.log on Debian' syslog.syslog(level, message) if __name__=='__main__': sys.path.insert(0, os.getcwd()) main(sys.argv[1:])