<s> import os <EOL> import time <EOL> import shutil <EOL> import msgpack <EOL> import tempfile <EOL> import subprocess <EOL> from nose . tools import ( assert_equal , assert_false , assert_greater , <EOL> assert_raises ) <EOL> from ssbench . scenario import Scenario <EOL> from ssbench . run_results import RunResults <EOL> from ssbench . tests . test_scenario import ScenarioFixture <EOL> class TestRunResults ( ScenarioFixture ) : <EOL> def setUp ( self ) : <EOL> self . temp_dir = tempfile . mkdtemp ( ) <EOL> self . result_file_path = os . path . join ( self . temp_dir , '<STR_LIT>' ) <EOL> self . stub_scenario_file = os . path . join ( self . temp_dir , <EOL> '<STR_LIT>' ) <EOL> self . run_results = RunResults ( self . result_file_path ) <EOL> super ( TestRunResults , self ) . setUp ( ) <EOL> def tearDown ( self ) : <EOL> super ( TestRunResults , self ) . tearDown ( ) <EOL> shutil . rmtree ( self . temp_dir , ignore_errors = True ) <EOL> def _current_size ( self ) : <EOL> if not self . run_results . output_file . closed : <EOL> self . run_results . output_file . flush ( ) <EOL> return os . path . getsize ( self . result_file_path ) <EOL> def test_basic_instantiation ( self ) : <EOL> assert_false ( os . path . exists ( self . result_file_path ) ) <EOL> assert_equal ( <NUM_LIT> , self . run_results . write_threshold ) <EOL> with assert_raises ( TypeError ) : <EOL> RunResults ( ) <EOL> def test_start_run_requires_scenario ( self ) : <EOL> with assert_raises ( TypeError ) : <EOL> self . run_results . start_run ( ) <EOL> def test_start_run ( self ) : <EOL> assert_false ( os . path . exists ( self . result_file_path ) ) <EOL> self . run_results . start_run ( self . scenario ) <EOL> assert_equal ( len ( self . scenario . packb ( ) ) , self . _current_size ( ) ) <EOL> assert_greater ( self . _current_size ( ) , <NUM_LIT:0> ) <EOL> with open ( self . result_file_path , '<STR_LIT:rb>' ) as f : <EOL> unpacker = msgpack . Unpacker ( file_like = f ) <EOL> got_scenario = Scenario . unpackb ( unpacker ) <EOL> for attr in [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:version>' , <EOL> '<STR_LIT>' ] : <EOL> assert_equal ( getattr ( got_scenario , attr ) , <EOL> getattr ( self . scenario , attr ) ) <EOL> def test_process_raw_results ( self ) : <EOL> self . run_results . start_run ( self . scenario ) <EOL> start_size = self . _current_size ( ) <EOL> assert_greater ( start_size , <NUM_LIT:0> ) <EOL> self . run_results . write_threshold = <NUM_LIT> <EOL> res1 = msgpack . packb ( [ { '<STR_LIT>' : <NUM_LIT:1.0> } ] ) <EOL> res2 = msgpack . packb ( [ { '<STR_LIT>' : <NUM_LIT> } , { '<STR_LIT>' : <NUM_LIT> } ] ) <EOL> self . run_results . process_raw_results ( res1 ) <EOL> self . run_results . process_raw_results ( res2 ) <EOL> assert_equal ( self . _current_size ( ) - start_size , <NUM_LIT:0> ) <EOL> res3 = msgpack . packb ( [ { '<STR_LIT>' : '<STR_LIT:3>' * <NUM_LIT> } ] ) <EOL> assert_greater ( len ( res3 ) , <NUM_LIT> ) <EOL> self . run_results . process_raw_results ( res3 ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> assert_greater ( self . _current_size ( ) - start_size , <NUM_LIT> ) <EOL> def test_finalize ( self ) : <EOL> self . run_results . start_run ( self . scenario ) <EOL> start_size = self . _current_size ( ) <EOL> assert_greater ( start_size , <NUM_LIT:0> ) <EOL> self . run_results . write_threshold = <NUM_LIT> <EOL> res1 = msgpack . packb ( [ { '<STR_LIT>' : '<STR_LIT:3>' * <NUM_LIT> } ] ) <EOL> self . run_results . process_raw_results ( res1 ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> new_size = self . _current_size ( ) <EOL> assert_greater ( new_size - start_size , <NUM_LIT> ) <EOL> res2 = msgpack . packb ( [ { '<STR_LIT>' : <NUM_LIT> } , { '<STR_LIT>' : <NUM_LIT> } ] ) <EOL> res3 = msgpack . packb ( [ { '<STR_LIT>' : '<STR_LIT:3>' } ] ) <EOL> self . run_results . process_raw_results ( res2 ) <EOL> self . run_results . process_raw_results ( res3 ) <EOL> self . run_results . finalize ( ) <EOL> assert_equal ( len ( res2 ) + len ( res3 ) , <EOL> self . _current_size ( ) - new_size ) <EOL> def test_read_results ( self ) : <EOL> self . run_results . start_run ( self . scenario ) <EOL> res1 = msgpack . packb ( [ { '<STR_LIT>' : '<STR_LIT:3>' * <NUM_LIT> } ] ) <EOL> res2 = msgpack . packb ( [ { '<STR_LIT>' : <NUM_LIT> } , { '<STR_LIT>' : <NUM_LIT> } ] ) <EOL> res3 = msgpack . packb ( [ { '<STR_LIT>' : '<STR_LIT:3>' } ] ) <EOL> self . run_results . process_raw_results ( res1 ) <EOL> self . run_results . process_raw_results ( res2 ) <EOL> self . run_results . process_raw_results ( res3 ) <EOL> self . run_results . finalize ( ) <EOL> got_scenario , unpacker = RunResults ( <EOL> self . result_file_path ) . read_results ( ) <EOL> for attr in [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:version>' , '<STR_LIT>' ] : <EOL> assert_equal ( getattr ( got_scenario , attr ) , <EOL> getattr ( self . scenario , attr ) ) <EOL> assert_equal ( list ( unpacker ) , [ <EOL> [ { '<STR_LIT>' : '<STR_LIT:3>' * <NUM_LIT> } ] , <EOL> [ { '<STR_LIT>' : <NUM_LIT> } , { '<STR_LIT>' : <NUM_LIT> } ] , <EOL> [ { '<STR_LIT>' : '<STR_LIT:3>' } ] , <EOL> ] ) <EOL> subprocess . check_call ( [ '<STR_LIT>' , self . result_file_path ] ) <EOL> got_scenario , unpacker = RunResults ( <EOL> self . result_file_path + '<STR_LIT>' ) . read_results ( ) <EOL> for attr in [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:version>' , '<STR_LIT>' ] : <EOL> assert_equal ( getattr ( got_scenario , attr ) , <EOL> getattr ( self . scenario , attr ) ) <EOL> assert_equal ( list ( unpacker ) , [ <EOL> [ { '<STR_LIT>' : '<STR_LIT:3>' * <NUM_LIT> } ] , <EOL> [ { '<STR_LIT>' : <NUM_LIT> } , { '<STR_LIT>' : <NUM_LIT> } ] , <EOL> [ { '<STR_LIT>' : '<STR_LIT:3>' } ] , <EOL> ] ) </s>
