Python twisted.trial.unittest 模块，TestCase() 实例源码

def makeSQLTests(base, suffix, globals):
    """Make a test case for every db connector which can connect.

    @param base: Base class for test case. Additional base classes
                 will be a DBConnector subclass and unittest.TestCase
    @param suffix: A suffix used to create test case names. Prefixes
                   are defined in the DBConnector subclasses.
    """
    connectors = [GadflyConnector, SQLiteConnector, PyPgSQLConnector,
                  PsycopgConnector, MySQLConnector, FirebirdConnector]
    for connclass in connectors:
        name = connclass.TEST_PREFIX + suffix
        import new
        klass = new.classobj(name, (connclass, base, unittest.TestCase), {})
        globals[name] = klass

# GadflyADBAPITestCase SQLiteADBAPITestCase PyPgSQLADBAPITestCase
# PsycopgADBAPITestCase MySQLADBAPITestCase FirebirdADBAPITestCase

def test_assertFailure_masked(self):
        """A single wrong assertFailure should fail the whole test.
        """
        class ExampleFailure(Exception):
            pass

        class TC(unittest.TestCase):
            failureException = ExampleFailure
            def test_assertFailure(self):
                d = defer.maybeDeferred(lambda: 1/0)
                self.assertFailure(d, OverflowError)
                self.assertFailure(d, ZeroDivisionError)
                return d

        test = TC('test_assertFailure')
        result = reporter.TestResult()
        test.run(result)
        self.assertEqual(1, len(result.failures))

def suiteVisit(suite, visitor):
    """
    Visit each test in C{suite} with C{visitor}.

    @param visitor: A callable which takes a single argument, the L{TestCase}
    instance to visit.
    @return: None
    """
    for case in suite._tests:
        visit = getattr(case, 'visit', None)
        if visit is not None:
            visit(visitor)
        elif isinstance(case, pyunit.TestCase):
            case = PyUnitTestCase(case)
            case.visit(visitor)
        elif isinstance(case, pyunit.TestSuite):
            suiteVisit(case, visitor)
        else:
            case.visit(visitor)

def name(thing):
    """
    @param thing: an object from modules (instance of PythonModule,
    PythonAttribute), a TestCase subclass, or an instance of a TestCase.
    """
    if isTestCase(thing):
        # TestCase subclass
        theName = reflect.qual(thing)
    else:
        # thing from trial, or thing from modules.
        # this monstrosity exists so that modules' objects do not have to
        # implement id(). -jml
        try:
            theName = thing.id()
        except AttributeError:
            theName = thing.name
    return theName

def loadAnything(self, thing, recurse=False):
        """
        Given a Python object, return whatever tests that are in it. Whatever
        'in' might mean.

        @param thing: A Python object. A module, method, class or package.
        @param recurse: Whether or not to look in subpackages of packages.
        Defaults to False.

        @return: A C{TestCase} or C{TestSuite}.
        """
        if isinstance(thing, types.ModuleType):
            if isPackage(thing):
                return self.loadPackage(thing, recurse)
            return self.loadModule(thing)
        elif isinstance(thing, types.ClassType):
            return self.loadClass(thing)
        elif isinstance(thing, type):
            return self.loadClass(thing)
        elif isinstance(thing, types.MethodType):
            return self.loadMethod(thing)
        raise TypeError("No loader for %r. Unrecognized type" % (thing,))

def makeSQLTests(base, suffix, globals):
    """Make a test case for every db connector which can connect.

    @param base: Base class for test case. Additional base classes
                 will be a DBConnector subclass and unittest.TestCase
    @param suffix: A suffix used to create test case names. Prefixes
                   are defined in the DBConnector subclasses.
    """
    connectors = [GadflyConnector, SQLiteConnector, PyPgSQLConnector,
                  PsycopgConnector, MySQLConnector, FirebirdConnector]
    for connclass in connectors:
        name = connclass.TEST_PREFIX + suffix
        import new
        klass = new.classobj(name, (connclass, base, unittest.TestCase), {})
        globals[name] = klass

# GadflyADBAPITestCase SQLiteADBAPITestCase PyPgSQLADBAPITestCase
# PsycopgADBAPITestCase MySQLADBAPITestCase FirebirdADBAPITestCase

