<s> import pytest <EOL> import sys <EOL> import os <EOL> import traceback <EOL> import hashlib <EOL> import twisted <EOL> import shutil <EOL> import json <EOL> from calvin . utilities . calvin_callback import CalvinCB <EOL> from calvin . utilities import calvinlogger <EOL> from calvin . utilities . utils import get_home <EOL> from calvin . runtime . south . plugins . storage . twistedimpl . securedht . append_server import * <EOL> from calvin . runtime . south . plugins . storage . twistedimpl . securedht . dht_server import * <EOL> from calvin . runtime . south . plugins . storage . twistedimpl . securedht . service_discovery_ssdp import * <EOL> from calvin . runtime . south . plugins . storage . twistedimpl . securedht . dht_server_commons import drawNetworkState <EOL> from kademlia . node import Node <EOL> from kademlia . utils import deferredDict , digest <EOL> from calvin . runtime . south . plugins . async import threads <EOL> from calvin . utilities import calvinconfig <EOL> _conf = calvinconfig . get ( ) <EOL> _conf . add_section ( "<STR_LIT>" ) <EOL> _conf_file = os . path . join ( get_home ( ) , "<STR_LIT>" ) <EOL> _conf . set ( "<STR_LIT>" , "<STR_LIT>" , _conf_file ) <EOL> _conf . set ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:test>" ) <EOL> _cert_conf = None <EOL> _log = calvinlogger . get_logger ( __name__ ) <EOL> name = "<STR_LIT>" <EOL> @ pytest . fixture ( scope = "<STR_LIT>" , autouse = True ) <EOL> def cleanup ( request ) : <EOL> def fin ( ) : <EOL> reactor . callFromThread ( reactor . stop ) <EOL> request . addfinalizer ( fin ) <EOL> @ pytest . mark . interactive <EOL> @ pytest . mark . slow <EOL> class TestDHT ( object ) : <EOL> test_nodes = <NUM_LIT:2> <EOL> _sucess_start = ( True , ) <EOL> @ pytest . fixture ( autouse = True , scope = "<STR_LIT:class>" ) <EOL> def setup ( self , request ) : <EOL> global _cert_conf <EOL> _cert_conf = certificate . Config ( _conf_file , "<STR_LIT:test>" ) . configuration <EOL> @ pytest . inlineCallbacks <EOL> def test_dht_multi ( self , monkeypatch ) : <EOL> iface = "<STR_LIT>" <EOL> a = None <EOL> b = None <EOL> q = Queue . Queue ( ) <EOL> def server_started ( aa , * args ) : <EOL> for b in args : <EOL> if isinstance ( b , twisted . python . failure . Failure ) : <EOL> b . printTraceback ( ) <EOL> else : <EOL> _log . debug ( "<STR_LIT>" % b ) <EOL> q . put ( [ aa , args ] ) <EOL> try : <EOL> amount_of_servers = <NUM_LIT:5> <EOL> servers = [ ] <EOL> callbacks = [ ] <EOL> for servno in range ( <NUM_LIT:0> , amount_of_servers ) : <EOL> a = AutoDHTServer ( ) <EOL> servers . append ( a ) <EOL> callback = CalvinCB ( server_started , str ( servno ) ) <EOL> servers [ servno ] . start ( iface , network = "<STR_LIT>" , cb = callback , name = name + "<STR_LIT:{}>" . format ( servno ) ) <EOL> callbacks . append ( callback ) <EOL> started = [ ] <EOL> while len ( started ) < amount_of_servers : <EOL> try : <EOL> server = yield threads . defer_to_thread ( q . get ) <EOL> except Queue . Empty : <EOL> _log . debug ( "<STR_LIT>" ) <EOL> if server not in started : <EOL> started . append ( server ) <EOL> callbacks [ int ( server [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) ] . func = lambda * args , ** kvargs : None <EOL> else : <EOL> print ( "<STR_LIT>" "<STR_LIT>" . format ( started , <EOL> len ( started ) , <EOL> amount_of_servers ) ) <EOL> print ( "<STR_LIT>" . format ( started , <EOL> len ( started ) , <EOL> amount_of_servers ) ) <EOL> for servno in range ( <NUM_LIT:0> , amount_of_servers ) : <EOL> assert [ str ( servno ) , self . _sucess_start ] in started <EOL> yield threads . defer_to_thread ( q . queue . clear ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:8> ) <EOL> key = "<STR_LIT>" <EOL> value = json . dumps ( [ "<STR_LIT>" ] ) <EOL> set_def = servers [ <NUM_LIT:0> ] . append ( key = key , value = value ) <EOL> set_value = yield threads . defer_to_thread ( set_def . wait , <NUM_LIT:10> ) <EOL> assert set_value <EOL> print ( "<STR_LIT>" . format ( servers [ <NUM_LIT:0> ] . dht_server . port . getHost ( ) . port , key , value ) ) <EOL> value = json . dumps ( [ "<STR_LIT>" ] ) <EOL> set_def = servers [ <NUM_LIT:0> ] . append ( key = key , value = value ) <EOL> set_value = yield threads . defer_to_thread ( set_def . wait , <NUM_LIT:10> ) <EOL> assert set_value <EOL> print ( "<STR_LIT>" . format ( servers [ <NUM_LIT:0> ] . dht_server . port . getHost ( ) . port , key , value ) ) <EOL> get_def = servers [ <NUM_LIT:0> ] . get_concat ( key = key ) <EOL> get_value = yield threads . defer_to_thread ( get_def . wait , <NUM_LIT:10> ) <EOL> assert set ( json . loads ( get_value ) ) == set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> print ( "<STR_LIT>" . format ( servers [ <NUM_LIT:0> ] . dht_server . port . getHost ( ) . port , key , get_value ) ) <EOL> drawNetworkState ( "<STR_LIT>" , servers , amount_of_servers ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:7> ) <EOL> drawNetworkState ( "<STR_LIT>" , servers , amount_of_servers ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:7> ) <EOL> drawNetworkState ( "<STR_LIT>" , servers , amount_of_servers ) <EOL> get_def = servers [ <NUM_LIT:0> ] . get_concat ( key = key ) <EOL> get_value = yield threads . defer_to_thread ( get_def . wait , <NUM_LIT:10> ) <EOL> assert set ( json . loads ( get_value ) ) == set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> print ( "<STR_LIT>" . format ( servers [ <NUM_LIT:0> ] . dht_server . port . getHost ( ) . port , get_value ) ) <EOL> value = json . dumps ( [ "<STR_LIT>" ] ) <EOL> set_def = servers [ <NUM_LIT:0> ] . remove ( key = key , value = value ) <EOL> set_value = yield threads . defer_to_thread ( set_def . wait , <NUM_LIT:10> ) <EOL> assert set_value <EOL> print ( "<STR_LIT>" . format ( servers [ <NUM_LIT:0> ] . dht_server . port . getHost ( ) . port , key , value ) ) <EOL> get_def = servers [ <NUM_LIT:1> ] . get_concat ( key = key ) <EOL> get_value = yield threads . defer_to_thread ( get_def . wait , <NUM_LIT:10> ) <EOL> assert set ( json . loads ( get_value ) ) == set ( [ "<STR_LIT>" ] ) <EOL> print ( "<STR_LIT>" . format ( servers [ <NUM_LIT:0> ] . dht_server . port . getHost ( ) . port , get_value ) ) <EOL> for i in range ( <NUM_LIT:0> , amount_of_servers ) : <EOL> name_dir = os . path . join ( _cert_conf [ "<STR_LIT>" ] [ "<STR_LIT>" ] , "<STR_LIT>" . format ( name , i ) ) <EOL> filenames = os . listdir ( os . path . join ( name_dir , "<STR_LIT>" ) ) <EOL> print ( "<STR_LIT>" . format ( servers [ i ] . dht_server . port . getHost ( ) . port , len ( filenames ) ) ) <EOL> except AssertionError as e : <EOL> print ( "<STR_LIT>" . format ( servers [ <NUM_LIT:0> ] . dht_server . port . getHost ( ) . port , get_value , value ) ) <EOL> pytest . fail ( traceback . format_exc ( ) ) <EOL> except Exception as e : <EOL> traceback . print_exc ( ) <EOL> pytest . fail ( traceback . format_exc ( ) ) <EOL> finally : <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:10> ) <EOL> i = <NUM_LIT:0> <EOL> for server in servers : <EOL> name_dir = os . path . join ( _cert_conf [ "<STR_LIT>" ] [ "<STR_LIT>" ] , name + "<STR_LIT:{}>" . format ( i ) ) <EOL> shutil . rmtree ( os . path . join ( name_dir , "<STR_LIT>" ) , ignore_errors = True ) <EOL> os . mkdir ( os . path . join ( name_dir , "<STR_LIT>" ) ) <EOL> i += <NUM_LIT:1> <EOL> server . stop ( ) </s>
