<s> import copy <EOL> import unittest <EOL> from os import path <EOL> from agora . stats import PBSVideoStats <EOL> from dateutil . parser import parse <EOL> from mrjob . protocol import JSONProtocol <EOL> HERE = path . abspath ( path . dirname ( __file__ ) ) <EOL> class PBSVideoStatsTestcase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ staticmethod <EOL> def events_generator ( ) : <EOL> """<STR_LIT>""" <EOL> protocol = JSONProtocol ( ) <EOL> input_filename = '<STR_LIT>' <EOL> with open ( path . join ( HERE , input_filename ) , '<STR_LIT:r>' ) as f : <EOL> line = f . readline ( ) <EOL> while line : <EOL> key , events = protocol . read ( line ) <EOL> yield key , events <EOL> line = f . readline ( ) <EOL> @ classmethod <EOL> def setup_class ( cls ) : <EOL> """<STR_LIT>""" <EOL> cls . events = { } <EOL> for key , event in cls . events_generator ( ) : <EOL> if key : <EOL> key_events = cls . events . setdefault ( key , [ ] ) <EOL> key_events . append ( event ) <EOL> cls . events [ key ] = key_events <EOL> def test_finished_playback ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in self . events . items ( ) : <EOL> stats = PBSVideoStats ( ) <EOL> finished_playback = False <EOL> for event in events : <EOL> stats . add_event ( event ) <EOL> if event . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> finished_playback = True <EOL> results = stats . summary ( ) <EOL> self . assertEqual ( <EOL> results . get ( '<STR_LIT>' ) , finished_playback ) <EOL> def test_latest_time ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in self . events . items ( ) : <EOL> stats = PBSVideoStats ( ) <EOL> latest_time = None <EOL> for event in events : <EOL> stats . add_event ( event ) <EOL> if event . get ( '<STR_LIT>' ) > latest_time : <EOL> latest_time = event . get ( '<STR_LIT>' ) <EOL> results = stats . summary ( ) <EOL> self . assertEqual ( results . get ( '<STR_LIT>' ) , latest_time ) <EOL> def test_latest_time_none ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in copy . deepcopy ( self . events . items ( ) ) : <EOL> stats = PBSVideoStats ( ) <EOL> for event in events : <EOL> if event . get ( '<STR_LIT>' ) : <EOL> event [ '<STR_LIT>' ] = None <EOL> stats . add_event ( event ) <EOL> results = stats . summary ( ) <EOL> self . assertEqual ( results . get ( '<STR_LIT>' ) , None ) <EOL> def test_earliest_time ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in self . events . items ( ) : <EOL> stats = PBSVideoStats ( ) <EOL> earliest_time = None <EOL> for event in events : <EOL> stats . add_event ( event ) <EOL> if event . get ( '<STR_LIT>' ) < earliest_time : <EOL> earliest_time = event . get ( '<STR_LIT>' ) <EOL> results = stats . summary ( ) <EOL> self . assertEqual ( results . get ( '<STR_LIT>' ) , earliest_time ) <EOL> def test_earliest_time_none ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in copy . deepcopy ( self . events . items ( ) ) : <EOL> stats = PBSVideoStats ( ) <EOL> earliest_time = None <EOL> for event in events : <EOL> if event . get ( '<STR_LIT>' ) : <EOL> event [ '<STR_LIT>' ] = None <EOL> stats . add_event ( event ) <EOL> results = stats . summary ( ) <EOL> self . assertEqual ( results . get ( '<STR_LIT>' ) , earliest_time ) <EOL> def test_add_duration_event ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in self . events . items ( ) : <EOL> stats = PBSVideoStats ( ) <EOL> for event in events : <EOL> stats . add_event ( event ) <EOL> results = stats . duration_events <EOL> for edata in results : <EOL> self . assertTrue ( edata . get ( '<STR_LIT>' ) in stats . MEDIA_EVENTS ) <EOL> self . assertTrue ( edata . get ( '<STR_LIT>' ) is not None ) <EOL> def test_sort_duration_events_in_order ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in self . events . items ( ) : <EOL> stats = PBSVideoStats ( ) <EOL> for event in events : <EOL> stats . add_event ( event ) <EOL> stats . _calculate_duration ( ) <EOL> results = stats . duration_events <EOL> for count , edata in enumerate ( results ) : <EOL> if count + <NUM_LIT:1> < len ( results ) : <EOL> self . assertTrue ( <EOL> edata . get ( '<STR_LIT>' ) <= results [ count + <NUM_LIT:1> ] . get ( '<STR_LIT>' ) ) <EOL> def test_sort_duration_events_reverse_order ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in self . events . items ( ) : <EOL> stats = PBSVideoStats ( ) <EOL> for event in reversed ( events ) : <EOL> stats . add_event ( event ) <EOL> stats . _calculate_duration ( ) <EOL> results = stats . duration_events <EOL> for count , edata in enumerate ( results ) : <EOL> if count + <NUM_LIT:1> < len ( results ) : <EOL> self . assertTrue ( <EOL> edata . get ( '<STR_LIT>' ) <= results [ count + <NUM_LIT:1> ] . get ( '<STR_LIT>' ) ) <EOL> def test_media_quality_change_auto ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in self . events . items ( ) : <EOL> stats = PBSVideoStats ( ) <EOL> auto_bitrate_events = <NUM_LIT:0> <EOL> for event in events : <EOL> stats . add_event ( event ) <EOL> if event . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> auto_bitrate_events += <NUM_LIT:1> <EOL> results = stats . summary ( ) <EOL> self . assertEqual ( <EOL> results . get ( '<STR_LIT>' ) , auto_bitrate_events ) <EOL> def test_media_quality_change_user ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , events in self . events . items ( ) : <EOL> stats = PBSVideoStats ( ) <EOL> user_bitrate_events = <NUM_LIT:0> <EOL> for event in events : <EOL> stats . add_event ( event ) <EOL> if event . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> user_bitrate_events += <NUM_LIT:1> <EOL> results = stats . summary ( ) <EOL> self . assertEqual ( <EOL> results . get ( '<STR_LIT>' ) , user_bitrate_events ) <EOL> def test_buffering_length ( self ) : <EOL> """<STR_LIT>""" <EOL> streams_with_buffer_length = <NUM_LIT:0> <EOL> enum = - <NUM_LIT:1> <EOL> for key , events in self . events . items ( ) : <EOL> enum += <NUM_LIT:1> <EOL> stats = PBSVideoStats ( ) <EOL> stream_buffer_length = None <EOL> open_start_buffer = False <EOL> is_valid_buffering = True <EOL> media_buffering_start_time = None <EOL> media_buffering_start_location = None <EOL> for event in events : <EOL> stats . add_event ( event ) <EOL> event_type = event . get ( '<STR_LIT>' ) <EOL> if not event_type : <EOL> is_valid_buffering = False <EOL> break <EOL> if event_type == '<STR_LIT>' : <EOL> if open_start_buffer : <EOL> is_valid_buffering = False <EOL> break <EOL> open_start_buffer = True <EOL> media_buffering_start_time = parse ( event . get ( '<STR_LIT>' ) ) <EOL> media_buffering_start_location = event . get ( '<STR_LIT>' ) <EOL> if event_type == '<STR_LIT>' : <EOL> if not open_start_buffer : <EOL> is_valid_buffering = False <EOL> break <EOL> open_start_buffer = False <EOL> if event . get ( '<STR_LIT>' ) != media_buffering_start_location : <EOL> is_valid_buffering = False <EOL> break <EOL> if not stream_buffer_length : <EOL> stream_buffer_length = <NUM_LIT:0> <EOL> media_buffering_end_time = parse ( event . get ( '<STR_LIT>' ) ) <EOL> buffer_delta = media_buffering_end_time - media_buffering_start_time <EOL> stream_buffer_length += self . _total_seconds ( buffer_delta ) <EOL> if is_valid_buffering : <EOL> results = stats . summary ( ) <EOL> results_buffering_length = results . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( results_buffering_length , stream_buffer_length ) <EOL> if stream_buffer_length is not None : <EOL> self . assertTrue ( <EOL> results_buffering_length >= <NUM_LIT:0> , <EOL> '<STR_LIT>' % results_buffering_length ) <EOL> streams_with_buffer_length += <NUM_LIT:1> <EOL> self . assertTrue ( streams_with_buffer_length > <NUM_LIT:0> ) <EOL> def _total_seconds ( self , timedelta ) : <EOL> """<STR_LIT>""" <EOL> delta_seconds = timedelta . days * <NUM_LIT> * <NUM_LIT> <EOL> delta_microseconds = ( timedelta . seconds + delta_seconds ) * <NUM_LIT:10> ** <NUM_LIT:6> <EOL> return ( timedelta . microseconds + delta_microseconds ) / <NUM_LIT:10> ** <NUM_LIT:6> </s>