def test_assertFailure_masked(self):
        """A single wrong assertFailure should fail the whole test.
        """
        class ExampleFailure(Exception):
            pass

        class TC(unittest.TestCase):
            failureException = ExampleFailure
            def test_assertFailure(self):
                d = defer.maybeDeferred(lambda: 1/0)
                self.assertFailure(d, OverflowError)
                self.assertFailure(d, ZeroDivisionError)
                return d

        test = TC('test_assertFailure')
        result = reporter.TestResult()
        test.run(result)
        self.assertEqual(1, len(result.failures))

def suiteVisit(suite, visitor):
    """
    Visit each test in C{suite} with C{visitor}.

    @param visitor: A callable which takes a single argument, the L{TestCase}
    instance to visit.
    @return: None
    """
    for case in suite._tests:
        visit = getattr(case, 'visit', None)
        if visit is not None:
            visit(visitor)
        elif isinstance(case, pyunit.TestCase):
            case = PyUnitTestCase(case)
            case.visit(visitor)
        elif isinstance(case, pyunit.TestSuite):
            suiteVisit(case, visitor)
        else:
            case.visit(visitor)

def name(thing):
    """
    @param thing: an object from modules (instance of PythonModule,
    PythonAttribute), a TestCase subclass, or an instance of a TestCase.
    """
    if isTestCase(thing):
        # TestCase subclass
        theName = reflect.qual(thing)
    else:
        # thing from trial, or thing from modules.
        # this monstrosity exists so that modules' objects do not have to
        # implement id(). -jml
        try:
            theName = thing.id()
        except AttributeError:
            theName = thing.name
    return theName

def loadAnything(self, thing, recurse=False):
        """
        Given a Python object, return whatever tests that are in it. Whatever
        'in' might mean.

        @param thing: A Python object. A module, method, class or package.
        @param recurse: Whether or not to look in subpackages of packages.
        Defaults to False.

        @return: A C{TestCase} or C{TestSuite}.
        """
        if isinstance(thing, types.ModuleType):
            if isPackage(thing):
                return self.loadPackage(thing, recurse)
            return self.loadModule(thing)
        elif isinstance(thing, types.ClassType):
            return self.loadClass(thing)
        elif isinstance(thing, type):
            return self.loadClass(thing)
        elif isinstance(thing, types.MethodType):
            return self.loadMethod(thing)
        raise TypeError("No loader for %r. Unrecognized type" % (thing,))

def test_buildConnectionPool(self):
        """
        L{buildConnectionPool} returns a L{ConnectionPool} which will be
        running only for the duration of the test.
        """
        collect = []

        class SampleTest(TestCase):

            def setUp(self):
                self.pool = buildConnectionPool(self)

            def test_sample(self):
                collect.append(self.pool.running)

            def tearDown(self):
                collect.append(self.pool.running)

        r = TestResult()
        t = SampleTest("test_sample")
        t.run(r)
        self.assertIsInstance(t.pool, ConnectionPool)
        self.assertEqual([True, False], collect)

def assertLoggedIn(self, d, username):
        """
        Assert that the L{Deferred} passed in is called back with the value
        'username'.  This represents a valid login for this TestCase.

        NOTE: To work, this method's return value must be returned from the
        test method, or otherwise hooked up to the test machinery.

        @param d: a L{Deferred} from an L{IChecker.requestAvatarId} method.
        @type d: L{Deferred}
        @rtype: L{Deferred}
        """
        result = []
        d.addBoth(result.append)
        self.assertEqual(len(result), 1, "login incomplete")
        if isinstance(result[0], Failure):
            result[0].raiseException()
        self.assertEqual(result[0], username)

def assertReading(testCase, reactor, transport):
    """
    Use the given test to assert that the given transport is actively reading
    in the given reactor.

    @note: Maintainers; for more information on why this is a function rather
        than a method on a test case, see U{this document on how we structure
        test tools
        <http://twistedmatrix.com/trac/wiki/Design/KeepTestToolsOutOfFixtures>}

    @param testCase: a test case to perform the assertion upon.
    @type testCase: L{TestCase}

    @param reactor: A reactor, possibly one providing L{IReactorFDSet}, or an
        IOCP reactor.

    @param transport: An L{ITCPTransport}
    """
    if IReactorFDSet.providedBy(reactor):
        testCase.assertIn(transport, reactor.getReaders())
    else:
        # IOCP.
        testCase.assertIn(transport, reactor.handles)
        testCase.assertTrue(transport.reading)

