<s> import uuid <EOL> import pickle <EOL> import pytest <EOL> import argparse <EOL> from collections import namedtuple <EOL> from _pytest . junitxml import mangle_testnames <EOL> from six import text_type <EOL> from allure . common import AllureImpl , StepContext <EOL> from allure . constants import Status , AttachmentType , Severity , FAILED_STATUSES , Label , SKIPPED_STATUSES <EOL> from allure . utils import parent_module , parent_down_from_module , labels_of , all_of , get_exception_message , now <EOL> from allure . structure import TestCase , TestStep , Attach , TestSuite , Failure , TestLabel <EOL> def pytest_addoption ( parser ) : <EOL> parser . getgroup ( "<STR_LIT>" ) . addoption ( '<STR_LIT>' , <EOL> action = "<STR_LIT:store>" , <EOL> dest = "<STR_LIT>" , <EOL> metavar = "<STR_LIT>" , <EOL> default = None , <EOL> help = "<STR_LIT>" ) <EOL> severities = [ v for ( _ , v ) in all_of ( Severity ) ] <EOL> def label_type ( name , legal_values = set ( ) ) : <EOL> """<STR_LIT>""" <EOL> def a_label_type ( string ) : <EOL> atoms = set ( string . split ( '<STR_LIT:U+002C>' ) ) <EOL> if legal_values and not atoms < legal_values : <EOL> raise argparse . ArgumentTypeError ( '<STR_LIT>' . format ( name , '<STR_LIT:U+002CU+0020>' . join ( atoms - legal_values ) , '<STR_LIT:U+002CU+0020>' . join ( legal_values ) ) ) <EOL> return set ( ( name , v ) for v in atoms ) <EOL> return a_label_type <EOL> parser . getgroup ( "<STR_LIT>" ) . addoption ( '<STR_LIT>' , <EOL> action = "<STR_LIT:store>" , <EOL> dest = "<STR_LIT>" , <EOL> metavar = "<STR_LIT>" , <EOL> default = { } , <EOL> type = label_type ( name = Label . SEVERITY , legal_values = set ( severities ) ) , <EOL> help = """<STR_LIT>""" % '<STR_LIT:U+002CU+0020>' . join ( severities ) ) <EOL> parser . getgroup ( "<STR_LIT>" ) . addoption ( '<STR_LIT>' , <EOL> action = "<STR_LIT:store>" , <EOL> dest = "<STR_LIT>" , <EOL> metavar = "<STR_LIT>" , <EOL> default = { } , <EOL> type = label_type ( name = Label . FEATURE ) , <EOL> help = """<STR_LIT>""" ) <EOL> parser . getgroup ( "<STR_LIT>" ) . addoption ( '<STR_LIT>' , <EOL> action = "<STR_LIT:store>" , <EOL> dest = "<STR_LIT>" , <EOL> metavar = "<STR_LIT>" , <EOL> default = { } , <EOL> type = label_type ( name = Label . STORY ) , <EOL> help = """<STR_LIT>""" ) <EOL> def pytest_configure ( config ) : <EOL> reportdir = config . option . allurereportdir <EOL> if reportdir : <EOL> allure_impl = AllureImpl ( reportdir ) <EOL> testlistener = AllureTestListener ( config ) <EOL> pytest . allure . _allurelistener = testlistener <EOL> config . pluginmanager . register ( testlistener ) <EOL> if not hasattr ( config , '<STR_LIT>' ) : <EOL> config . pluginmanager . register ( AllureAgregatingListener ( allure_impl , config ) ) <EOL> config . pluginmanager . register ( AllureCollectionListener ( allure_impl ) ) <EOL> class AllureTestListener ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , config ) : <EOL> self . config = config <EOL> self . environment = { } <EOL> self . test = None <EOL> self . _magicaldoublereport = hasattr ( self . config , '<STR_LIT>' ) and self . config . getvalue ( "<STR_LIT>" ) <EOL> @ pytest . mark . hookwrapper <EOL> def pytest_runtest_protocol ( self , item , nextitem ) : <EOL> try : <EOL> description = item . function . __doc__ <EOL> except AttributeError : <EOL> description = item . reportinfo ( ) [ <NUM_LIT:2> ] <EOL> self . test = TestCase ( name = '<STR_LIT:.>' . join ( mangle_testnames ( [ x . name for x in parent_down_from_module ( item ) ] ) ) , <EOL> description = description , <EOL> start = now ( ) , <EOL> attachments = [ ] , <EOL> labels = labels_of ( item ) , <EOL> status = None , <EOL> steps = [ ] , <EOL> id = str ( uuid . uuid4 ( ) ) ) <EOL> self . stack = [ self . test ] <EOL> yield <EOL> self . test = None <EOL> self . stack = [ ] <EOL> def attach ( self , title , contents , attach_type ) : <EOL> """<STR_LIT>""" <EOL> attach = Attach ( source = contents , <EOL> title = title , <EOL> type = attach_type ) <EOL> self . stack [ - <NUM_LIT:1> ] . attachments . append ( attach ) <EOL> def dynamic_issue ( self , * issues ) : <EOL> """<STR_LIT>""" <EOL> if self . test : <EOL> self . test . labels . extend ( [ TestLabel ( name = Label . ISSUE , value = issue ) for issue in issues ] ) <EOL> def start_step ( self , name ) : <EOL> """<STR_LIT>""" <EOL> step = TestStep ( name = name , <EOL> title = name , <EOL> start = now ( ) , <EOL> attachments = [ ] , <EOL> steps = [ ] ) <EOL> self . stack [ - <NUM_LIT:1> ] . steps . append ( step ) <EOL> self . stack . append ( step ) <EOL> return step <EOL> def stop_step ( self ) : <EOL> """<STR_LIT>""" <EOL> step = self . stack . pop ( ) <EOL> step . stop = now ( ) <EOL> def _fill_case ( self , report , call , pyteststatus , status ) : <EOL> """<STR_LIT>""" <EOL> [ self . attach ( name , contents , AttachmentType . TEXT ) for ( name , contents ) in dict ( report . sections ) . items ( ) ] <EOL> self . test . stop = now ( ) <EOL> self . test . status = status <EOL> if status in FAILED_STATUSES : <EOL> self . test . failure = Failure ( message = get_exception_message ( call . excinfo , pyteststatus , report ) , <EOL> trace = report . longrepr or hasattr ( report , '<STR_LIT>' ) and report . wasxfail ) <EOL> elif status in SKIPPED_STATUSES : <EOL> skip_message = type ( report . longrepr ) == tuple and report . longrepr [ <NUM_LIT:2> ] or report . wasxfail <EOL> trim_msg_len = <NUM_LIT> <EOL> short_message = skip_message . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:0> ] [ : trim_msg_len ] <EOL> self . test . failure = Failure ( message = ( short_message + '<STR_LIT>' * ( len ( skip_message ) > trim_msg_len ) ) , <EOL> trace = status == Status . PENDING and report . longrepr or short_message != skip_message and skip_message or '<STR_LIT>' ) <EOL> def report_case ( self , item , report ) : <EOL> """<STR_LIT>""" <EOL> parent = parent_module ( item ) <EOL> report . __dict__ . update ( _allure_result = pickle . dumps ( ( parent . nodeid , <EOL> parent . module . __name__ , <EOL> parent . module . __doc__ or '<STR_LIT>' , <EOL> self . environment , <EOL> self . test ) ) ) <EOL> @ pytest . mark . hookwrapper <EOL> def pytest_runtest_makereport ( self , item , call ) : <EOL> """<STR_LIT>""" <EOL> report = ( yield ) . get_result ( ) <EOL> status = self . config . hook . pytest_report_teststatus ( report = report ) <EOL> status = status and status [ <NUM_LIT:0> ] <EOL> if report . when == '<STR_LIT>' : <EOL> if report . passed : <EOL> self . _fill_case ( report , call , status , Status . PASSED ) <EOL> elif report . failed : <EOL> self . _fill_case ( report , call , status , Status . FAILED ) <EOL> if self . _magicaldoublereport : <EOL> self . report_case ( item , report ) <EOL> elif report . skipped : <EOL> if hasattr ( report , '<STR_LIT>' ) : <EOL> self . _fill_case ( report , call , status , Status . PENDING ) <EOL> else : <EOL> self . _fill_case ( report , call , status , Status . CANCELED ) <EOL> elif report . when == '<STR_LIT>' : <EOL> if report . failed : <EOL> self . _fill_case ( report , call , status , Status . BROKEN ) <EOL> elif report . skipped : <EOL> if hasattr ( report , '<STR_LIT>' ) : <EOL> self . _fill_case ( report , call , status , Status . PENDING ) <EOL> else : <EOL> self . _fill_case ( report , call , status , Status . CANCELED ) <EOL> elif report . when == '<STR_LIT>' : <EOL> if not report . passed : <EOL> if self . test . status not in FAILED_STATUSES : <EOL> self . _fill_case ( report , call , status , Status . BROKEN ) <EOL> else : <EOL> self . test . status = Status . BROKEN <EOL> self . report_case ( item , report ) <EOL> def pytest_runtest_setup ( item ) : <EOL> item_labels = set ( ( l . name , l . value ) for l in labels_of ( item ) ) <EOL> arg_labels = set ( ) . union ( item . config . option . allurefeatures , <EOL> item . config . option . allurestories , <EOL> item . config . option . allureseverities ) <EOL> if arg_labels and not item_labels & arg_labels : <EOL> pytest . skip ( '<STR_LIT>' % '<STR_LIT:U+002CU+0020>' . join ( text_type ( l ) for l in sorted ( arg_labels ) ) ) <EOL> class LazyInitStepContext ( StepContext ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , allure_helper , title ) : <EOL> self . allure_helper = allure_helper <EOL> self . title = title <EOL> self . step = None <EOL> @ property <EOL> def allure ( self ) : <EOL> return self . allure_helper . get_listener ( ) <EOL> class AllureHelper ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _allurelistener = None <EOL> def get_listener ( self ) : <EOL> return self . _allurelistener <EOL> def attach ( self , name , contents , type = AttachmentType . TEXT ) : <EOL> """<STR_LIT>""" <EOL> if self . _allurelistener : <EOL> self . _allurelistener . attach ( name , contents , type ) <EOL> def label ( self , name , * value ) : <EOL> """<STR_LIT>""" <EOL> allure_label = getattr ( pytest . mark , '<STR_LIT>' % ( Label . DEFAULT , name ) ) <EOL> return allure_label ( * value ) <EOL> def severity ( self , severity ) : <EOL> """<STR_LIT>""" <EOL> return self . label ( Label . SEVERITY , severity ) <EOL> def feature ( self , * features ) : <EOL> """<STR_LIT>""" <EOL> return self . label ( Label . FEATURE , * features ) <EOL> def story ( self , * stories ) : <EOL> """<STR_LIT>""" <EOL> return self . label ( Label . STORY , * stories ) <EOL> def issue ( self , * issues ) : <EOL> """<STR_LIT>""" <EOL> return self . label ( Label . ISSUE , * issues ) <EOL> def dynamic_issue ( self , * issues ) : <EOL> """<STR_LIT>""" <EOL> if self . _allurelistener : <EOL> self . _allurelistener . dynamic_issue ( * issues ) <EOL> def testcase ( self , * testcases ) : <EOL> """<STR_LIT>""" <EOL> return self . label ( Label . TESTCASE , * testcases ) <EOL> def step ( self , title ) : <EOL> """<STR_LIT>""" <EOL> if callable ( title ) : <EOL> return LazyInitStepContext ( self , title . __name__ ) ( title ) <EOL> else : <EOL> return LazyInitStepContext ( self , title ) <EOL> def single_step ( self , text ) : <EOL> """<STR_LIT>""" <EOL> if self . _allurelistener : <EOL> with self . step ( text ) : <EOL> pass <EOL> def environment ( self , ** env_dict ) : <EOL> if self . _allurelistener : <EOL> self . _allurelistener . environment . update ( env_dict ) <EOL> @ property <EOL> def attach_type ( self ) : <EOL> return AttachmentType <EOL> @ property <EOL> def severity_level ( self ) : <EOL> return Severity <EOL> def __getattr__ ( self , attr ) : <EOL> """<STR_LIT>""" <EOL> if attr in dir ( Severity ) and not attr . startswith ( '<STR_LIT:_>' ) : <EOL> return self . severity ( getattr ( Severity , attr ) ) <EOL> else : <EOL> raise AttributeError <EOL> MASTER_HELPER = AllureHelper ( ) <EOL> def pytest_namespace ( ) : <EOL> return { '<STR_LIT>' : MASTER_HELPER } <EOL> class AllureAgregatingListener ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , impl , config ) : <EOL> self . impl = impl <EOL> self . suites = { } <EOL> def pytest_sessionfinish ( self ) : <EOL> """<STR_LIT>""" <EOL> for s in self . suites . values ( ) : <EOL> if s . tests : <EOL> s . stop = max ( case . stop for case in s . tests ) <EOL> known_ids = set ( ) <EOL> refined_tests = [ ] <EOL> for t in s . tests [ : : - <NUM_LIT:1> ] : <EOL> if t . id not in known_ids : <EOL> known_ids . add ( t . id ) <EOL> refined_tests . append ( t ) <EOL> s . tests [ : : - <NUM_LIT:1> ] = refined_tests <EOL> with self . impl . _reportfile ( '<STR_LIT>' % uuid . uuid4 ( ) ) as f : <EOL> self . impl . _write_xml ( f , s ) <EOL> self . impl . store_environment ( ) <EOL> def write_attach ( self , attachment ) : <EOL> """<STR_LIT>""" <EOL> attachment . source = self . impl . _save_attach ( attachment . source , attachment . type ) <EOL> attachment . type = attachment . type . mime_type <EOL> def pytest_runtest_logreport ( self , report ) : <EOL> if hasattr ( report , '<STR_LIT>' ) : <EOL> module_id , module_name , module_doc , environment , testcase = pickle . loads ( report . _allure_result ) <EOL> self . impl . environment . update ( environment ) <EOL> for a in testcase . iter_attachments ( ) : <EOL> self . write_attach ( a ) <EOL> self . suites . setdefault ( module_id , TestSuite ( name = module_name , <EOL> description = module_doc , <EOL> tests = [ ] , <EOL> labels = [ ] , <EOL> start = testcase . start , <EOL> stop = None ) ) . tests . append ( testcase ) <EOL> CollectFail = namedtuple ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class AllureCollectionListener ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , impl ) : <EOL> self . impl = impl <EOL> self . fails = [ ] <EOL> def pytest_collectreport ( self , report ) : <EOL> if not report . passed : <EOL> if report . failed : <EOL> status = Status . BROKEN <EOL> else : <EOL> status = Status . CANCELED <EOL> self . fails . append ( CollectFail ( name = mangle_testnames ( report . nodeid . split ( "<STR_LIT>" ) ) [ - <NUM_LIT:1> ] , <EOL> status = status , <EOL> message = get_exception_message ( None , None , report ) , <EOL> trace = report . longrepr ) ) <EOL> def pytest_sessionfinish ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . fails : <EOL> self . impl . start_suite ( name = '<STR_LIT>' , <EOL> title = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) <EOL> for fail in self . fails : <EOL> self . impl . start_case ( name = fail . name . split ( "<STR_LIT:.>" ) [ - <NUM_LIT:1> ] ) <EOL> self . impl . stop_case ( status = fail . status , message = fail . message , trace = fail . trace ) <EOL> self . impl . stop_suite ( ) </s>
