root/projects/WinDictator/trunk/test/mock.py

# WinDictator:
# Dictate in Windows, have the text typed in Linux via X faked keystroke events
#
# Copyright (C) 2005-2006 by Edwin A. Suominen, http://www.eepatents.com
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
# 
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the file COPYING for more details.
# 
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA


"""
Mock objects for unit tests
"""

import os, time
from twisted.internet import defer, reactor, utils

DELAY_CONFIRMATION = 0.01
DELAY_TYPED = 0.005
VERBOSE = False


class Mock(object):
    def __init__(self):
        self.N = 10000
        self.checks = 0

    def ok_get(self):
        self.checks += 1
        if self.checks <= self.N:
            return True
        else:
            return False
    ok = property(ok_get)

    def failAfter(self, N):
        self.N = N


class MockObserver(Mock):
    def confirm(self, keycode):
        def confirmNow(d):
            d.callback(self.ok)

        if isinstance(keycode, int):
            d = defer.Deferred()
            reactor.callLater(DELAY_CONFIRMATION, confirmNow, d)
        else:
            d = defer.succeed(False)
        return d
    

class MockTransport(object):
    def close(self):
        pass


class MockKeyer(Mock):
    def __init__(self):
        Mock.__init__(self)
        self.typed = []
        self.stdin = MockTransport()

    def whatTyped(self):
        return self.typed
    
    def pressKey(self, keySym):
        self.typed.append(keySym)
        return self.key('d', keySym)

    def releaseKey(self, keySym):
        return self.key('u', keySym)

    def runTypistForOneKeySym(self, keySym):
        t1 = time.time()
        
        def gotResult(result):
            try:
                keycode = int(result[0])
            except:
                keycode = None
            if VERBOSE:
                print "\nKEYED:", time.time() - t1, keycode
            return keycode

        d = utils.getProcessOutputAndValue(
            '/bin/bash',
            args=('-c', "echo 'c %s' |../bin/typist/typist" % keySym),
            env = os.environ)
        d.addCallback(gotResult)
        return d

    def key(self, du, keySym):
        def gotKeyCode(keycode):
            self.keycode = keycode
            d2 = defer.Deferred()
            d2.addCallback(returnKeyCode)
            reactor.callLater(DELAY_TYPED, d2.callback, None)
            return d2
            
        def returnKeyCode(null):
            return self.keycode
        
        # Deferred d1 for pressing
        if self.ok:
            d1 = self.runTypistForOneKeySym(keySym)
        else:
            d1 = defer.succeed(None)
        d1.addCallback(gotKeyCode)
        return d1


class MockHistory(object):
    def __init__(self, N=None):
        self.events = []

    def backspace(self, N):
        self.events = self.events[:-N]
        
    def reportKeyEvent(self, *args):
        self.events.append(args)


class MockUpdater(object):
    def __init__(self, clientRoot):
        self.clientRoot = clientRoot

    def update(self, currentText):
        return defer.succeed(True)


class MockClientRoot(object):
    def __init__(self):
        self.operations = []
    
    def callRemote(self, funcName, *args, **kw):
        self.verbose = kw.pop('verbose', False)
        return getattr(self, funcName)(*args, **kw)
    
    def backspace(self, N):
        if self.verbose:
            print "\nBACKSPACE", N
        self.operations.append(('backspace', N))
        return defer.succeed(True)

    def insert(self, text):
        if self.verbose:
            print "\nINSERT", text
        self.operations.append(('insert', text))
        return defer.succeed(True)