def _patchTextFileLogObserver(patch):
    """
    Patch L{logger.textFileLogObserver} to record every call and keep a
    reference to the passed log file for tests.

    @param patch: a callback for patching (usually L{unittest.TestCase.patch}).

    @return: the list that keeps track of the log files.
    @rtype: C{list}
    """
    logFiles = []
    oldFileLogObserver = logger.textFileLogObserver

    def observer(logFile, *args, **kwargs):
        logFiles.append(logFile)
        return oldFileLogObserver(logFile, *args, **kwargs)

    patch(logger, 'textFileLogObserver', observer)
    return logFiles

def pathContainingDumpOf(testCase, *dumpables):
    """
    Create a temporary file to store some serializable-as-PEM objects in, and
    return its name.

    @param testCase: a test case to use for generating a temporary directory.
    @type testCase: L{twisted.trial.unittest.TestCase}

    @param dumpables: arguments are objects from pyOpenSSL with a C{dump}
        method, taking a pyOpenSSL file-type constant, such as
        L{OpenSSL.crypto.FILETYPE_PEM} or L{OpenSSL.crypto.FILETYPE_ASN1}.
    @type dumpables: L{tuple} of L{object} with C{dump} method taking L{int}
        returning L{bytes}

    @return: the path to a file where all of the dumpables were dumped in PEM
        format.
    @rtype: L{str}
    """
    fname = testCase.mktemp()
    with open(fname, "wb") as f:
        for dumpable in dumpables:
            f.write(dumpable.dump(FILETYPE_PEM))
    return fname

def test_decorateTestSuiteReferences(self):
        """
        When decorating a test suite in-place, the number of references to the
        test objects in that test suite should stay the same.

        Previously, L{unittest.decorate} recreated a test suite, so the
        original suite kept references to the test objects. This test is here
        to ensure the problem doesn't reappear again.
        """
        getrefcount = getattr(sys, 'getrefcount', None)
        if getrefcount is None:
            raise unittest.SkipTest(
                "getrefcount not supported on this platform")
        test = self.TestCase()
        suite = unittest.TestSuite([test])
        count1 = getrefcount(test)
        unittest.decorate(suite, unittest.TestDecorator)
        count2 = getrefcount(test)
        self.assertEqual(count1, count2)

def test_shouldStop(self):
        """
        Test the C{shouldStop} management: raising a C{KeyboardInterrupt} must
        interrupt the suite.
        """
        called = []
        class MockTest(unittest.TestCase):
            def test_foo1(test):
                called.append(1)
            def test_foo2(test):
                raise KeyboardInterrupt()
            def test_foo3(test):
                called.append(2)
        result = reporter.TestResult()
        loader = runner.TestLoader()
        loader.suiteFactory = runner.DestructiveTestSuite
        suite = loader.loadClass(MockTest)
        self.assertEqual(called, [])
        suite.run(result)
        self.assertEqual(called, [1])
        # The last test shouldn't have been run
        self.assertEqual(suite.countTestCases(), 1)

def test_reporterDeprecations(self):
        """
        The runner emits a warning if it is using a result that doesn't
        implement 'done'.
        """
        trialRunner = runner.TrialRunner(None)
        result = self.FakeReporter()
        trialRunner._makeResult = lambda: result
        def f():
            # We have to use a pyunit test, otherwise we'll get deprecation
            # warnings about using iterate() in a test.
            trialRunner.run(pyunit.TestCase('id'))

        f()
        warnings = self.flushWarnings([self.test_reporterDeprecations])

        self.assertEqual(warnings[0]['category'], DeprecationWarning)
        self.assertEqual(warnings[0]['message'],
            "%s should implement done() but doesn't. Falling back to "
            "printErrors() and friends." % reflect.qual(result.__class__))
        self.assertTrue(__file__.startswith(warnings[0]['filename']))
        self.assertEqual(len(warnings), 1)

