<s> """<STR_LIT>""" <EOL> import httplib2 <EOL> import json <EOL> import logging <EOL> import numbers <EOL> import oauth2client . client <EOL> import threading <EOL> import zlib <EOL> from openhtf . io . output import json_factory <EOL> from openhtf . io . proto import testrun_pb2 <EOL> from openhtf . io . proto import units_pb2 <EOL> from openhtf . io import test_record <EOL> from openhtf . util import data <EOL> from openhtf . util import measurements <EOL> from openhtf . util import validators <EOL> MIMETYPE_MAP = { <EOL> '<STR_LIT>' : testrun_pb2 . InformationParameter . JPG , <EOL> '<STR_LIT>' : testrun_pb2 . InformationParameter . PNG , <EOL> '<STR_LIT>' : testrun_pb2 . InformationParameter . WAV , <EOL> '<STR_LIT>' : testrun_pb2 . InformationParameter . TEXT_UTF8 , <EOL> '<STR_LIT>' : testrun_pb2 . InformationParameter . TIFF , <EOL> '<STR_LIT>' : testrun_pb2 . InformationParameter . MP4 , <EOL> } <EOL> OUTCOME_MAP = { <EOL> test_record . Outcome . ERROR : testrun_pb2 . ERROR , <EOL> test_record . Outcome . FAIL : testrun_pb2 . FAIL , <EOL> test_record . Outcome . PASS : testrun_pb2 . PASS , <EOL> test_record . Outcome . TIMEOUT : testrun_pb2 . ERROR , <EOL> } <EOL> UOM_CODE_MAP = { <EOL> u . GetOptions ( ) . Extensions [ units_pb2 . uom_code ] : num <EOL> for num , u in units_pb2 . Units . UnitCode . DESCRIPTOR . values_by_number . iteritems ( ) <EOL> } <EOL> MAX_PARAMS_PER_MEASUREMENT = <NUM_LIT:100> <EOL> class UploadFailedError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def _PopulateHeader ( record , testrun ) : <EOL> """<STR_LIT>""" <EOL> testrun . dut_serial = record . dut_id <EOL> testrun . tester_name = record . station_id <EOL> if '<STR_LIT>' in record . metadata : <EOL> testrun . test_info . name = record . metadata [ '<STR_LIT>' ] <EOL> else : <EOL> testrun . test_info . name = record . station_id <EOL> if '<STR_LIT>' in record . metadata : <EOL> testrun . test_info . description = record . metadata [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in record . metadata : <EOL> testrun . test_info . version_string = record . metadata [ '<STR_LIT>' ] <EOL> testrun . test_status = OUTCOME_MAP [ record . outcome ] <EOL> testrun . start_time_millis = record . start_time_millis <EOL> testrun . end_time_millis = record . end_time_millis <EOL> if '<STR_LIT>' in record . metadata : <EOL> testrun . run_name = record . metadata [ '<STR_LIT>' ] <EOL> for details in record . outcome_details : <EOL> testrun_code = testrun . failure_codes . add ( ) <EOL> testrun_code . code = details . code <EOL> testrun_code . details = details . description <EOL> def _AttachJson ( record , testrun ) : <EOL> """<STR_LIT>""" <EOL> record_dict = data . ConvertToBaseTypes ( record , ignore_keys = ( '<STR_LIT>' , ) ) <EOL> record_json = json_factory . OutputToJSON ( sort_keys = True ) . encode ( record_dict ) <EOL> testrun_param = testrun . info_parameters . add ( ) <EOL> testrun_param . name = '<STR_LIT>' <EOL> testrun_param . value_binary = record_json <EOL> testrun_param . type = testrun_pb2 . InformationParameter . TEXT_UTF8 <EOL> def _ExtractAttachments ( phase , testrun , used_parameter_names ) : <EOL> """<STR_LIT>""" <EOL> for name , ( data , mimetype ) in sorted ( phase . attachments . items ( ) ) : <EOL> while name in used_parameter_names : <EOL> name += '<STR_LIT:_>' <EOL> used_parameter_names . add ( name ) <EOL> testrun_param = testrun . info_parameters . add ( ) <EOL> testrun_param . name = name <EOL> testrun_param . value_binary = data <EOL> if mimetype in MIMETYPE_MAP : <EOL> testrun_param . type = MIMETYPE_MAP [ mimetype ] <EOL> else : <EOL> testrun_param . type = testrun_pb2 . InformationParameter . BINARY <EOL> def _MangleMeasurement ( name , value , measurement , mangled_parameters ) : <EOL> """<STR_LIT>""" <EOL> for current_value in value [ : MAX_PARAMS_PER_MEASUREMENT ] : <EOL> mangled_name = '<STR_LIT:_>' . join ( [ name ] + [ <EOL> '<STR_LIT>' % ( <EOL> dim_val , <EOL> dim_units . uom_suffix if dim_units . uom_suffix else '<STR_LIT>' ) for <EOL> dim_val , dim_units in zip ( <EOL> current_value [ : - <NUM_LIT:1> ] , measurement . dimensions ) ] ) <EOL> while mangled_name in mangled_parameters : <EOL> logging . warning ( '<STR_LIT>' , mangled_name ) <EOL> mangled_name += '<STR_LIT:_>' <EOL> mangled_param = testrun_pb2 . TestParameter ( ) <EOL> mangled_param . name = mangled_name <EOL> mangled_param . numeric_value = float ( current_value [ - <NUM_LIT:1> ] ) <EOL> if measurement . units : <EOL> mangled_param . unit_code = UOM_CODE_MAP [ measurement . units . uom_code ] <EOL> mangled_param . description = ( <EOL> '<STR_LIT>' % ( <EOL> name , tuple ( d . uom_suffix for d in measurement . dimensions ) ) ) <EOL> for validator in measurement . validators : <EOL> mangled_param . description += '<STR_LIT>' + str ( validator ) <EOL> mangled_parameters [ mangled_name ] = mangled_param <EOL> def _ExtractParameters ( record , testrun , used_parameter_names ) : <EOL> """<STR_LIT>""" <EOL> mangled_parameters = { } <EOL> for phase in record . phases : <EOL> testrun_phase = testrun . phases . add ( ) <EOL> testrun_phase . name = phase . name <EOL> testrun_phase . description = phase . codeinfo . sourcecode <EOL> testrun_phase . timing . start_time_millis = phase . start_time_millis <EOL> testrun_phase . timing . end_time_millis = phase . end_time_millis <EOL> _ExtractAttachments ( phase , testrun , used_parameter_names ) <EOL> for name , measurement in sorted ( phase . measurements . items ( ) ) : <EOL> tr_name = name <EOL> while tr_name in used_parameter_names : <EOL> tr_name += '<STR_LIT:_>' <EOL> used_parameter_names . add ( tr_name ) <EOL> testrun_param = testrun . test_parameters . add ( ) <EOL> testrun_param . name = tr_name <EOL> if measurement . outcome == measurements . Outcome . PASS : <EOL> testrun_param . status = testrun_pb2 . PASS <EOL> else : <EOL> testrun_param . status = testrun_pb2 . FAIL <EOL> if measurement . docstring : <EOL> testrun_param . description = measurement . docstring <EOL> if measurement . units : <EOL> testrun_param . unit_code = UOM_CODE_MAP [ measurement . units . uom_code ] <EOL> if name not in phase . measured_values : <EOL> testrun_param . status = testrun_pb2 . ERROR <EOL> continue <EOL> value = phase . measured_values [ name ] <EOL> if measurement . dimensions is None : <EOL> if isinstance ( value , numbers . Number ) : <EOL> testrun_param . numeric_value = float ( value ) <EOL> else : <EOL> testrun_param . text_value = str ( value ) <EOL> for validator in measurement . validators : <EOL> if isinstance ( validator , validators . InRange ) : <EOL> if validator . minimum is not None : <EOL> testrun_param . numeric_minimum = float ( validator . minimum ) <EOL> if validator . maximum is not None : <EOL> testrun_param . numeric_maximum = float ( validator . maximum ) <EOL> elif isinstance ( validator , validators . RegexMatcher ) : <EOL> testrun_param . expected_text = validator . regex <EOL> else : <EOL> testrun_param . description += '<STR_LIT>' + str ( validator ) <EOL> else : <EOL> _MangleMeasurement ( name , value , measurement , mangled_parameters ) <EOL> if testrun_param . status == testrun_pb2 . FAIL : <EOL> testrun_code = testrun . failure_codes . add ( ) <EOL> testrun_code . code = testrun_param . name <EOL> if measurement . dimensions is None : <EOL> if testrun_param . numeric_value : <EOL> testrun_code . details = str ( testrun_param . numeric_value ) <EOL> else : <EOL> testrun_code . details = testrun_param . text_value <EOL> return mangled_parameters <EOL> def _AddMangledParameters ( testrun , mangled_parameters , used_parameter_names ) : <EOL> """<STR_LIT>""" <EOL> for mangled_name , mangled_param in sorted ( mangled_parameters . items ( ) ) : <EOL> while mangled_name in used_parameter_names : <EOL> logging . warning ( '<STR_LIT>' , <EOL> mangled_name ) <EOL> mangled_name += '<STR_LIT:_>' <EOL> testrun_param = testrun . test_parameters . add ( ) <EOL> testrun_param . CopyFrom ( mangled_param ) <EOL> def _AddLogLines ( record , testrun ) : <EOL> """<STR_LIT>""" <EOL> for log in record . log_records : <EOL> testrun_log = testrun . test_logs . add ( ) <EOL> testrun_log . timestamp_millis = log . timestamp_millis <EOL> testrun_log . log_message = log . message <EOL> testrun_log . logger_name = log . logger_name <EOL> testrun_log . levelno = log . level <EOL> if log . level <= logging . DEBUG : <EOL> testrun_log . level = testrun_pb2 . TestRunLogMessage . DEBUG <EOL> elif log . level <= logging . INFO : <EOL> testrun_log . level = testrun_pb2 . TestRunLogMessage . INFO <EOL> elif log . level <= logging . WARNING : <EOL> testrun_log . level = testrun_pb2 . TestRunLogMessage . WARNING <EOL> elif log . level <= logging . ERROR : <EOL> testrun_log . level = testrun_pb2 . TestRunLogMessage . ERROR <EOL> elif log . level <= logging . CRITICAL : <EOL> testrun_log . level = testrun_pb2 . TestRunLogMessage . CRITICAL <EOL> testrun_log . log_source = log . source <EOL> testrun_log . lineno = log . lineno <EOL> def _TestRunFromTestRecord ( record ) : <EOL> """<STR_LIT>""" <EOL> testrun = testrun_pb2 . TestRun ( ) <EOL> _PopulateHeader ( record , testrun ) <EOL> _AttachJson ( record , testrun ) <EOL> used_parameter_names = set ( '<STR_LIT>' ) <EOL> mangled_parameters = _ExtractParameters ( record , testrun , used_parameter_names ) <EOL> _AddMangledParameters ( testrun , mangled_parameters , used_parameter_names ) <EOL> _AddLogLines ( record , testrun ) <EOL> return testrun <EOL> class OutputToTestRunProto ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , filename_pattern ) : <EOL> self . filename_pattern = filename_pattern <EOL> def __call__ ( self , test_record ) : <EOL> as_dict = data . ConvertToBaseTypes ( test_record ) <EOL> serialized = _TestRunFromTestRecord ( test_record ) . SerializeToString ( ) <EOL> if isinstance ( self . filename_pattern , basestring ) : <EOL> with open ( self . filename_pattern % as_dict , '<STR_LIT:w>' ) as outfile : <EOL> outfile . write ( serialized ) <EOL> else : <EOL> self . filename_pattern . write ( serialized ) <EOL> class UploadToMfgInspector ( object ) : <EOL> """<STR_LIT>""" <EOL> TOKEN_URI = '<STR_LIT>' <EOL> SCOPE_CODE_URI = '<STR_LIT>' <EOL> DESTINATION_URL = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> class _MemStorage ( oauth2client . client . Storage ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _lock = threading . Lock ( ) <EOL> self . _credentials = None <EOL> def acquire_lock ( self ) : <EOL> self . _lock . acquire ( True ) <EOL> def release_lock ( self ) : <EOL> self . _lock . release ( ) <EOL> def locked_get ( self ) : <EOL> return self . _credentials <EOL> def locked_put ( self , credentials ) : <EOL> self . _credentials = credentials <EOL> def __init__ ( self , user , keydata , token_uri = TOKEN_URI ) : <EOL> self . user = user <EOL> self . keydata = keydata <EOL> self . token_uri = token_uri <EOL> @ classmethod <EOL> def from_json ( cls , json_data ) : <EOL> """<STR_LIT>""" <EOL> return cls ( user = json_data [ '<STR_LIT>' ] , <EOL> keydata = json_data [ '<STR_LIT>' ] , <EOL> token_uri = json_data [ '<STR_LIT>' ] ) <EOL> @ staticmethod <EOL> def UploadTestRun ( testrun , destination , credentials = None ) : <EOL> """<STR_LIT>""" <EOL> http = httplib2 . Http ( ) <EOL> if credentials : <EOL> if credentials . access_token_expired : <EOL> credentials . refresh ( http ) <EOL> credentials . authorize ( http ) <EOL> test_run_envelope = testrun_pb2 . TestRunEnvelope ( ) <EOL> test_run_envelope . payload = zlib . compress ( testrun . SerializeToString ( ) ) <EOL> test_run_envelope . payload_type = testrun_pb2 . COMPRESSED_TEST_RUN <EOL> serialized_envelope = test_run_envelope . SerializeToString ( ) <EOL> _ , content = http . request ( destination , '<STR_LIT:POST>' , serialized_envelope ) <EOL> if content . split ( '<STR_LIT:\n>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] != '<STR_LIT:OK>' : <EOL> results = json . loads ( content ) <EOL> raise UploadFailedError ( results [ '<STR_LIT:error>' ] , results ) <EOL> def __call__ ( self , test_record ) : <EOL> credentials = oauth2client . client . SignedJwtAssertionCredentials ( <EOL> service_account_name = self . user , <EOL> private_key = self . keydata , <EOL> scope = self . SCOPE_CODE_URI , <EOL> user_agent = '<STR_LIT>' , <EOL> token_uri = self . token_uri ) <EOL> credentials . set_store ( self . _MemStorage ( ) ) <EOL> testrun = _TestRunFromTestRecord ( test_record ) <EOL> try : <EOL> self . UploadTestRun ( testrun , self . DESTINATION_URL , credentials ) <EOL> except UploadFailedError : <EOL> logging . exception ( '<STR_LIT>' ) </s>
