root/projects/WinDictator/trunk/test/observer.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


"""
Unit tests for windicator.linux.observer
"""

import os, time
from twisted.internet import defer, task, reactor
from twisted.trial.unittest import TestCase

import windictator.linux.observer as observer
import mock

REAL_BINARY = observer.BINARY
DELAY = 0.1
# Interval of 0.1 sec between wait loop iterations
WAIT_INTERVAL = 0.05
# No more than 15 iterations, or 1.5 sec of waiting
MAX_INTERVALS = 30
# Key events "typed" less than 0.7 sec after starting xnee don't register for
# some reason.
FIRST_TYPING_WAIT = 0.7
                        

class TestKeyHistory(TestCase):
    def setUp(self):
        self.fakeStartTime = divmod(int(500*time.time()),2**32)[1]
        self.fakeTimeCount = 0
        self.history = observer.KeyHistory(16)

    def fakeTime(self):
        self.fakeTimeCount += 1
        return self.fakeStartTime + 300*self.fakeTimeCount

    def putStuff(self):
        toPut = (
            ((38,), True),
            ((56, 50), False),
            ((54,), False),
            ((40, 50), True))
        for keyCombo, isRealTyping in toPut:
            self.history.put(keyCombo, isRealTyping, self.fakeTime())

    def testFakeVsRealBuffer(self):
        self.putStuff()
        # Real buffer
        values = self.history.realHistory.values()
        values.sort()
        self.failUnlessEqual(values, [(38,), (40,50)])
        # Fake buffer
        values = self.history.fakeHistory.values()
        values.sort()
        self.failUnlessEqual(values, [(54,), (56,50)])

    def testGetKeys(self):
        self.putStuff()
        for N in (4,100):
            self.failUnlessEqual(
                self.history.getKeys(N), [(38,), (56,50), (54,), (40,50)])
        self.failUnlessEqual(
            self.history.getKeys(2), [(54,), (40,50)])
        self.failUnlessEqual(
            self.history.getKeys(3, realOnly=True), [(38,), (40,50)])
        self.failUnlessEqual(
            self.history.getKeys(1, fakeOnly=True), [(54,)])

    def testGracefulOverflow(self):
        for k in xrange(10):
            # Putting 40 items total, lots of overflow going on
            self.putStuff()
        lastSixExpected = [(54,), (40,50), (38,), (56,50), (54,), (40,50)]
        self.failUnlessEqual(
            self.history.getKeys(6), lastSixExpected)

    def testSameTimeValues(self):
        timeValue = self.fakeTime()
        self.history.put((38,), True, timeValue)
        self.history.put((54,), True, timeValue)
        values = self.history.realHistory.values()
        values.sort()
        self.failUnlessEqual(values, [(38,), (54,)])

    def testReportKeyEvent(self):
        # keycode, isKeyPress, isRealTyping, timeReported
        keyEvents = (
            (38, True),
            (38, False),
            (50, True),
            (56, True),
            (56, False),
            (50, False),
            (54, True),
            (54, False)
            )
        for keycode, isKeypress in keyEvents:
            self.history.reportKeyEvent(
                keycode, isKeypress, True, self.fakeTime())
        timeKeys = self.history.realHistory.keys()
        timeKeys.sort()
        for k, shouldBe in enumerate([(38,), (50,56), (54,)]):
            value = self.history.realHistory[timeKeys[k]]
            self.failUnless(
                [True for x in shouldBe if x in value],
                "Keycode combo '%s' doesn't match expected combination '%s'" \
                % (value, shouldBe))


class LoopForReport:
    def __init__(self, history):
        self.count = 0
        self.events = history.events
        self.loop = task.LoopingCall(self.checkIfDone)

    def start(self, N):
        self.N = N
        return self.loop.start(WAIT_INTERVAL)

    def checkIfDone(self):
        self.count += 1
        if self.count >= MAX_INTERVALS or len(self.events) >= self.N:
            self.loop.stop()