def _testUnequalPair(self, first, second):
        """
        Assert that when called with unequal arguments, C{assertEqual} raises a
        failure exception with the same message as the standard library
        C{assertEqual} would have raised.
        """
        raised = False
        try:
            self.assertEqual(first, second)
        except self.failureException as ourFailure:
            case = pyunit.TestCase("setUp")
            try:
                case.assertEqual(first, second)
            except case.failureException as theirFailure:
                raised = True
                got = str(ourFailure)
                expected = str(theirFailure)
                if expected != got:
                    self.fail("Expected: %r; Got: %r" % (expected, got))

        if not raised:
            self.fail(
                "Call to assertEqual(%r, %r) didn't fail" % (first, second)
            )

def test_assertFailure_masked(self):
        """
        A single wrong assertFailure should fail the whole test.
        """
        class ExampleFailure(Exception):
            pass

        class TC(unittest.TestCase):
            failureException = ExampleFailure
            def test_assertFailure(self):
                d = defer.maybeDeferred(lambda: 1/0)
                self.assertFailure(d, OverflowError)
                self.assertFailure(d, ZeroDivisionError)
                return d

        test = TC('test_assertFailure')
        result = pyunit.TestResult()
        test.run(result)
        self.assertEqual(1, len(result.failures))

def savedJSONInvariants(testCase, savedJSON):
    """
    Assert a few things about the result of L{eventAsJSON}, then return it.

    @param testCase: The L{TestCase} with which to perform the assertions.
    @type testCase: L{TestCase}

    @param savedJSON: The result of L{eventAsJSON}.
    @type savedJSON: L{unicode} (we hope)

    @return: C{savedJSON}
    @rtype: L{unicode}

    @raise AssertionError: If any of the preconditions fail.
    """
    testCase.assertIsInstance(savedJSON, unicode)
    testCase.assertEqual(savedJSON.count("\n"), 0)
    return savedJSON

def test_runTest_method(testdir):
    testdir.makepyfile("""
        import unittest
        class MyTestCaseWithRunTest(unittest.TestCase):
            def runTest(self):
                self.assertEquals('foo', 'foo')
        class MyTestCaseWithoutRunTest(unittest.TestCase):
            def runTest(self):
                self.assertEquals('foo', 'foo')
            def test_something(self):
                pass
        """)
    result = testdir.runpytest("-v")
    result.stdout.fnmatch_lines("""
        *MyTestCaseWithRunTest::runTest*
        *MyTestCaseWithoutRunTest::test_something*
        *2 passed*
    """)

def test_setup(testdir):
    testpath = testdir.makepyfile("""
        import unittest
        class MyTestCase(unittest.TestCase):
            def setUp(self):
                self.foo = 1
            def setup_method(self, method):
                self.foo2 = 1
            def test_both(self):
                self.assertEquals(1, self.foo)
                assert self.foo2 == 1
            def teardown_method(self, method):
                assert 0, "42"

    """)
    reprec = testdir.inline_run("-s", testpath)
    assert reprec.matchreport("test_both", when="call").passed
    rep = reprec.matchreport("test_both", when="teardown")
    assert rep.failed and '42' in str(rep.longrepr)

def test_teardown(testdir):
    testpath = testdir.makepyfile("""
        import unittest
        class MyTestCase(unittest.TestCase):
            l = []
            def test_one(self):
                pass
            def tearDown(self):
                self.l.append(None)
        class Second(unittest.TestCase):
            def test_check(self):
                self.assertEquals(MyTestCase.l, [None])
    """)
    reprec = testdir.inline_run(testpath)
    passed, skipped, failed = reprec.countoutcomes()
    assert failed == 0, failed
    assert passed == 2
    assert passed + skipped + failed == 2

def test_method_and_teardown_failing_reporting(testdir):
    testdir.makepyfile("""
        import unittest, pytest
        class TC(unittest.TestCase):
            def tearDown(self):
                assert 0, "down1"
            def test_method(self):
                assert False, "down2"
    """)
    result = testdir.runpytest("-s")
    assert result.ret == 1
    result.stdout.fnmatch_lines([
        "*tearDown*",
        "*assert 0*",
        "*test_method*",
        "*assert False*",
        "*1 failed*1 error*",
    ])

def test_setup_failure_is_shown(testdir):
    testdir.makepyfile("""
        import unittest
        import pytest
        class TC(unittest.TestCase):
            def setUp(self):
                assert 0, "down1"
            def test_method(self):
                print ("never42")
                xyz
    """)
    result = testdir.runpytest("-s")
    assert result.ret == 1
    result.stdout.fnmatch_lines([
        "*setUp*",
        "*assert 0*down1*",
        "*1 failed*",
    ])
    assert 'never42' not in result.stdout.str()

