<s> from __future__ import with_statement <EOL> from datetime import datetime <EOL> from hashlib import sha1 <EOL> from robot . api import ExecutionResult <EOL> from sqlite3 import IntegrityError <EOL> from dbbot import Logger <EOL> class RobotResultsParser ( object ) : <EOL> def __init__ ( self , include_keywords , db , verbose_stream ) : <EOL> self . _verbose = Logger ( '<STR_LIT>' , verbose_stream ) <EOL> self . _include_keywords = include_keywords <EOL> self . _db = db <EOL> def xml_to_db ( self , xml_file ) : <EOL> self . _verbose ( '<STR_LIT>' % xml_file ) <EOL> test_run = ExecutionResult ( xml_file , include_keywords = self . _include_keywords ) <EOL> hash = self . _hash ( xml_file ) <EOL> try : <EOL> test_run_id = self . _db . insert ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : hash , <EOL> '<STR_LIT>' : datetime . utcnow ( ) . strftime ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : test_run . source , <EOL> '<STR_LIT>' : self . _format_robot_timestamp ( test_run . suite . starttime ) , <EOL> '<STR_LIT>' : self . _format_robot_timestamp ( test_run . suite . endtime ) <EOL> } ) <EOL> except IntegrityError : <EOL> test_run_id = self . _db . fetch_id ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : test_run . source , <EOL> '<STR_LIT>' : self . _format_robot_timestamp ( test_run . suite . starttime ) , <EOL> '<STR_LIT>' : self . _format_robot_timestamp ( test_run . suite . endtime ) <EOL> } ) <EOL> self . _parse_errors ( test_run . errors . messages , test_run_id ) <EOL> self . _parse_statistics ( test_run . statistics , test_run_id ) <EOL> self . _parse_suite ( test_run . suite , test_run_id ) <EOL> def _hash ( self , xml_file ) : <EOL> block_size = <NUM_LIT> <EOL> hasher = sha1 ( ) <EOL> with open ( xml_file , '<STR_LIT:rb>' ) as f : <EOL> chunk = f . read ( block_size ) <EOL> while len ( chunk ) > <NUM_LIT:0> : <EOL> hasher . update ( chunk ) <EOL> chunk = f . read ( block_size ) <EOL> return hasher . hexdigest ( ) <EOL> def _parse_errors ( self , errors , test_run_id ) : <EOL> self . _db . insert_many_or_ignore ( '<STR_LIT>' , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:content>' ) , <EOL> [ ( test_run_id , error . level , self . _format_robot_timestamp ( error . timestamp ) , error . message ) <EOL> for error in errors ] <EOL> ) <EOL> def _parse_statistics ( self , statistics , test_run_id ) : <EOL> self . _parse_test_run_statistics ( statistics . total , test_run_id ) <EOL> self . _parse_tag_statistics ( statistics . tags , test_run_id ) <EOL> def _parse_test_run_statistics ( self , test_run_statistics , test_run_id ) : <EOL> self . _verbose ( '<STR_LIT>' ) <EOL> [ self . _parse_test_run_stats ( stat , test_run_id ) for stat in test_run_statistics ] <EOL> def _parse_tag_statistics ( self , tag_statistics , test_run_id ) : <EOL> self . _verbose ( '<STR_LIT>' ) <EOL> [ self . _parse_tag_stats ( stat , test_run_id ) for stat in tag_statistics . tags . values ( ) ] <EOL> def _parse_tag_stats ( self , stat , test_run_id ) : <EOL> self . _db . insert_or_ignore ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : test_run_id , <EOL> '<STR_LIT:name>' : stat . name , <EOL> '<STR_LIT>' : stat . critical , <EOL> '<STR_LIT>' : getattr ( stat , '<STR_LIT>' , None ) , <EOL> '<STR_LIT>' : stat . failed , <EOL> '<STR_LIT>' : stat . passed <EOL> } ) <EOL> def _parse_test_run_stats ( self , stat , test_run_id ) : <EOL> self . _db . insert_or_ignore ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : test_run_id , <EOL> '<STR_LIT:name>' : stat . name , <EOL> '<STR_LIT>' : getattr ( stat , '<STR_LIT>' , None ) , <EOL> '<STR_LIT>' : stat . failed , <EOL> '<STR_LIT>' : stat . passed <EOL> } ) <EOL> def _parse_suite ( self , suite , test_run_id , parent_suite_id = None ) : <EOL> self . _verbose ( '<STR_LIT>' % suite . name ) <EOL> try : <EOL> suite_id = self . _db . insert ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : parent_suite_id , <EOL> '<STR_LIT>' : suite . id , <EOL> '<STR_LIT:name>' : suite . name , <EOL> '<STR_LIT:source>' : suite . source , <EOL> '<STR_LIT>' : suite . doc <EOL> } ) <EOL> except IntegrityError : <EOL> suite_id = self . _db . fetch_id ( '<STR_LIT>' , { <EOL> '<STR_LIT:name>' : suite . name , <EOL> '<STR_LIT:source>' : suite . source <EOL> } ) <EOL> self . _parse_suite_status ( test_run_id , suite_id , suite ) <EOL> self . _parse_suites ( suite , test_run_id , suite_id ) <EOL> self . _parse_tests ( suite . tests , test_run_id , suite_id ) <EOL> self . _parse_keywords ( suite . keywords , test_run_id , suite_id , None ) <EOL> def _parse_suite_status ( self , test_run_id , suite_id , suite ) : <EOL> self . _db . insert_or_ignore ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : test_run_id , <EOL> '<STR_LIT>' : suite_id , <EOL> '<STR_LIT>' : suite . statistics . all . passed , <EOL> '<STR_LIT>' : suite . statistics . all . failed , <EOL> '<STR_LIT>' : suite . elapsedtime , <EOL> '<STR_LIT:status>' : suite . status <EOL> } ) <EOL> def _parse_suites ( self , suite , test_run_id , parent_suite_id ) : <EOL> [ self . _parse_suite ( subsuite , test_run_id , parent_suite_id ) for subsuite in suite . suites ] <EOL> def _parse_tests ( self , tests , test_run_id , suite_id ) : <EOL> [ self . _parse_test ( test , test_run_id , suite_id ) for test in tests ] <EOL> def _parse_test ( self , test , test_run_id , suite_id ) : <EOL> self . _verbose ( '<STR_LIT>' % test . name ) <EOL> try : <EOL> test_id = self . _db . insert ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : suite_id , <EOL> '<STR_LIT>' : test . id , <EOL> '<STR_LIT:name>' : test . name , <EOL> '<STR_LIT>' : test . timeout , <EOL> '<STR_LIT>' : test . doc <EOL> } ) <EOL> except IntegrityError : <EOL> test_id = self . _db . fetch_id ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : suite_id , <EOL> '<STR_LIT:name>' : test . name <EOL> } ) <EOL> self . _parse_test_status ( test_run_id , test_id , test ) <EOL> self . _parse_tags ( test . tags , test_id ) <EOL> self . _parse_keywords ( test . keywords , test_run_id , None , test_id ) <EOL> def _parse_test_status ( self , test_run_id , test_id , test ) : <EOL> self . _db . insert_or_ignore ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : test_run_id , <EOL> '<STR_LIT>' : test_id , <EOL> '<STR_LIT:status>' : test . status , <EOL> '<STR_LIT>' : test . elapsedtime <EOL> } ) <EOL> def _parse_tags ( self , tags , test_id ) : <EOL> self . _db . insert_many_or_ignore ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT:content>' ) , <EOL> [ ( test_id , tag ) for tag in tags ] <EOL> ) <EOL> def _parse_keywords ( self , keywords , test_run_id , suite_id , test_id , keyword_id = None ) : <EOL> if self . _include_keywords : <EOL> [ self . _parse_keyword ( keyword , test_run_id , suite_id , test_id , keyword_id ) <EOL> for keyword in keywords ] <EOL> def _parse_keyword ( self , keyword , test_run_id , suite_id , test_id , keyword_id ) : <EOL> try : <EOL> keyword_id = self . _db . insert ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : suite_id , <EOL> '<STR_LIT>' : test_id , <EOL> '<STR_LIT>' : keyword_id , <EOL> '<STR_LIT:name>' : keyword . name , <EOL> '<STR_LIT:type>' : keyword . type , <EOL> '<STR_LIT>' : keyword . timeout , <EOL> '<STR_LIT>' : keyword . doc <EOL> } ) <EOL> except IntegrityError : <EOL> keyword_id = self . _db . fetch_id ( '<STR_LIT>' , { <EOL> '<STR_LIT:name>' : keyword . name , <EOL> '<STR_LIT:type>' : keyword . type <EOL> } ) <EOL> self . _parse_keyword_status ( test_run_id , keyword_id , keyword ) <EOL> self . _parse_messages ( keyword . messages , keyword_id ) <EOL> self . _parse_arguments ( keyword . args , keyword_id ) <EOL> self . _parse_keywords ( keyword . keywords , test_run_id , None , None , keyword_id ) <EOL> def _parse_keyword_status ( self , test_run_id , keyword_id , keyword ) : <EOL> self . _db . insert_or_ignore ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : test_run_id , <EOL> '<STR_LIT>' : keyword_id , <EOL> '<STR_LIT:status>' : keyword . status , <EOL> '<STR_LIT>' : keyword . elapsedtime <EOL> } ) <EOL> def _parse_messages ( self , messages , keyword_id ) : <EOL> self . _db . insert_many_or_ignore ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:content>' ) , <EOL> [ ( keyword_id , message . level , self . _format_robot_timestamp ( message . timestamp ) , <EOL> message . message ) for message in messages ] <EOL> ) <EOL> def _parse_arguments ( self , args , keyword_id ) : <EOL> self . _db . insert_many_or_ignore ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT:content>' ) , <EOL> [ ( keyword_id , arg ) for arg in args ] <EOL> ) <EOL> def _format_robot_timestamp ( self , timestamp ) : <EOL> return datetime . strptime ( timestamp , '<STR_LIT>' ) </s>
