<s> from unittest import TestCase <EOL> import zmq . green as zmq <EOL> import gevent <EOL> import gevent . socket as socket <EOL> from gevent . queue import Queue <EOL> import random <EOL> import os <EOL> from logcabin . event import Event <EOL> from logcabin . context import DummyContext <EOL> from logcabin . inputs import udp , zeromq , file as fileinput <EOL> from testhelper import TempDirectory , assertEventEquals <EOL> class InputTests ( TestCase ) : <EOL> def create ( self , conf ) : <EOL> self . output = Queue ( ) <EOL> with DummyContext ( ) : <EOL> self . i = i = self . cls ( ** conf ) <EOL> i . setup ( self . output ) <EOL> i . start ( ) <EOL> return i <EOL> def waitForQueue ( self , timeout = <NUM_LIT:1.0> , events = <NUM_LIT:1> ) : <EOL> with gevent . Timeout ( timeout ) : <EOL> while self . output . qsize ( ) < events : <EOL> gevent . sleep ( <NUM_LIT:0.0> ) <EOL> self . i . stop ( ) <EOL> if events : <EOL> return [ self . output . get ( ) for n in xrange ( events ) ] <EOL> class ZeromqTests ( InputTests ) : <EOL> cls = zeromq . Zeromq <EOL> def test_event ( self ) : <EOL> conf = { '<STR_LIT:address>' : '<STR_LIT>' } <EOL> self . create ( conf ) <EOL> ctx = zmq . Context ( ) <EOL> sock = ctx . socket ( zmq . PUSH ) <EOL> sock . connect ( conf [ '<STR_LIT:address>' ] ) <EOL> sock . send ( '<STR_LIT:abc>' ) <EOL> q = self . waitForQueue ( ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:0> ] ) <EOL> class UdpTests ( InputTests ) : <EOL> cls = udp . Udp <EOL> def test_event ( self ) : <EOL> conf = { '<STR_LIT:port>' : random . randint ( <NUM_LIT> , <NUM_LIT> ) } <EOL> self . create ( conf ) <EOL> self . sock = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) <EOL> self . sock . sendto ( '<STR_LIT:abc>' , ( '<STR_LIT>' , conf [ '<STR_LIT:port>' ] ) ) <EOL> q = self . waitForQueue ( ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:0> ] ) <EOL> class FileTests ( InputTests ) : <EOL> cls = fileinput . File <EOL> def test_success ( self ) : <EOL> with TempDirectory ( ) : <EOL> conf = { '<STR_LIT:path>' : '<STR_LIT>' } <EOL> self . create ( conf ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT:abc>' <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> print >> fin , '<STR_LIT>' <EOL> q = self . waitForQueue ( events = <NUM_LIT:2> ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:0> ] ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT>' ) , q [ <NUM_LIT:1> ] ) <EOL> def test_rolling ( self ) : <EOL> with TempDirectory ( ) : <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT:abc>' <EOL> conf = { '<STR_LIT:path>' : '<STR_LIT>' } <EOL> self . create ( conf ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> os . rename ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT>' <EOL> q = self . waitForQueue ( events = <NUM_LIT:2> ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:0> ] ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT>' ) , q [ <NUM_LIT:1> ] ) <EOL> def test_multiple ( self ) : <EOL> with TempDirectory ( ) : <EOL> conf = { '<STR_LIT:path>' : '<STR_LIT>' } <EOL> self . create ( conf ) <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT:abc>' <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT:abc>' <EOL> q = self . waitForQueue ( events = <NUM_LIT:2> ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:0> ] ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:1> ] ) <EOL> def test_resume ( self ) : <EOL> with TempDirectory ( ) : <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT:abc>' <EOL> conf = { '<STR_LIT:path>' : '<STR_LIT>' } <EOL> self . create ( conf ) <EOL> q = self . waitForQueue ( events = <NUM_LIT:1> ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:0> ] ) <EOL> with file ( '<STR_LIT>' , '<STR_LIT:a>' ) as fin : <EOL> print >> fin , '<STR_LIT>' <EOL> self . create ( conf ) <EOL> q = self . waitForQueue ( events = <NUM_LIT:1> ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT>' ) , q [ <NUM_LIT:0> ] ) <EOL> def test_resume_statedir ( self ) : <EOL> with TempDirectory ( ) : <EOL> os . mkdir ( '<STR_LIT:state>' ) <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT:abc>' <EOL> conf = { '<STR_LIT:path>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT:state>' } <EOL> self . create ( conf ) <EOL> q = self . waitForQueue ( events = <NUM_LIT:1> ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:0> ] ) <EOL> with file ( '<STR_LIT>' , '<STR_LIT:a>' ) as fin : <EOL> print >> fin , '<STR_LIT>' <EOL> self . create ( conf ) <EOL> q = self . waitForQueue ( events = <NUM_LIT:1> ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT>' ) , q [ <NUM_LIT:0> ] ) <EOL> def test_truncated ( self ) : <EOL> with TempDirectory ( ) : <EOL> conf = { '<STR_LIT:path>' : '<STR_LIT>' } <EOL> self . create ( conf ) <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT:abc>' <EOL> print >> fin , '<STR_LIT>' <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> with file ( '<STR_LIT>' , '<STR_LIT:w>' ) as fin : <EOL> print >> fin , '<STR_LIT>' <EOL> q = self . waitForQueue ( events = <NUM_LIT:3> ) <EOL> assertEventEquals ( self , Event ( data = '<STR_LIT:abc>' ) , q [ <NUM_LIT:0> ] ) <EOL> def test_missing ( self ) : <EOL> with TempDirectory ( ) : <EOL> conf = { '<STR_LIT:path>' : '<STR_LIT>' } <EOL> self . create ( conf ) <EOL> self . waitForQueue ( events = <NUM_LIT:0> ) </s>
