<s> """<STR_LIT>""" <EOL> import unittest <EOL> import pandas as pd <EOL> from perfkitbenchmarker import object_storage_multistream_analysis as analysis <EOL> from perfkitbenchmarker import units <EOL> SAMPLE_TABLE = pd . DataFrame ( [ <EOL> { '<STR_LIT>' : <NUM_LIT:0.0> , '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT:size>' : <NUM_LIT:1> } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT:size>' : <NUM_LIT:4> } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:1.0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT:size>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT:size>' : <NUM_LIT:8> } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT:size>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:size>' : <NUM_LIT:4> } ] ) <EOL> class TestInterval ( unittest . TestCase ) : <EOL> def testStartAndDuration ( self ) : <EOL> interval = analysis . Interval ( <NUM_LIT:2> , duration = <NUM_LIT:5> ) <EOL> self . assertEqual ( interval . end , <NUM_LIT:7> ) <EOL> def testStartAndEnd ( self ) : <EOL> interval = analysis . Interval ( <NUM_LIT:2> , end = <NUM_LIT:7> ) <EOL> self . assertEqual ( interval . duration , <NUM_LIT:5> ) <EOL> def testThreeArgsGood ( self ) : <EOL> analysis . Interval ( <NUM_LIT:2> , duration = <NUM_LIT:5> , end = <NUM_LIT:7> ) <EOL> def testThreeArgsBad ( self ) : <EOL> with self . assertRaises ( ValueError ) : <EOL> analysis . Interval ( <NUM_LIT:2> , duration = <NUM_LIT:5> , end = <NUM_LIT:8> ) <EOL> def testEndBeforeStart ( self ) : <EOL> with self . assertRaises ( ValueError ) : <EOL> analysis . Interval ( <NUM_LIT:2> , end = <NUM_LIT:1> ) <EOL> class TestGetStreamActiveIntervals ( unittest . TestCase ) : <EOL> def testGetStreamActiveIntervals ( self ) : <EOL> any_stream , all_streams = analysis . GetStreamActiveIntervals ( <EOL> SAMPLE_TABLE [ '<STR_LIT>' ] , <EOL> SAMPLE_TABLE [ '<STR_LIT>' ] , <EOL> SAMPLE_TABLE [ '<STR_LIT>' ] ) <EOL> print "<STR_LIT>" , any_stream , "<STR_LIT>" , all_streams <EOL> self . assertEqual ( any_stream , analysis . Interval ( <NUM_LIT:0> , end = <NUM_LIT> ) ) <EOL> self . assertEqual ( all_streams , analysis . Interval ( <NUM_LIT> , end = <NUM_LIT> ) ) <EOL> class TestStreamStartAndEndGaps ( unittest . TestCase ) : <EOL> def testStreamStartAndEndGaps ( self ) : <EOL> start_gap , stop_gap = analysis . StreamStartAndEndGaps ( <EOL> SAMPLE_TABLE [ '<STR_LIT>' ] , SAMPLE_TABLE [ '<STR_LIT>' ] , <EOL> analysis . Interval ( <NUM_LIT> , duration = <NUM_LIT> ) ) <EOL> self . assertEqual ( start_gap , <NUM_LIT> ) <EOL> self . assertEqual ( stop_gap , <NUM_LIT> ) <EOL> class TestFullyInInterval ( unittest . TestCase ) : <EOL> def testFullyInInterval ( self ) : <EOL> overlaps = analysis . FullyInInterval ( <EOL> SAMPLE_TABLE [ '<STR_LIT>' ] , <EOL> SAMPLE_TABLE [ '<STR_LIT>' ] , <EOL> analysis . Interval ( <NUM_LIT> , duration = <NUM_LIT> ) ) <EOL> self . assertTrue ( <EOL> ( overlaps == pd . Series ( [ False , False , True , False , False , True ] ) ) . all ( ) ) <EOL> class TestThroughputStats ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . byte = units . byte <EOL> self . sec = units . second <EOL> self . percent = units . percent <EOL> def doTest ( self , data , num_streams , correct_answer ) : <EOL> output = analysis . ThroughputStats ( <EOL> data [ '<STR_LIT>' ] , data [ '<STR_LIT>' ] , data [ '<STR_LIT:size>' ] , data [ '<STR_LIT>' ] , <EOL> num_streams ) <EOL> print '<STR_LIT>' , output <EOL> for name , value in output . iteritems ( ) : <EOL> if name not in correct_answer : <EOL> raise KeyError ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( name , str ( correct_answer ) ) ) <EOL> self . assertEqual ( value , correct_answer [ name ] ) <EOL> def testOneObject ( self ) : <EOL> one_op = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> ] , <EOL> '<STR_LIT:size>' : [ <NUM_LIT:2> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> ] } ) <EOL> self . doTest ( one_op , <NUM_LIT:1> , <EOL> { '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec , <EOL> '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec } ) <EOL> def testSecondObjectSameSpeed ( self ) : <EOL> no_gap = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:1> ] , <EOL> '<STR_LIT:size>' : [ <NUM_LIT:2> , <NUM_LIT:2> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:0> ] } ) <EOL> self . doTest ( no_gap , <NUM_LIT:1> , <EOL> { '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec , <EOL> '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec } ) <EOL> def testSecondObjectDifferentSpeed ( self ) : <EOL> different_speeds = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:3> ] , <EOL> '<STR_LIT:size>' : [ <NUM_LIT:2> , <NUM_LIT:8> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:0> ] } ) <EOL> self . doTest ( different_speeds , <NUM_LIT:1> , <EOL> { '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec , <EOL> '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec } ) <EOL> def testGapBetweenObjects ( self ) : <EOL> with_gap = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:3> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:1> ] , <EOL> '<STR_LIT:size>' : [ <NUM_LIT:2> , <NUM_LIT:2> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:0> ] } ) <EOL> self . doTest ( with_gap , <NUM_LIT:1> , <EOL> { '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec , <EOL> '<STR_LIT>' : <NUM_LIT:1.0> * self . byte / self . sec } ) <EOL> def testSimultaneousObjects ( self ) : <EOL> two_streams = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:1> ] , <EOL> '<STR_LIT:size>' : [ <NUM_LIT:2> , <NUM_LIT:2> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:1> ] } ) <EOL> self . doTest ( two_streams , <NUM_LIT:2> , <EOL> { '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec , <EOL> '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec } ) <EOL> def testTwoStreamGaps ( self ) : <EOL> two_streams_with_gap = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:3> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] , <EOL> '<STR_LIT:size>' : [ <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> ] } ) <EOL> self . doTest ( two_streams_with_gap , <NUM_LIT:2> , <EOL> { '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec , <EOL> '<STR_LIT>' : <NUM_LIT> * self . byte / self . sec } ) <EOL> class TestGapStats ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . sec = units . second <EOL> self . percent = units . percent <EOL> def doTest ( self , data , num_streams , interval , correct_answer ) : <EOL> output = analysis . GapStats ( <EOL> data [ '<STR_LIT>' ] , data [ '<STR_LIT>' ] , data [ '<STR_LIT>' ] , <EOL> interval , num_streams ) <EOL> for name , value in output . iteritems ( ) : <EOL> if name not in correct_answer : <EOL> raise KeyError ( '<STR_LIT>' % <EOL> ( name , str ( correct_answer ) ) ) <EOL> self . assertEqual ( value , correct_answer [ name ] ) <EOL> def testOneRecord ( self ) : <EOL> one_record = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> ] } ) <EOL> self . doTest ( one_record , <NUM_LIT:1> , analysis . Interval ( <NUM_LIT:0> , duration = <NUM_LIT:1> ) , <EOL> { '<STR_LIT>' : <NUM_LIT:0.0> * self . sec , <EOL> '<STR_LIT>' : <NUM_LIT:0.0> * self . percent } ) <EOL> def testOneStream ( self ) : <EOL> one_stream = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:4> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] } ) <EOL> self . doTest ( one_stream , <NUM_LIT:1> , analysis . Interval ( <NUM_LIT:0> , duration = <NUM_LIT:5> ) , <EOL> { '<STR_LIT>' : <NUM_LIT> * self . sec , <EOL> '<STR_LIT>' : <NUM_LIT> * self . percent } ) <EOL> def testOverlapInterval ( self ) : <EOL> overlap = pd . DataFrame ( { '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:5> , <NUM_LIT:10> , <NUM_LIT> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] } ) <EOL> self . doTest ( overlap , <NUM_LIT:1> , analysis . Interval ( <NUM_LIT:3> , duration = <NUM_LIT:8> ) , <EOL> { '<STR_LIT>' : <NUM_LIT> * self . sec , <EOL> '<STR_LIT>' : <NUM_LIT> * self . percent } ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
