<s> """<STR_LIT>""" <EOL> import itertools <EOL> import unittest <EOL> import contextlib2 <EOL> from mock import patch <EOL> from perfkitbenchmarker import benchmark_spec <EOL> from perfkitbenchmarker import configs <EOL> from perfkitbenchmarker import context <EOL> from perfkitbenchmarker import errors <EOL> from perfkitbenchmarker import os_types <EOL> from perfkitbenchmarker import providers <EOL> from perfkitbenchmarker . configs import benchmark_config_spec <EOL> from perfkitbenchmarker . linux_benchmarks import ping_benchmark <EOL> from tests import mock_flags <EOL> NAME = '<STR_LIT>' <EOL> UID = '<STR_LIT>' <EOL> CONFIG_WITH_BACKGROUND_NETWORK = """<STR_LIT>""" <EOL> CONFIG_WITH_BACKGROUND_NETWORK_BAD_IPFLAG = """<STR_LIT>""" <EOL> CONFIG_WITH_BACKGROUND_NETWORK_IPFLAG = """<STR_LIT>""" <EOL> _GROUP_1 = '<STR_LIT>' <EOL> _GROUP_2 = '<STR_LIT>' <EOL> _MOCKED_VM_FUNCTIONS = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> class TestBackgroundNetworkWorkload ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestBackgroundNetworkWorkload , self ) . setUp ( ) <EOL> self . mocked_flags = mock_flags . PatchTestCaseFlags ( self ) <EOL> self . mocked_flags . os_type = os_types . DEBIAN <EOL> self . mocked_flags . cloud = providers . GCP <EOL> self . addCleanup ( context . SetThreadBenchmarkSpec , None ) <EOL> def _CheckVmCallCounts ( self , spec , working_groups , working_expected_counts , <EOL> non_working_groups , non_working_expected_counts ) : <EOL> expected_call_counts = { group : working_expected_counts <EOL> for group in working_groups } <EOL> expected_call_counts . update ( { group : non_working_expected_counts <EOL> for group in non_working_groups } ) <EOL> for group_name , vm_expected_call_counts in expected_call_counts . iteritems ( ) : <EOL> group_vms = spec . vm_groups [ group_name ] <EOL> self . assertEqual ( len ( group_vms ) , <NUM_LIT:1> , <EOL> msg = '<STR_LIT>' . format ( group_name , <EOL> len ( group_vms ) ) ) <EOL> vm = group_vms [ <NUM_LIT:0> ] <EOL> iter_mocked_functions = itertools . izip_longest ( _MOCKED_VM_FUNCTIONS , <EOL> vm_expected_call_counts ) <EOL> for function_name , expected_call_count in iter_mocked_functions : <EOL> call_count = getattr ( vm , function_name ) . call_count <EOL> self . assertEqual ( call_count , expected_call_count , msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( function_name , group_name , <EOL> expected_call_count , call_count ) ) ) <EOL> def _CheckVMFromSpec ( self , spec , working_groups = ( ) , non_working_groups = ( ) ) : <EOL> with contextlib2 . ExitStack ( ) as stack : <EOL> for vm in spec . vms : <EOL> for function_name in _MOCKED_VM_FUNCTIONS : <EOL> stack . enter_context ( patch . object ( vm , function_name ) ) <EOL> vm . RemoteCommand . side_effect = itertools . repeat ( ( '<STR_LIT:0>' , '<STR_LIT>' ) ) <EOL> working , non_working = working_groups , non_working_groups <EOL> self . _CheckVmCallCounts ( spec , working , ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , non_working , ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> spec . Prepare ( ) <EOL> self . _CheckVmCallCounts ( spec , working , ( <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> ) , non_working , ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> spec . StartBackgroundWorkload ( ) <EOL> self . _CheckVmCallCounts ( spec , working , ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:4> ) , non_working , ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> spec . StopBackgroundWorkload ( ) <EOL> self . _CheckVmCallCounts ( spec , working , ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:6> ) , non_working , ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> def makeSpec ( self , yaml_benchmark_config = ping_benchmark . BENCHMARK_CONFIG ) : <EOL> config = configs . LoadConfig ( yaml_benchmark_config , { } , NAME ) <EOL> config_spec = benchmark_config_spec . BenchmarkConfigSpec ( <EOL> NAME , flag_values = self . mocked_flags , ** config ) <EOL> spec = benchmark_spec . BenchmarkSpec ( config_spec , NAME , UID ) <EOL> spec . ConstructVirtualMachines ( ) <EOL> return spec <EOL> def testWindowsVMCausesError ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mocked_flags [ '<STR_LIT>' ] . Parse ( <NUM_LIT:200> ) <EOL> self . mocked_flags [ '<STR_LIT>' ] . Parse ( os_types . WINDOWS ) <EOL> spec = self . makeSpec ( ) <EOL> with self . assertRaisesRegexp ( Exception , '<STR_LIT>' ) : <EOL> spec . Prepare ( ) <EOL> with self . assertRaisesRegexp ( Exception , '<STR_LIT>' ) : <EOL> spec . StartBackgroundWorkload ( ) <EOL> with self . assertRaisesRegexp ( Exception , '<STR_LIT>' ) : <EOL> spec . StopBackgroundWorkload ( ) <EOL> def testBackgroundWorkloadVM ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mocked_flags [ '<STR_LIT>' ] . Parse ( <NUM_LIT:200> ) <EOL> spec = self . makeSpec ( ) <EOL> self . _CheckVMFromSpec ( spec , working_groups = ( _GROUP_1 , _GROUP_2 ) ) <EOL> def testBackgroundWorkloadVanillaConfig ( self ) : <EOL> """<STR_LIT>""" <EOL> spec = self . makeSpec ( ) <EOL> for vm in spec . vms : <EOL> self . assertIsNone ( vm . background_network_mbits_per_sec ) <EOL> self . _CheckVMFromSpec ( spec , non_working_groups = ( _GROUP_1 , _GROUP_2 ) ) <EOL> def testBackgroundWorkloadWindows ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mocked_flags . os_type = os_types . WINDOWS <EOL> spec = self . makeSpec ( ) <EOL> for vm in spec . vms : <EOL> self . assertIsNone ( vm . background_network_mbits_per_sec ) <EOL> self . _CheckVMFromSpec ( spec , non_working_groups = ( _GROUP_1 , _GROUP_2 ) ) <EOL> def testBackgroundWorkloadVanillaConfigFlag ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mocked_flags [ '<STR_LIT>' ] . Parse ( <NUM_LIT:200> ) <EOL> spec = self . makeSpec ( ) <EOL> for vm in spec . vms : <EOL> self . assertEqual ( vm . background_network_mbits_per_sec , <NUM_LIT:200> ) <EOL> self . assertEqual ( vm . background_network_ip_type , '<STR_LIT>' ) <EOL> self . _CheckVMFromSpec ( spec , working_groups = ( _GROUP_1 , _GROUP_2 ) ) <EOL> def testBackgroundWorkloadVanillaConfigFlagIpType ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mocked_flags [ '<STR_LIT>' ] . Parse ( <NUM_LIT:200> ) <EOL> self . mocked_flags [ '<STR_LIT>' ] . Parse ( '<STR_LIT>' ) <EOL> spec = self . makeSpec ( ) <EOL> for vm in spec . vms : <EOL> self . assertEqual ( vm . background_network_mbits_per_sec , <NUM_LIT:200> ) <EOL> self . assertEqual ( vm . background_network_ip_type , '<STR_LIT>' ) <EOL> self . _CheckVMFromSpec ( spec , working_groups = ( _GROUP_1 , _GROUP_2 ) ) <EOL> def testBackgroundWorkloadVanillaConfigBadIpTypeFlag ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mocked_flags [ '<STR_LIT>' ] . Parse ( <NUM_LIT:200> ) <EOL> self . mocked_flags [ '<STR_LIT>' ] . Parse ( '<STR_LIT>' ) <EOL> config = configs . LoadConfig ( ping_benchmark . BENCHMARK_CONFIG , { } , NAME ) <EOL> with self . assertRaises ( errors . Config . InvalidValue ) : <EOL> benchmark_config_spec . BenchmarkConfigSpec ( <EOL> NAME , flag_values = self . mocked_flags , ** config ) <EOL> def testBackgroundWorkloadConfig ( self ) : <EOL> """<STR_LIT>""" <EOL> spec = self . makeSpec ( <EOL> yaml_benchmark_config = CONFIG_WITH_BACKGROUND_NETWORK ) <EOL> for vm in spec . vm_groups [ _GROUP_1 ] : <EOL> self . assertIsNone ( vm . background_network_mbits_per_sec ) <EOL> for vm in spec . vm_groups [ _GROUP_2 ] : <EOL> self . assertEqual ( vm . background_network_mbits_per_sec , <NUM_LIT> ) <EOL> self . assertEqual ( vm . background_network_ip_type , '<STR_LIT>' ) <EOL> self . _CheckVMFromSpec ( spec , working_groups = [ _GROUP_2 ] , <EOL> non_working_groups = [ _GROUP_1 ] ) <EOL> def testBackgroundWorkloadConfigBadIp ( self ) : <EOL> """<STR_LIT>""" <EOL> config = configs . LoadConfig ( CONFIG_WITH_BACKGROUND_NETWORK_BAD_IPFLAG , <EOL> { } , NAME ) <EOL> with self . assertRaises ( errors . Config . InvalidValue ) : <EOL> benchmark_config_spec . BenchmarkConfigSpec ( <EOL> NAME , flag_values = self . mocked_flags , ** config ) <EOL> def testBackgroundWorkloadConfigGoodIp ( self ) : <EOL> """<STR_LIT>""" <EOL> spec = self . makeSpec ( <EOL> yaml_benchmark_config = CONFIG_WITH_BACKGROUND_NETWORK_IPFLAG ) <EOL> for vm in spec . vm_groups [ _GROUP_1 ] : <EOL> self . assertIsNone ( vm . background_network_mbits_per_sec ) <EOL> for vm in spec . vm_groups [ _GROUP_2 ] : <EOL> self . assertEqual ( vm . background_network_mbits_per_sec , <NUM_LIT> ) <EOL> self . assertEqual ( vm . background_network_ip_type , '<STR_LIT>' ) <EOL> self . _CheckVMFromSpec ( spec , working_groups = [ _GROUP_2 ] , <EOL> non_working_groups = [ _GROUP_1 ] ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
