<s> import unittest <EOL> import yaml <EOL> import threading <EOL> import pettingzoo . local_config <EOL> import pettingzoo . discovery <EOL> from pettingzoo . utils import connect_to_zk <EOL> class DiscoveryTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> pettingzoo . utils . configure_logger ( ) <EOL> self . mock = True <EOL> self . conn_string = '<STR_LIT>' <EOL> if self . mock : <EOL> import zc . zk . testing <EOL> zc . zk . testing . setUp ( self , connection_string = self . conn_string ) <EOL> self . connection = connect_to_zk ( self . conn_string ) <EOL> self . path = '<STR_LIT>' <EOL> pettingzoo . discovery . CONFIG_PATH = self . path <EOL> self . sample = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT:localhost>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' <EOL> } <EOL> def test_write_distributed_config ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample , '<STR_LIT:127.0.0.1>' ) <EOL> znode = self . connection . get ( self . path + '<STR_LIT>' ) <EOL> config = yaml . load ( znode [ <NUM_LIT:0> ] ) <EOL> mismatch_keys = [ <EOL> key for key in self . sample <EOL> if not key in config or self . sample [ key ] != config [ key ] ] <EOL> self . assertEqual ( len ( mismatch_keys ) , <NUM_LIT:0> ) <EOL> def test_write_distributed_config_no_ip ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample , interface = '<STR_LIT>' ) <EOL> znode = self . connection . get ( self . path + '<STR_LIT>' ) <EOL> config = yaml . load ( znode [ <NUM_LIT:0> ] ) <EOL> mismatch_keys = [ <EOL> key for key in self . sample <EOL> if not key in config or self . sample [ key ] != config [ key ] ] <EOL> self . assertEqual ( len ( mismatch_keys ) , <NUM_LIT:0> ) <EOL> def test_remove_stale_config ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , <EOL> self . sample , '<STR_LIT:127.0.0.1>' , ephemeral = False ) <EOL> test = self . connection . exists ( self . path + '<STR_LIT>' ) <EOL> self . assertTrue ( test ) <EOL> pettingzoo . discovery . remove_stale_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:127.0.0.1>' ) <EOL> test = self . connection . exists ( self . path + '<STR_LIT>' ) <EOL> self . assertTrue ( not test ) <EOL> def tearDown ( self ) : <EOL> pettingzoo . discovery . CONFIG_PATH = '<STR_LIT>' <EOL> self . connection . close ( ) <EOL> self . connection = connect_to_zk ( '<STR_LIT>' ) <EOL> self . connection . delete_recursive ( self . path ) <EOL> self . connection . close ( ) <EOL> if self . mock : <EOL> import zc . zk . testing <EOL> zc . zk . testing . tearDown ( self ) <EOL> class DistributedDiscoveryTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> pettingzoo . utils . configure_logger ( ) <EOL> self . mock = True <EOL> self . conn_string = '<STR_LIT>' <EOL> if self . mock : <EOL> import zc . zk . testing <EOL> zc . zk . testing . setUp ( self , connection_string = self . conn_string ) <EOL> self . connection = connect_to_zk ( self . conn_string ) <EOL> self . path = '<STR_LIT>' <EOL> pettingzoo . discovery . CONFIG_PATH = self . path <EOL> self . sample = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT:localhost>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' <EOL> } <EOL> self . sample_slash = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:host>' : '<STR_LIT:localhost>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> } <EOL> def test_distributed_config_dne ( self ) : <EOL> ddc = pettingzoo . discovery . DistributedDiscovery ( self . connection ) <EOL> self . assertRaises ( IOError , ddc . load_config , "<STR_LIT>" , "<STR_LIT>" ) <EOL> def test_load_config ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample , '<STR_LIT:127.0.0.1>' ) <EOL> ddc = pettingzoo . discovery . DistributedDiscovery ( self . connection ) <EOL> config = ddc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mismatch_keys = [ <EOL> key for key in self . sample <EOL> if not key in config or self . sample [ key ] != config [ key ] ] <EOL> self . assertEqual ( len ( mismatch_keys ) , <NUM_LIT:0> ) <EOL> def test_load_config_via_path ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample , '<STR_LIT:127.0.0.1>' ) <EOL> ddc = pettingzoo . discovery . DistributedDiscovery ( self . connection ) <EOL> config = ddc . load_config_via_path ( '<STR_LIT>' ) <EOL> mismatch_keys = [ <EOL> key for key in self . sample <EOL> if not key in config or self . sample [ key ] != config [ key ] ] <EOL> self . assertEqual ( len ( mismatch_keys ) , <NUM_LIT:0> ) <EOL> def test_load_config_with_slash ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , <EOL> '<STR_LIT>' , '<STR_LIT>' , self . sample_slash , '<STR_LIT:127.0.0.1>' ) <EOL> ddc = pettingzoo . discovery . DistributedDiscovery ( self . connection ) <EOL> config = ddc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mismatch_keys = [ <EOL> key for key in self . sample_slash <EOL> if not key in config or self . sample_slash [ key ] != config [ key ] ] <EOL> self . assertEqual ( len ( mismatch_keys ) , <NUM_LIT:0> ) <EOL> def test_load_config_with_callback ( self ) : <EOL> self . touched = False <EOL> self . cbpath = None <EOL> self . cbconfig = None <EOL> event = threading . Event ( ) <EOL> def callback ( path , config ) : <EOL> self . touched = True <EOL> self . cbpath = path <EOL> self . cbconfig = config <EOL> event . set ( ) <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample , '<STR_LIT:127.0.0.1>' ) <EOL> ddc = pettingzoo . discovery . DistributedDiscovery ( self . connection ) <EOL> config = ddc . load_config ( '<STR_LIT>' , '<STR_LIT>' , callback = callback ) <EOL> sample2 = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' <EOL> } <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , sample2 , '<STR_LIT>' ) <EOL> event . wait ( <NUM_LIT> ) <EOL> self . assertTrue ( self . touched ) <EOL> self . assertEqual ( self . cbpath , '<STR_LIT>' ) <EOL> config = self . cbconfig <EOL> self . assertTrue ( config [ '<STR_LIT:host>' ] in [ '<STR_LIT:localhost>' , '<STR_LIT>' ] ) <EOL> mismatch_keys = [ <EOL> key for key in self . sample <EOL> if not key in config or self . sample [ key ] != config [ key ] ] <EOL> for key in mismatch_keys : <EOL> self . assertTrue ( key in [ '<STR_LIT:host>' , '<STR_LIT>' ] ) <EOL> def test_load_config_with_multiple_entries ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample , '<STR_LIT:127.0.0.1>' ) <EOL> sample2 = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' <EOL> } <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , sample2 , '<STR_LIT>' ) <EOL> ddc = pettingzoo . discovery . DistributedDiscovery ( self . connection ) <EOL> config = ddc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mismatch_keys = [ <EOL> key for key in self . sample <EOL> if not key in config or self . sample [ key ] != config [ key ] ] <EOL> for key in mismatch_keys : <EOL> self . assertTrue ( key in [ '<STR_LIT:host>' , '<STR_LIT>' ] ) <EOL> def tearDown ( self ) : <EOL> pettingzoo . discovery . CONFIG_PATH = '<STR_LIT>' <EOL> self . connection . close ( ) <EOL> self . connection = connect_to_zk ( '<STR_LIT>' ) <EOL> self . connection . delete_recursive ( self . path ) <EOL> self . connection . close ( ) <EOL> if self . mock : <EOL> import zc . zk . testing <EOL> zc . zk . testing . tearDown ( self ) <EOL> class DistributedMultiDiscovery ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> pettingzoo . utils . configure_logger ( ) <EOL> self . mock = True <EOL> self . conn_string = '<STR_LIT>' <EOL> if self . mock : <EOL> import zc . zk . testing <EOL> zc . zk . testing . setUp ( self , connection_string = self . conn_string ) <EOL> self . connection = connect_to_zk ( self . conn_string ) <EOL> self . path = '<STR_LIT>' <EOL> pettingzoo . discovery . CONFIG_PATH = self . path <EOL> self . sample = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT:localhost>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' <EOL> } <EOL> self . sample2 = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' <EOL> } <EOL> def test_config_dne ( self ) : <EOL> dmc = pettingzoo . discovery . DistributedMultiDiscovery ( self . connection ) <EOL> self . assertRaises ( IOError , dmc . load_config , "<STR_LIT>" , "<STR_LIT>" ) <EOL> def test_load_config ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample , '<STR_LIT:127.0.0.1>' ) <EOL> dmc = pettingzoo . discovery . DistributedMultiDiscovery ( self . connection ) <EOL> configs = dmc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertTrue ( len ( configs ) , <NUM_LIT:1> ) <EOL> config = configs [ <NUM_LIT:0> ] <EOL> mismatch_keys = [ <EOL> key for key in self . sample <EOL> if not key in config or self . sample [ key ] != config [ key ] ] <EOL> self . assertEqual ( len ( mismatch_keys ) , <NUM_LIT:0> ) <EOL> def test_load_config_with_multiple_entries ( self ) : <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample , '<STR_LIT:127.0.0.1>' ) <EOL> pettingzoo . discovery . write_distributed_config ( <EOL> self . connection , '<STR_LIT>' , '<STR_LIT>' , self . sample2 , '<STR_LIT>' ) <EOL> dmc = pettingzoo . discovery . DistributedMultiDiscovery ( self . connection ) <EOL> configs = dmc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertTrue ( len ( configs ) , <NUM_LIT:2> ) <EOL> ips = [ '<STR_LIT:127.0.0.1>' , '<STR_LIT>' ] <EOL> for config in configs : <EOL> mismatch_keys = [ <EOL> key for key in self . sample <EOL> if not key in config or self . sample [ key ] != config [ key ] ] <EOL> ips . remove ( config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:key>' ] ) <EOL> self . assertEquals ( len ( ips ) , <NUM_LIT:0> ) <EOL> def tearDown ( self ) : <EOL> pettingzoo . discovery . CONFIG_PATH = '<STR_LIT>' <EOL> self . connection . close ( ) <EOL> self . connection = connect_to_zk ( '<STR_LIT>' ) <EOL> self . connection . delete_recursive ( self . path ) <EOL> self . connection . close ( ) <EOL> if self . mock : <EOL> import zc . zk . testing <EOL> zc . zk . testing . tearDown ( self ) <EOL> class FileFallbackTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> pettingzoo . utils . configure_logger ( ) <EOL> self . mock = True <EOL> self . conn_string = '<STR_LIT>' <EOL> if self . mock : <EOL> import zc . zk . testing <EOL> zc . zk . testing . setUp ( self , connection_string = self . conn_string ) <EOL> self . connection = connect_to_zk ( self . conn_string ) <EOL> self . path = '<STR_LIT>' <EOL> pettingzoo . discovery . CONFIG_PATH = self . path <EOL> self . sample = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT:localhost>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' <EOL> } <EOL> self . sample2 = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT:version>' : <NUM_LIT:1.0> } <EOL> } , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' <EOL> } <EOL> pettingzoo . local_config . LocalConfig = pettingzoo . local_config . LocalConfigTest ( ) <EOL> pettingzoo . local_config . LocalConfig ( ) . add_config ( <EOL> self . sample , '<STR_LIT>' ) <EOL> self . list_sample = { '<STR_LIT>' : [ self . sample ] } <EOL> pettingzoo . local_config . LocalConfig ( ) . add_config ( <EOL> self . list_sample , '<STR_LIT>' ) <EOL> self . list_sample2 = { '<STR_LIT>' : [ self . sample , self . sample2 ] } <EOL> pettingzoo . local_config . LocalConfig ( ) . add_config ( <EOL> self . list_sample2 , '<STR_LIT>' ) <EOL> def test_config_file_fallback_dc ( self ) : <EOL> ddc = pettingzoo . discovery . DistributedDiscovery ( self . connection ) <EOL> config = ddc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEquals ( config , self . sample ) <EOL> def test_config_file_list_fallback_dc ( self ) : <EOL> ddc = pettingzoo . discovery . DistributedDiscovery ( self . connection ) <EOL> config = ddc . load_config ( '<STR_LIT>' , '<STR_LIT:list>' ) <EOL> self . assertEquals ( config , self . sample ) <EOL> config = ddc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertTrue ( config in [ self . sample , self . sample2 ] ) <EOL> def test_config_file_fallback_dmc ( self ) : <EOL> dmc = pettingzoo . discovery . DistributedMultiDiscovery ( self . connection ) <EOL> config = dmc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEquals ( config [ <NUM_LIT:0> ] , self . sample ) <EOL> def test_config_file_fallback_list_dmc ( self ) : <EOL> dmc = pettingzoo . discovery . DistributedMultiDiscovery ( self . connection ) <EOL> config = dmc . load_config ( '<STR_LIT>' , '<STR_LIT:list>' ) <EOL> self . assertEquals ( config [ <NUM_LIT:0> ] , self . sample ) <EOL> config = dmc . load_config ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEquals ( config [ <NUM_LIT:0> ] , self . sample ) <EOL> self . assertEquals ( config [ <NUM_LIT:1> ] , self . sample2 ) <EOL> def tearDown ( self ) : <EOL> pettingzoo . local_config . LocalConfig = pettingzoo . local_config . LocalConfigDefault ( ) <EOL> pettingzoo . discovery . CONFIG_PATH = '<STR_LIT>' <EOL> self . connection . close ( ) <EOL> self . connection = connect_to_zk ( '<STR_LIT>' ) <EOL> self . connection . delete_recursive ( self . path ) <EOL> self . connection . close ( ) <EOL> if self . mock : <EOL> import zc . zk . testing <EOL> zc . zk . testing . tearDown ( self ) </s>