def test_setup_setUpClass(testdir):
    testpath = testdir.makepyfile("""
        import unittest
        import pytest
        class MyTestCase(unittest.TestCase):
            x = 0
            @classmethod
            def setUpClass(cls):
                cls.x += 1
            def test_func1(self):
                assert self.x == 1
            def test_func2(self):
                assert self.x == 1
            @classmethod
            def tearDownClass(cls):
                cls.x -= 1
        def test_teareddown():
            assert MyTestCase.x == 0
    """)
    reprec = testdir.inline_run(testpath)
    reprec.assertoutcome(passed=3)

def test_setup_class(testdir):
    testpath = testdir.makepyfile("""
        import unittest
        import pytest
        class MyTestCase(unittest.TestCase):
            x = 0
            def setup_class(cls):
                cls.x += 1
            def test_func1(self):
                assert self.x == 1
            def test_func2(self):
                assert self.x == 1
            def teardown_class(cls):
                cls.x -= 1
        def test_teareddown():
            assert MyTestCase.x == 0
    """)
    reprec = testdir.inline_run(testpath)
    reprec.assertoutcome(passed=3)

def test_testcase_adderrorandfailure_defers(testdir, type):
    testdir.makepyfile("""
        from unittest import TestCase
        import pytest
        class MyTestCase(TestCase):
            def run(self, result):
                excinfo = pytest.raises(ZeroDivisionError, lambda: 0/0)
                try:
                    result.add%s(self, excinfo._excinfo)
                except KeyboardInterrupt:
                    raise
                except:
                    pytest.fail("add%s should not raise")
            def test_hello(self):
                pass
    """ % (type, type))
    result = testdir.runpytest()
    assert 'should not raise' not in result.stdout.str()

def test_trial_testcase_runtest_not_collected(self, testdir):
        testdir.makepyfile("""
            from twisted.trial.unittest import TestCase

            class TC(TestCase):
                def test_hello(self):
                    pass
        """)
        reprec = testdir.inline_run()
        reprec.assertoutcome(passed=1)
        testdir.makepyfile("""
            from twisted.trial.unittest import TestCase

            class TC(TestCase):
                def runTest(self):
                    pass
        """)
        reprec = testdir.inline_run()
        reprec.assertoutcome(passed=1)

def test_unorderable_types(testdir):
    testdir.makepyfile("""
        import unittest
        class TestJoinEmpty(unittest.TestCase):
            pass

        def make_test():
            class Test(unittest.TestCase):
                pass
            Test.__name__ = "TestFoo"
            return Test
        TestFoo = make_test()
    """)
    result = testdir.runpytest()
    assert "TypeError" not in result.stdout.str()
    assert result.ret == EXIT_NOTESTSCOLLECTED

def test_unittest_unexpected_failure(testdir):
    testdir.makepyfile("""
        import unittest
        class MyTestCase(unittest.TestCase):
            @unittest.expectedFailure
            def test_func1(self):
                assert 0
            @unittest.expectedFailure
            def test_func2(self):
                assert 1
    """)
    result = testdir.runpytest("-rxX")
    result.stdout.fnmatch_lines([
        "*XFAIL*MyTestCase*test_func1*",
        "*XPASS*MyTestCase*test_func2*",
        "*1 xfailed*1 xpass*",
    ])

def test_no_teardown_if_setupclass_failed(testdir):
    testpath = testdir.makepyfile("""
        import unittest

        class MyTestCase(unittest.TestCase):
            x = 0

            @classmethod
            def setUpClass(cls):
                cls.x = 1
                assert False

            def test_func1(self):
                cls.x = 10

            @classmethod
            def tearDownClass(cls):
                cls.x = 100

        def test_notTornDown():
            assert MyTestCase.x == 1
    """)
    reprec = testdir.inline_run(testpath)
    reprec.assertoutcome(passed=1, failed=1)

def testIsinstance(self):
        self.assert_(isinstance(u'hi', types.StringTypes))
        self.assert_(isinstance(self, unittest.TestCase))
        # I'm pretty sure it's impossible to implement this
        # without replacing isinstance on 2.2 as well :(
        # self.assert_(isinstance({}, dict))

