<s> import unittest <EOL> import time <EOL> import glob <EOL> import sys <EOL> import threading <EOL> import tempfile <EOL> import signal <EOL> import logging <EOL> from os . path import abspath , dirname , join , exists , curdir <EOL> from os import chdir <EOL> from robot import run , rebot <EOL> from robot . model import SuiteVisitor <EOL> from robot . running import namespace <EOL> from robot . utils import StringIO <EOL> from robot . utils . asserts import assert_equal , assert_true <EOL> from resources . runningtestcase import RunningTestCase <EOL> from resources . Listener import Listener <EOL> ROOT = dirname ( dirname ( dirname ( abspath ( __file__ ) ) ) ) <EOL> TEMP = tempfile . gettempdir ( ) <EOL> OUTPUT_PATH = join ( TEMP , '<STR_LIT>' ) <EOL> REPORT_PATH = join ( TEMP , '<STR_LIT>' ) <EOL> LOG_PATH = join ( TEMP , '<STR_LIT>' ) <EOL> LOG = '<STR_LIT>' % LOG_PATH <EOL> def run_without_outputs ( * args , ** kwargs ) : <EOL> kwargs . update ( output = '<STR_LIT>' , log = '<STR_LIT>' , report = None ) <EOL> return run ( * args , ** kwargs ) <EOL> class StreamWithOnlyWriteAndFlush ( object ) : <EOL> def __init__ ( self ) : <EOL> self . _buffer = [ ] <EOL> def write ( self , msg ) : <EOL> self . _buffer . append ( msg ) <EOL> def flush ( self ) : <EOL> pass <EOL> def getvalue ( self ) : <EOL> return '<STR_LIT>' . join ( self . _buffer ) <EOL> class TestRun ( RunningTestCase ) : <EOL> data = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> warn = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> nonex = join ( TEMP , '<STR_LIT>' ) <EOL> remove_files = [ LOG_PATH , REPORT_PATH , OUTPUT_PATH ] <EOL> def test_run_once ( self ) : <EOL> assert_equal ( run ( self . data , outputdir = TEMP , report = '<STR_LIT:none>' ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( [ ( '<STR_LIT>' , <NUM_LIT:2> ) , ( LOG , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> assert exists ( LOG_PATH ) <EOL> def test_run_multiple_times ( self ) : <EOL> assert_equal ( run_without_outputs ( self . data , critical = '<STR_LIT>' ) , <NUM_LIT:0> ) <EOL> assert_equal ( run_without_outputs ( self . data , name = '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( [ ( '<STR_LIT>' , <NUM_LIT:2> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) , ( LOG , <NUM_LIT:0> ) ] ) <EOL> def test_run_fail ( self ) : <EOL> assert_equal ( run ( self . data , outputdir = TEMP ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( stdout = [ ( '<STR_LIT>' , <NUM_LIT:2> ) , ( LOG , <NUM_LIT:1> ) ] ) <EOL> def test_run_error ( self ) : <EOL> assert_equal ( run ( self . nonex ) , <NUM_LIT> ) <EOL> self . _assert_outputs ( stderr = [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( self . nonex , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> ) ] ) <EOL> def test_custom_stdout ( self ) : <EOL> stdout = StringIO ( ) <EOL> assert_equal ( run_without_outputs ( self . data , stdout = stdout ) , <NUM_LIT:1> ) <EOL> self . _assert_output ( stdout , [ ( '<STR_LIT>' , <NUM_LIT:2> ) , ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:0> ) , ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> self . _assert_outputs ( ) <EOL> def test_custom_stderr ( self ) : <EOL> stderr = StringIO ( ) <EOL> assert_equal ( run_without_outputs ( self . warn , stderr = stderr ) , <NUM_LIT:0> ) <EOL> self . _assert_output ( stderr , [ ( '<STR_LIT>' , <NUM_LIT:4> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) ] ) <EOL> self . _assert_outputs ( [ ( '<STR_LIT>' , <NUM_LIT:2> ) , ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:0> ) , ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> def test_custom_stdout_and_stderr_with_minimal_implementation ( self ) : <EOL> output = StreamWithOnlyWriteAndFlush ( ) <EOL> assert_equal ( run_without_outputs ( self . warn , stdout = output , stderr = output ) , <NUM_LIT:0> ) <EOL> self . _assert_output ( output , [ ( '<STR_LIT>' , <NUM_LIT:4> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:3> ) , ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:0> ) , ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> self . _assert_outputs ( ) <EOL> def test_multi_options_as_single_string ( self ) : <EOL> assert_equal ( run_without_outputs ( self . data , exclude = '<STR_LIT>' ) , <NUM_LIT:0> ) <EOL> self . _assert_outputs ( [ ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> def test_listener_gets_notification_about_log_report_and_output ( self ) : <EOL> listener = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert_equal ( run ( self . data , output = OUTPUT_PATH , report = REPORT_PATH , <EOL> log = LOG_PATH , listener = listener ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( stdout = [ ( '<STR_LIT>' . format ( OUTPUT_PATH ) , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' . format ( REPORT_PATH ) , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' . format ( LOG_PATH ) , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> ) ] ) <EOL> def test_pass_listener_as_instance ( self ) : <EOL> assert_equal ( run_without_outputs ( self . data , listener = Listener ( <NUM_LIT:1> ) ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( [ ( "<STR_LIT>" , <NUM_LIT:1> ) ] ) <EOL> def test_pass_listener_as_string ( self ) : <EOL> module_file = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert_equal ( run_without_outputs ( self . data , listener = module_file + "<STR_LIT>" ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( [ ( "<STR_LIT>" , <NUM_LIT:1> ) ] ) <EOL> def test_pass_listener_as_list ( self ) : <EOL> module_file = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert_equal ( run_without_outputs ( self . data , listener = [ module_file + "<STR_LIT>" , Listener ( <NUM_LIT:2> ) ] ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( [ ( "<STR_LIT>" , <NUM_LIT:1> ) , ( "<STR_LIT>" , <NUM_LIT:1> ) ] ) <EOL> def test_pre_run_modifier_as_instance ( self ) : <EOL> class Modifier ( SuiteVisitor ) : <EOL> def start_suite ( self , suite ) : <EOL> suite . tests = [ t for t in suite . tests if t . tags . match ( '<STR_LIT>' ) ] <EOL> assert_equal ( run_without_outputs ( self . data , prerunmodifier = Modifier ( ) ) , <NUM_LIT:0> ) <EOL> self . _assert_outputs ( [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> def test_pre_rebot_modifier_as_instance ( self ) : <EOL> class Modifier ( SuiteVisitor ) : <EOL> def __init__ ( self ) : <EOL> self . tests = [ ] <EOL> def visit_test ( self , test ) : <EOL> self . tests . append ( test . name ) <EOL> modifier = Modifier ( ) <EOL> assert_equal ( run ( self . data , outputdir = TEMP , log = LOG_PATH , prerebotmodifier = modifier ) , <NUM_LIT:1> ) <EOL> assert_equal ( modifier . tests , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . _assert_outputs ( [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:1> ) ] ) <EOL> def test_invalid_modifier ( self ) : <EOL> assert_equal ( run_without_outputs ( self . data , prerunmodifier = <NUM_LIT> ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:1> ) ] , <EOL> [ ( "<STR_LIT>" <EOL> "<STR_LIT>" , <NUM_LIT:1> ) ] ) <EOL> class TestRebot ( RunningTestCase ) : <EOL> data = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> nonex = join ( TEMP , '<STR_LIT>' ) <EOL> remove_files = [ LOG_PATH , REPORT_PATH ] <EOL> def test_run_once ( self ) : <EOL> assert_equal ( rebot ( self . data , outputdir = TEMP , report = '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( [ ( LOG , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> assert exists ( LOG_PATH ) <EOL> def test_run_multiple_times ( self ) : <EOL> assert_equal ( rebot ( self . data , outputdir = TEMP , critical = '<STR_LIT>' ) , <NUM_LIT:0> ) <EOL> assert_equal ( rebot ( self . data , outputdir = TEMP , name = '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( [ ( LOG , <NUM_LIT:2> ) ] ) <EOL> def test_run_fails ( self ) : <EOL> assert_equal ( rebot ( self . nonex ) , <NUM_LIT> ) <EOL> assert_equal ( rebot ( self . data , outputdir = TEMP ) , <NUM_LIT:1> ) <EOL> self . _assert_outputs ( stdout = [ ( LOG , <NUM_LIT:1> ) ] , <EOL> stderr = [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( self . nonex , ( <NUM_LIT:1> , <NUM_LIT:2> ) ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> ) ] ) <EOL> def test_custom_stdout ( self ) : <EOL> stdout = StringIO ( ) <EOL> assert_equal ( rebot ( self . data , report = '<STR_LIT:None>' , stdout = stdout , <EOL> outputdir = TEMP ) , <NUM_LIT:1> ) <EOL> self . _assert_output ( stdout , [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> self . _assert_outputs ( ) <EOL> def test_custom_stdout_and_stderr_with_minimal_implementation ( self ) : <EOL> output = StreamWithOnlyWriteAndFlush ( ) <EOL> assert_equal ( rebot ( self . data , log = '<STR_LIT>' , report = '<STR_LIT>' , stdout = output , <EOL> stderr = output ) , <NUM_LIT> ) <EOL> assert_equal ( rebot ( self . data , report = '<STR_LIT>' , stdout = output , <EOL> stderr = output , outputdir = TEMP ) , <NUM_LIT:1> ) <EOL> self . _assert_output ( output , [ ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:0> ) ] ) <EOL> self . _assert_outputs ( ) <EOL> def test_pre_rebot_modifier_as_instance ( self ) : <EOL> class Modifier ( SuiteVisitor ) : <EOL> def __init__ ( self ) : <EOL> self . tests = [ ] <EOL> def visit_test ( self , test ) : <EOL> self . tests . append ( test . name ) <EOL> test . status = '<STR_LIT>' <EOL> modifier = Modifier ( ) <EOL> assert_equal ( rebot ( self . data , outputdir = TEMP , <EOL> prerebotmodifier = modifier ) , <NUM_LIT:3> ) <EOL> assert_equal ( modifier . tests , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class TestStateBetweenTestRuns ( RunningTestCase ) : <EOL> data = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_importer_caches_are_cleared_between_runs ( self ) : <EOL> self . _run ( self . data ) <EOL> lib = self . _import_library ( ) <EOL> res = self . _import_resource ( ) <EOL> self . _run ( self . data ) <EOL> assert_true ( lib is not self . _import_library ( ) ) <EOL> assert_true ( res is not self . _import_resource ( ) ) <EOL> def _run ( self , data , ** config ) : <EOL> return run_without_outputs ( data , outputdir = TEMP , ** config ) <EOL> def _import_library ( self ) : <EOL> return namespace . IMPORTER . import_library ( '<STR_LIT>' , None , None , None ) <EOL> def _import_resource ( self ) : <EOL> resource = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> return namespace . IMPORTER . import_resource ( resource ) <EOL> def test_clear_namespace_between_runs ( self ) : <EOL> data = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> rc = self . _run ( data , test = [ '<STR_LIT>' ] , variable = [ '<STR_LIT>' ] ) <EOL> assert_equal ( rc , <NUM_LIT:0> ) <EOL> rc = self . _run ( data , test = [ '<STR_LIT>' ] ) <EOL> assert_equal ( rc , <NUM_LIT:1> ) <EOL> def test_reset_logging_conf ( self ) : <EOL> assert_equal ( logging . getLogger ( ) . handlers , [ ] ) <EOL> assert_equal ( logging . raiseExceptions , <NUM_LIT:1> ) <EOL> self . _run ( join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> assert_equal ( logging . getLogger ( ) . handlers , [ ] ) <EOL> assert_equal ( logging . raiseExceptions , <NUM_LIT:1> ) <EOL> def test_listener_unregistration ( self ) : <EOL> listener = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert_equal ( run_without_outputs ( self . data , listener = listener + '<STR_LIT>' ) , <NUM_LIT:0> ) <EOL> self . _assert_outputs ( [ ( "<STR_LIT>" , <NUM_LIT:1> ) , ( "<STR_LIT>" , <NUM_LIT:1> ) ] ) <EOL> self . _clear_outputs ( ) <EOL> assert_equal ( run_without_outputs ( self . data ) , <NUM_LIT:0> ) <EOL> self . _assert_outputs ( [ ( "<STR_LIT>" , <NUM_LIT:0> ) , ( "<STR_LIT>" , <NUM_LIT:0> ) ] ) <EOL> class TestTimestampOutputs ( RunningTestCase ) : <EOL> output = join ( TEMP , '<STR_LIT>' ) <EOL> report = join ( TEMP , '<STR_LIT>' ) <EOL> log = join ( TEMP , '<STR_LIT>' ) <EOL> remove_files = [ output , report , log ] <EOL> def test_different_timestamps_when_run_multiple_times ( self ) : <EOL> self . run_tests ( ) <EOL> output1 , = self . find_results ( self . output , <NUM_LIT:1> ) <EOL> report1 , = self . find_results ( self . report , <NUM_LIT:1> ) <EOL> log1 , = self . find_results ( self . log , <NUM_LIT:1> ) <EOL> self . wait_until_next_second ( ) <EOL> self . run_tests ( ) <EOL> output21 , output22 = self . find_results ( self . output , <NUM_LIT:2> ) <EOL> report21 , report22 = self . find_results ( self . report , <NUM_LIT:2> ) <EOL> log21 , log22 = self . find_results ( self . log , <NUM_LIT:2> ) <EOL> assert_equal ( output1 , output21 ) <EOL> assert_equal ( report1 , report21 ) <EOL> assert_equal ( log1 , log21 ) <EOL> def run_tests ( self ) : <EOL> data = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert_equal ( run ( data , timestampoutputs = True , outputdir = TEMP , <EOL> output = '<STR_LIT>' , report = '<STR_LIT>' , <EOL> log = '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> def find_results ( self , pattern , expected ) : <EOL> matches = glob . glob ( pattern ) <EOL> assert_equal ( len ( matches ) , expected ) <EOL> return sorted ( matches ) <EOL> def wait_until_next_second ( self ) : <EOL> start = time . localtime ( ) [ <NUM_LIT:5> ] <EOL> while time . localtime ( ) [ <NUM_LIT:5> ] == start : <EOL> time . sleep ( <NUM_LIT> ) <EOL> class TestSignalHandlers ( unittest . TestCase ) : <EOL> data = join ( ROOT , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_original_signal_handlers_are_restored ( self ) : <EOL> orig_sigint = signal . getsignal ( signal . SIGINT ) <EOL> orig_sigterm = signal . getsignal ( signal . SIGTERM ) <EOL> my_sigterm = lambda signum , frame : None <EOL> signal . signal ( signal . SIGTERM , my_sigterm ) <EOL> try : <EOL> run_without_outputs ( self . data , stdout = StringIO ( ) ) <EOL> assert_equal ( signal . getsignal ( signal . SIGINT ) , orig_sigint ) <EOL> assert_equal ( signal . getsignal ( signal . SIGTERM ) , my_sigterm ) <EOL> finally : <EOL> signal . signal ( signal . SIGINT , orig_sigint ) <EOL> signal . signal ( signal . SIGTERM , orig_sigterm ) <EOL> def test_dont_register_signal_handlers_then_run_on_thread ( self ) : <EOL> stream = StringIO ( ) <EOL> thread = threading . Thread ( target = run_without_outputs , args = ( self . data , ) , <EOL> kwargs = dict ( stdout = stream , stderr = stream ) ) <EOL> thread . start ( ) <EOL> thread . join ( ) <EOL> output = stream . getvalue ( ) <EOL> assert_true ( '<STR_LIT>' not in output . upper ( ) , '<STR_LIT>' % output ) <EOL> class TestRelativeImportsFromPythonpath ( RunningTestCase ) : <EOL> data = join ( abspath ( dirname ( __file__ ) ) , '<STR_LIT>' ) <EOL> def setUp ( self ) : <EOL> self . _orig_path = abspath ( curdir ) <EOL> chdir ( ROOT ) <EOL> sys . path . append ( join ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def tearDown ( self ) : <EOL> chdir ( self . _orig_path ) <EOL> sys . path . pop ( ) <EOL> def test_importing_library_from_pythonpath ( self ) : <EOL> errors = StringIO ( ) <EOL> run ( self . data , outputdir = TEMP , stdout = StringIO ( ) , stderr = errors ) <EOL> self . _assert_output ( errors , '<STR_LIT>' ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
