<s> from __future__ import print_function , unicode_literals <EOL> from mock import patch <EOL> import json <EOL> import mock <EOL> import experiments . ud_xilinx . server as UdXilinxExperiment <EOL> import experiments . ud_xilinx . exc as UdXilinxExperimentErrors <EOL> from experiments . ud_xilinx import command_senders as UdXilinxCommandSenders <EOL> import unittest <EOL> from experiments . ud_xilinx . watertank_simulation import Watertank <EOL> from test . unit . experiments . ud_xilinx . test_server import FakeSerialPort , FakeHttpDevice <EOL> import voodoo . configuration as ConfigurationManager <EOL> import test . unit . configuration as configuration_module <EOL> import time <EOL> from mock import patch , Mock <EOL> import time <EOL> _time = <NUM_LIT:100> <EOL> def _sleep_mock_handler ( * args , ** kwargs ) : <EOL> global _time <EOL> _time += args [ <NUM_LIT:0> ] <EOL> def _time_mock_handler ( * args , ** kwargs ) : <EOL> global _time <EOL> return _time <EOL> class BasicUdXilinxExperimentTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> from voodoo . configuration import ConfigurationManager <EOL> from voodoo . sessions . session_id import SessionId <EOL> self . cfg_manager = ConfigurationManager ( ) <EOL> self . cfg_manager . append_module ( configuration_module ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> UdXilinxCommandSenders . _SerialPort = FakeSerialPort <EOL> UdXilinxCommandSenders . _HttpDevice = FakeHttpDevice <EOL> self . experiment = UdXilinxExperiment . UdXilinxExperiment ( None , None , self . cfg_manager ) <EOL> self . lab_session_id = SessionId ( '<STR_LIT>' ) <EOL> def cleanUp ( self ) : <EOL> pass <EOL> def test_basic_start ( self ) : <EOL> configjson = self . experiment . do_start_experiment ( ) <EOL> self . assertIsNotNone ( configjson ) <EOL> def test_basic_start_config ( self ) : <EOL> configjson = self . experiment . do_start_experiment ( ) <EOL> self . assertEquals ( UdXilinxExperiment . STATE_NOT_READY , self . experiment . _current_state ) <EOL> initconfig = json . loads ( configjson ) <EOL> self . assertIsNotNone ( initconfig ) <EOL> self . assertTrue ( "<STR_LIT>" in initconfig ) <EOL> config = json . loads ( initconfig [ "<STR_LIT>" ] ) <EOL> self . assertTrue ( "<STR_LIT>" in config ) <EOL> def test_basic_start_end ( self ) : <EOL> self . experiment . do_start_experiment ( ) <EOL> self . experiment . do_dispose ( ) <EOL> def test_check_state ( self ) : <EOL> self . experiment . do_start_experiment ( ) <EOL> resp = self . experiment . do_send_command_to_device ( "<STR_LIT>" ) <EOL> self . assertEquals ( resp , "<STR_LIT>" + UdXilinxExperiment . STATE_NOT_READY ) <EOL> def _test_report_switches ( self ) : <EOL> self . experiment . do_start_experiment ( ) <EOL> resp = self . experiment . do_send_command_to_device ( "<STR_LIT>" ) <EOL> self . assertEquals ( "<STR_LIT>" , resp ) <EOL> def test_api_version ( self ) : <EOL> api = self . experiment . do_get_api ( ) <EOL> self . assertEquals ( "<STR_LIT:2>" , api ) <EOL> class EarlyKickingXilinxExperimentTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> from voodoo . configuration import ConfigurationManager <EOL> from voodoo . sessions . session_id import SessionId <EOL> self . cfg_manager = ConfigurationManager ( ) <EOL> self . cfg_manager . append_module ( configuration_module ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> UdXilinxCommandSenders . _SerialPort = FakeSerialPort <EOL> UdXilinxCommandSenders . _HttpDevice = FakeHttpDevice <EOL> self . experiment = UdXilinxExperiment . UdXilinxExperiment ( None , None , self . cfg_manager ) <EOL> self . lab_session_id = SessionId ( '<STR_LIT>' ) <EOL> def test_basic_start_config ( self ) : <EOL> configjson = self . experiment . do_start_experiment ( ) <EOL> self . assertEquals ( UdXilinxExperiment . STATE_NOT_READY , self . experiment . _current_state ) <EOL> initconfig = json . loads ( configjson ) <EOL> self . assertIsNotNone ( initconfig ) <EOL> self . assertTrue ( "<STR_LIT>" in initconfig ) <EOL> config = json . loads ( initconfig [ "<STR_LIT>" ] ) <EOL> self . assertTrue ( "<STR_LIT>" in config ) <EOL> self . assertTrue ( "<STR_LIT>" in config ) <EOL> def test_cleanup ( self ) : <EOL> self . experiment . do_start_experiment ( ) <EOL> self . experiment . do_dispose ( ) <EOL> self . assertIsNone ( self . experiment . _use_time_start ) <EOL> def test_report ( self ) : <EOL> self . experiment . do_start_experiment ( ) <EOL> resp = self . experiment . do_send_command_to_device ( "<STR_LIT>" ) <EOL> self . assertEquals ( "<STR_LIT>" , resp ) <EOL> class PermissionsXilinxExperimentTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> from voodoo . configuration import ConfigurationManager <EOL> from voodoo . sessions . session_id import SessionId <EOL> self . cfg_manager = ConfigurationManager ( ) <EOL> self . cfg_manager . append_module ( configuration_module ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , False ) <EOL> UdXilinxCommandSenders . _SerialPort = FakeSerialPort <EOL> UdXilinxCommandSenders . _HttpDevice = FakeHttpDevice <EOL> self . experiment = UdXilinxExperiment . UdXilinxExperiment ( None , None , self . cfg_manager ) <EOL> self . lab_session_id = SessionId ( '<STR_LIT>' ) <EOL> def cleanUp ( self ) : <EOL> self . experiment . do_dispose ( ) <EOL> def test_bit_disallowed ( self ) : <EOL> new_state = self . experiment . do_send_file_to_device ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( "<STR_LIT>" , new_state ) <EOL> def test_vhd_allowed ( self ) : <EOL> new_state = self . experiment . do_send_file_to_device ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertNotEquals ( "<STR_LIT>" , new_state ) <EOL> def test_all_disallowed ( self ) : <EOL> new_state = self . experiment . do_send_file_to_device ( "<STR_LIT>" , "<STR_LIT>" ) ; <EOL> self . assertEquals ( "<STR_LIT>" , new_state ) <EOL> class CompilerXilinxExperimentTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> from voodoo . configuration import ConfigurationManager <EOL> from voodoo . sessions . session_id import SessionId <EOL> self . cfg_manager = ConfigurationManager ( ) <EOL> self . cfg_manager . append_module ( configuration_module ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> UdXilinxCommandSenders . _SerialPort = FakeSerialPort <EOL> UdXilinxCommandSenders . _HttpDevice = FakeHttpDevice <EOL> self . experiment = UdXilinxExperiment . UdXilinxExperiment ( None , None , self . cfg_manager ) <EOL> self . configjson = self . experiment . do_start_experiment ( ) <EOL> self . lab_session_id = SessionId ( '<STR_LIT>' ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> def test_compile ( self , compiler_class ) : <EOL> pass <EOL> def cleanUp ( self ) : <EOL> self . experiment . do_dispose ( ) <EOL> class VirtualWorldXilinxExperimentTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> from voodoo . configuration import ConfigurationManager <EOL> from voodoo . sessions . session_id import SessionId <EOL> self . cfg_manager = ConfigurationManager ( ) <EOL> self . cfg_manager . append_module ( configuration_module ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> UdXilinxCommandSenders . _SerialPort = FakeSerialPort <EOL> UdXilinxCommandSenders . _HttpDevice = FakeHttpDevice <EOL> self . experiment = UdXilinxExperiment . UdXilinxExperiment ( None , None , self . cfg_manager ) <EOL> self . configjson = self . experiment . do_start_experiment ( ) <EOL> self . lab_session_id = SessionId ( '<STR_LIT>' ) <EOL> def cleanUp ( self ) : <EOL> self . experiment . do_dispose ( ) <EOL> def test_basic_state_command ( self ) : <EOL> resp = self . experiment . do_send_command_to_device ( "<STR_LIT>" ) <EOL> self . assertEquals ( "<STR_LIT:{}>" , resp ) <EOL> def _test_virtualworld_mode_command ( self ) : <EOL> resp = self . experiment . do_send_command_to_device ( "<STR_LIT>" ) <EOL> self . assertEquals ( "<STR_LIT>" , resp ) <EOL> def test_virtualworld_non_existant ( self ) : <EOL> resp = self . experiment . do_send_command_to_device ( "<STR_LIT>" ) <EOL> self . assertEquals ( "<STR_LIT>" , resp ) <EOL> class WatertankSimulationTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . watertank = Watertank ( <NUM_LIT:1000> , [ <NUM_LIT:10> , <NUM_LIT:10> ] , [ <NUM_LIT:10> ] , <NUM_LIT:0.5> ) <EOL> def test_water_raises ( self ) : <EOL> so = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> so = json . loads ( so ) <EOL> sf = json . loads ( sf ) <EOL> self . assertGreater ( sf [ "<STR_LIT>" ] , so [ "<STR_LIT>" ] ) <EOL> def test_water_lowers ( self ) : <EOL> self . watertank . set_inputs ( [ <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> so = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> so = json . loads ( so ) <EOL> sf = json . loads ( sf ) <EOL> self . assertLess ( sf [ "<STR_LIT>" ] , so [ "<STR_LIT>" ] ) <EOL> def cleanUp ( self ) : <EOL> pass <EOL> class WatertankSimulationTestCaseTemperatures ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . watertank = Watertank ( <NUM_LIT:1000> , [ <NUM_LIT:10> , <NUM_LIT:10> ] , [ <NUM_LIT:10> ] , <NUM_LIT:0.5> , True ) <EOL> def test_not_overheat ( self ) : <EOL> self . assertFalse ( self . watertank . firstPumpOverheated ) <EOL> self . assertFalse ( self . watertank . secondPumpOverheated ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> self . assertFalse ( self . watertank . firstPumpOverheated ) <EOL> self . assertFalse ( self . watertank . secondPumpOverheated ) <EOL> def test_stay_within_temp_work_range ( self ) : <EOL> self . assertTrue ( self . watertank . firstPumpWorkRange [ <NUM_LIT:1> ] >= self . watertank . firstPumpTemperature >= <EOL> self . watertank . secondPumpWorkRange [ <NUM_LIT:0> ] ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> self . assertTrue ( self . watertank . firstPumpWorkRange [ <NUM_LIT:1> ] >= self . watertank . firstPumpTemperature >= <EOL> self . watertank . secondPumpWorkRange [ <NUM_LIT:0> ] ) <EOL> def test_water_raises ( self ) : <EOL> so = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> so = json . loads ( so ) <EOL> sf = json . loads ( sf ) <EOL> self . assertGreater ( sf [ "<STR_LIT>" ] , so [ "<STR_LIT>" ] ) <EOL> def test_water_lowers ( self ) : <EOL> self . watertank . set_inputs ( [ <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> so = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> so = json . loads ( so ) <EOL> sf = json . loads ( sf ) <EOL> self . assertLess ( sf [ "<STR_LIT>" ] , so [ "<STR_LIT>" ] ) <EOL> def test_left_temperature_raises ( self ) : <EOL> self . watertank . set_inputs ( [ <NUM_LIT:10> , <NUM_LIT:0> ] ) <EOL> so = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> so = json . loads ( so ) <EOL> sf = json . loads ( sf ) <EOL> self . assertGreater ( sf [ "<STR_LIT>" ] [ <NUM_LIT:0> ] , so [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ) <EOL> def test_left_temperature_remains ( self ) : <EOL> self . watertank . set_inputs ( [ <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> so = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> rso = self . watertank . firstPumpTemperature <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> rsf = self . watertank . firstPumpTemperature <EOL> so = json . loads ( so ) <EOL> sf = json . loads ( sf ) <EOL> self . assertEquals ( rsf , rso ) <EOL> self . assertEquals ( sf [ "<STR_LIT>" ] [ <NUM_LIT:0> ] , so [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ) <EOL> def test_right_temperature_raises ( self ) : <EOL> self . watertank . set_inputs ( [ <NUM_LIT:0> , <NUM_LIT:10> ] ) <EOL> so = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> so = json . loads ( so ) <EOL> sf = json . loads ( sf ) <EOL> self . assertGreater ( sf [ "<STR_LIT>" ] [ <NUM_LIT:1> ] , so [ "<STR_LIT>" ] [ <NUM_LIT:1> ] ) <EOL> def test_right_temperature_remains ( self ) : <EOL> self . watertank . set_inputs ( [ <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> so = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> rso = self . watertank . secondPumpTemperature <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> sf = self . watertank . get_json_state ( [ <NUM_LIT:20> , <NUM_LIT:20> ] , [ <NUM_LIT:20> ] ) <EOL> rsf = self . watertank . secondPumpTemperature <EOL> so = json . loads ( so ) <EOL> sf = json . loads ( sf ) <EOL> self . assertEquals ( rsf , rso ) <EOL> self . assertEquals ( sf [ "<STR_LIT>" ] [ <NUM_LIT:1> ] , so [ "<STR_LIT>" ] [ <NUM_LIT:1> ] ) <EOL> def test_all_overheat ( self ) : <EOL> self . watertank . set_inputs ( [ <NUM_LIT:10> , <NUM_LIT:10> ] ) <EOL> for i in range ( <NUM_LIT:30> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> self . assertTrue ( self . watertank . firstPumpOverheated ) <EOL> self . assertTrue ( self . watertank . secondPumpOverheated ) <EOL> def test_all_recover_after_overheat ( self ) : <EOL> self . watertank . set_inputs ( [ <NUM_LIT:10> , <NUM_LIT:10> ] ) <EOL> for i in range ( <NUM_LIT:30> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> self . watertank . set_inputs ( [ <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> for i in range ( <NUM_LIT:20> ) : <EOL> self . watertank . update ( <NUM_LIT:1> ) <EOL> self . assertFalse ( self . watertank . firstPumpOverheated ) <EOL> self . assertFalse ( self . watertank . secondPumpOverheated ) <EOL> def cleanUp ( self ) : <EOL> pass <EOL> class FakeOpen ( object ) : <EOL> def read ( self ) : <EOL> return """<STR_LIT>""" <EOL> def _ret_fake ( self ) : <EOL> return VirtualWorldXilinxExperimentTestCase . FakeOpen ( ) <EOL> @ patch ( "<STR_LIT>" , new = _ret_fake ) <EOL> def test_read_leds ( self ) : <EOL> resp = self . experiment . query_leds_from_json ( ) <EOL> self . assertEquals ( [ "<STR_LIT:0>" , "<STR_LIT:1>" , "<STR_LIT:1>" , "<STR_LIT:0>" , "<STR_LIT:1>" , "<STR_LIT:1>" , "<STR_LIT:0>" , "<STR_LIT:1>" ] , resp ) <EOL> resp = self . experiment . do_send_command_to_device ( "<STR_LIT>" ) <EOL> self . assertEquals ( "<STR_LIT>" , resp ) <EOL> def suite ( ) : <EOL> return unittest . TestSuite ( <EOL> ( <EOL> ) <EOL> ) <EOL> class FakeImpact ( object ) : <EOL> def __init__ ( self ) : <EOL> super ( FakeImpact , self ) . __init__ ( ) <EOL> def program_device ( self , program_path ) : <EOL> pass <EOL> def source2svf ( self , program_path ) : <EOL> pass <EOL> def get_suffix ( self ) : <EOL> return "<STR_LIT>" <EOL> class FakeSerialPort ( object ) : <EOL> def __init__ ( self ) : <EOL> super ( FakeSerialPort , self ) . __init__ ( ) <EOL> self . dict = { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:0> } <EOL> self . codes = [ ] <EOL> def _increment ( self , sth ) : <EOL> self . dict [ sth ] = self . dict [ sth ] + <NUM_LIT:1> <EOL> def open_serial_port ( self , number ) : <EOL> self . _increment ( '<STR_LIT>' ) <EOL> def send_code ( self , n ) : <EOL> self . _increment ( '<STR_LIT>' ) <EOL> self . codes . append ( n ) <EOL> def close_serial_port ( self ) : <EOL> self . _increment ( '<STR_LIT>' ) <EOL> class FakeHttpDevice ( object ) : <EOL> def __init__ ( self , * args , ** kargs ) : <EOL> super ( FakeHttpDevice , self ) . __init__ ( ) <EOL> self . sent = <NUM_LIT:0> <EOL> self . msgs = [ ] <EOL> def send_message ( self , msg ) : <EOL> self . sent += len ( msg ) <EOL> self . msgs . append ( msg ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