def test_visitPyunitCase(self):
        """
        Test that a stdlib test case in a suite gets visited.
        """
        class PyunitCase(pyunit.TestCase):
            def test_foo(self):
                pass
        test = PyunitCase('test_foo')
        TestSuite([test]).visit(self.visitor)
        self.assertEqual(self.visitor.calls, [test])

def test_defaultIsSuccessful(self):
        """
        Test that L{unittest.TestCase} itself can be instantiated, run, and
        reported as being successful.
        """
        test = unittest.TestCase()
        test.run(self.result)
        self.assertSuccessful(test, self.result)

def makeTestFixtures(self):
        class MockTest(unittest.TestCase):
            def test_foo(test):
                self.log.append('test_foo')
        self.test = MockTest('test_foo')
        self.suite = runner.TestSuite()

def test_failIf_matches_assertNot(self):
        asserts = reflect.prefixedMethods(unittest.TestCase, 'assertNot')
        failIfs = reflect.prefixedMethods(unittest.TestCase, 'failIf')
        self.failUnlessEqual(dsu(asserts, self._name),
                             dsu(failIfs, self._name))

def test_success(self):
        class SuccessTest(TestCase):
            ran = False
            def test_foo(s):
                s.ran = True
        test = SuccessTest('test_foo')
        result = pyunit.TestResult()
        test.run(result)

        self.failUnless(test.ran)
        self.assertEqual(1, result.testsRun)
        self.failUnless(result.wasSuccessful())

def test_failure(self):
        class FailureTest(TestCase):
            ran = False
            def test_foo(s):
                s.ran = True
                s.fail('boom!')
        test = FailureTest('test_foo')
        result = pyunit.TestResult()
        test.run(result)

        self.failUnless(test.ran)
        self.assertEqual(1, result.testsRun)
        self.assertEqual(1, len(result.failures))
        self.failIf(result.wasSuccessful())

def test_error(self):
        class ErrorTest(TestCase):
            ran = False
            def test_foo(s):
                s.ran = True
                1/0
        test = ErrorTest('test_foo')
        result = pyunit.TestResult()
        test.run(result)

        self.failUnless(test.ran)
        self.assertEqual(1, result.testsRun)
        self.assertEqual(1, len(result.errors))
        self.failIf(result.wasSuccessful())

def visit(self, visitor):
        """
        Call the given visitor with the original, standard library, test case
        that C{self} wraps. See L{unittest.TestCase.visit}.
        """
        visitor(self._test)

def _bail(self):
        from twisted.internet import reactor
        d = defer.Deferred()
        reactor.addSystemEventTrigger('after', 'shutdown',
                                      lambda: d.callback(None))
        reactor.fireSystemEvent('shutdown') # radix's suggestion
        treactor = interfaces.IReactorThreads(reactor, None)
        if treactor is not None:
            treactor.suggestThreadPoolSize(0)
        # As long as TestCase does crap stuff with the reactor we need to
        # manually shutdown the reactor here, and that requires util.wait
        # :(
        # so that the shutdown event completes
        unittest.TestCase('mktemp')._wait(d)

def visit(self, visitor):
        """
        See L{unittest.TestCase.visit}.
        """
        visitor(self)

def loadClass(self, klass):
        """
        Given a class which contains test cases, return a sorted list of
        C{TestCase} instances.
        """
        if not (isinstance(klass, type) or isinstance(klass, types.ClassType)):
            raise TypeError("%r is not a class" % (klass,))
        if not isTestCase(klass):
            raise ValueError("%r is not a test case" % (klass,))
        names = self.getTestCaseNames(klass)
        tests = self.sort([self._makeCase(klass, self.methodPrefix+name)
                           for name in names])
        return self.suiteFactory(tests)

def getTestCaseNames(self, klass):
        """
        Given a class that contains C{TestCase}s, return a list of names of
        methods that probably contain tests.
        """
        return reflect.prefixedMethodNames(klass, self.methodPrefix)

def loadMethod(self, method):
        """
        Given a method of a C{TestCase} that represents a test, return a
        C{TestCase} instance for that test.
        """
        if not isinstance(method, types.MethodType):
            raise TypeError("%r not a method" % (method,))
        return self._makeCase(method.im_class, method.__name__)