<s> from collections import namedtuple <EOL> import os <EOL> import errno <EOL> import threading <EOL> import time <EOL> import uuid <EOL> import struct <EOL> from nose import SkipTest <EOL> from nose . tools import eq_ <EOL> from nose . tools import raises <EOL> import mock <EOL> from kazoo . exceptions import ConnectionLoss <EOL> from kazoo . protocol . serialization import ( <EOL> Connect , <EOL> int_struct , <EOL> write_string , <EOL> ) <EOL> from kazoo . protocol . states import KazooState <EOL> from kazoo . protocol . connection import _CONNECTION_DROP <EOL> from kazoo . testing import KazooTestCase <EOL> from kazoo . tests . util import wait <EOL> from kazoo . tests . util import TRAVIS_ZK_VERSION <EOL> class Delete ( namedtuple ( '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> type = <NUM_LIT:2> <EOL> def serialize ( self ) : <EOL> b = bytearray ( ) <EOL> b . extend ( write_string ( self . path ) ) <EOL> b . extend ( int_struct . pack ( self . version ) ) <EOL> return b <EOL> @ classmethod <EOL> def deserialize ( self , bytes , offset ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> class TestConnectionHandler ( KazooTestCase ) : <EOL> def test_bad_deserialization ( self ) : <EOL> async_object = self . client . handler . async_result ( ) <EOL> self . client . _queue . append ( ( Delete ( self . client . chroot , - <NUM_LIT:1> ) , async_object ) ) <EOL> os . write ( self . client . _connection . _write_pipe , b'<STR_LIT>' ) <EOL> @ raises ( ValueError ) <EOL> def testit ( ) : <EOL> async_object . get ( ) <EOL> testit ( ) <EOL> def test_with_bad_sessionid ( self ) : <EOL> ev = threading . Event ( ) <EOL> def expired ( state ) : <EOL> if state == KazooState . CONNECTED : <EOL> ev . set ( ) <EOL> password = os . urandom ( <NUM_LIT:16> ) <EOL> client = self . _get_client ( client_id = ( <NUM_LIT> , password ) ) <EOL> client . add_listener ( expired ) <EOL> client . start ( ) <EOL> try : <EOL> ev . wait ( <NUM_LIT:15> ) <EOL> eq_ ( ev . is_set ( ) , True ) <EOL> finally : <EOL> client . stop ( ) <EOL> def test_connection_read_timeout ( self ) : <EOL> client = self . client <EOL> ev = threading . Event ( ) <EOL> path = "<STR_LIT:/>" + uuid . uuid4 ( ) . hex <EOL> handler = client . handler <EOL> _select = handler . select <EOL> _socket = client . _connection . _socket <EOL> def delayed_select ( * args , ** kwargs ) : <EOL> result = _select ( * args , ** kwargs ) <EOL> if len ( args [ <NUM_LIT:0> ] ) == <NUM_LIT:1> and _socket in args [ <NUM_LIT:0> ] : <EOL> return [ ] , [ ] , [ ] <EOL> return result <EOL> def back ( state ) : <EOL> if state == KazooState . CONNECTED : <EOL> ev . set ( ) <EOL> client . add_listener ( back ) <EOL> client . create ( path , b"<STR_LIT:1>" ) <EOL> try : <EOL> handler . select = delayed_select <EOL> self . assertRaises ( ConnectionLoss , client . get , path ) <EOL> finally : <EOL> handler . select = _select <EOL> ev . wait ( <NUM_LIT:5> ) <EOL> eq_ ( ev . is_set ( ) , True ) <EOL> eq_ ( client . get ( path ) [ <NUM_LIT:0> ] , b"<STR_LIT:1>" ) <EOL> def test_connection_write_timeout ( self ) : <EOL> client = self . client <EOL> ev = threading . Event ( ) <EOL> path = "<STR_LIT:/>" + uuid . uuid4 ( ) . hex <EOL> handler = client . handler <EOL> _select = handler . select <EOL> _socket = client . _connection . _socket <EOL> def delayed_select ( * args , ** kwargs ) : <EOL> result = _select ( * args , ** kwargs ) <EOL> if _socket in args [ <NUM_LIT:1> ] : <EOL> return [ ] , [ ] , [ ] <EOL> return result <EOL> def back ( state ) : <EOL> if state == KazooState . CONNECTED : <EOL> ev . set ( ) <EOL> client . add_listener ( back ) <EOL> try : <EOL> handler . select = delayed_select <EOL> self . assertRaises ( ConnectionLoss , client . create , path ) <EOL> finally : <EOL> handler . select = _select <EOL> ev . wait ( <NUM_LIT:5> ) <EOL> eq_ ( ev . is_set ( ) , True ) <EOL> eq_ ( client . exists ( path ) , None ) <EOL> def test_connection_deserialize_fail ( self ) : <EOL> client = self . client <EOL> ev = threading . Event ( ) <EOL> path = "<STR_LIT:/>" + uuid . uuid4 ( ) . hex <EOL> handler = client . handler <EOL> _select = handler . select <EOL> _socket = client . _connection . _socket <EOL> def delayed_select ( * args , ** kwargs ) : <EOL> result = _select ( * args , ** kwargs ) <EOL> if _socket in args [ <NUM_LIT:1> ] : <EOL> return [ ] , [ ] , [ ] <EOL> return result <EOL> def back ( state ) : <EOL> if state == KazooState . CONNECTED : <EOL> ev . set ( ) <EOL> client . add_listener ( back ) <EOL> deserialize_ev = threading . Event ( ) <EOL> def bad_deserialize ( bytes , offset ) : <EOL> deserialize_ev . set ( ) <EOL> raise struct . error ( ) <EOL> with mock . patch . object ( Connect , '<STR_LIT>' ) as mock_deserialize : <EOL> mock_deserialize . side_effect = bad_deserialize <EOL> try : <EOL> handler . select = delayed_select <EOL> self . assertRaises ( ConnectionLoss , client . create , path ) <EOL> finally : <EOL> handler . select = _select <EOL> deserialize_ev . wait ( <NUM_LIT:5> ) <EOL> eq_ ( deserialize_ev . is_set ( ) , True ) <EOL> ev . wait ( <NUM_LIT:5> ) <EOL> eq_ ( ev . is_set ( ) , True ) <EOL> eq_ ( client . exists ( path ) , None ) <EOL> def test_connection_close ( self ) : <EOL> self . assertRaises ( Exception , self . client . close ) <EOL> self . client . stop ( ) <EOL> self . client . close ( ) <EOL> self . client . start ( ) <EOL> def test_connection_pipe ( self ) : <EOL> client = self . client <EOL> read_pipe = client . _connection . _read_pipe <EOL> write_pipe = client . _connection . _write_pipe <EOL> assert read_pipe is not None <EOL> assert write_pipe is not None <EOL> client . stop ( ) <EOL> assert read_pipe is not None <EOL> assert write_pipe is not None <EOL> os . fstat ( read_pipe ) <EOL> os . fstat ( write_pipe ) <EOL> client . close ( ) <EOL> try : <EOL> os . fstat ( read_pipe ) <EOL> except OSError as e : <EOL> if not e . errno == errno . EBADF : <EOL> raise <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> try : <EOL> os . fstat ( write_pipe ) <EOL> except OSError as e : <EOL> if not e . errno == errno . EBADF : <EOL> raise <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> client . start ( ) <EOL> read_pipe = client . _connection . _read_pipe <EOL> write_pipe = client . _connection . _write_pipe <EOL> assert read_pipe is not None <EOL> assert write_pipe is not None <EOL> os . fstat ( read_pipe ) <EOL> os . fstat ( write_pipe ) <EOL> def test_dirty_pipe ( self ) : <EOL> client = self . client <EOL> read_pipe = client . _connection . _read_pipe <EOL> write_pipe = client . _connection . _write_pipe <EOL> os . write ( write_pipe , b'<STR_LIT>' ) <EOL> wait ( lambda : client . handler . select ( [ read_pipe ] , [ ] , [ ] , <NUM_LIT:0> ) [ <NUM_LIT:0> ] == [ ] ) <EOL> class TestConnectionDrop ( KazooTestCase ) : <EOL> def test_connection_dropped ( self ) : <EOL> ev = threading . Event ( ) <EOL> def back ( state ) : <EOL> if state == KazooState . CONNECTED : <EOL> ev . set ( ) <EOL> path = "<STR_LIT:/>" + uuid . uuid4 ( ) . hex <EOL> self . client . create ( path ) <EOL> self . client . add_listener ( back ) <EOL> result = self . client . set_async ( path , b'<STR_LIT:a>' * <NUM_LIT:1000> * <NUM_LIT> ) <EOL> self . client . _call ( _CONNECTION_DROP , None ) <EOL> self . assertRaises ( ConnectionLoss , result . get ) <EOL> ev . wait ( <NUM_LIT:30> ) <EOL> eq_ ( ev . is_set ( ) , True ) <EOL> class TestReadOnlyMode ( KazooTestCase ) : <EOL> def setUp ( self ) : <EOL> self . setup_zookeeper ( read_only = True ) <EOL> skip = False <EOL> if TRAVIS_ZK_VERSION and TRAVIS_ZK_VERSION < ( <NUM_LIT:3> , <NUM_LIT:4> ) : <EOL> skip = True <EOL> elif TRAVIS_ZK_VERSION and TRAVIS_ZK_VERSION >= ( <NUM_LIT:3> , <NUM_LIT:4> ) : <EOL> skip = False <EOL> else : <EOL> ver = self . client . server_version ( ) <EOL> if ver [ <NUM_LIT:1> ] < <NUM_LIT:4> : <EOL> skip = True <EOL> if skip : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> def tearDown ( self ) : <EOL> self . client . stop ( ) <EOL> def test_read_only ( self ) : <EOL> from kazoo . exceptions import NotReadOnlyCallError <EOL> from kazoo . protocol . states import KeeperState <EOL> client = self . client <EOL> states = [ ] <EOL> ev = threading . Event ( ) <EOL> @ client . add_listener <EOL> def listen ( state ) : <EOL> states . append ( state ) <EOL> if client . client_state == KeeperState . CONNECTED_RO : <EOL> ev . set ( ) <EOL> try : <EOL> self . cluster [ <NUM_LIT:1> ] . stop ( ) <EOL> self . cluster [ <NUM_LIT:2> ] . stop ( ) <EOL> ev . wait ( <NUM_LIT:6> ) <EOL> eq_ ( ev . is_set ( ) , True ) <EOL> eq_ ( client . client_state , KeeperState . CONNECTED_RO ) <EOL> eq_ ( client . get_children ( '<STR_LIT:/>' ) , [ ] ) <EOL> @ raises ( NotReadOnlyCallError ) <EOL> def testit ( ) : <EOL> client . create ( '<STR_LIT>' ) <EOL> testit ( ) <EOL> time . sleep ( <NUM_LIT:15> ) <EOL> finally : <EOL> client . remove_listener ( listen ) <EOL> self . cluster [ <NUM_LIT:1> ] . run ( ) <EOL> self . cluster [ <NUM_LIT:2> ] . run ( ) </s>
