<s> import unittest <EOL> import sys <EOL> import os <EOL> import signal <EOL> from config . experiment_config_lib import ControllerConfig <EOL> from sts . control_flow . replayer import Replayer <EOL> from sts . topology import FatTree , BufferedPatchPanel , MeshTopology <EOL> from sts . simulation_state import SimulationConfig <EOL> from sts . entities import Host <EOL> from sts . util . convenience import IPAddressSpace <EOL> sys . path . append ( os . path . dirname ( __file__ ) + "<STR_LIT>" ) <EOL> _running_simulation = None <EOL> def handle_int ( sigspec , frame ) : <EOL> print >> sys . stderr , "<STR_LIT>" % sigspec <EOL> if ( _running_simulation is not None and <EOL> _running_simulation . current_simulation is not None ) : <EOL> _running_simulation . current_simulation . clean_up ( ) <EOL> raise RuntimeError ( "<STR_LIT>" % sigspec ) <EOL> signal . signal ( signal . SIGINT , handle_int ) <EOL> signal . signal ( signal . SIGTERM , handle_int ) <EOL> class ReplayerTest ( unittest . TestCase ) : <EOL> tmp_basic_superlog = '<STR_LIT>' <EOL> tmp_controller_superlog = '<STR_LIT>' <EOL> tmp_dataplane_superlog = '<STR_LIT>' <EOL> tmp_migration_superlog = '<STR_LIT>' <EOL> def write_simple_superlog ( self ) : <EOL> '''<STR_LIT>''' <EOL> superlog = open ( self . tmp_basic_superlog , '<STR_LIT:w>' ) <EOL> e1 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e1 + '<STR_LIT:\n>' ) <EOL> e2 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e2 + '<STR_LIT:\n>' ) <EOL> e3 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e3 + '<STR_LIT:\n>' ) <EOL> e4 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e4 + '<STR_LIT:\n>' ) <EOL> superlog . close ( ) <EOL> def setup_simple_simulation ( self ) : <EOL> controllers = [ ] <EOL> topology_class = FatTree <EOL> topology_params = "<STR_LIT>" <EOL> patch_panel_class = BufferedPatchPanel <EOL> sim = SimulationConfig ( controllers , topology_class , topology_params , patch_panel_class ) <EOL> global _running_simulation <EOL> _running_simulation = sim <EOL> return sim <EOL> def test_basic ( self ) : <EOL> simulation = None <EOL> try : <EOL> self . write_simple_superlog ( ) <EOL> simulation_cfg = self . setup_simple_simulation ( ) <EOL> replayer = Replayer ( simulation_cfg , self . tmp_basic_superlog ) <EOL> simulation = replayer . simulate ( ) <EOL> finally : <EOL> os . unlink ( self . tmp_basic_superlog ) <EOL> if simulation is not None : <EOL> simulation . clean_up ( ) <EOL> def write_controller_crash_superlog ( self ) : <EOL> superlog = open ( self . tmp_controller_superlog , '<STR_LIT:w>' ) <EOL> e1 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e1 + '<STR_LIT:\n>' ) <EOL> e2 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e2 + '<STR_LIT:\n>' ) <EOL> superlog . close ( ) <EOL> def setup_controller_simulation ( self ) : <EOL> start_cmd = "<STR_LIT>" <EOL> ControllerConfig . _controller_labels . clear ( ) <EOL> IPAddressSpace . _claimed_addresses . clear ( ) <EOL> controllers = [ ControllerConfig ( cwd = '<STR_LIT>' , label = "<STR_LIT>" , start_cmd = start_cmd , address = "<STR_LIT:127.0.0.1>" , port = <NUM_LIT> , sync = "<STR_LIT>" ) ] <EOL> topology_class = MeshTopology <EOL> topology_params = "<STR_LIT>" <EOL> patch_panel_class = BufferedPatchPanel <EOL> return SimulationConfig ( controllers , <EOL> topology_class , <EOL> topology_params , <EOL> patch_panel_class ) <EOL> def test_controller_crash ( self ) : <EOL> simulation = None <EOL> try : <EOL> self . write_controller_crash_superlog ( ) <EOL> simulation_cfg = self . setup_controller_simulation ( ) <EOL> replayer = Replayer ( simulation_cfg , self . tmp_controller_superlog ) <EOL> simulation = replayer . simulate ( ) <EOL> finally : <EOL> os . unlink ( self . tmp_controller_superlog ) <EOL> if simulation is not None : <EOL> simulation . clean_up ( ) <EOL> def write_dataplane_trace_superlog ( self ) : <EOL> superlog = open ( self . tmp_dataplane_superlog , '<STR_LIT:w>' ) <EOL> e1 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e1 + '<STR_LIT:\n>' ) <EOL> e2 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e2 + '<STR_LIT:\n>' ) <EOL> superlog . close ( ) <EOL> def setup_dataplane_simulation ( self ) : <EOL> controllers = [ ] <EOL> topology_class = MeshTopology <EOL> topology_params = "<STR_LIT>" <EOL> patch_panel_class = BufferedPatchPanel <EOL> dataplane_trace_path = "<STR_LIT>" <EOL> return SimulationConfig ( controllers , topology_class , topology_params , <EOL> patch_panel_class , dataplane_trace = dataplane_trace_path ) <EOL> def test_dataplane_injection ( self ) : <EOL> simulation = None <EOL> try : <EOL> self . write_dataplane_trace_superlog ( ) <EOL> simulation_cfg = self . setup_dataplane_simulation ( ) <EOL> replayer = Replayer ( simulation_cfg , self . tmp_dataplane_superlog ) <EOL> simulation = replayer . simulate ( ) <EOL> finally : <EOL> os . unlink ( self . tmp_dataplane_superlog ) <EOL> if simulation is not None : <EOL> simulation . clean_up ( ) <EOL> def write_migration_superlog ( self ) : <EOL> superlog = open ( self . tmp_migration_superlog , '<STR_LIT:w>' ) <EOL> e1 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e1 + '<STR_LIT:\n>' ) <EOL> e2 = str ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) <EOL> superlog . write ( e2 + '<STR_LIT:\n>' ) <EOL> superlog . close ( ) <EOL> def setup_migration_simulation ( self ) : <EOL> controllers = [ ] <EOL> topology_class = FatTree <EOL> topology_params = "<STR_LIT>" <EOL> patch_panel_class = BufferedPatchPanel <EOL> return SimulationConfig ( controllers , topology_class , topology_params , <EOL> patch_panel_class ) <EOL> def test_migration ( self ) : <EOL> simulation = None <EOL> try : <EOL> self . write_migration_superlog ( ) <EOL> simulation_cfg = self . setup_migration_simulation ( ) <EOL> replayer = Replayer ( simulation_cfg , self . tmp_migration_superlog ) <EOL> simulation = replayer . simulate ( ) <EOL> latest_switch = simulation . topology . get_switch ( <NUM_LIT:7> ) <EOL> latest_port = latest_switch . ports [ <NUM_LIT> ] <EOL> ( host , interface ) = simulation . topology . get_connected_port ( latest_switch , <EOL> latest_port ) <EOL> self . assertTrue ( type ( host ) == Host ) <EOL> finally : <EOL> os . unlink ( self . tmp_migration_superlog ) <EOL> if simulation is not None : <EOL> simulation . clean_up ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
