<s> from __future__ import unicode_literals <EOL> import copy <EOL> import os <EOL> import platform <EOL> import shutil <EOL> import signal <EOL> import sys <EOL> import tempfile <EOL> from textwrap import dedent <EOL> import unittest <EOL> import weakref <EOL> from green . config import default_args <EOL> from green . exceptions import InitializerOrFinalizerError <EOL> from green . loader import loadTargets <EOL> from green . output import GreenStream <EOL> from green . runner import InitializerOrFinalizer , run <EOL> from green . suite import GreenTestSuite <EOL> try : <EOL> from unittest . mock import MagicMock <EOL> except : <EOL> from mock import MagicMock <EOL> try : <EOL> from io import StringIO <EOL> except : <EOL> from StringIO import StringIO <EOL> global skip_testtools <EOL> skip_testtools = False <EOL> try : <EOL> import testtools ; testtools <EOL> except : <EOL> skip_testtools = True <EOL> global importable_function_worked <EOL> importable_function_worked = False <EOL> def _importableFunction ( ) : <EOL> """<STR_LIT>""" <EOL> global importable_function_worked <EOL> importable_function_worked = True <EOL> non_callable = None <EOL> def _crashy ( ) : <EOL> """<STR_LIT>""" <EOL> raise Exception ( '<STR_LIT>' ) <EOL> class TestInitializerOrFinalizer ( unittest . TestCase ) : <EOL> def test_blank ( self ) : <EOL> """<STR_LIT>""" <EOL> initializer = InitializerOrFinalizer ( '<STR_LIT>' ) <EOL> initializer ( ) <EOL> def test_unimportable ( self ) : <EOL> """<STR_LIT>""" <EOL> initializer = InitializerOrFinalizer ( '<STR_LIT>' ) <EOL> self . assertRaises ( InitializerOrFinalizerError , initializer ) <EOL> def test_importable ( self ) : <EOL> """<STR_LIT>""" <EOL> global importable_function_worked <EOL> importable_function_worked = False <EOL> InitializerOrFinalizer ( '<STR_LIT>' ) ( ) <EOL> self . assertTrue ( importable_function_worked ) <EOL> def test_not_callable ( self ) : <EOL> """<STR_LIT>""" <EOL> initializer = InitializerOrFinalizer ( '<STR_LIT>' ) <EOL> self . assertRaises ( InitializerOrFinalizerError , initializer ) <EOL> def test_crash ( self ) : <EOL> """<STR_LIT>""" <EOL> initializer = InitializerOrFinalizer ( '<STR_LIT>' ) <EOL> self . assertRaises ( InitializerOrFinalizerError , initializer ) <EOL> class TestRun ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . startdir = os . getcwd ( ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> if os . getcwd ( ) != cls . startdir : <EOL> os . chdir ( cls . startdir ) <EOL> cls . startdir = None <EOL> def setUp ( self ) : <EOL> self . args = copy . deepcopy ( default_args ) <EOL> self . stream = StringIO ( ) <EOL> self . tmpdir = tempfile . mkdtemp ( ) <EOL> def tearDown ( self ) : <EOL> del ( self . tmpdir ) <EOL> del ( self . stream ) <EOL> def test_stdout ( self ) : <EOL> """<STR_LIT>""" <EOL> saved_stdout = sys . stdout <EOL> sys . stdout = self . stream <EOL> self . addCleanup ( setattr , sys , '<STR_LIT>' , saved_stdout ) <EOL> run ( GreenTestSuite ( ) , sys . stdout , args = self . args ) <EOL> self . assertIn ( '<STR_LIT>' , self . stream . getvalue ( ) ) <EOL> def test_GreenStream ( self ) : <EOL> """<STR_LIT>""" <EOL> gs = GreenStream ( self . stream ) <EOL> run ( GreenTestSuite ( ) , gs , args = self . args ) <EOL> self . assertIn ( '<STR_LIT>' , self . stream . getvalue ( ) ) <EOL> def test_verbose3 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . args . verbose = <NUM_LIT:3> <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" . format ( os . getpid ( ) ) ) ) <EOL> fh . close ( ) <EOL> os . chdir ( sub_tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT>' ) <EOL> result = run ( tests , self . stream , self . args ) <EOL> os . chdir ( self . startdir ) <EOL> self . assertEqual ( result . testsRun , <NUM_LIT:1> ) <EOL> self . assertIn ( '<STR_LIT>' , self . stream . getvalue ( ) ) <EOL> self . assertIn ( '<STR_LIT:OK>' , self . stream . getvalue ( ) ) <EOL> def test_warnings ( self ) : <EOL> """<STR_LIT>""" <EOL> self . args . warnings = '<STR_LIT>' <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" . format ( os . getpid ( ) ) ) ) <EOL> fh . close ( ) <EOL> os . chdir ( sub_tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT>' ) <EOL> result = run ( tests , self . stream , self . args ) <EOL> os . chdir ( self . startdir ) <EOL> self . assertEqual ( result . testsRun , <NUM_LIT:1> ) <EOL> self . assertIn ( '<STR_LIT:OK>' , self . stream . getvalue ( ) ) <EOL> def test_noTestsFound ( self ) : <EOL> """<STR_LIT>""" <EOL> run ( GreenTestSuite ( ) , self . stream , self . args ) <EOL> self . assertIn ( '<STR_LIT>' , self . stream . getvalue ( ) ) <EOL> def test_failedSaysSo ( self ) : <EOL> """<STR_LIT>""" <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" . format ( os . getpid ( ) ) ) ) <EOL> fh . close ( ) <EOL> os . chdir ( sub_tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT>' ) <EOL> result = run ( tests , self . stream , self . args ) <EOL> os . chdir ( self . startdir ) <EOL> self . assertEqual ( result . testsRun , <NUM_LIT:1> ) <EOL> self . assertIn ( '<STR_LIT>' , self . stream . getvalue ( ) ) <EOL> def test_failfast ( self ) : <EOL> """<STR_LIT>""" <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" . format ( os . getpid ( ) ) ) ) <EOL> fh . close ( ) <EOL> os . chdir ( sub_tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT>' ) <EOL> self . args . failfast = True <EOL> result = run ( tests , self . stream , self . args ) <EOL> os . chdir ( self . startdir ) <EOL> self . assertEqual ( result . testsRun , <NUM_LIT:1> ) <EOL> def test_systemExit ( self ) : <EOL> """<STR_LIT>""" <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" . format ( os . getpid ( ) ) ) ) <EOL> fh . close ( ) <EOL> os . chdir ( sub_tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT>' ) <EOL> result = run ( tests , self . stream , self . args ) <EOL> os . chdir ( self . startdir ) <EOL> self . assertEqual ( result . testsRun , <NUM_LIT:2> ) <EOL> class TestProcesses ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . startdir = os . getcwd ( ) <EOL> cls . container_dir = tempfile . mkdtemp ( ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> if os . getcwd ( ) != cls . startdir : <EOL> os . chdir ( cls . startdir ) <EOL> cls . startdir = None <EOL> shutil . rmtree ( cls . container_dir ) <EOL> def setUp ( self ) : <EOL> os . chdir ( self . container_dir ) <EOL> self . tmpdir = tempfile . mkdtemp ( dir = self . container_dir ) <EOL> self . stream = StringIO ( ) <EOL> self . args = copy . deepcopy ( default_args ) <EOL> def tearDown ( self ) : <EOL> os . chdir ( self . startdir ) <EOL> shutil . rmtree ( self . tmpdir , ignore_errors = True ) <EOL> del ( self . stream ) <EOL> def test_catchProcessSIGINT ( self ) : <EOL> """<STR_LIT>""" <EOL> if platform . system ( ) == '<STR_LIT>' : <EOL> self . skipTest ( '<STR_LIT>' ) <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> saved__results = unittest . signals . _results <EOL> unittest . signals . _results = weakref . WeakKeyDictionary ( ) <EOL> self . addCleanup ( setattr , unittest . signals , '<STR_LIT>' , saved__results ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" . format ( os . getpid ( ) ) ) ) <EOL> fh . close ( ) <EOL> os . chdir ( sub_tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT>' ) <EOL> self . args . processes = <NUM_LIT:2> <EOL> run ( tests , self . stream , self . args ) <EOL> os . chdir ( TestProcesses . startdir ) <EOL> def test_collisionProtection ( self ) : <EOL> """<STR_LIT>""" <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( '<STR_LIT:\n>' ) <EOL> fh . close ( ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( '<STR_LIT>' ) <EOL> fh . close ( ) <EOL> os . mkdir ( os . path . join ( sub_tmpdir , '<STR_LIT:test>' ) ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT:test>' , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( '<STR_LIT:\n>' ) <EOL> fh . close ( ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT:test>' , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" . format ( os . path . basename ( sub_tmpdir ) ) ) ) <EOL> fh . close ( ) <EOL> os . chdir ( self . tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT:.>' ) <EOL> self . args . processes = <NUM_LIT:2> <EOL> self . args . termcolor = False <EOL> try : <EOL> run ( tests , self . stream , self . args ) <EOL> except KeyboardInterrupt : <EOL> os . kill ( os . getpid ( ) , signal . SIGINT ) <EOL> os . chdir ( TestProcesses . startdir ) <EOL> self . assertIn ( '<STR_LIT:OK>' , self . stream . getvalue ( ) ) <EOL> def test_detectNumProcesses ( self ) : <EOL> """<STR_LIT>""" <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( '<STR_LIT:\n>' ) <EOL> fh . close ( ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" ) ) <EOL> fh . close ( ) <EOL> os . chdir ( self . tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT:.>' ) <EOL> self . args . processes = <NUM_LIT:0> <EOL> run ( tests , self . stream , self . args ) <EOL> os . chdir ( TestProcesses . startdir ) <EOL> self . assertIn ( '<STR_LIT:OK>' , self . stream . getvalue ( ) ) <EOL> def test_runCoverage ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> import coverage ; coverage <EOL> except : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( '<STR_LIT:\n>' ) <EOL> fh . close ( ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" ) ) <EOL> fh . close ( ) <EOL> os . chdir ( self . tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT:.>' ) <EOL> self . args . processes = <NUM_LIT:2> <EOL> self . args . run_coverage = True <EOL> self . args . cov = MagicMock ( ) <EOL> run ( tests , self . stream , self . args , testing = True ) <EOL> os . chdir ( TestProcesses . startdir ) <EOL> self . assertIn ( '<STR_LIT:OK>' , self . stream . getvalue ( ) ) <EOL> def test_badTest ( self ) : <EOL> """<STR_LIT>""" <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( '<STR_LIT:\n>' ) <EOL> fh . close ( ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( "<STR_LIT>" ) <EOL> fh . close ( ) <EOL> os . chdir ( self . tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT:.>' ) <EOL> self . args . processes = <NUM_LIT:2> <EOL> os . chdir ( TestProcesses . startdir ) <EOL> self . assertRaises ( ImportError , run , tests , self . stream , self . args ) <EOL> def test_uncaughtException ( self ) : <EOL> """<STR_LIT>""" <EOL> global skip_testtools <EOL> if skip_testtools : <EOL> self . skipTest ( '<STR_LIT>' ) <EOL> sub_tmpdir = tempfile . mkdtemp ( dir = self . tmpdir ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( '<STR_LIT:\n>' ) <EOL> fh . close ( ) <EOL> fh = open ( os . path . join ( sub_tmpdir , '<STR_LIT>' ) , '<STR_LIT:w>' ) <EOL> fh . write ( dedent ( <EOL> """<STR_LIT>""" ) ) <EOL> fh . close ( ) <EOL> os . chdir ( self . tmpdir ) <EOL> tests = loadTargets ( '<STR_LIT:.>' ) <EOL> self . args . processes = <NUM_LIT:2> <EOL> run ( tests , self . stream , self . args ) <EOL> os . chdir ( TestProcesses . startdir ) <EOL> self . assertIn ( '<STR_LIT>' , self . stream . getvalue ( ) ) <EOL> def test_empty ( self ) : <EOL> """<STR_LIT>""" <EOL> suite = GreenTestSuite ( ) <EOL> self . args . processes = <NUM_LIT:2> <EOL> self . args . termcolor = False <EOL> run ( suite , self . stream , self . args ) </s>
