Source

gadgets / gadgets / tests / test_input.py

import os, time, random, threading
from nose.tools import eq_
from gadgets.io.gpio import GPIO
from gadgets.pins.beaglebone import pins
from gadgets import Addresses, Sockets, Gadgets
from gadgets.devices.input.factory import input_factory
from gadgets.devices.input.io import InputPoller


def setup():
    InputPoller.PollerClass = FakePoller
    

class FakePoller(object):

    def __init__(self, pin, timeout, edge):
        pass
    
    def wait(self):
        time.sleep(1)
        return '1\n'

    def close(self):
        pass

    @property
    def value(self):
        return False

class TestInput(object):

    def setup(self):
        port = random.randint(5000, 50000)
        self.addresses = Addresses(in_port=port, out_port=port+1, req_port=port+2)
        pin = None
        self.sockets = Sockets(self.addresses, events=['update'])
        self.input = input_factory('left', 'input', {'pin':None}, self.addresses)
        self.gadgets = Gadgets([self.input], self.addresses)

    def test_create(self):
        pass

    def test_run(self):
        t = threading.Thread(target=self.gadgets.start)
        t.start()
        time.sleep(1)
        event, message = self.sockets.recv()
        eq_(event, 'update')
        expected = {u'left': {u'input': {u'input': {u'value': False, u'units': u'boolean'}}}}
        eq_(message, expected)
        self.sockets.send('shutdown')