<s> import sys <EOL> import os . path <EOL> import heapq <EOL> import unittest <EOL> this_dir = os . path . dirname ( os . path . abspath ( __file__ ) ) <EOL> sys . path . append ( os . path . dirname ( this_dir ) ) <EOL> from paxos import external <EOL> import test_practical <EOL> from test_practical import PID <EOL> class ExternalMessenger ( test_practical . PracticalMessenger ) : <EOL> procs = <NUM_LIT:0> <EOL> tleader = None <EOL> def send_leadership_proclamation ( self , pnum ) : <EOL> self . procs += <NUM_LIT:1> <EOL> def on_leadership_acquired ( self ) : <EOL> super ( ExternalMessenger , self ) . on_leadership_acquired ( ) <EOL> self . tleader = '<STR_LIT>' <EOL> def on_leadership_lost ( self ) : <EOL> self . tleader = '<STR_LIT>' <EOL> def on_leadership_change ( self , old_uid , new_uid ) : <EOL> pass <EOL> class ExternalTests ( object ) : <EOL> node_factory = None <EOL> def create_node ( self ) : <EOL> self . l = self . node_factory ( self , '<STR_LIT:A>' , <NUM_LIT:3> ) <EOL> def test_initial_leader ( self ) : <EOL> self . l = self . node_factory ( self , '<STR_LIT:A>' , <NUM_LIT:3> , '<STR_LIT:A>' ) <EOL> self . assertTrue ( self . l . leader ) <EOL> self . assertEquals ( self . procs , <NUM_LIT:0> ) <EOL> def test_gain_leader ( self ) : <EOL> self . l . set_proposal ( '<STR_LIT:foo>' ) <EOL> self . l . prepare ( ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) ) <EOL> self . l . recv_promise ( '<STR_LIT:A>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , None , None ) <EOL> self . l . recv_promise ( '<STR_LIT:B>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , None , None ) <EOL> self . an ( ) <EOL> self . l . recv_promise ( '<STR_LIT:C>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , None , None ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , '<STR_LIT:foo>' ) <EOL> self . assertEquals ( self . tleader , '<STR_LIT>' ) <EOL> self . assertEquals ( self . l . leader_uid , '<STR_LIT:A>' ) <EOL> def test_gain_leader_nack ( self ) : <EOL> self . l . set_proposal ( '<STR_LIT:foo>' ) <EOL> self . l . prepare ( ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) ) <EOL> self . l . recv_promise ( '<STR_LIT:A>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , None , None ) <EOL> self . l . recv_prepare_nack ( '<STR_LIT:B>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , PID ( <NUM_LIT:2> , '<STR_LIT:C>' ) ) <EOL> self . l . prepare ( ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:3> , '<STR_LIT:A>' ) ) <EOL> def test_lose_leader ( self ) : <EOL> self . test_gain_leader ( ) <EOL> self . assertEquals ( self . l . leader_proposal_id , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) ) <EOL> self . l . recv_leadership_proclamation ( '<STR_LIT:B>' , PID ( <NUM_LIT:5> , '<STR_LIT:B>' ) ) <EOL> self . assertEquals ( self . l . leader_proposal_id , PID ( <NUM_LIT:5> , '<STR_LIT:B>' ) ) <EOL> self . assertEquals ( self . tleader , '<STR_LIT>' ) <EOL> def test_lose_leader_via_nacks ( self ) : <EOL> self . test_gain_leader ( ) <EOL> self . assertEquals ( self . l . leader_proposal_id , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) ) <EOL> self . l . recv_accept_nack ( '<STR_LIT:B>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , PID ( <NUM_LIT:2> , '<STR_LIT:B>' ) ) <EOL> self . l . recv_accept_nack ( '<STR_LIT:C>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , PID ( <NUM_LIT:2> , '<STR_LIT:B>' ) ) <EOL> self . assertEquals ( self . tleader , '<STR_LIT>' ) <EOL> self . assertEquals ( self . l . leader_proposal_id , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) ) <EOL> self . l . recv_accept_nack ( '<STR_LIT:D>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , PID ( <NUM_LIT:2> , '<STR_LIT:B>' ) ) <EOL> self . assertEquals ( self . l . leader_proposal_id , None ) <EOL> self . assertEquals ( self . tleader , '<STR_LIT>' ) <EOL> def test_regain_leader ( self ) : <EOL> self . test_lose_leader ( ) <EOL> self . l . prepare ( ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:6> , '<STR_LIT:A>' ) ) <EOL> self . l . recv_promise ( '<STR_LIT:B>' , PID ( <NUM_LIT:6> , '<STR_LIT:A>' ) , None , None ) <EOL> self . l . recv_promise ( '<STR_LIT:C>' , PID ( <NUM_LIT:6> , '<STR_LIT:A>' ) , None , None ) <EOL> self . an ( ) <EOL> self . l . recv_promise ( '<STR_LIT:D>' , PID ( <NUM_LIT:6> , '<STR_LIT:A>' ) , None , None ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:6> , '<STR_LIT:A>' ) , '<STR_LIT:foo>' ) <EOL> self . assertEquals ( self . tleader , '<STR_LIT>' ) <EOL> def test_ignore_old_leader_proclamation ( self ) : <EOL> self . test_lose_leader ( ) <EOL> self . l . recv_leadership_proclamation ( '<STR_LIT:A>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) ) <EOL> self . assertEquals ( self . l . leader_proposal_id , PID ( <NUM_LIT:5> , '<STR_LIT:B>' ) ) <EOL> def test_proposal_id_increment ( self ) : <EOL> self . l . set_proposal ( '<STR_LIT:foo>' ) <EOL> self . l . prepare ( ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) ) <EOL> self . l . recv_promise ( '<STR_LIT:A>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , None , None ) <EOL> self . l . recv_promise ( '<STR_LIT:B>' , PID ( <NUM_LIT:1> , '<STR_LIT:A>' ) , None , None ) <EOL> self . an ( ) <EOL> self . l . prepare ( ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:2> , '<STR_LIT:A>' ) ) <EOL> self . l . recv_promise ( '<STR_LIT:A>' , PID ( <NUM_LIT:2> , '<STR_LIT:A>' ) , None , None ) <EOL> self . l . recv_promise ( '<STR_LIT:B>' , PID ( <NUM_LIT:2> , '<STR_LIT:A>' ) , None , None ) <EOL> self . l . recv_promise ( '<STR_LIT:C>' , PID ( <NUM_LIT:2> , '<STR_LIT:A>' ) , None , None ) <EOL> self . am ( '<STR_LIT>' , PID ( <NUM_LIT:2> , '<STR_LIT:A>' ) , '<STR_LIT:foo>' ) <EOL> self . assertEquals ( self . tleader , '<STR_LIT>' ) <EOL> class ExternalTester ( ExternalTests , ExternalMessenger , unittest . TestCase ) : <EOL> node_factory = external . ExternalNode <EOL> def setUp ( self ) : <EOL> super ( ExternalTester , self ) . setUp ( ) <EOL> self . create_node ( ) <EOL> class TNode ( test_practical . AutoSaveMixin , external . ExternalNode ) : <EOL> pass <EOL> class ExternalProposerTester ( test_practical . PracticalProposerTests , <EOL> ExternalMessenger , <EOL> unittest . TestCase ) : <EOL> proposer_factory = TNode <EOL> def setUp ( self ) : <EOL> super ( ExternalProposerTester , self ) . setUp ( ) <EOL> class ExternalAcceptorTester ( test_practical . PracticalAcceptorTests , <EOL> ExternalMessenger , <EOL> unittest . TestCase ) : <EOL> acceptor_factory = TNode <EOL> def setUp ( self ) : <EOL> super ( ExternalAcceptorTester , self ) . setUp ( ) <EOL> class ExternalLearnerTester ( test_practical . PracticalLearnerTests , <EOL> ExternalMessenger , <EOL> unittest . TestCase ) : <EOL> learner_factory = TNode <EOL> def setUp ( self ) : <EOL> super ( ExternalLearnerTester , self ) . setUp ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