class TestObserverInVitro(TestCase):
    def setUp(self):
        observer.BINARY = ("../test/mockxnee.py",)
        self.mockHistory = mock.MockHistory()
        self.observer = observer.Observer(self.mockHistory)
        self.loop = LoopForReport(self.mockHistory)

    def tearDown(self):
        self.observer.shutdown()

    def testRealSingle(self):
        def started(null):
            self.observer.process.write("d 38\n")

        def done(null):
            # Event tuple:
            # (keycode, isKeyPress, isRealTyping, timeReported)
            self.failUnlessEqual(
                self.mockHistory.events[0][0:3],
                (38, True, True))
            
        d1 = self.loop.start(1)
        d2 = self.observer.startup()
        d2.addCallback(started)
        return defer.DeferredList([d1,d2]).addCallback(done)

    def testRealMultiple(self):
        cmdList = (
            ("d", 38), ("d", 50), ("u", 50), ("u", 38),
            ("d", 56), ("u", 56)
            )
        def started(null):
            for cmd in cmdList:
                self.observer.process.write("%s %d\n" % cmd)

        def done(null):
            # Event tuple:
            # (keycode, isKeyPress, isRealTyping, timeReported)
            for k, event in enumerate(self.mockHistory.events):
                cmd = cmdList[k]
                self.failUnlessEqual(
                    event[0:3], (cmd[1], (cmd[0] == "d"), True))

        d1 = self.loop.start(len(cmdList))
        d2 = self.observer.startup()
        d2.addCallback(started)
        return defer.DeferredList([d1,d2]).addCallback(done)

    def testFakeSingle(self):
        def started(null):
            d = self.observer.confirm(38)
            self.observer.process.write("d 38\n")
            return d

        def done(null):
            # Event tuple:
            # (keycode, isKeyPress, isRealTyping, timeReported)
            self.failUnlessEqual(
                self.mockHistory.events[0][0:3],
                (38, True, False))
            
        d1 = self.loop.start(1)
        d2 = self.observer.startup()
        d2.addCallback(started)
        return defer.DeferredList([d1,d2]).addCallback(done)
    
    def testMixedMultiple(self):
        cmdList = (
            ("d", 38), ("u", 38),                               # a
            ("d", 56), ("d", 50), ("u", 50), ("u", 56),         # B
            ("d", 54), ("u", 54),                               # c
            ("d", 40), ("u", 40)                                # d
            )
        def started(null):
            dList = [self.observer.confirm(54), self.observer.confirm(54)]
            for cmd in cmdList:
                self.observer.process.write("%s %d\n" % cmd)
            return defer.DeferredList(dList)

        def done(null):
            # Event tuple:
            # (keycode, isKeyPress, isRealTyping, timeReported)
            for k, event in enumerate(self.mockHistory.events):
                cmd = cmdList[k]
                shouldBeReal = (cmd[1] != 54)
                self.failUnlessEqual(
                    event[0:3], (cmd[1], (cmd[0] == "d"), shouldBeReal))
        
        d1 = self.loop.start(len(cmdList))
        d2 = self.observer.startup()
        d2.addCallback(started)
        return defer.DeferredList([d1,d2]).addCallback(done)


class TestObserverInVivo(TestCase):
    def setUp(self):
        import windictator.linux.keyer as keyer

        observer.BINARY = REAL_BINARY
        self.keyer = keyer.Keyer()
        self.mockHistory = mock.MockHistory()
        self.observer = observer.Observer(self.mockHistory)
        self.loop = LoopForReport(self.mockHistory)

    def tearDown(self):
        self.keyer.shutdown()
        self.observer.shutdown()

    def testRealSingle(self):
        def started(null):
            d = defer.Deferred()
            d.addCallback(pressing)
            reactor.callLater(FIRST_TYPING_WAIT, d.callback, None)
            return d

        def pressing(null):
            d = self.keyer.pressKey('a')
            d.addCallback(pressed)
            return d
        
        def pressed(null):
            d = self.keyer.releaseKey('a')
            return d
        
        def done(null):
            # Event tuple:
            # (keycode, isKeyPress, isRealTyping, timeReported)
            self.failUnlessEqual(
                self.mockHistory.events[0][0:3], (38, True, True))
            self.failUnlessEqual(
                self.mockHistory.events[1][0:3], (38, False, True))

        d1 = self.loop.start(2)
        d2 = self.observer.startup()
        d2.addCallback(started)
        return defer.DeferredList([d1,d2]).addCallback(done)