def testSecondInterrupt(self):
        if due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
            return

        result = unittest.TestResult()
        unittest.installHandler()
        unittest.registerResult(result)

        def test(result):
            pid = os.getpid()
            os.kill(pid, signal.SIGINT)
            result.breakCaught = True
            if not due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
                self.assertTrue(result.shouldStop)
            os.kill(pid, signal.SIGINT)
            if not due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
                self.fail("Second KeyboardInterrupt not raised")

        try:
            test(result)
        except KeyboardInterrupt:
            pass
        else:
            if not due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
                self.fail("Second KeyboardInterrupt not raised")
        self.assertTrue(result.breakCaught)

    def testTwoResults(self):
        if due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
            return

        unittest.installHandler()

        result = unittest.TestResult()
        unittest.registerResult(result)
        new_handler = signal.getsignal(signal.SIGINT)

        result2 = unittest.TestResult()
        unittest.registerResult(result2)
        self.assertEqual(signal.getsignal(signal.SIGINT), new_handler)

        result3 = unittest.TestResult()

        def test(result):
            pid = os.getpid()
            os.kill(pid, signal.SIGINT)

        try:
            test(result)
        except KeyboardInterrupt:
            self.fail("KeyboardInterrupt not handled")

        if not due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
            self.assertTrue(result.shouldStop)
            self.assertTrue(result2.shouldStop)
        self.assertFalse(result3.shouldStop)

    def testTwoResults(self):
        unittest.installHandler()

        result = unittest.TestResult()
        unittest.registerResult(result)
        new_handler = signal.getsignal(signal.SIGINT)

        result2 = unittest.TestResult()
        unittest.registerResult(result2)
        self.assertEqual(signal.getsignal(signal.SIGINT), new_handler)

        result3 = unittest.TestResult()

        def test(result):
            pid = os.getpid()
            os.kill(pid, signal.SIGINT)

        try:
            test(result)
        except KeyboardInterrupt:
            self.fail("KeyboardInterrupt not handled")

        self.assertTrue(result.shouldStop)
        self.assertTrue(result2.shouldStop)
        self.assertFalse(result3.shouldStop)

    def run(self, test):
        "Run the given test case or test suite."
        result = self._makeResult()
        registerResult(result)
        result.failfast = self.failfast
        result.buffer = self.buffer
        startTime = time.time()
        startTestRun = getattr(result, 'startTestRun', None)
        if startTestRun is not None:
            startTestRun()
        try:
            test(result)
        finally:
            stopTestRun = getattr(result, 'stopTestRun', None)
            if stopTestRun is not None:
                stopTestRun()
        stopTime = time.time()
        timeTaken = stopTime - startTime
        #stopTestRun = getattr(result, 'logRunTime', None)
        #stopTestRun(test, timeTaken)

        result.printErrors()
        if hasattr(result, 'separator2'):
            self.stream.writeln(result.separator2)
        run = result.testsRun
        self.stream.writeln("Ran %d test%s in %.3fs" %
                            (run, run != 1 and "s" or "", timeTaken))
        self.stream.writeln()
        return result

 def testWeakReferences(self):
     result = unittest.TestResult()
     unittest.registerResult(result)
     ref = weakref.ref(result)
     del result
     gc.collect()
     gc.collect()
     self.assertIsNone(ref())

 def run(self, test):
   result = unittest.TestResult()
   unittest.registerResult(result)
   test(result)
   for kind, errors in [('FAIL', result.failures), ('ERROR', result.errors)]:
     for test, err in errors:
       sys.stderr.write('{} {}\n{}'.format(test, kind, err))
   return result

    def run(self,
            test_modules=__all__,
            pretend=False,
            logfile='test/test.log',
            loglevel=forcebalance.output.INFO,
            **kwargs):
            
        self.check()

        self.logger.setLevel(loglevel)

        # first install unittest interrupt handler which gracefully finishes current test on Ctrl+C
        unittest.installHandler()

        # create blank test suite and fill it with test suites loaded from each test module
        tests = unittest.TestSuite()
        systemTests = unittest.TestSuite()
        for suite in unittest.defaultTestLoader.discover('test'):
            for module in suite:
                for test in module:
                    modName,caseName,testName = test.id().split('.')
                    if modName in test_modules:
                        if modName=="test_system": systemTests.addTest(test)
                        else: tests.addTest(test)

        tests.addTests(systemTests) # integration tests should be run after other tests

        result = ForceBalanceTestResult()
        
        forcebalance.output.getLogger("forcebalance").addHandler(forcebalance.output.NullHandler())

        ### START TESTING ###
        # run any pretest tasks before first test
        result.startTestRun(tests)

        # if pretend option is enabled, skip all tests instead of running them
        if pretend:
            for test in tests:
                result.addSkip(test)

        # otherwise do a normal test run
        else:
            unittest.registerResult(result)
            try:
                tests.run(result)
            except KeyboardInterrupt:
                # Adding this allows us to determine
                # what is causing tests to hang indefinitely.
                import traceback
                traceback.print_exc()
                self.logger.exception(msg="Test run cancelled by user")
            except:
                self.logger.exception(msg="An unexpected exception occurred while running tests\n")

        result.stopTestRun(tests)
        ### STOP TESTING ###

        return result

    def testRegisterResult(self):
        result = unittest.TestResult()
        self.assertNotIn(result, unittest.signals._results)

        unittest.registerResult(result)
        try:
            self.assertIn(result, unittest.signals._results)
        finally:
            unittest.removeResult(result)

 def testRegisterResult(self):
     result = unittest.TestResult()
     unittest.registerResult(result)
     for ref in unittest.signals._results:
         if ref is result:
             break
         else:
             while ref is not result:
                 self.fail('odd object in result set')
     self.fail('result not found')

    def run(self,
            test_modules=__all__,
            pretend=False,
            logfile='test/test.log',
            loglevel=logging.INFO,
            **kwargs):

        self.logger.setLevel(loglevel)

        # first install unittest interrupt handler which gracefully finishes current test on Ctrl+C
        unittest.installHandler()

        # create blank test suite and fill it with test suites loaded from each test module
        tests = unittest.TestSuite()
        for module in test_modules:
            try:
                m=__import__(module)
                module_tests=unittest.defaultTestLoader.loadTestsFromModule(m)
                tests.addTest(module_tests)
            except ImportError:
                self.logger.error("No such test module: %s\n" % module)
            except:
                self.logger.critical("Error loading '%s'\n" % module)
                print traceback.print_exc()

        result = ForceBalanceTestResult()

        ### START TESTING ###
        # run any pretest tasks before first test
        result.startTestRun(tests)

        # if pretend option is enabled, skip all tests instead of running them
        if pretend:
            for module in tests:
                for test in module:
                    try:
                        result.addSkip(test)
                    # addSkip will fail if run on TestSuite objects
                    except AttributeError: continue

        # otherwise do a normal test run
        else:
            self.console = sys.stdout
            sys.stdout = open(logfile, 'w')

            unittest.registerResult(result)
            tests.run(result)

            sys.stdout.close()
            sys.stdout = self.console

        result.stopTestRun(tests)
        ### STOP TESTING ###

        return result

    def testRegisterResult(self):
        result = unittest.TestResult()
        unittest.registerResult(result)

        for ref in unittest.signals._results:
            if ref is result:
                break
            elif ref is not result:
                self.fail("odd object in result set")
        else:
            self.fail("result not found")

    def testWeakReferences(self):
        # Calling registerResult on a result should not keep it alive
        result = unittest.TestResult()
        unittest.registerResult(result)

        ref = weakref.ref(result)
        del result

        # For non-reference counting implementations
        gc.collect();gc.collect()
        self.assertIsNone(ref())

 def testRemoveResult(self):
     result = unittest.TestResult()
     unittest.registerResult(result)
     unittest.installHandler()
     self.assertTrue(unittest.removeResult(result))
     self.assertFalse(unittest.removeResult(unittest.TestResult()))
     try:
         pid = os.getpid()
         os.kill(pid, signal.SIGINT)
     except KeyboardInterrupt:
         pass
     self.assertFalse(result.shouldStop)

 def run(self, test):
     result = self._makeResult()
     # diffs between Python 2.6 and Python 2.7
     if hasattr(unittest, 'registerResult'):
         unittest.registerResult(result)
     if hasattr(result, 'failfast'):
         result.failfast = self.failfast
     if hasattr(result, 'buffer'):
         result.buffer = self.buffer
     test(result)
     self.stream.writeln((_testname + ": %d tests, %d failures, %d errors") %
                         (result.testsRun, len(result.failures), len(result.errors)))
     result.printErrors()
     return result

    def testInterruptCaught(self):
        default_handler = signal.getsignal(signal.SIGINT)
        result = unittest.TestResult()
        unittest.installHandler()
        unittest.registerResult(result)
        self.assertNotEqual(signal.getsignal(signal.SIGINT), default_handler)

        def test(result):
            pid = os.getpid()
            os.kill(pid, signal.SIGINT)
            result.breakCaught = True
            self.assertTrue(result.shouldStop)

        try:
            test(result)
        except KeyboardInterrupt:
            self.fail('KeyboardInterrupt not handled')
        self.assertTrue(result.breakCaught)

    def testSecondInterrupt(self):
        result = unittest.TestResult()
        unittest.installHandler()
        unittest.registerResult(result)

        def test(result):
            pid = os.getpid()
            os.kill(pid, signal.SIGINT)
            result.breakCaught = True
            self.assertTrue(result.shouldStop)
            os.kill(pid, signal.SIGINT)
            self.fail("Second KeyboardInterrupt not raised")

        try:
            test(result)
        except KeyboardInterrupt:
            pass
        else:
            self.fail("Second KeyboardInterrupt not raised")
        self.assertTrue(result.breakCaught)

    def testRemoveResult(self):
        if due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
            return

        result = unittest.TestResult()
        unittest.registerResult(result)

        unittest.installHandler()
        self.assertTrue(unittest.removeResult(result))

        # Should this raise an error instead?
        self.assertFalse(unittest.removeResult(unittest.TestResult()))

        try:
            pid = os.getpid()
            os.kill(pid, signal.SIGINT)
        except KeyboardInterrupt:
            pass

        self.assertFalse(result.shouldStop)

    def testSecondInterrupt(self):
        if signal.getsignal(signal.SIGINT) == signal.SIG_IGN:
            self.skipTest('test requires SIGINT to not be ignored')
        result = unittest.TestResult()
        unittest.installHandler()
        unittest.registerResult(result)

        def test(result):
            pid = os.getpid()
            os.kill(pid, signal.SIGINT)
            result.breakCaught = True
            self.assertTrue(result.shouldStop)
            os.kill(pid, signal.SIGINT)
            self.fail('Second KeyboardInterrupt not raised')

        try:
            test(result)
        except KeyboardInterrupt:
            pass
        self.fail('Second KeyboardInterrupt not raised')
        self.assertTrue(result.breakCaught)

    def run(self, test):
        "Run the given test case or test suite."
        result = self._result
        unittest.registerResult(result)
        result.failfast = self.failfast
        result.buffer = self.buffer
        startTime = time.time()
        startTestRun = getattr(result, 'startTestRun', None)
        if startTestRun is not None:
            startTestRun()
        try:
            test(result)
        finally:
            stopTime = time.time()
            timeTaken = stopTime - startTime
            setTimes = getattr(result, 'setTimes', None)
            if setTimes is not None:
                setTimes(startTime, stopTime, timeTaken)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if stopTestRun is not None:
                stopTestRun()

        return result

    def testInterruptCaught(self):
        if due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
            return

        default_handler = signal.getsignal(signal.SIGINT)

        result = unittest.TestResult()
        unittest.installHandler()
        unittest.registerResult(result)

        self.assertNotEqual(signal.getsignal(signal.SIGINT), default_handler)

        def test(result):
            pid = os.getpid()
            os.kill(pid, signal.SIGINT)
            result.breakCaught = True
            if not due_to_ironpython_bug("http://ironpython.codeplex.com/workitem/28171"):
                self.assertTrue(result.shouldStop)

        try:
            test(result)
        except KeyboardInterrupt:
            self.fail("KeyboardInterrupt not handled")
        self.assertTrue(result.breakCaught)