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
python类TestCase()的实例源码
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)
