<s> import testtools <EOL> from shaker . engine . executors import base <EOL> from shaker . engine . executors import iperf <EOL> IP = '<STR_LIT>' <EOL> AGENT = { '<STR_LIT>' : { '<STR_LIT>' : IP } } <EOL> class TestIperfGraphExecutor ( testtools . TestCase ) : <EOL> def test_get_command ( self ) : <EOL> executor = iperf . IperfGraphExecutor ( { } , AGENT ) <EOL> expected = { '<STR_LIT:data>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % IP , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } <EOL> self . assertEqual ( expected , executor . get_command ( ) ) <EOL> def test_get_command_udp ( self ) : <EOL> executor = iperf . IperfGraphExecutor ( <EOL> { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:time>' : <NUM_LIT:30> , <EOL> '<STR_LIT>' : <NUM_LIT> } , AGENT ) <EOL> expected = { '<STR_LIT:data>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % IP , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } <EOL> self . assertEqual ( expected , executor . get_command ( ) ) <EOL> def test_get_command_static_host ( self ) : <EOL> executor = iperf . IperfGraphExecutor ( { '<STR_LIT:host>' : '<STR_LIT>' } , { } ) <EOL> expected = { '<STR_LIT:data>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } <EOL> self . assertEqual ( expected , executor . get_command ( ) ) <EOL> def test_process_reply ( self ) : <EOL> executor = iperf . IperfGraphExecutor ( { } , AGENT ) <EOL> message = { <EOL> '<STR_LIT>' : """<STR_LIT>""" <EOL> } <EOL> expected = { <EOL> '<STR_LIT>' : [ <EOL> [ <NUM_LIT:1.0> , <NUM_LIT> ] , <EOL> [ <NUM_LIT> , <NUM_LIT> ] , <EOL> [ <NUM_LIT> , <NUM_LIT> ] , <EOL> ] , <EOL> '<STR_LIT>' : [ <EOL> [ '<STR_LIT:time>' , '<STR_LIT:s>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ] <EOL> } <EOL> reply = executor . process_reply ( message ) <EOL> self . assertEqual ( expected [ '<STR_LIT>' ] , reply [ '<STR_LIT>' ] , <EOL> message = '<STR_LIT>' ) <EOL> self . assertEqual ( expected [ '<STR_LIT>' ] , reply [ '<STR_LIT>' ] , <EOL> message = '<STR_LIT>' ) <EOL> def test_process_reply_multiple_threads ( self ) : <EOL> executor = iperf . IperfGraphExecutor ( { '<STR_LIT>' : <NUM_LIT:2> } , AGENT ) <EOL> message = { <EOL> '<STR_LIT>' : """<STR_LIT>""" <EOL> } <EOL> expected = { <EOL> '<STR_LIT>' : [ <EOL> [ <NUM_LIT:1.0> , <NUM_LIT> ] , <EOL> [ <NUM_LIT> , <NUM_LIT> ] , <EOL> ] , <EOL> '<STR_LIT>' : [ <EOL> [ '<STR_LIT:time>' , '<STR_LIT:s>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ] <EOL> } <EOL> reply = executor . process_reply ( message ) <EOL> self . assertEqual ( expected [ '<STR_LIT>' ] , reply [ '<STR_LIT>' ] , <EOL> message = '<STR_LIT>' ) <EOL> self . assertEqual ( expected [ '<STR_LIT>' ] , reply [ '<STR_LIT>' ] , <EOL> message = '<STR_LIT>' ) <EOL> def test_process_empty_reply ( self ) : <EOL> executor = iperf . IperfGraphExecutor ( { } , AGENT ) <EOL> message = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . assertRaises ( <EOL> base . ExecutorException , executor . process_reply , message ) <EOL> class TestIperf3Executor ( testtools . TestCase ) : <EOL> def test_get_command ( self ) : <EOL> executor = iperf . Iperf3Executor ( { } , AGENT ) <EOL> expected = { '<STR_LIT:data>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % IP , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } <EOL> self . assertEqual ( expected , executor . get_command ( ) ) <EOL> def test_get_command_udp ( self ) : <EOL> executor = iperf . Iperf3Executor ( <EOL> { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:time>' : <NUM_LIT:30> , <EOL> '<STR_LIT>' : <NUM_LIT> } , AGENT ) <EOL> expected = { '<STR_LIT:data>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % IP , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } <EOL> self . assertEqual ( expected , executor . get_command ( ) ) <EOL> def test_get_command_udp_bandwidth_0 ( self ) : <EOL> executor = iperf . Iperf3Executor ( { '<STR_LIT>' : True , '<STR_LIT>' : <NUM_LIT:0> } , AGENT ) <EOL> expected = { '<STR_LIT:data>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % IP , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } <EOL> self . assertEqual ( expected , executor . get_command ( ) ) <EOL> def test_get_command_reverse ( self ) : <EOL> executor = iperf . Iperf3Executor ( { '<STR_LIT>' : '<STR_LIT:yes>' } , AGENT ) <EOL> expected = { '<STR_LIT:data>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % IP , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } <EOL> self . assertEqual ( expected , executor . get_command ( ) ) <EOL> def test_process_reply_tcp ( self ) : <EOL> executor = iperf . Iperf3Executor ( { } , AGENT ) <EOL> message = { <EOL> '<STR_LIT>' : """<STR_LIT>""" <EOL> } <EOL> expected = { <EOL> '<STR_LIT>' : [ <EOL> [ <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT:0> ] , <EOL> [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> ] , <EOL> [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT:0> ] , <EOL> ] , <EOL> '<STR_LIT>' : [ <EOL> [ '<STR_LIT:time>' , '<STR_LIT:s>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ] <EOL> } <EOL> reply = executor . process_reply ( message ) <EOL> self . assertEqual ( expected [ '<STR_LIT>' ] , reply [ '<STR_LIT>' ] , <EOL> message = '<STR_LIT>' ) <EOL> self . assertEqual ( expected [ '<STR_LIT>' ] , reply [ '<STR_LIT>' ] , <EOL> message = '<STR_LIT>' ) <EOL> def test_process_reply_udp ( self ) : <EOL> executor = iperf . Iperf3Executor ( { '<STR_LIT>' : True } , AGENT ) <EOL> message = { <EOL> '<STR_LIT>' : """<STR_LIT>""" <EOL> } <EOL> expected = { <EOL> '<STR_LIT>' : [ <EOL> [ <NUM_LIT:1.0> , <NUM_LIT> ] , <EOL> [ <NUM_LIT> , <NUM_LIT> ] , <EOL> [ <NUM_LIT> , <NUM_LIT> ] , <EOL> ] , <EOL> '<STR_LIT>' : [ <EOL> [ '<STR_LIT:time>' , '<STR_LIT:s>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ] <EOL> } <EOL> reply = executor . process_reply ( message ) <EOL> self . assertEqual ( expected [ '<STR_LIT>' ] , reply [ '<STR_LIT>' ] , <EOL> message = '<STR_LIT>' ) <EOL> self . assertEqual ( expected [ '<STR_LIT>' ] , reply [ '<STR_LIT>' ] , <EOL> message = '<STR_LIT>' ) </s>
