<s> import socket <EOL> from . base import Test , passive <EOL> class Redis ( Test ) : <EOL> def setUp ( self ) : <EOL> super ( ) . setUp ( ) <EOL> import zorro . redis <EOL> self . r = zorro . redis . Redis ( db = <NUM_LIT> ) <EOL> class SingleThread ( Redis ) : <EOL> @ passive <EOL> def test_execute ( self ) : <EOL> self . assertEquals ( self . r . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:value>' ) , '<STR_LIT:OK>' ) <EOL> self . assertEquals ( self . r . execute ( '<STR_LIT:GET>' , '<STR_LIT>' ) , b'<STR_LIT:value>' ) <EOL> self . assertEquals ( self . r . execute ( '<STR_LIT>' , '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> @ passive <EOL> def test_reconnect ( self ) : <EOL> self . assertEquals ( self . r . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:value>' ) , '<STR_LIT:OK>' ) <EOL> self . r . _channel . _sock . shutdown ( socket . SHUT_RDWR ) <EOL> self . z . sleep ( <NUM_LIT> ) <EOL> self . assertEquals ( self . r . execute ( '<STR_LIT:GET>' , '<STR_LIT>' ) , b'<STR_LIT:value>' ) <EOL> self . assertEquals ( self . r . execute ( '<STR_LIT>' , '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> @ passive <EOL> def test_disconnect ( self ) : <EOL> fut = self . r . future ( '<STR_LIT>' , <EOL> '<STR_LIT>' , b'<STR_LIT>' * <NUM_LIT> ) <EOL> self . r . _channel . _sock . shutdown ( socket . SHUT_RDWR ) <EOL> self . z . sleep ( <NUM_LIT> ) <EOL> with self . assertRaises ( self . z . channel . PipeError ) : <EOL> fut . get ( ) <EOL> @ passive <EOL> def test_bulk ( self ) : <EOL> self . assertEquals ( self . r . bulk ( [ <EOL> ( '<STR_LIT>' , ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , ) , <EOL> ] ) , [ '<STR_LIT:OK>' , <NUM_LIT:11> , <NUM_LIT:1> ] ) <EOL> @ passive <EOL> def test_keys ( self ) : <EOL> self . r . execute ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( self . r . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:value>' ) , '<STR_LIT:OK>' ) <EOL> self . assertEqual ( self . r . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:value>' ) , '<STR_LIT:OK>' ) <EOL> self . assertEqual ( set ( map ( bytes , self . r . execute ( '<STR_LIT>' , '<STR_LIT:*>' ) ) ) , <EOL> set ( [ b'<STR_LIT>' , b'<STR_LIT>' ] ) ) <EOL> val = self . r . bulk ( [ ( '<STR_LIT>' , ) , <EOL> ( '<STR_LIT:GET>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:*>' ) , <EOL> ( '<STR_LIT>' , ) ] ) <EOL> self . assertEqual ( val [ <NUM_LIT:0> ] , b'<STR_LIT:value>' ) <EOL> self . assertEqual ( val [ <NUM_LIT:1> ] , [ b'<STR_LIT:value>' , b'<STR_LIT:value>' ] ) <EOL> self . assertSetEqual ( set ( map ( bytes , val [ <NUM_LIT:2> ] ) ) , <EOL> set ( [ b'<STR_LIT>' , b'<STR_LIT>' ] ) ) <EOL> self . assertEquals ( self . r . execute ( '<STR_LIT>' , '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . r . execute ( '<STR_LIT>' , '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> class BigTest ( Redis ) : <EOL> test_timeout = <NUM_LIT:10> <EOL> @ passive <EOL> def test_time ( self ) : <EOL> def get100 ( ) : <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> self . r . execute ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> import time <EOL> old = time . time ( ) <EOL> f = [ ] <EOL> for i in range ( <NUM_LIT:200> ) : <EOL> f . append ( self . z . Future ( get100 ) ) <EOL> for i in f : <EOL> i . get ( ) <EOL> print ( "<STR_LIT>" , time . time ( ) - old ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import unittest <EOL> unittest . main ( ) </s>
