<s> import etcd <EOL> import unittest <EOL> try : <EOL> import mock <EOL> except ImportError : <EOL> from unittest import mock <EOL> from etcd import EtcdException <EOL> class FakeHTTPResponse ( object ) : <EOL> def __init__ ( self , status , data = '<STR_LIT>' , headers = None ) : <EOL> self . status = status <EOL> self . data = data . encode ( '<STR_LIT:utf-8>' ) <EOL> self . headers = headers or { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> def getheaders ( self ) : <EOL> return self . headers <EOL> def getheader ( self , header ) : <EOL> return self . headers [ header ] <EOL> class TestClientRequest ( unittest . TestCase ) : <EOL> def test_set ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> return_value = FakeHTTPResponse ( <NUM_LIT> , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ) <EOL> result = client . set ( '<STR_LIT>' , '<STR_LIT:test>' , ttl = <NUM_LIT> ) <EOL> self . assertEquals ( <EOL> etcd . EtcdResult ( <EOL> ** { u'<STR_LIT:action>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:key>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : True , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:value>' : u'<STR_LIT:test>' } } ) , result ) <EOL> def test_test_and_set ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> return_value = FakeHTTPResponse ( <NUM_LIT:200> , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ) <EOL> result = client . test_and_set ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:test>' , ttl = <NUM_LIT> ) <EOL> self . assertEquals ( <EOL> etcd . EtcdResult ( <EOL> ** { u'<STR_LIT:action>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:key>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT:test>' , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:value>' : u'<STR_LIT>' } <EOL> } ) , result ) <EOL> def test_test_and_test_failure ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> side_effect = ValueError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> try : <EOL> result = client . test_and_set ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:test>' , ttl = <NUM_LIT> ) <EOL> except ValueError as e : <EOL> self . assertEquals ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , str ( e ) ) <EOL> def test_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> return_value = FakeHTTPResponse ( <NUM_LIT:200> , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ) <EOL> result = client . delete ( '<STR_LIT>' ) <EOL> self . assertEquals ( etcd . EtcdResult ( <EOL> ** { u'<STR_LIT:action>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:key>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT:test>' } <EOL> } ) , result ) <EOL> def test_get ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> return_value = FakeHTTPResponse ( <NUM_LIT:200> , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ) <EOL> result = client . get ( '<STR_LIT>' ) <EOL> self . assertEquals ( etcd . EtcdResult ( <EOL> ** { u'<STR_LIT:action>' : u'<STR_LIT:GET>' , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:key>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:value>' : u'<STR_LIT:test>' } <EOL> } ) , result ) <EOL> def test_get_multi ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def test_get_subdirs ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def test_not_in ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . get = mock . Mock ( side_effect = etcd . EtcdKeyNotFound ( ) ) <EOL> result = '<STR_LIT>' not in client <EOL> self . assertEquals ( True , result ) <EOL> def test_in ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> return_value = FakeHTTPResponse ( <NUM_LIT:200> , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ) <EOL> result = '<STR_LIT>' in client <EOL> self . assertEquals ( True , result ) <EOL> def test_simple_watch ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> return_value = FakeHTTPResponse ( <NUM_LIT:200> , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ) <EOL> result = client . watch ( '<STR_LIT>' ) <EOL> self . assertEquals ( <EOL> etcd . EtcdResult ( <EOL> ** { u'<STR_LIT:action>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:key>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : True , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:value>' : u'<STR_LIT:test>' } <EOL> } ) , result ) <EOL> def test_index_watch ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> return_value = FakeHTTPResponse ( <NUM_LIT:200> , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ) <EOL> result = client . watch ( '<STR_LIT>' , index = <NUM_LIT> ) <EOL> self . assertEquals ( <EOL> etcd . EtcdResult ( <EOL> ** { u'<STR_LIT:action>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:key>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : True , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:value>' : u'<STR_LIT:test>' } <EOL> } ) , result ) <EOL> class TestEventGenerator ( object ) : <EOL> def check_watch ( self , result ) : <EOL> assert etcd . EtcdResult ( <EOL> ** { u'<STR_LIT:action>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:key>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : True , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT:value>' : u'<STR_LIT:test>' } <EOL> } ) == result <EOL> def test_eternal_watch ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> client . api_execute = mock . Mock ( <EOL> return_value = FakeHTTPResponse ( <NUM_LIT:200> , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ) <EOL> for result in range ( <NUM_LIT:1> , <NUM_LIT:5> ) : <EOL> result = next ( client . eternal_watch ( '<STR_LIT>' , index = <NUM_LIT> ) ) <EOL> yield self . check_watch , result <EOL> class TestClientApiExecutor ( unittest . TestCase ) : <EOL> def test_get ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( status = <NUM_LIT:200> , data = '<STR_LIT>' ) <EOL> client . http . request = mock . Mock ( return_value = response ) <EOL> result = client . api_execute ( '<STR_LIT>' , client . _MGET ) <EOL> self . assertEquals ( '<STR_LIT>' . encode ( '<STR_LIT:utf-8>' ) , result . data ) <EOL> def test_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( status = <NUM_LIT:200> , data = '<STR_LIT>' ) <EOL> client . http . request = mock . Mock ( return_value = response ) <EOL> result = client . api_execute ( '<STR_LIT>' , client . _MDELETE ) <EOL> self . assertEquals ( '<STR_LIT>' . encode ( '<STR_LIT:utf-8>' ) , result . data ) <EOL> def test_get_error ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( status = <NUM_LIT> , <EOL> data = '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> client . http . request = mock . Mock ( return_value = response ) <EOL> try : <EOL> client . api_execute ( '<STR_LIT>' , client . _MGET ) <EOL> assert False <EOL> except etcd . EtcdKeyNotFound as e : <EOL> self . assertEquals ( str ( e ) , '<STR_LIT>' ) <EOL> def test_put ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( status = <NUM_LIT:200> , data = '<STR_LIT>' ) <EOL> client . http . request_encode_body = mock . Mock ( return_value = response ) <EOL> result = client . api_execute ( '<STR_LIT>' , client . _MPUT ) <EOL> self . assertEquals ( '<STR_LIT>' . encode ( '<STR_LIT:utf-8>' ) , result . data ) <EOL> def test_test_and_set_error ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( <EOL> status = <NUM_LIT> , <EOL> data = '<STR_LIT>' ) <EOL> client . http . request_encode_body = mock . Mock ( return_value = response ) <EOL> payload = { '<STR_LIT:value>' : '<STR_LIT:value>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> try : <EOL> client . api_execute ( '<STR_LIT>' , client . _MPUT , payload ) <EOL> self . fail ( ) <EOL> except ValueError as e : <EOL> self . assertEquals ( '<STR_LIT>' , str ( e ) ) <EOL> def test_set_not_file_error ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( <EOL> status = <NUM_LIT> , <EOL> data = '<STR_LIT>' ) <EOL> client . http . request_encode_body = mock . Mock ( return_value = response ) <EOL> payload = { '<STR_LIT:value>' : '<STR_LIT:value>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> try : <EOL> client . api_execute ( '<STR_LIT>' , client . _MPUT , payload ) <EOL> self . fail ( ) <EOL> except etcd . EtcdNotFile as e : <EOL> self . assertEquals ( '<STR_LIT>' , str ( e ) ) <EOL> def test_get_error_unknown ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( status = <NUM_LIT> , <EOL> data = '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> client . http . request = mock . Mock ( return_value = response ) <EOL> try : <EOL> client . api_execute ( '<STR_LIT>' , client . _MGET ) <EOL> self . fail ( ) <EOL> except etcd . EtcdException as e : <EOL> self . assertEqual ( str ( e ) , "<STR_LIT>" ) <EOL> def test_get_error_request_invalid ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( status = <NUM_LIT> , <EOL> data = '<STR_LIT>' ) <EOL> client . http . request = mock . Mock ( return_value = response ) <EOL> try : <EOL> client . api_execute ( '<STR_LIT>' , client . _MGET ) <EOL> self . fail ( ) <EOL> except etcd . EtcdException as e : <EOL> self . assertEqual ( str ( e ) , <EOL> "<STR_LIT>" ) <EOL> def test_get_error_invalid ( self ) : <EOL> """<STR_LIT>""" <EOL> client = etcd . Client ( ) <EOL> response = FakeHTTPResponse ( status = <NUM_LIT> , <EOL> data = '<STR_LIT>' ) <EOL> client . http . request = mock . Mock ( return_value = response ) <EOL> self . assertRaises ( etcd . EtcdException , client . api_execute , <EOL> '<STR_LIT>' , client . _MGET ) </s>
