Update unittest partially (#6051)

* Update `test_unittest` from 3.13.5

* Remove old `test_unittest.py`
This commit is contained in:
Shahar Naveh authored and GitHub committed 2025-07-31 10:53:39 +09:00
1 parent 1c55f9eee2
commit f32a5b105a
41 files changed
+1612 -680

No files matched your search

-16
View File
@@ -1,16 +0,0 @@
import unittest.test
from test import support
def load_tests(*_):
# used by unittest
return unittest.test.suite()
def tearDownModule():
support.reap_children()
if __name__ == "__main__":
unittest.main()
+6
View File
@@ -0,0 +1,6 @@
import os.path
from test.support import load_package_tests
def load_tests(*args):
return load_package_tests(os.path.dirname(__file__), *args)
+4
View File
@@ -0,0 +1,4 @@
from . import load_tests
import unittest
unittest.main()
@@ -18,17 +18,6 @@ def warnfun():
warnings.warn('rw', RuntimeWarning)
class TestWarnings(unittest.TestCase):
# unittest warnings will be printed at most once per type (max one message
# for the fail* methods, and one for the assert* methods)
def test_assert(self):
self.assertEquals(2+2, 4)
self.assertEquals(2*2, 4)
self.assertEquals(2**2, 4)
def test_fail(self):
self.failUnless(1)
self.failUnless(True)
def test_other_unittest(self):
self.assertAlmostEqual(2+2, 4)
self.assertNotAlmostEqual(4+4, 2)
File renamed without changes.
@@ -136,3 +136,19 @@ class ResultWithNoStartTestRunStopTestRun(object):
def wasSuccessful(self):
return True
class BufferedWriter:
def __init__(self):
self.result = ''
self.buffer = ''
def write(self, arg):
self.buffer += arg
def flush(self):
self.result += self.buffer
self.buffer = ''
def getvalue(self):
return self.result
@@ -271,20 +271,11 @@ class TestLongMessage(unittest.TestCase):
r"\+ \{'key': 'value'\}$",
r"\+ \{'key': 'value'\} : oops$"])
def testAssertDictContainsSubset(self):
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
self.assertMessages('assertDictContainsSubset', ({'key': 'value'}, {}),
["^Missing: 'key'$", "^oops$",
"^Missing: 'key'$",
"^Missing: 'key' : oops$"])
def testAssertMultiLineEqual(self):
self.assertMessages('assertMultiLineEqual', ("", "foo"),
[r"\+ foo$", "^oops$",
r"\+ foo$",
r"\+ foo : oops$"])
[r"\+ foo\n$", "^oops$",
r"\+ foo\n$",
r"\+ foo\n : oops$"])
def testAssertLess(self):
self.assertMessages('assertLess', (2, 1),
@@ -395,6 +386,16 @@ class TestLongMessage(unittest.TestCase):
'^UserWarning not triggered$',
'^UserWarning not triggered : oops$'])
def test_assertNotWarns(self):
def warn_future():
warnings.warn('xyz', FutureWarning, stacklevel=2)
self.assertMessagesCM('_assertNotWarns', (FutureWarning,),
warn_future,
['^FutureWarning triggered$',
'^oops$',
'^FutureWarning triggered$',
'^FutureWarning triggered : oops$'])
def testAssertWarnsRegex(self):
# test error not raised
self.assertMessagesCM('assertWarnsRegex', (UserWarning, 'unused regex'),
@@ -37,11 +37,9 @@ class LacksExit:
pass
# TODO: RUSTPYTHON; used by following test suite
# VAR = contextvars.ContextVar('VAR', default=())
VAR = contextvars.ContextVar('VAR', default=())
@unittest.skip("TODO: RUSTPYTHON; requires sys.get_coroutine_origin_tracking_depth()")
class TestAsyncCase(unittest.TestCase):
maxDiff = None
@@ -50,6 +48,8 @@ class TestAsyncCase(unittest.TestCase):
# starting a new event loop
self.addCleanup(support.gc_collect)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_full_cycle(self):
expected = ['setUp',
'asyncSetUp',
@@ -296,6 +296,8 @@ class TestAsyncCase(unittest.TestCase):
test.doCleanups()
self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup2', 'cleanup1'])
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_deprecation_of_return_val_from_test(self):
# Issue 41322 - deprecate return of value that is not None from a test
class Nothing:
@@ -486,5 +488,21 @@ class TestAsyncCase(unittest.TestCase):
result = test.run()
self.assertTrue(result.wasSuccessful())
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_loop_factory(self):
asyncio.set_event_loop_policy(None)
class TestCase1(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.EventLoop
async def test_demo1(self):
pass
test = TestCase1('test_demo1')
result = test.run()
self.assertTrue(result.wasSuccessful())
self.assertIsNone(support.maybe_get_event_loop_policy())
if __name__ == "__main__":
unittest.main()
@@ -236,6 +236,7 @@ class TestBreak(unittest.TestCase):
self.testRunner = FakeRunner
self.test = test
self.result = None
self.durations = None
p = Program(False)
p.runTests()
@@ -244,7 +245,8 @@ class TestBreak(unittest.TestCase):
'verbosity': verbosity,
'failfast': failfast,
'tb_locals': False,
'warnings': None})])
'warnings': None,
'durations': None})])
self.assertEqual(FakeRunner.runArgs, [test])
self.assertEqual(p.result, result)
@@ -259,7 +261,8 @@ class TestBreak(unittest.TestCase):
'verbosity': verbosity,
'failfast': failfast,
'tb_locals': False,
'warnings': None})])
'warnings': None,
'durations': None})])
self.assertEqual(FakeRunner.runArgs, [test])
self.assertEqual(p.result, result)
@@ -15,7 +15,7 @@ from test import support
import unittest
from unittest.test.support import (
from test.test_unittest.support import (
TestEquality, TestHashing, LoggingResult, LegacyLoggingResult,
ResultWithNoStartTestRunStopTestRun
)
@@ -304,7 +304,8 @@ class Test_TestCase(unittest.TestCase, TestEquality, TestHashing):
def test(self):
pass
Foo('test').run()
with self.assertWarns(RuntimeWarning):
Foo('test').run()
def test_deprecation_of_return_val_from_test(self):
# Issue 41322 - deprecate return of value that is not None from a test
@@ -709,36 +710,6 @@ class Test_TestCase(unittest.TestCase, TestEquality, TestHashing):
self.assertRaises(self.failureException, self.assertNotIn, 'cow',
animals)
def testAssertDictContainsSubset(self):
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
self.assertDictContainsSubset({}, {})
self.assertDictContainsSubset({}, {'a': 1})
self.assertDictContainsSubset({'a': 1}, {'a': 1})
self.assertDictContainsSubset({'a': 1}, {'a': 1, 'b': 2})
self.assertDictContainsSubset({'a': 1, 'b': 2}, {'a': 1, 'b': 2})
with self.assertRaises(self.failureException):
self.assertDictContainsSubset({1: "one"}, {})
with self.assertRaises(self.failureException):
self.assertDictContainsSubset({'a': 2}, {'a': 1})
with self.assertRaises(self.failureException):
self.assertDictContainsSubset({'c': 1}, {'a': 1})
with self.assertRaises(self.failureException):
self.assertDictContainsSubset({'a': 1, 'c': 1}, {'a': 1})
with self.assertRaises(self.failureException):
self.assertDictContainsSubset({'a': 1, 'c': 1}, {'a': 1})
one = ''.join(chr(i) for i in range(255))
# this used to cause a UnicodeDecodeError constructing the failure msg
with self.assertRaises(self.failureException):
self.assertDictContainsSubset({'foo': one}, {'foo': '\uFFFD'})
def testAssertEqual(self):
equal_pairs = [
((), ()),
@@ -1161,6 +1132,8 @@ test case
# need to remove the first line of the error message
error = str(e).split('\n', 1)[1]
self.assertEqual(sample_text_error, error)
else:
self.fail(f'{self.failureException} not raised')
def testAssertEqualSingleLine(self):
sample_text = "laden swallows fly slowly"
@@ -1177,6 +1150,74 @@ test case
# need to remove the first line of the error message
error = str(e).split('\n', 1)[1]
self.assertEqual(sample_text_error, error)
else:
self.fail(f'{self.failureException} not raised')
def testAssertEqualwithEmptyString(self):
'''Verify when there is an empty string involved, the diff output
does not treat the empty string as a single empty line. It should
instead be handled as a non-line.
'''
sample_text = ''
revised_sample_text = 'unladen swallows fly quickly'
sample_text_error = '''\
+ unladen swallows fly quickly
'''
try:
self.assertEqual(sample_text, revised_sample_text)
except self.failureException as e:
# need to remove the first line of the error message
error = str(e).split('\n', 1)[1]
self.assertEqual(sample_text_error, error)
else:
self.fail(f'{self.failureException} not raised')
def testAssertEqualMultipleLinesMissingNewlineTerminator(self):
'''Verifying format of diff output from assertEqual involving strings
with multiple lines, but missing the terminating newline on both.
'''
sample_text = 'laden swallows\nfly sloely'
revised_sample_text = 'laden swallows\nfly slowly'
sample_text_error = '''\
laden swallows
- fly sloely
? ^
+ fly slowly
? ^
'''
try:
self.assertEqual(sample_text, revised_sample_text)
except self.failureException as e:
# need to remove the first line of the error message
error = str(e).split('\n', 1)[1]
self.assertEqual(sample_text_error, error)
else:
self.fail(f'{self.failureException} not raised')
def testAssertEqualMultipleLinesMismatchedNewlinesTerminators(self):
'''Verifying format of diff output from assertEqual involving strings
with multiple lines and mismatched newlines. The output should
include a - on it's own line to indicate the newline difference
between the two strings
'''
sample_text = 'laden swallows\nfly sloely\n'
revised_sample_text = 'laden swallows\nfly slowly'
sample_text_error = '''\
laden swallows
- fly sloely
? ^
+ fly slowly
? ^
-\x20
'''
try:
self.assertEqual(sample_text, revised_sample_text)
except self.failureException as e:
# need to remove the first line of the error message
error = str(e).split('\n', 1)[1]
self.assertEqual(sample_text_error, error)
else:
self.fail(f'{self.failureException} not raised')
def testEqualityBytesWarning(self):
if sys.flags.bytes_warning:
@@ -1801,45 +1842,18 @@ test case
pass
self.assertIsNone(value)
def testDeprecatedMethodNames(self):
"""
Test that the deprecated methods raise a DeprecationWarning. See #9424.
"""
old = (
(self.failIfEqual, (3, 5)),
(self.assertNotEquals, (3, 5)),
(self.failUnlessEqual, (3, 3)),
(self.assertEquals, (3, 3)),
(self.failUnlessAlmostEqual, (2.0, 2.0)),
(self.assertAlmostEquals, (2.0, 2.0)),
(self.failIfAlmostEqual, (3.0, 5.0)),
(self.assertNotAlmostEquals, (3.0, 5.0)),
(self.failUnless, (True,)),
(self.assert_, (True,)),
(self.failUnlessRaises, (TypeError, lambda _: 3.14 + 'spam')),
(self.failIf, (False,)),
(self.assertDictContainsSubset, (dict(a=1, b=2), dict(a=1, b=2, c=3))),
(self.assertRaisesRegexp, (KeyError, 'foo', lambda: {}['foo'])),
(self.assertRegexpMatches, ('bar', 'bar')),
)
for meth, args in old:
with self.assertWarns(DeprecationWarning):
meth(*args)
# disable this test for now. When the version where the fail* methods will
# be removed is decided, re-enable it and update the version
def _testDeprecatedFailMethods(self):
"""Test that the deprecated fail* methods get removed in 3.x"""
if sys.version_info[:2] < (3, 3):
return
def testDeprecatedFailMethods(self):
"""Test that the deprecated fail* methods get removed in 3.12"""
deprecated_names = [
'failIfEqual', 'failUnlessEqual', 'failUnlessAlmostEqual',
'failIfAlmostEqual', 'failUnless', 'failUnlessRaises', 'failIf',
'assertDictContainsSubset',
'assertNotEquals', 'assertEquals', 'assertAlmostEquals',
'assertNotAlmostEquals', 'assert_', 'assertDictContainsSubset',
'assertRaisesRegexp', 'assertRegexpMatches'
]
for deprecated_name in deprecated_names:
with self.assertRaises(AttributeError):
getattr(self, deprecated_name) # remove these in 3.x
getattr(self, deprecated_name)
def testDeepcopy(self):
# Issue: 5660
@@ -1956,7 +1970,7 @@ test case
del case
self.assertFalse(wr())
# TODO: RUSTPYTHON; destructors
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_no_exception_leak(self):
# Issue #19880: TestCase.run() should not keep a reference
@@ -6,11 +6,10 @@ import types
import pickle
from test import support
from test.support import import_helper
import test.test_importlib.util
import unittest
import unittest.mock
import unittest.test
import test.test_unittest
class TestableTestProgram(unittest.TestProgram):
@@ -365,6 +364,8 @@ class TestDiscovery(unittest.TestCase):
[(loader, [], 'test*.py'),
(loader, [], 'test*.py')])
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_discover(self):
loader = unittest.TestLoader()
@@ -407,10 +408,36 @@ class TestDiscovery(unittest.TestCase):
top_level_dir = os.path.abspath('/foo/bar')
start_dir = os.path.abspath('/foo/bar/baz')
self.assertEqual(suite, "['tests']")
self.assertEqual(loader._top_level_dir, top_level_dir)
self.assertEqual(loader._top_level_dir, os.path.abspath('/foo'))
self.assertEqual(_find_tests_args, [(start_dir, 'pattern')])
self.assertIn(top_level_dir, sys.path)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_discover_should_not_persist_top_level_dir_between_calls(self):
original_isfile = os.path.isfile
original_isdir = os.path.isdir
original_sys_path = sys.path[:]
def restore():
os.path.isfile = original_isfile
os.path.isdir = original_isdir
sys.path[:] = original_sys_path
self.addCleanup(restore)
os.path.isfile = lambda path: True
os.path.isdir = lambda path: True
loader = unittest.TestLoader()
loader.suiteClass = str
dir = '/foo/bar'
top_level_dir = '/foo'
loader.discover(dir, top_level_dir=top_level_dir)
self.assertEqual(loader._top_level_dir, None)
loader._top_level_dir = dir2 = '/previous/dir'
loader.discover(dir, top_level_dir=top_level_dir)
self.assertEqual(loader._top_level_dir, dir2)
def test_discover_start_dir_is_package_calls_package_load_tests(self):
# This test verifies that the package load_tests in a package is indeed
# invoked when the start_dir is a package (and not the top level).
@@ -789,7 +816,7 @@ class TestDiscovery(unittest.TestCase):
loader = unittest.TestLoader()
tests = [self]
expectedPath = os.path.abspath(os.path.dirname(unittest.test.__file__))
expectedPath = os.path.abspath(os.path.dirname(test.test_unittest.__file__))
self.wasRun = False
def _find_tests(start_dir, pattern):
@@ -797,7 +824,7 @@ class TestDiscovery(unittest.TestCase):
self.assertEqual(start_dir, expectedPath)
return tests
loader._find_tests = _find_tests
suite = loader.discover('unittest.test')
suite = loader.discover('test.test_unittest')
self.assertTrue(self.wasRun)
self.assertEqual(suite._tests, tests)
@@ -826,6 +853,8 @@ class TestDiscovery(unittest.TestCase):
'as dotted module names')
def test_discovery_failed_discovery(self):
from test.test_importlib import util
loader = unittest.TestLoader()
package = types.ModuleType('package')
@@ -837,7 +866,7 @@ class TestDiscovery(unittest.TestCase):
# Since loader.discover() can modify sys.path, restore it when done.
with import_helper.DirsOnSysPath():
# Make sure to remove 'package' from sys.modules when done.
with test.test_importlib.util.uncache('package'):
with util.uncache('package'):
with self.assertRaises(TypeError) as cm:
loader.discover('package')
self.assertEqual(str(cm.exception),
@@ -1,6 +1,6 @@
import unittest
from unittest.test.support import LoggingResult
from test.test_unittest.support import LoggingResult
class Test_FunctionTestCase(unittest.TestCase):
@@ -1,29 +1,9 @@
import functools
import sys
import types
import warnings
import unittest
# Decorator used in the deprecation tests to reset the warning registry for
# test isolation and reproducibility.
def warningregistry(func):
def wrapper(*args, **kws):
missing = []
saved = getattr(warnings, '__warningregistry__', missing).copy()
try:
return func(*args, **kws)
finally:
if saved is missing:
try:
del warnings.__warningregistry__
except AttributeError:
pass
else:
warnings.__warningregistry__ = saved
return wrapper
class Test_TestLoader(unittest.TestCase):
### Basic object tests
@@ -102,6 +82,24 @@ class Test_TestLoader(unittest.TestCase):
self.assertIsInstance(suite, loader.suiteClass)
self.assertEqual(list(suite), [Foo('runTest')])
# "Do not load any tests from `TestCase` class itself."
def test_loadTestsFromTestCase__from_TestCase(self):
loader = unittest.TestLoader()
suite = loader.loadTestsFromTestCase(unittest.TestCase)
self.assertIsInstance(suite, loader.suiteClass)
self.assertEqual(list(suite), [])
# TODO: RUSTPYTHON
@unittest.expectedFailure
# "Do not load any tests from `FunctionTestCase` class."
def test_loadTestsFromTestCase__from_FunctionTestCase(self):
loader = unittest.TestLoader()
suite = loader.loadTestsFromTestCase(unittest.FunctionTestCase)
self.assertIsInstance(suite, loader.suiteClass)
self.assertEqual(list(suite), [])
################################################################
### /Tests for TestLoader.loadTestsFromTestCase
@@ -123,6 +121,21 @@ class Test_TestLoader(unittest.TestCase):
expected = [loader.suiteClass([MyTestCase('test')])]
self.assertEqual(list(suite), expected)
# TODO: RUSTPYTHON
@unittest.expectedFailure
# "This test ensures that internal `TestCase` subclasses are not loaded"
def test_loadTestsFromModule__TestCase_subclass_internals(self):
# See https://github.com/python/cpython/issues/84867
m = types.ModuleType('m')
# Simulate imported names:
m.TestCase = unittest.TestCase
m.FunctionTestCase = unittest.FunctionTestCase
loader = unittest.TestLoader()
suite = loader.loadTestsFromModule(m)
self.assertIsInstance(suite, loader.suiteClass)
self.assertEqual(list(suite), [])
# "This method searches `module` for classes derived from TestCase"
#
# What happens if no tests are found (no TestCase instances)?
@@ -174,9 +187,10 @@ class Test_TestLoader(unittest.TestCase):
self.assertEqual(list(suite), reference)
# Check that loadTestsFromModule honors (or not) a module
# TODO: RUSTPYTHON
@unittest.expectedFailure
# Check that loadTestsFromModule honors a module
# with a load_tests function.
@warningregistry
def test_loadTestsFromModule__load_tests(self):
m = types.ModuleType('m')
class MyTestCase(unittest.TestCase):
@@ -195,126 +209,13 @@ class Test_TestLoader(unittest.TestCase):
suite = loader.loadTestsFromModule(m)
self.assertIsInstance(suite, unittest.TestSuite)
self.assertEqual(load_tests_args, [loader, suite, None])
# With Python 3.5, the undocumented and unofficial use_load_tests is
# ignored (and deprecated).
load_tests_args = []
with warnings.catch_warnings(record=False):
warnings.simplefilter('ignore')
suite = loader.loadTestsFromModule(m, use_load_tests=False)
self.assertEqual(load_tests_args, [loader, suite, None])
@warningregistry
def test_loadTestsFromModule__use_load_tests_deprecated_positional(self):
m = types.ModuleType('m')
class MyTestCase(unittest.TestCase):
def test(self):
pass
m.testcase_1 = MyTestCase
load_tests_args = []
def load_tests(loader, tests, pattern):
self.assertIsInstance(tests, unittest.TestSuite)
load_tests_args.extend((loader, tests, pattern))
return tests
m.load_tests = load_tests
# The method still works.
loader = unittest.TestLoader()
# use_load_tests=True as a positional argument.
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter('always')
suite = loader.loadTestsFromModule(m, False)
self.assertIsInstance(suite, unittest.TestSuite)
# load_tests was still called because use_load_tests is deprecated
# and ignored.
self.assertEqual(load_tests_args, [loader, suite, None])
# We got a warning.
self.assertIs(w[-1].category, DeprecationWarning)
self.assertEqual(str(w[-1].message),
'use_load_tests is deprecated and ignored')
@warningregistry
def test_loadTestsFromModule__use_load_tests_deprecated_keyword(self):
m = types.ModuleType('m')
class MyTestCase(unittest.TestCase):
def test(self):
pass
m.testcase_1 = MyTestCase
load_tests_args = []
def load_tests(loader, tests, pattern):
self.assertIsInstance(tests, unittest.TestSuite)
load_tests_args.extend((loader, tests, pattern))
return tests
m.load_tests = load_tests
# The method still works.
loader = unittest.TestLoader()
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter('always')
suite = loader.loadTestsFromModule(m, use_load_tests=False)
self.assertIsInstance(suite, unittest.TestSuite)
# load_tests was still called because use_load_tests is deprecated
# and ignored.
self.assertEqual(load_tests_args, [loader, suite, None])
# We got a warning.
self.assertIs(w[-1].category, DeprecationWarning)
self.assertEqual(str(w[-1].message),
'use_load_tests is deprecated and ignored')
@warningregistry
def test_loadTestsFromModule__too_many_positional_args(self):
m = types.ModuleType('m')
class MyTestCase(unittest.TestCase):
def test(self):
pass
m.testcase_1 = MyTestCase
load_tests_args = []
def load_tests(loader, tests, pattern):
self.assertIsInstance(tests, unittest.TestSuite)
load_tests_args.extend((loader, tests, pattern))
return tests
m.load_tests = load_tests
loader = unittest.TestLoader()
with self.assertRaises(TypeError) as cm, \
warnings.catch_warnings(record=True) as w:
warnings.simplefilter('always')
loader.loadTestsFromModule(m, False, 'testme.*')
# We still got the deprecation warning.
self.assertIs(w[-1].category, DeprecationWarning)
self.assertEqual(str(w[-1].message),
'use_load_tests is deprecated and ignored')
# We also got a TypeError for too many positional arguments.
self.assertEqual(type(cm.exception), TypeError)
self.assertEqual(
str(cm.exception),
'loadTestsFromModule() takes 1 positional argument but 3 were given')
@warningregistry
def test_loadTestsFromModule__use_load_tests_other_bad_keyword(self):
m = types.ModuleType('m')
class MyTestCase(unittest.TestCase):
def test(self):
pass
m.testcase_1 = MyTestCase
load_tests_args = []
def load_tests(loader, tests, pattern):
self.assertIsInstance(tests, unittest.TestSuite)
load_tests_args.extend((loader, tests, pattern))
return tests
m.load_tests = load_tests
loader = unittest.TestLoader()
with warnings.catch_warnings():
warnings.simplefilter('ignore')
with self.assertRaises(TypeError) as cm:
loader.loadTestsFromModule(
m, use_load_tests=False, very_bad=True, worse=False)
self.assertEqual(type(cm.exception), TypeError)
# The error message names the first bad argument alphabetically,
# however use_load_tests (which sorts first) is ignored.
self.assertEqual(
str(cm.exception),
"loadTestsFromModule() got an unexpected keyword argument 'very_bad'")
# In Python 3.12, the undocumented and unofficial use_load_tests has
# been removed.
with self.assertRaises(TypeError):
loader.loadTestsFromModule(m, False)
with self.assertRaises(TypeError):
loader.loadTestsFromModule(m, use_load_tests=False)
def test_loadTestsFromModule__pattern(self):
m = types.ModuleType('m')
@@ -716,7 +617,7 @@ class Test_TestLoader(unittest.TestCase):
# We're going to try to load this module as a side-effect, so it
# better not be loaded before we try.
#
module_name = 'unittest.test.dummy'
module_name = 'test.test_unittest.dummy'
sys.modules.pop(module_name, None)
loader = unittest.TestLoader()
@@ -844,7 +745,7 @@ class Test_TestLoader(unittest.TestCase):
loader = unittest.TestLoader()
suite = loader.loadTestsFromNames(
['unittest.loader.sdasfasfasdf', 'unittest.test.dummy'])
['unittest.loader.sdasfasfasdf', 'test.test_unittest.dummy'])
error, test = self.check_deferred_error(loader, list(suite)[0])
expected = "module 'unittest.loader' has no attribute 'sdasfasfasdf'"
self.assertIn(
@@ -1141,7 +1042,7 @@ class Test_TestLoader(unittest.TestCase):
# We're going to try to load this module as a side-effect, so it
# better not be loaded before we try.
#
module_name = 'unittest.test.dummy'
module_name = 'test.test_unittest.dummy'
sys.modules.pop(module_name, None)
loader = unittest.TestLoader()
@@ -1604,39 +1505,6 @@ class TestObsoleteFunctions(unittest.TestCase):
def reverse_three_way_cmp(a, b):
return unittest.util.three_way_cmp(b, a)
def test_getTestCaseNames(self):
with self.assertWarns(DeprecationWarning) as w:
tests = unittest.getTestCaseNames(self.MyTestCase,
prefix='check', sortUsing=self.reverse_three_way_cmp,
testNamePatterns=None)
self.assertEqual(w.filename, __file__)
self.assertEqual(tests, ['check_2', 'check_1'])
def test_makeSuite(self):
with self.assertWarns(DeprecationWarning) as w:
suite = unittest.makeSuite(self.MyTestCase,
prefix='check', sortUsing=self.reverse_three_way_cmp,
suiteClass=self.MyTestSuite)
self.assertEqual(w.filename, __file__)
self.assertIsInstance(suite, self.MyTestSuite)
expected = self.MyTestSuite([self.MyTestCase('check_2'),
self.MyTestCase('check_1')])
self.assertEqual(suite, expected)
def test_findTestCases(self):
m = types.ModuleType('m')
m.testcase_1 = self.MyTestCase
with self.assertWarns(DeprecationWarning) as w:
suite = unittest.findTestCases(m,
prefix='check', sortUsing=self.reverse_three_way_cmp,
suiteClass=self.MyTestSuite)
self.assertEqual(w.filename, __file__)
self.assertIsInstance(suite, self.MyTestSuite)
expected = [self.MyTestSuite([self.MyTestCase('check_2'),
self.MyTestCase('check_1')])]
self.assertEqual(list(suite), expected)
if __name__ == "__main__":
unittest.main()
@@ -1,21 +1,20 @@
import io
import os
import sys
import subprocess
from test import support
import unittest
import unittest.test
from unittest.test.test_result import BufferedWriter
import test.test_unittest
from test.test_unittest.test_result import BufferedWriter
@support.force_not_colorized_test_class
class Test_TestProgram(unittest.TestCase):
def test_discovery_from_dotted_path(self):
loader = unittest.TestLoader()
tests = [self]
expectedPath = os.path.abspath(os.path.dirname(unittest.test.__file__))
expectedPath = os.path.abspath(os.path.dirname(test.test_unittest.__file__))
self.wasRun = False
def _find_tests(start_dir, pattern):
@@ -23,7 +22,7 @@ class Test_TestProgram(unittest.TestCase):
self.assertEqual(start_dir, expectedPath)
return tests
loader._find_tests = _find_tests
suite = loader.discover('unittest.test')
suite = loader.discover('test.test_unittest')
self.assertTrue(self.wasRun)
self.assertEqual(suite._tests, tests)
@@ -73,15 +72,22 @@ class Test_TestProgram(unittest.TestCase):
def testUnexpectedSuccess(self):
pass
class FooBarLoader(unittest.TestLoader):
"""Test loader that returns a suite containing FooBar."""
class Empty(unittest.TestCase):
pass
class TestLoader(unittest.TestLoader):
"""Test loader that returns a suite containing the supplied testcase."""
def __init__(self, testcase):
self.testcase = testcase
def loadTestsFromModule(self, module):
return self.suiteClass(
[self.loadTestsFromTestCase(Test_TestProgram.FooBar)])
[self.loadTestsFromTestCase(self.testcase)])
def loadTestsFromNames(self, names, module):
return self.suiteClass(
[self.loadTestsFromTestCase(Test_TestProgram.FooBar)])
[self.loadTestsFromTestCase(self.testcase)])
def test_defaultTest_with_string(self):
class FakeRunner(object):
@@ -93,10 +99,10 @@ class Test_TestProgram(unittest.TestCase):
sys.argv = ['faketest']
runner = FakeRunner()
program = unittest.TestProgram(testRunner=runner, exit=False,
defaultTest='unittest.test',
testLoader=self.FooBarLoader())
defaultTest='test.test_unittest',
testLoader=self.TestLoader(self.FooBar))
sys.argv = old_argv
self.assertEqual(('unittest.test',), program.testNames)
self.assertEqual(('test.test_unittest',), program.testNames)
def test_defaultTest_with_iterable(self):
class FakeRunner(object):
@@ -109,10 +115,10 @@ class Test_TestProgram(unittest.TestCase):
runner = FakeRunner()
program = unittest.TestProgram(
testRunner=runner, exit=False,
defaultTest=['unittest.test', 'unittest.test2'],
testLoader=self.FooBarLoader())
defaultTest=['test.test_unittest', 'test.test_unittest2'],
testLoader=self.TestLoader(self.FooBar))
sys.argv = old_argv
self.assertEqual(['unittest.test', 'unittest.test2'],
self.assertEqual(['test.test_unittest', 'test.test_unittest2'],
program.testNames)
def test_NonExit(self):
@@ -120,7 +126,7 @@ class Test_TestProgram(unittest.TestCase):
program = unittest.main(exit=False,
argv=["foobar"],
testRunner=unittest.TextTestRunner(stream=stream),
testLoader=self.FooBarLoader())
testLoader=self.TestLoader(self.FooBar))
self.assertTrue(hasattr(program, 'result'))
out = stream.getvalue()
self.assertIn('\nFAIL: testFail ', out)
@@ -132,13 +138,13 @@ class Test_TestProgram(unittest.TestCase):
def test_Exit(self):
stream = BufferedWriter()
self.assertRaises(
SystemExit,
unittest.main,
argv=["foobar"],
testRunner=unittest.TextTestRunner(stream=stream),
exit=True,
testLoader=self.FooBarLoader())
with self.assertRaises(SystemExit) as cm:
unittest.main(
argv=["foobar"],
testRunner=unittest.TextTestRunner(stream=stream),
exit=True,
testLoader=self.TestLoader(self.FooBar))
self.assertEqual(cm.exception.code, 1)
out = stream.getvalue()
self.assertIn('\nFAIL: testFail ', out)
self.assertIn('\nERROR: testError ', out)
@@ -149,12 +155,11 @@ class Test_TestProgram(unittest.TestCase):
def test_ExitAsDefault(self):
stream = BufferedWriter()
self.assertRaises(
SystemExit,
unittest.main,
argv=["foobar"],
testRunner=unittest.TextTestRunner(stream=stream),
testLoader=self.FooBarLoader())
with self.assertRaises(SystemExit):
unittest.main(
argv=["foobar"],
testRunner=unittest.TextTestRunner(stream=stream),
testLoader=self.TestLoader(self.FooBar))
out = stream.getvalue()
self.assertIn('\nFAIL: testFail ', out)
self.assertIn('\nERROR: testError ', out)
@@ -163,6 +168,29 @@ class Test_TestProgram(unittest.TestCase):
'expected failures=1, unexpected successes=1)\n')
self.assertTrue(out.endswith(expected))
def test_ExitSkippedSuite(self):
stream = BufferedWriter()
with self.assertRaises(SystemExit) as cm:
unittest.main(
argv=["foobar", "-k", "testSkipped"],
testRunner=unittest.TextTestRunner(stream=stream),
testLoader=self.TestLoader(self.FooBar))
self.assertEqual(cm.exception.code, 0)
out = stream.getvalue()
expected = '\n\nOK (skipped=1)\n'
self.assertTrue(out.endswith(expected))
def test_ExitEmptySuite(self):
stream = BufferedWriter()
with self.assertRaises(SystemExit) as cm:
unittest.main(
argv=["empty"],
testRunner=unittest.TextTestRunner(stream=stream),
testLoader=self.TestLoader(self.Empty))
self.assertEqual(cm.exception.code, 5)
out = stream.getvalue()
self.assertIn('\nNO TESTS RAN\n', out)
class InitialisableProgram(unittest.TestProgram):
exit = False
@@ -286,6 +314,7 @@ class TestCommandLineArgs(unittest.TestCase):
program.failfast = 'failfast'
program.buffer = 'buffer'
program.warnings = 'warnings'
program.durations = '5'
program.runTests()
@@ -293,7 +322,8 @@ class TestCommandLineArgs(unittest.TestCase):
'failfast': 'failfast',
'buffer': 'buffer',
'tb_locals': False,
'warnings': 'warnings'})
'warnings': 'warnings',
'durations': '5'})
self.assertEqual(FakeRunner.test, 'test')
self.assertIs(program.result, RESULT)
@@ -322,7 +352,8 @@ class TestCommandLineArgs(unittest.TestCase):
'failfast': False,
'tb_locals': True,
'verbosity': 1,
'warnings': None})
'warnings': None,
'durations': None})
def testRunTestsOldRunnerClass(self):
program = self.program
@@ -335,6 +366,7 @@ class TestCommandLineArgs(unittest.TestCase):
program.failfast = 'failfast'
program.buffer = 'buffer'
program.test = 'test'
program.durations = '0'
program.runTests()
@@ -358,6 +390,7 @@ class TestCommandLineArgs(unittest.TestCase):
program = self.program
program.catchbreak = True
program.durations = None
program.testRunner = FakeRunner
@@ -427,8 +460,8 @@ class TestCommandLineArgs(unittest.TestCase):
def testParseArgsAbsolutePathsThatCannotBeConverted(self):
program = self.program
# even on Windows '/...' is considered absolute by os.path.abspath
argv = ['progname', '/foo/bar/baz.py', '/green/red.py']
drive = os.path.splitdrive(os.getcwd())[0]
argv = ['progname', f'{drive}/foo/bar/baz.py', f'{drive}/green/red.py']
self._patch_isfile(argv)
program.createTests = lambda: None
@@ -452,6 +485,8 @@ class TestCommandLineArgs(unittest.TestCase):
self.assertEqual(program.testNamePatterns, ['*foo*', '*bar*', '*pat*'])
# TODO: RUSTPYTHON
@unittest.expectedFailureIf(sys.platform != 'win32', "TODO: RUSTPYTHON")
def testSelectedTestNamesFunctionalTest(self):
def run_unittest(args):
# Use -E to ignore PYTHONSAFEPATH env var
@@ -463,14 +498,14 @@ class TestCommandLineArgs(unittest.TestCase):
return stderr.decode()
t = '_test_warnings'
self.assertIn('Ran 7 tests', run_unittest([t]))
self.assertIn('Ran 7 tests', run_unittest(['-k', 'TestWarnings', t]))
self.assertIn('Ran 7 tests', run_unittest(['discover', '-p', '*_test*', '-k', 'TestWarnings']))
self.assertIn('Ran 2 tests', run_unittest(['-k', 'f', t]))
self.assertIn('Ran 7 tests', run_unittest(['-k', 't', t]))
self.assertIn('Ran 3 tests', run_unittest(['-k', '*t', t]))
self.assertIn('Ran 7 tests', run_unittest(['-k', '*test_warnings.*Warning*', t]))
self.assertIn('Ran 1 test', run_unittest(['-k', '*test_warnings.*warning*', t]))
self.assertIn('Ran 5 tests', run_unittest([t]))
self.assertIn('Ran 5 tests', run_unittest(['-k', 'TestWarnings', t]))
self.assertIn('Ran 5 tests', run_unittest(['discover', '-p', '*_test*', '-k', 'TestWarnings']))
self.assertIn('Ran 1 test ', run_unittest(['-k', 'f', t]))
self.assertIn('Ran 5 tests', run_unittest(['-k', 't', t]))
self.assertIn('Ran 2 tests', run_unittest(['-k', '*t', t]))
self.assertIn('Ran 5 tests', run_unittest(['-k', '*test_warnings.*Warning*', t]))
self.assertIn('Ran 1 test ', run_unittest(['-k', '*test_warnings.*warning*', t]))
if __name__ == '__main__':
@@ -1,12 +1,12 @@
import io
import sys
import textwrap
from test.support import warnings_helper, captured_stdout, captured_stderr
import traceback
import unittest
from unittest.util import strclass
from test.support import warnings_helper
from test.support import captured_stdout, force_not_colorized_test_class
from test.test_unittest.support import BufferedWriter
class MockTraceback(object):
@@ -33,22 +33,7 @@ def bad_cleanup2():
raise ValueError('bad cleanup2')
class BufferedWriter:
def __init__(self):
self.result = ''
self.buffer = ''
def write(self, arg):
self.buffer += arg
def flush(self):
self.result += self.buffer
self.buffer = ''
def getvalue(self):
return self.result
@force_not_colorized_test_class
class Test_TestResult(unittest.TestCase):
# Note: there are not separate tests for TestResult.wasSuccessful(),
# TestResult.errors, TestResult.failures, TestResult.testsRun or
@@ -465,12 +450,14 @@ class Test_TestResult(unittest.TestCase):
stream = BufferedWriter()
runner = unittest.TextTestRunner(stream=stream, failfast=True)
def test(result):
result.testsRun += 1
self.assertTrue(result.failfast)
result = runner.run(test)
stream.flush()
self.assertTrue(stream.getvalue().endswith('\n\nOK\n'))
@force_not_colorized_test_class
class Test_TextTestResult(unittest.TestCase):
maxDiff = None
@@ -505,8 +492,8 @@ class Test_TextTestResult(unittest.TestCase):
'(' + __name__ + '.Test_TextTestResult.testGetSubTestDescriptionWithoutDocstringAndParams) '
'(<subtest>)')
def testGetSubTestDescriptionForFalsyValues(self):
expected = 'testGetSubTestDescriptionForFalsyValues (%s.Test_TextTestResult.testGetSubTestDescriptionForFalsyValues) [%s]'
def testGetSubTestDescriptionForFalseValues(self):
expected = 'testGetSubTestDescriptionForFalseValues (%s.Test_TextTestResult.testGetSubTestDescriptionForFalseValues) [%s]'
result = unittest.TextTestResult(None, True, 1)
for arg in [0, None, []]:
with self.subTest(arg):
@@ -772,6 +759,7 @@ class Test_OldTestResult(unittest.TestCase):
runner.run(Test('testFoo'))
@force_not_colorized_test_class
class TestOutputBuffering(unittest.TestCase):
def setUp(self):
@@ -8,8 +8,11 @@ from test import support
import unittest
from unittest.case import _Outcome
from unittest.test.support import (LoggingResult,
ResultWithNoStartTestRunStopTestRun)
from test.test_unittest.support import (
BufferedWriter,
LoggingResult,
ResultWithNoStartTestRunStopTestRun,
)
def resultFactory(*_):
@@ -21,6 +24,13 @@ def getRunner():
stream=io.StringIO())
class CustomError(Exception):
pass
# For test output compat:
CustomErrorRepr = f"{__name__ + '.' if __name__ != '__main__' else ''}CustomError"
def runTests(*cases):
suite = unittest.TestSuite()
for case in cases:
@@ -43,7 +53,7 @@ def cleanup(ordering, blowUp=False):
ordering.append('cleanup_good')
else:
ordering.append('cleanup_exc')
raise Exception('CleanUpExc')
raise CustomError('CleanUpExc')
class TestCM:
@@ -96,6 +106,7 @@ class TestCleanUp(unittest.TestCase):
self.assertTrue(test.doCleanups())
self.assertEqual(cleanups, [(2, (), {}), (1, (1, 2, 3), dict(four='hello', five='goodbye'))])
@support.force_not_colorized
def testCleanUpWithErrors(self):
class TestableTest(unittest.TestCase):
def testNothing(self):
@@ -105,8 +116,8 @@ class TestCleanUp(unittest.TestCase):
result = unittest.TestResult()
outcome = test._outcome = _Outcome(result=result)
CleanUpExc = Exception('foo')
exc2 = Exception('bar')
CleanUpExc = CustomError('foo')
exc2 = CustomError('bar')
def cleanup1():
raise CleanUpExc
@@ -122,10 +133,10 @@ class TestCleanUp(unittest.TestCase):
(_, msg2), (_, msg1) = result.errors
self.assertIn('in cleanup1', msg1)
self.assertIn('raise CleanUpExc', msg1)
self.assertIn('Exception: foo', msg1)
self.assertIn(f'{CustomErrorRepr}: foo', msg1)
self.assertIn('in cleanup2', msg2)
self.assertIn('raise exc2', msg2)
self.assertIn('Exception: bar', msg2)
self.assertIn(f'{CustomErrorRepr}: bar', msg2)
def testCleanupInRun(self):
blowUp = False
@@ -136,7 +147,7 @@ class TestCleanUp(unittest.TestCase):
ordering.append('setUp')
test.addCleanup(cleanup2)
if blowUp:
raise Exception('foo')
raise CustomError('foo')
def testNothing(self):
ordering.append('test')
@@ -239,6 +250,7 @@ class TestCleanUp(unittest.TestCase):
self.assertEqual(test._cleanups, [])
@support.force_not_colorized_test_class
class TestClassCleanup(unittest.TestCase):
def test_addClassCleanUp(self):
class TestableTest(unittest.TestCase):
@@ -277,7 +289,7 @@ class TestClassCleanup(unittest.TestCase):
ordering.append('setUpClass')
cls.addClassCleanup(cleanup, ordering)
if blowUp:
raise Exception()
raise CustomError()
def testNothing(self):
ordering.append('test')
@classmethod
@@ -303,7 +315,7 @@ class TestClassCleanup(unittest.TestCase):
ordering.append('setUpClass')
cls.addClassCleanup(cleanup, ordering)
if blowUp:
raise Exception()
raise CustomError()
def testNothing(self):
ordering.append('test')
@classmethod
@@ -343,7 +355,7 @@ class TestClassCleanup(unittest.TestCase):
ordering = []
blowUp = True
suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestableTest)
with self.assertRaises(Exception) as cm:
with self.assertRaises(CustomError) as cm:
suite.debug()
self.assertEqual(str(cm.exception), 'CleanUpExc')
self.assertEqual(ordering,
@@ -363,10 +375,10 @@ class TestClassCleanup(unittest.TestCase):
@classmethod
def tearDownClass(cls):
ordering.append('tearDownClass')
raise Exception('TearDownClassExc')
raise CustomError('TearDownClassExc')
suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestableTest)
with self.assertRaises(Exception) as cm:
with self.assertRaises(CustomError) as cm:
suite.debug()
self.assertEqual(str(cm.exception), 'TearDownClassExc')
self.assertEqual(ordering, ['setUpClass', 'test', 'tearDownClass'])
@@ -376,7 +388,7 @@ class TestClassCleanup(unittest.TestCase):
ordering = []
blowUp = True
suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestableTest)
with self.assertRaises(Exception) as cm:
with self.assertRaises(CustomError) as cm:
suite.debug()
self.assertEqual(str(cm.exception), 'TearDownClassExc')
self.assertEqual(ordering, ['setUpClass', 'test', 'tearDownClass'])
@@ -389,16 +401,22 @@ class TestClassCleanup(unittest.TestCase):
pass
def cleanup1():
raise Exception('cleanup1')
raise CustomError('cleanup1')
def cleanup2():
raise Exception('cleanup2')
raise CustomError('cleanup2')
TestableTest.addClassCleanup(cleanup1)
TestableTest.addClassCleanup(cleanup2)
with self.assertRaises(Exception) as e:
TestableTest.doClassCleanups()
self.assertEqual(e, 'cleanup1')
TestableTest.doClassCleanups()
self.assertEqual(len(TestableTest.tearDown_exceptions), 2)
e1, e2 = TestableTest.tearDown_exceptions
self.assertIsInstance(e1[1], CustomError)
self.assertEqual(str(e1[1]), 'cleanup2')
self.assertIsInstance(e2[1], CustomError)
self.assertEqual(str(e2[1]), 'cleanup1')
def test_with_errors_addCleanUp(self):
ordering = []
@@ -418,7 +436,7 @@ class TestClassCleanup(unittest.TestCase):
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering,
['setUpClass', 'setUp', 'cleanup_exc',
'tearDownClass', 'cleanup_good'])
@@ -441,7 +459,7 @@ class TestClassCleanup(unittest.TestCase):
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering,
['setUpClass', 'setUp', 'test', 'cleanup_good',
'tearDownClass', 'cleanup_exc'])
@@ -457,11 +475,11 @@ class TestClassCleanup(unittest.TestCase):
ordering.append('setUpClass')
cls.addClassCleanup(cleanup, ordering, blowUp=True)
if class_blow_up:
raise Exception('ClassExc')
raise CustomError('ClassExc')
def setUp(self):
ordering.append('setUp')
if method_blow_up:
raise Exception('MethodExc')
raise CustomError('MethodExc')
def testNothing(self):
ordering.append('test')
@classmethod
@@ -470,7 +488,7 @@ class TestClassCleanup(unittest.TestCase):
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering,
['setUpClass', 'setUp', 'test',
'tearDownClass', 'cleanup_exc'])
@@ -480,9 +498,9 @@ class TestClassCleanup(unittest.TestCase):
method_blow_up = False
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: ClassExc')
f'{CustomErrorRepr}: ClassExc')
self.assertEqual(result.errors[1][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering,
['setUpClass', 'cleanup_exc'])
@@ -491,9 +509,9 @@ class TestClassCleanup(unittest.TestCase):
method_blow_up = True
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: MethodExc')
f'{CustomErrorRepr}: MethodExc')
self.assertEqual(result.errors[1][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering,
['setUpClass', 'setUp', 'tearDownClass',
'cleanup_exc'])
@@ -510,11 +528,11 @@ class TestClassCleanup(unittest.TestCase):
@classmethod
def tearDownClass(cls):
ordering.append('tearDownClass')
raise Exception('TearDownExc')
raise CustomError('TearDownExc')
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: TearDownExc')
f'{CustomErrorRepr}: TearDownExc')
self.assertEqual(ordering,
['setUpClass', 'test', 'tearDownClass', 'cleanup_good'])
@@ -574,7 +592,18 @@ class TestClassCleanup(unittest.TestCase):
'inner setup', 'inner test', 'inner cleanup',
'end outer test', 'outer cleanup'])
def test_run_empty_suite_error_message(self):
class EmptyTest(unittest.TestCase):
pass
suite = unittest.defaultTestLoader.loadTestsFromTestCase(EmptyTest)
runner = getRunner()
runner.run(suite)
self.assertIn("\nNO TESTS RAN\n", runner.stream.getvalue())
@support.force_not_colorized_test_class
class TestModuleCleanUp(unittest.TestCase):
def test_add_and_do_ModuleCleanup(self):
module_cleanups = []
@@ -607,7 +636,7 @@ class TestModuleCleanUp(unittest.TestCase):
module_cleanups.append((3, args, kwargs))
def module_cleanup_bad(*args, **kwargs):
raise Exception('CleanUpExc')
raise CustomError('CleanUpExc')
class Module(object):
unittest.addModuleCleanup(module_cleanup_good, 1, 2, 3,
@@ -617,7 +646,7 @@ class TestModuleCleanUp(unittest.TestCase):
[(module_cleanup_good, (1, 2, 3),
dict(four='hello', five='goodbye')),
(module_cleanup_bad, (), {})])
with self.assertRaises(Exception) as e:
with self.assertRaises(CustomError) as e:
unittest.case.doModuleCleanups()
self.assertEqual(str(e.exception), 'CleanUpExc')
self.assertEqual(unittest.case._module_cleanups, [])
@@ -646,7 +675,7 @@ class TestModuleCleanUp(unittest.TestCase):
ordering.append('setUpModule')
unittest.addModuleCleanup(cleanup, ordering)
if blowUp:
raise Exception('setUpModule Exc')
raise CustomError('setUpModule Exc')
@staticmethod
def tearDownModule():
ordering.append('tearDownModule')
@@ -666,7 +695,7 @@ class TestModuleCleanUp(unittest.TestCase):
result = runTests(TestableTest)
self.assertEqual(ordering, ['setUpModule', 'cleanup_good'])
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: setUpModule Exc')
f'{CustomErrorRepr}: setUpModule Exc')
ordering = []
blowUp = False
@@ -686,7 +715,7 @@ class TestModuleCleanUp(unittest.TestCase):
ordering.append('setUpModule')
unittest.addModuleCleanup(cleanup, ordering)
if blowUp:
raise Exception()
raise CustomError()
@staticmethod
def tearDownModule():
ordering.append('tearDownModule')
@@ -697,7 +726,7 @@ class TestModuleCleanUp(unittest.TestCase):
ordering.append('setUpModule2')
unittest.addModuleCleanup(cleanup, ordering)
if blowUp2:
raise Exception()
raise CustomError()
@staticmethod
def tearDownModule():
ordering.append('tearDownModule2')
@@ -786,7 +815,7 @@ class TestModuleCleanUp(unittest.TestCase):
@staticmethod
def tearDownModule():
ordering.append('tearDownModule')
raise Exception('CleanUpExc')
raise CustomError('CleanUpExc')
class TestableTest(unittest.TestCase):
@classmethod
@@ -802,7 +831,7 @@ class TestModuleCleanUp(unittest.TestCase):
sys.modules['Module'] = Module
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering, ['setUpModule', 'setUpClass', 'test',
'tearDownClass', 'tearDownModule',
'cleanup_good'])
@@ -842,7 +871,7 @@ class TestModuleCleanUp(unittest.TestCase):
ordering = []
blowUp = True
suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestableTest)
with self.assertRaises(Exception) as cm:
with self.assertRaises(CustomError) as cm:
suite.debug()
self.assertEqual(str(cm.exception), 'CleanUpExc')
self.assertEqual(ordering, ['setUpModule', 'setUpClass', 'test',
@@ -860,7 +889,7 @@ class TestModuleCleanUp(unittest.TestCase):
@staticmethod
def tearDownModule():
ordering.append('tearDownModule')
raise Exception('TearDownModuleExc')
raise CustomError('TearDownModuleExc')
class TestableTest(unittest.TestCase):
@classmethod
@@ -875,7 +904,7 @@ class TestModuleCleanUp(unittest.TestCase):
TestableTest.__module__ = 'Module'
sys.modules['Module'] = Module
suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestableTest)
with self.assertRaises(Exception) as cm:
with self.assertRaises(CustomError) as cm:
suite.debug()
self.assertEqual(str(cm.exception), 'TearDownModuleExc')
self.assertEqual(ordering, ['setUpModule', 'setUpClass', 'test',
@@ -886,7 +915,7 @@ class TestModuleCleanUp(unittest.TestCase):
ordering = []
blowUp = True
suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestableTest)
with self.assertRaises(Exception) as cm:
with self.assertRaises(CustomError) as cm:
suite.debug()
self.assertEqual(str(cm.exception), 'TearDownModuleExc')
self.assertEqual(ordering, ['setUpModule', 'setUpClass', 'test',
@@ -965,7 +994,7 @@ class TestModuleCleanUp(unittest.TestCase):
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering,
['setUpModule', 'setUpClass', 'test', 'tearDownClass',
'cleanup_exc', 'tearDownModule', 'cleanup_good'])
@@ -995,7 +1024,7 @@ class TestModuleCleanUp(unittest.TestCase):
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering,
['setUpModule', 'setUp', 'test', 'tearDown',
'cleanup_exc', 'tearDownModule', 'cleanup_good'])
@@ -1011,7 +1040,7 @@ class TestModuleCleanUp(unittest.TestCase):
ordering.append('setUpModule')
unittest.addModuleCleanup(cleanup, ordering, blowUp=True)
if module_blow_up:
raise Exception('ModuleExc')
raise CustomError('ModuleExc')
@staticmethod
def tearDownModule():
ordering.append('tearDownModule')
@@ -1021,11 +1050,11 @@ class TestModuleCleanUp(unittest.TestCase):
def setUpClass(cls):
ordering.append('setUpClass')
if class_blow_up:
raise Exception('ClassExc')
raise CustomError('ClassExc')
def setUp(self):
ordering.append('setUp')
if method_blow_up:
raise Exception('MethodExc')
raise CustomError('MethodExc')
def testNothing(self):
ordering.append('test')
@classmethod
@@ -1037,7 +1066,7 @@ class TestModuleCleanUp(unittest.TestCase):
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering,
['setUpModule', 'setUpClass', 'setUp', 'test',
'tearDownClass', 'tearDownModule',
@@ -1049,9 +1078,9 @@ class TestModuleCleanUp(unittest.TestCase):
method_blow_up = False
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: ModuleExc')
f'{CustomErrorRepr}: ModuleExc')
self.assertEqual(result.errors[1][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering, ['setUpModule', 'cleanup_exc'])
ordering = []
@@ -1060,9 +1089,9 @@ class TestModuleCleanUp(unittest.TestCase):
method_blow_up = False
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: ClassExc')
f'{CustomErrorRepr}: ClassExc')
self.assertEqual(result.errors[1][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering, ['setUpModule', 'setUpClass',
'tearDownModule', 'cleanup_exc'])
@@ -1072,9 +1101,9 @@ class TestModuleCleanUp(unittest.TestCase):
method_blow_up = True
result = runTests(TestableTest)
self.assertEqual(result.errors[0][1].splitlines()[-1],
'Exception: MethodExc')
f'{CustomErrorRepr}: MethodExc')
self.assertEqual(result.errors[1][1].splitlines()[-1],
'Exception: CleanUpExc')
f'{CustomErrorRepr}: CleanUpExc')
self.assertEqual(ordering, ['setUpModule', 'setUpClass', 'setUp',
'tearDownClass', 'tearDownModule',
'cleanup_exc'])
@@ -1176,6 +1205,7 @@ class Test_TextTestRunner(unittest.TestCase):
self.assertTrue(runner.descriptions)
self.assertEqual(runner.resultclass, unittest.TextTestResult)
self.assertFalse(runner.tb_locals)
self.assertIsNone(runner.durations)
def test_multiple_inheritance(self):
class AResult(unittest.TestResult):
@@ -1267,7 +1297,7 @@ class Test_TextTestRunner(unittest.TestCase):
expected = ['startTestRun', 'stopTestRun']
self.assertEqual(events, expected)
# TODO: RUSTPYTHON; fix pickling with io objects
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_pickle_unpickle(self):
# Issue #7197: a TextTestRunner should be (un)pickleable. This is
@@ -1293,6 +1323,7 @@ class Test_TextTestRunner(unittest.TestCase):
expectedresult = (runner.stream, DESCRIPTIONS, VERBOSITY)
self.assertEqual(runner._makeResult(), expectedresult)
@support.force_not_colorized
@support.requires_subprocess()
def test_warnings(self):
"""
@@ -1305,8 +1336,6 @@ class Test_TextTestRunner(unittest.TestCase):
return [b.splitlines() for b in p.communicate()]
opts = dict(stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=os.path.dirname(__file__))
ae_msg = b'Please use assertEqual instead.'
at_msg = b'Please use assertTrue instead.'
# no args -> all the warnings are printed, unittest warnings only once
p = subprocess.Popen([sys.executable, '-E', '_test_warnings.py'], **opts)
@@ -1314,11 +1343,11 @@ class Test_TextTestRunner(unittest.TestCase):
out, err = get_parse_out_err(p)
self.assertIn(b'OK', err)
# check that the total number of warnings in the output is correct
self.assertEqual(len(out), 12)
self.assertEqual(len(out), 10)
# check that the numbers of the different kind of warnings is correct
for msg in [b'dw', b'iw', b'uw']:
self.assertEqual(out.count(msg), 3)
for msg in [ae_msg, at_msg, b'rw']:
for msg in [b'rw']:
self.assertEqual(out.count(msg), 1)
args_list = (
@@ -1345,11 +1374,9 @@ class Test_TextTestRunner(unittest.TestCase):
with p:
out, err = get_parse_out_err(p)
self.assertIn(b'OK', err)
self.assertEqual(len(out), 14)
self.assertEqual(len(out), 12)
for msg in [b'dw', b'iw', b'uw', b'rw']:
self.assertEqual(out.count(msg), 3)
for msg in [ae_msg, at_msg]:
self.assertEqual(out.count(msg), 1)
def testStdErrLookedUpAtInstantiationTime(self):
# see issue 10786
@@ -1368,6 +1395,65 @@ class Test_TextTestRunner(unittest.TestCase):
runner = unittest.TextTestRunner(f)
self.assertTrue(runner.stream.stream is f)
def test_durations(self):
def run(test, *, expect_durations=True):
stream = BufferedWriter()
runner = unittest.TextTestRunner(stream=stream, durations=5, verbosity=2)
result = runner.run(test)
self.assertEqual(result.durations, 5)
stream.flush()
text = stream.getvalue()
regex = r"\n\d+.\d\d\ds"
if expect_durations:
self.assertEqual(len(result.collectedDurations), 1)
self.assertIn('Slowest test durations', text)
self.assertRegex(text, regex)
else:
self.assertEqual(len(result.collectedDurations), 0)
self.assertNotIn('Slowest test durations', text)
self.assertNotRegex(text, regex)
# success
class Foo(unittest.TestCase):
def test_1(self):
pass
run(Foo('test_1'), expect_durations=True)
# failure
class Foo(unittest.TestCase):
def test_1(self):
self.assertEqual(0, 1)
run(Foo('test_1'), expect_durations=True)
# error
class Foo(unittest.TestCase):
def test_1(self):
1 / 0
run(Foo('test_1'), expect_durations=True)
# error in setUp and tearDown
class Foo(unittest.TestCase):
def setUp(self):
1 / 0
tearDown = setUp
def test_1(self):
pass
run(Foo('test_1'), expect_durations=True)
# skip (expect no durations)
class Foo(unittest.TestCase):
@unittest.skip("reason")
def test_1(self):
pass
run(Foo('test_1'), expect_durations=False)
if __name__ == "__main__":
unittest.main()
File renamed without changes.
@@ -1,6 +1,6 @@
import unittest
from unittest.test.support import LoggingResult
from test.test_unittest.support import LoggingResult
class Test_TestSkipping(unittest.TestCase):
@@ -3,7 +3,7 @@ import unittest
import gc
import sys
import weakref
from unittest.test.support import LoggingResult, TestEquality
from test.test_unittest.support import LoggingResult, TestEquality
### Support code for Test_TestSuite
+33
View File
@@ -0,0 +1,33 @@
import unittest
from unittest.util import safe_repr, sorted_list_difference, unorderable_list_difference
class TestUtil(unittest.TestCase):
def test_safe_repr(self):
class RaisingRepr:
def __repr__(self):
raise ValueError("Invalid repr()")
class LongRepr:
def __repr__(self):
return 'x' * 100
safe_repr(RaisingRepr())
self.assertEqual(safe_repr('foo'), "'foo'")
self.assertEqual(safe_repr(LongRepr(), short=True), 'x'*80 + ' [truncated]...')
def test_sorted_list_difference(self):
self.assertEqual(sorted_list_difference([], []), ([], []))
self.assertEqual(sorted_list_difference([1, 2], [2, 3]), ([1], [3]))
self.assertEqual(sorted_list_difference([1, 2], [1, 3]), ([2], [3]))
self.assertEqual(sorted_list_difference([1, 1, 1], [1, 2, 3]), ([], [2, 3]))
self.assertEqual(sorted_list_difference([4], [1, 2, 3, 4]), ([], [1, 2, 3]))
self.assertEqual(sorted_list_difference([1, 1], [2]), ([1], [2]))
self.assertEqual(sorted_list_difference([2], [1, 1]), ([2], [1]))
self.assertEqual(sorted_list_difference([1, 2], [1, 1]), ([2], []))
def test_unorderable_list_difference(self):
self.assertEqual(unorderable_list_difference([], []), ([], []))
self.assertEqual(unorderable_list_difference([1, 2], []), ([2, 1], []))
self.assertEqual(unorderable_list_difference([], [1, 2]), ([], [1, 2]))
self.assertEqual(unorderable_list_difference([1, 2], [1, 3]), ([2], [3]))
+6
View File
@@ -0,0 +1,6 @@
import os.path
from test.support import load_package_tests
def load_tests(*args):
return load_package_tests(os.path.dirname(__file__), *args)
@@ -6,7 +6,7 @@ def load_tests(loader, standard_tests, pattern):
# top level directory cached on loader instance
this_dir = os.path.dirname(__file__)
pattern = pattern or "test*.py"
# We are inside unittest.test.testmock, so the top-level is three notches up
# We are inside test.test_unittest.testmock, so the top-level is three notches up
top_level_dir = os.path.dirname(os.path.dirname(os.path.dirname(this_dir)))
package_tests = loader.discover(start_dir=this_dir, pattern=pattern,
top_level_dir=top_level_dir)
@@ -14,3 +14,14 @@ class SomeClass(object):
class X(object):
pass
# A standin for weurkzeug.local.LocalProxy - issue 119600
def _inaccessible(*args, **kwargs):
raise AttributeError
class OpaqueProxy:
__getattribute__ = _inaccessible
g = OpaqueProxy()
@@ -218,10 +218,6 @@ class AsyncAutospecTest(unittest.TestCase):
with self.assertRaises(RuntimeError):
create_autospec(async_func, instance=True)
@unittest.skip('Broken test from https://bugs.python.org/issue37251')
def test_create_autospec_awaitable_class(self):
self.assertIsInstance(create_autospec(AwaitableClass), AsyncMock)
def test_create_autospec(self):
spec = create_autospec(async_func_args)
awaitable = spec(1, 2, c=3)
@@ -236,7 +232,9 @@ class AsyncAutospecTest(unittest.TestCase):
run(main())
self.assertTrue(iscoroutinefunction(spec))
self.assertTrue(inspect.iscoroutinefunction(spec))
self.assertTrue(asyncio.iscoroutine(awaitable))
self.assertTrue(inspect.iscoroutine(awaitable))
self.assertEqual(spec.await_count, 1)
self.assertEqual(spec.await_args, call(1, 2, c=3))
self.assertEqual(spec.await_args_list, [call(1, 2, c=3)])
@@ -248,6 +246,25 @@ class AsyncAutospecTest(unittest.TestCase):
with self.assertRaises(AssertionError):
spec.assert_any_await(e=1)
def test_autospec_checks_signature(self):
spec = create_autospec(async_func_args)
# signature is not checked when called
awaitable = spec()
self.assertListEqual(spec.mock_calls, [])
async def main():
await awaitable
# but it is checked when awaited
with self.assertRaises(TypeError):
run(main())
# _checksig_ raises before running or awaiting the mock
self.assertListEqual(spec.mock_calls, [])
self.assertEqual(spec.await_count, 0)
self.assertIsNone(spec.await_args)
self.assertEqual(spec.await_args_list, [])
spec.assert_not_awaited()
def test_patch_with_autospec(self):
@@ -257,7 +274,9 @@ class AsyncAutospecTest(unittest.TestCase):
self.assertIsInstance(mock_method.mock, AsyncMock)
self.assertTrue(iscoroutinefunction(mock_method))
self.assertTrue(inspect.iscoroutinefunction(mock_method))
self.assertTrue(asyncio.iscoroutine(awaitable))
self.assertTrue(inspect.iscoroutine(awaitable))
self.assertTrue(inspect.isawaitable(awaitable))
# Verify the default values during mock setup
@@ -308,6 +327,21 @@ class AsyncSpecTest(unittest.TestCase):
with self.assertRaises(AttributeError):
mock.async_method
def test_spec_async_attributes_instance(self):
async_instance = AsyncClass()
async_instance.async_func_attr = async_func
async_instance.later_async_func_attr = normal_func
mock_async_instance = Mock(spec_set=async_instance)
async_instance.later_async_func_attr = async_func
self.assertIsInstance(mock_async_instance.async_func_attr, AsyncMock)
# only the shape of the spec at the time of mock construction matters
self.assertNotIsInstance(mock_async_instance.later_async_func_attr, AsyncMock)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_spec_mock_type_kw(self):
def inner_test(mock_type):
async_mock = mock_type(spec=async_func)
@@ -322,6 +356,8 @@ class AsyncSpecTest(unittest.TestCase):
with self.subTest(f"test spec kwarg with {mock_type}"):
inner_test(mock_type)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_spec_mock_type_positional(self):
def inner_test(mock_type):
async_mock = mock_type(async_func)
@@ -427,9 +463,10 @@ class AsyncArguments(IsolatedAsyncioTestCase):
self.assertEqual(output, 10)
async def test_add_side_effect_exception(self):
class CustomError(Exception): pass
async def addition(var): pass
mock = AsyncMock(addition, side_effect=Exception('err'))
with self.assertRaises(Exception):
mock = AsyncMock(addition, side_effect=CustomError('side-effect'))
with self.assertRaisesRegex(CustomError, 'side-effect'):
await mock(5)
async def test_add_side_effect_coroutine(self):
@@ -699,8 +736,8 @@ class AsyncIteratorTest(unittest.TestCase):
async def __anext__(self): pass
# TODO: RUSTPYTHON; async for
"""
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_aiter_set_return_value(self):
mock_iter = AsyncMock(name="tester")
mock_iter.__aiter__.return_value = [1, 2, 3]
@@ -708,7 +745,6 @@ class AsyncIteratorTest(unittest.TestCase):
return [i async for i in mock_iter]
result = run(main())
self.assertEqual(result, [1, 2, 3])
"""
def test_mock_aiter_and_anext_asyncmock(self):
def inner_test(mock_type):
@@ -726,8 +762,9 @@ class AsyncIteratorTest(unittest.TestCase):
with self.subTest(f"test aiter and anext corourtine with {mock_type}"):
inner_test(mock_type)
# TODO: RUSTPYTHON; async for
"""
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_mock_async_for(self):
async def iterate(iterator):
accumulator = []
@@ -761,7 +798,7 @@ class AsyncIteratorTest(unittest.TestCase):
with self.subTest(f"set return_value iterator with {mock_type}"):
test_set_return_value_iter(mock_type)
"""
class AsyncMockAssert(unittest.TestCase):
def setUp(self):
@@ -773,6 +810,8 @@ class AsyncMockAssert(unittest.TestCase):
async def _await_coroutine(self, coroutine):
return await coroutine
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_assert_called_but_not_awaited(self):
mock = AsyncMock(AsyncClass)
with assertNeverAwaited(self):
@@ -813,6 +852,8 @@ class AsyncMockAssert(unittest.TestCase):
self.mock.assert_called_once()
self.mock.assert_awaited_once()
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_assert_called_twice_and_awaited_once(self):
mock = AsyncMock(AsyncClass)
coroutine = mock.async_method()
@@ -827,6 +868,8 @@ class AsyncMockAssert(unittest.TestCase):
mock.async_method.assert_awaited()
mock.async_method.assert_awaited_once()
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_assert_called_once_and_awaited_twice(self):
mock = AsyncMock(AsyncClass)
coroutine = mock.async_method()
@@ -851,6 +894,8 @@ class AsyncMockAssert(unittest.TestCase):
with self.assertRaises(AssertionError):
self.mock.assert_called()
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_assert_has_calls_not_awaits(self):
kalls = [call('foo')]
with assertNeverAwaited(self):
@@ -859,6 +904,8 @@ class AsyncMockAssert(unittest.TestCase):
with self.assertRaises(AssertionError):
self.mock.assert_has_awaits(kalls)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_assert_has_mock_calls_on_async_mock_no_spec(self):
with assertNeverAwaited(self):
self.mock()
@@ -872,6 +919,8 @@ class AsyncMockAssert(unittest.TestCase):
mock_kalls = ([call(), call('foo'), call('baz')])
self.assertEqual(self.mock.mock_calls, mock_kalls)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_assert_has_mock_calls_on_async_mock_with_spec(self):
a_class_mock = AsyncMock(AsyncClass)
with assertNeverAwaited(self):
@@ -887,6 +936,8 @@ class AsyncMockAssert(unittest.TestCase):
self.assertEqual(a_class_mock.async_method.mock_calls, method_kalls)
self.assertEqual(a_class_mock.mock_calls, mock_kalls)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_async_method_calls_recorded(self):
with assertNeverAwaited(self):
self.mock.something(3, fish=None)
@@ -902,6 +953,8 @@ class AsyncMockAssert(unittest.TestCase):
[("something", (6,), {'cake': sentinel.Cake})],
"method calls not recorded correctly")
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_async_arg_lists(self):
def assert_attrs(mock):
names = ('call_args_list', 'method_calls', 'mock_calls')
@@ -3,7 +3,7 @@
# http://www.voidspace.org.uk/python/mock/
import unittest
from unittest.test.testmock.support import is_instance, X, SomeClass
from test.test_unittest.testmock.support import is_instance, X, SomeClass
from unittest.mock import (
Mock, MagicMock, NonCallableMagicMock,
@@ -43,6 +43,8 @@ class AnyTest(unittest.TestCase):
mock.assert_called_with(ANY, foo=ANY)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_any_mock_calls_comparison_order(self):
mock = Mock()
class Foo(object):
@@ -882,6 +884,8 @@ class SpecSignatureTest(unittest.TestCase):
a.f.assert_called_with(self=10)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_autospec_data_descriptor(self):
class Descriptor(object):
def __init__(self, value):
@@ -926,6 +930,8 @@ class SpecSignatureTest(unittest.TestCase):
check_data_descriptor(foo.desc)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_autospec_on_bound_builtin_function(self):
meth = types.MethodType(time.ctime, time.time())
self.assertIsInstance(meth(), str)
@@ -952,6 +958,24 @@ class SpecSignatureTest(unittest.TestCase):
self.assertFalse(hasattr(autospec, '__name__'))
def test_autospec_signature_staticmethod(self):
class Foo:
@staticmethod
def static_method(a, b=10, *, c): pass
mock = create_autospec(Foo.__dict__['static_method'])
self.assertEqual(inspect.signature(Foo.static_method), inspect.signature(mock))
def test_autospec_signature_classmethod(self):
class Foo:
@classmethod
def class_method(cls, a, b=10, *, c): pass
mock = create_autospec(Foo.__dict__['class_method'])
self.assertEqual(inspect.signature(Foo.class_method), inspect.signature(mock))
def test_spec_inspect_signature(self):
def myfunc(x, y): pass
@@ -1077,7 +1101,7 @@ class TestCallList(unittest.TestCase):
p.stop()
def test_propertymock_returnvalue(self):
def test_propertymock_bare(self):
m = MagicMock()
p = PropertyMock()
type(m).foo = p
@@ -1088,6 +1112,35 @@ class TestCallList(unittest.TestCase):
self.assertNotIsInstance(returned, PropertyMock)
def test_propertymock_returnvalue(self):
m = MagicMock()
p = PropertyMock(return_value=42)
type(m).foo = p
returned = m.foo
p.assert_called_once_with()
self.assertEqual(returned, 42)
self.assertNotIsInstance(returned, PropertyMock)
def test_propertymock_side_effect(self):
m = MagicMock()
p = PropertyMock(side_effect=ValueError)
type(m).foo = p
with self.assertRaises(ValueError):
m.foo
p.assert_called_once_with()
def test_propertymock_attach(self):
m = Mock()
p = PropertyMock()
type(m).foo = p
m.attach_mock(p, 'foo')
self.assertEqual(m.mock_calls, [])
class TestCallablePredicate(unittest.TestCase):
def test_type(self):
@@ -331,6 +331,45 @@ class TestMockingMagicMethods(unittest.TestCase):
self.assertEqual(os.fspath(mock), expected_path)
mock.__fspath__.assert_called_once()
def test_magic_mock_does_not_reset_magic_returns(self):
# https://github.com/python/cpython/issues/123934
for reset in (True, False):
with self.subTest(reset=reset):
mm = MagicMock()
self.assertIs(type(mm.__str__()), str)
mm.__str__.assert_called_once()
self.assertIs(type(mm.__hash__()), int)
mm.__hash__.assert_called_once()
for _ in range(3):
# Repeat reset several times to be sure:
mm.reset_mock(return_value=reset)
self.assertIs(type(mm.__str__()), str)
mm.__str__.assert_called_once()
self.assertIs(type(mm.__hash__()), int)
mm.__hash__.assert_called_once()
def test_magic_mock_resets_manual_mocks(self):
mm = MagicMock()
mm.__iter__ = MagicMock(return_value=iter([1]))
mm.custom = MagicMock(return_value=2)
self.assertEqual(list(iter(mm)), [1])
self.assertEqual(mm.custom(), 2)
mm.reset_mock(return_value=True)
self.assertEqual(list(iter(mm)), [])
self.assertIsInstance(mm.custom(), MagicMock)
def test_magic_mock_resets_manual_mocks_empty_iter(self):
mm = MagicMock()
mm.__iter__.return_value = []
self.assertEqual(list(iter(mm)), [])
mm.reset_mock(return_value=True)
self.assertEqual(list(iter(mm)), [])
def test_magic_methods_and_spec(self):
class Iterable(object):
@@ -5,7 +5,7 @@ import tempfile
from test.support import ALWAYS_EQ
import unittest
from unittest.test.testmock.support import is_instance
from test.test_unittest.testmock.support import is_instance
from unittest import mock
from unittest.mock import (
call, DEFAULT, patch, sentinel,
@@ -38,6 +38,17 @@ class Something(object):
def smeth(a, b, c, d=None): pass
class SomethingElse(object):
def __init__(self):
self._instance = None
@property
def instance(self):
if not self._instance:
self._instance = 'object'
return self._instance
class Typos():
autospect = None
auto_spec = None
@@ -104,6 +115,24 @@ class MockTest(unittest.TestCase):
with self.assertRaises(TypeError):
mock()
def test_create_autospec_should_be_configurable_by_kwargs(self):
"""If kwargs are given to configure mock, the function must configure
the parent mock during initialization."""
mocked_result = 'mocked value'
class_mock = create_autospec(spec=Something, **{
'return_value.meth.side_effect': [ValueError, DEFAULT],
'return_value.meth.return_value': mocked_result})
with self.assertRaises(ValueError):
class_mock().meth(a=None, b=None, c=None)
self.assertEqual(class_mock().meth(a=None, b=None, c=None), mocked_result)
# Only the parent mock should be configurable because the user will
# pass kwargs with respect to the parent mock.
self.assertEqual(class_mock().return_value.meth.side_effect, None)
def test_create_autospec_correctly_handles_name(self):
class X: ...
mock = create_autospec(X, spec_set=True, name="Y")
self.assertEqual(mock._mock_name, "Y")
def test_repr(self):
mock = Mock(name='foo')
@@ -234,6 +263,73 @@ class MockTest(unittest.TestCase):
with mock.patch('builtins.open', mock.mock_open()):
mock.mock_open() # should still be valid with open() mocked
def test_create_autospec_wraps_class(self):
"""Autospec a class with wraps & test if the call is passed to the
wrapped object."""
result = "real result"
class Result:
def get_result(self):
return result
class_mock = create_autospec(spec=Result, wraps=Result)
# Have to reassign the return_value to DEFAULT to return the real
# result (actual instance of "Result") when the mock is called.
class_mock.return_value = mock.DEFAULT
self.assertEqual(class_mock().get_result(), result)
# Autospec should also wrap child attributes of parent.
self.assertEqual(class_mock.get_result._mock_wraps, Result.get_result)
def test_create_autospec_instance_wraps_class(self):
"""Autospec a class instance with wraps & test if the call is passed
to the wrapped object."""
result = "real result"
class Result:
@staticmethod
def get_result():
"""This is a static method because when the mocked instance of
'Result' will call this method, it won't be able to consume
'self' argument."""
return result
instance_mock = create_autospec(spec=Result, instance=True, wraps=Result)
# Have to reassign the return_value to DEFAULT to return the real
# result from "Result.get_result" when the mocked instance of "Result"
# calls "get_result".
instance_mock.get_result.return_value = mock.DEFAULT
self.assertEqual(instance_mock.get_result(), result)
# Autospec should also wrap child attributes of the instance.
self.assertEqual(instance_mock.get_result._mock_wraps, Result.get_result)
def test_create_autospec_wraps_function_type(self):
"""Autospec a function or a method with wraps & test if the call is
passed to the wrapped object."""
result = "real result"
class Result:
def get_result(self):
return result
func_mock = create_autospec(spec=Result.get_result, wraps=Result.get_result)
self.assertEqual(func_mock(Result()), result)
def test_explicit_return_value_even_if_mock_wraps_object(self):
"""If the mock has an explicit return_value set then calls are not
passed to the wrapped object and the return_value is returned instead.
"""
def my_func():
return None
func_mock = create_autospec(spec=my_func, wraps=my_func)
return_value = "explicit return value"
func_mock.return_value = return_value
self.assertEqual(func_mock(), return_value)
def test_explicit_parent(self):
parent = Mock()
mock1 = Mock(parent=parent, return_value=None)
mock1(1, 2, 3)
mock2 = Mock(parent=parent, return_value=None)
mock2(4, 5, 6)
self.assertEqual(parent.mock_calls, [call(1, 2, 3), call(4, 5, 6)])
def test_reset_mock(self):
parent = Mock()
@@ -603,6 +699,14 @@ class MockTest(unittest.TestCase):
real = Mock()
mock = Mock(wraps=real)
# If "Mock" wraps an object, just accessing its
# "return_value" ("NonCallableMock.__get_return_value") should not
# trigger its descriptor ("NonCallableMock.__set_return_value") so
# the default "return_value" should always be "sentinel.DEFAULT".
self.assertEqual(mock.return_value, DEFAULT)
# It will not be "sentinel.DEFAULT" if the mock is not wrapping any
# object.
self.assertNotEqual(real.return_value, DEFAULT)
self.assertEqual(mock(), real())
real.reset_mock()
@@ -1039,7 +1143,7 @@ class MockTest(unittest.TestCase):
actual = 'not called.'
expected = "mock(1, '2', 3, bar='foo')"
message = 'expected call not found.\nExpected: %s\nActual: %s'
message = 'expected call not found.\nExpected: %s\n Actual: %s'
self.assertRaisesWithMsg(
AssertionError, message % (expected, actual),
mock.assert_called_with, 1, '2', 3, bar='foo'
@@ -1054,7 +1158,7 @@ class MockTest(unittest.TestCase):
for meth in asserters:
actual = "foo(1, '2', 3, foo='foo')"
expected = "foo(1, '2', 3, bar='foo')"
message = 'expected call not found.\nExpected: %s\nActual: %s'
message = 'expected call not found.\nExpected: %s\n Actual: %s'
self.assertRaisesWithMsg(
AssertionError, message % (expected, actual),
meth, 1, '2', 3, bar='foo'
@@ -1064,7 +1168,7 @@ class MockTest(unittest.TestCase):
for meth in asserters:
actual = "foo(1, '2', 3, foo='foo')"
expected = "foo(bar='foo')"
message = 'expected call not found.\nExpected: %s\nActual: %s'
message = 'expected call not found.\nExpected: %s\n Actual: %s'
self.assertRaisesWithMsg(
AssertionError, message % (expected, actual),
meth, bar='foo'
@@ -1074,7 +1178,7 @@ class MockTest(unittest.TestCase):
for meth in asserters:
actual = "foo(1, '2', 3, foo='foo')"
expected = "foo(1, 2, 3)"
message = 'expected call not found.\nExpected: %s\nActual: %s'
message = 'expected call not found.\nExpected: %s\n Actual: %s'
self.assertRaisesWithMsg(
AssertionError, message % (expected, actual),
meth, 1, 2, 3
@@ -1084,7 +1188,7 @@ class MockTest(unittest.TestCase):
for meth in asserters:
actual = "foo(1, '2', 3, foo='foo')"
expected = "foo()"
message = 'expected call not found.\nExpected: %s\nActual: %s'
message = 'expected call not found.\nExpected: %s\n Actual: %s'
self.assertRaisesWithMsg(
AssertionError, message % (expected, actual), meth
)
@@ -1528,25 +1632,33 @@ class MockTest(unittest.TestCase):
mock = Mock(spec=f)
mock(1)
with self.assertRaisesRegex(
AssertionError,
'^{}$'.format(
re.escape('Calls not found.\n'
'Expected: [call()]\n'
'Actual: [call(1)]'))) as cm:
with self.assertRaises(AssertionError) as cm:
mock.assert_has_calls([call()])
self.assertEqual(str(cm.exception),
'Calls not found.\n'
'Expected: [call()]\n'
' Actual: [call(1)]'
)
self.assertIsNone(cm.exception.__cause__)
uncalled_mock = Mock()
with self.assertRaises(AssertionError) as cm:
uncalled_mock.assert_has_calls([call()])
self.assertEqual(str(cm.exception),
'Calls not found.\n'
'Expected: [call()]\n'
' Actual: []'
)
self.assertIsNone(cm.exception.__cause__)
with self.assertRaisesRegex(
AssertionError,
'^{}$'.format(
re.escape(
'Error processing expected calls.\n'
"Errors: [None, TypeError('too many positional arguments')]\n"
"Expected: [call(), call(1, 2)]\n"
'Actual: [call(1)]'))) as cm:
with self.assertRaises(AssertionError) as cm:
mock.assert_has_calls([call(), call(1, 2)])
self.assertEqual(str(cm.exception),
'Error processing expected calls.\n'
"Errors: [None, TypeError('too many positional arguments')]\n"
'Expected: [call(), call(1, 2)]\n'
' Actual: [call(1)]'
)
self.assertIsInstance(cm.exception.__cause__, TypeError)
def test_assert_any_call(self):
@@ -1645,21 +1757,41 @@ class MockTest(unittest.TestCase):
m.aseert_foo_call()
with self.assertRaisesRegex(AttributeError, msg):
m.assrt_foo_call()
with self.assertRaisesRegex(AttributeError, msg):
m.called_once_with()
with self.assertRaisesRegex(AttributeError, msg):
m.called_once()
with self.assertRaisesRegex(AttributeError, msg):
m.has_calls()
class Foo(object):
def called_once(self): pass
def has_calls(self): pass
m = Mock(spec=Foo)
m.called_once()
m.has_calls()
m.called_once.assert_called_once()
m.has_calls.assert_called_once()
m = Mock(unsafe=True)
m.assert_foo_call()
m.assret_foo_call()
m.asert_foo_call()
m.aseert_foo_call()
m.assrt_foo_call()
m.called_once()
m.called_once_with()
m.has_calls()
# gh-100739
def test_mock_safe_with_spec(self):
class Foo(object):
def assert_bar(self):
pass
def assert_bar(self): pass
def assertSome(self):
pass
def assertSome(self): pass
m = Mock(spec=Foo)
m.assert_bar()
@@ -2231,7 +2363,7 @@ class MockTest(unittest.TestCase):
class Foo():
one = 'one'
# patch, patch.object and create_autospec need to check for misspelled
# arguments explicitly and throw a RuntimError if found.
# arguments explicitly and throw a RuntimeError if found.
with self.assertRaises(RuntimeError):
with patch(f'{__name__}.Something.meth', autospect=True): pass
with self.assertRaises(RuntimeError):
@@ -2273,6 +2405,26 @@ class MockTest(unittest.TestCase):
f'{__name__}.Typos', autospect=True, set_spec=True, auto_spec=True):
pass
def test_property_not_called_with_spec_mock(self):
obj = SomethingElse()
self.assertIsNone(obj._instance, msg='before mock')
mock = Mock(spec=obj)
self.assertIsNone(obj._instance, msg='after mock')
self.assertEqual('object', obj.instance)
def test_decorated_async_methods_with_spec_mock(self):
class Foo():
@classmethod
async def class_method(cls):
pass
@staticmethod
async def static_method():
pass
async def method(self):
pass
mock = Mock(spec=Foo)
for m in (mock.method, mock.class_method, mock.static_method):
self.assertIsInstance(m, AsyncMock)
if __name__ == '__main__':
unittest.main()
@@ -7,9 +7,11 @@ import sys
from collections import OrderedDict
import unittest
from unittest.test.testmock import support
from unittest.test.testmock.support import SomeClass, is_instance
import test
from test.test_unittest.testmock import support
from test.test_unittest.testmock.support import SomeClass, is_instance
from test.support.import_helper import DirsOnSysPath
from test.test_importlib.util import uncache
from unittest.mock import (
NonCallableMock, CallableMixin, sentinel,
@@ -656,6 +658,8 @@ class PatchTest(unittest.TestCase):
self.assertEqual(foo, {})
# TODO: RUSTPYTHON
@unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
def test_patch_dict_with_string(self):
@patch.dict('os.environ', {'konrad_delong': 'some value'})
def test():
@@ -669,7 +673,7 @@ class PatchTest(unittest.TestCase):
# the new dictionary during function call
original = support.target.copy()
@patch.dict('unittest.test.testmock.support.target', {'bar': 'BAR'})
@patch.dict('test.test_unittest.testmock.support.target', {'bar': 'BAR'})
def test():
self.assertEqual(support.target, {'foo': 'BAZ', 'bar': 'BAR'})
@@ -743,6 +747,54 @@ class PatchTest(unittest.TestCase):
self.assertIsNone(patcher.stop())
def test_exit_idempotent(self):
patcher = patch(foo_name, 'bar', 3)
with patcher:
patcher.stop()
def test_second_start_failure(self):
patcher = patch(foo_name, 'bar', 3)
patcher.start()
try:
self.assertRaises(RuntimeError, patcher.start)
finally:
patcher.stop()
def test_second_enter_failure(self):
patcher = patch(foo_name, 'bar', 3)
with patcher:
self.assertRaises(RuntimeError, patcher.start)
def test_second_start_after_stop(self):
patcher = patch(foo_name, 'bar', 3)
patcher.start()
patcher.stop()
patcher.start()
patcher.stop()
def test_property_setters(self):
mock_object = Mock()
mock_bar = mock_object.bar
patcher = patch.object(mock_object, 'bar', 'x')
with patcher:
self.assertEqual(patcher.is_local, False)
self.assertIs(patcher.target, mock_object)
self.assertEqual(patcher.temp_original, mock_bar)
patcher.is_local = True
patcher.target = mock_bar
patcher.temp_original = mock_object
self.assertEqual(patcher.is_local, True)
self.assertIs(patcher.target, mock_bar)
self.assertEqual(patcher.temp_original, mock_object)
# if changes are left intact, they may lead to disruption as shown below (it might be what someone needs though)
self.assertEqual(mock_bar.bar, mock_object)
self.assertEqual(mock_object.bar, 'x')
def test_patchobject_start_stop(self):
original = something
patcher = patch.object(PTModule, 'something', 'foo')
@@ -996,6 +1048,36 @@ class PatchTest(unittest.TestCase):
method.assert_called_once_with()
def test_autospec_staticmethod_signature(self):
# Patched methods which are decorated with @staticmethod should have the same signature
class Foo:
@staticmethod
def static_method(a, b=10, *, c): pass
Foo.static_method(1, 2, c=3)
with patch.object(Foo, 'static_method', autospec=True) as method:
method(1, 2, c=3)
self.assertRaises(TypeError, method)
self.assertRaises(TypeError, method, 1)
self.assertRaises(TypeError, method, 1, 2, 3, c=4)
def test_autospec_classmethod_signature(self):
# Patched methods which are decorated with @classmethod should have the same signature
class Foo:
@classmethod
def class_method(cls, a, b=10, *, c): pass
Foo.class_method(1, 2, c=3)
with patch.object(Foo, 'class_method', autospec=True) as method:
method(1, 2, c=3)
self.assertRaises(TypeError, method)
self.assertRaises(TypeError, method, 1)
self.assertRaises(TypeError, method, 1, 2, 3, c=4)
def test_autospec_with_new(self):
patcher = patch('%s.function' % __name__, new=3, autospec=True)
self.assertRaises(TypeError, patcher.start)
@@ -1066,7 +1148,7 @@ class PatchTest(unittest.TestCase):
self.assertIsNot(m1, m2)
for mock in m1, m2:
self.assertNotCallable(m1)
self.assertNotCallable(mock)
def test_new_callable_patch_object(self):
@@ -1079,7 +1161,7 @@ class PatchTest(unittest.TestCase):
self.assertIsNot(m1, m2)
for mock in m1, m2:
self.assertNotCallable(m1)
self.assertNotCallable(mock)
def test_new_callable_keyword_arguments(self):
@@ -1614,7 +1696,7 @@ class PatchTest(unittest.TestCase):
def test_patch_nested_autospec_repr(self):
with patch('unittest.test.testmock.support', autospec=True) as m:
with patch('test.test_unittest.testmock.support', autospec=True) as m:
self.assertIn(" name='support.SomeClass.wibble()'",
repr(m.SomeClass.wibble()))
self.assertIn(" name='support.SomeClass().wibble()'",
@@ -1698,6 +1780,75 @@ class PatchTest(unittest.TestCase):
'exception traceback not propagated')
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_name_resolution_import_rebinding(self):
# Currently mock.patch uses pkgutil.resolve_name(), but repeat
# similar tests just for the case.
# The same data is also used for testing import in test_import and
# pkgutil.resolve_name() in test_pkgutil.
path = os.path.join(os.path.dirname(test.__file__), 'test_import', 'data')
def check(name):
p = patch(name)
p.start()
p.stop()
def check_error(name):
p = patch(name)
self.assertRaises(AttributeError, p.start)
with uncache('package3', 'package3.submodule'), DirsOnSysPath(path):
check('package3.submodule.A.attr')
check_error('package3.submodule.B.attr')
with uncache('package3', 'package3.submodule'), DirsOnSysPath(path):
check('package3.submodule:A.attr')
check_error('package3.submodule:B.attr')
with uncache('package3', 'package3.submodule'), DirsOnSysPath(path):
check('package3:submodule.B.attr')
check_error('package3:submodule.A.attr')
check('package3.submodule.A.attr')
check_error('package3.submodule.B.attr')
check('package3:submodule.B.attr')
check_error('package3:submodule.A.attr')
with uncache('package3', 'package3.submodule'), DirsOnSysPath(path):
check('package3:submodule.B.attr')
check_error('package3:submodule.A.attr')
check('package3.submodule:A.attr')
check_error('package3.submodule:B.attr')
check('package3:submodule.B.attr')
check_error('package3:submodule.A.attr')
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_name_resolution_import_rebinding2(self):
path = os.path.join(os.path.dirname(test.__file__), 'test_import', 'data')
def check(name):
p = patch(name)
p.start()
p.stop()
def check_error(name):
p = patch(name)
self.assertRaises(AttributeError, p.start)
with uncache('package4', 'package4.submodule'), DirsOnSysPath(path):
check('package4.submodule.A.attr')
check_error('package4.submodule.B.attr')
with uncache('package4', 'package4.submodule'), DirsOnSysPath(path):
check('package4.submodule:A.attr')
check_error('package4.submodule:B.attr')
with uncache('package4', 'package4.submodule'), DirsOnSysPath(path):
check('package4:submodule.B.attr')
check_error('package4:submodule.A.attr')
check('package4.submodule.A.attr')
check_error('package4.submodule.B.attr')
check('package4:submodule.A.attr')
check_error('package4:submodule.B.attr')
with uncache('package4', 'package4.submodule'), DirsOnSysPath(path):
check('package4:submodule.B.attr')
check_error('package4:submodule.A.attr')
check('package4.submodule:A.attr')
check_error('package4.submodule:B.attr')
check('package4:submodule.A.attr')
check_error('package4:submodule.B.attr')
def test_create_and_specs(self):
for kwarg in ('spec', 'spec_set', 'autospec'):
p = patch('%s.doesnotexist' % __name__, create=True,
@@ -1867,6 +2018,8 @@ class PatchTest(unittest.TestCase):
self.assertEqual(dic2, origdic2)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_special_attrs(self):
def foo(x=0):
"""TEST"""
@@ -1882,7 +2035,7 @@ class PatchTest(unittest.TestCase):
with patch.object(foo, '__module__', "testpatch2"):
self.assertEqual(foo.__module__, "testpatch2")
self.assertEqual(foo.__module__, 'unittest.test.testmock.testpatch')
self.assertEqual(foo.__module__, __name__)
with patch.object(foo, '__annotations__', dict([('s', 1, )])):
self.assertEqual(foo.__annotations__, dict([('s', 1, )]))
@@ -1917,16 +2070,16 @@ class PatchTest(unittest.TestCase):
# This exercises the AttributeError branch of _dot_lookup.
# make sure it's there
import unittest.test.testmock.support
import test.test_unittest.testmock.support
# now make sure it's not:
with patch.dict('sys.modules'):
del sys.modules['unittest.test.testmock.support']
del sys.modules['unittest.test.testmock']
del sys.modules['unittest.test']
del sys.modules['unittest']
del sys.modules['test.test_unittest.testmock.support']
del sys.modules['test.test_unittest.testmock']
del sys.modules['test.test_unittest']
del sys.modules['test']
# now make sure we can patch based on a dotted path:
@patch('unittest.test.testmock.support.X')
@patch('test.test_unittest.testmock.support.X')
def test(mock):
pass
test()
@@ -1943,11 +2096,18 @@ class PatchTest(unittest.TestCase):
def test_cant_set_kwargs_when_passing_a_mock(self):
@patch('unittest.test.testmock.support.X', new=object(), x=1)
@patch('test.test_unittest.testmock.support.X', new=object(), x=1)
def test(): pass
with self.assertRaises(TypeError):
test()
def test_patch_proxy_object(self):
@patch("test.test_unittest.testmock.support.g", new_callable=MagicMock())
def test(_):
pass
test()
if __name__ == '__main__':
unittest.main()
@@ -175,15 +175,12 @@ class TestSealable(unittest.TestCase):
# https://bugs.python.org/issue45156
class Foo:
foo = 0
def bar1(self):
return 1
def bar2(self):
return 2
def bar1(self): pass
def bar2(self): pass
class Baz:
baz = 3
def ban(self):
return 4
def ban(self): pass
for spec_set in (True, False):
with self.subTest(spec_set=spec_set):
@@ -200,6 +197,9 @@ class TestSealable(unittest.TestCase):
self.assertIsInstance(foo.Baz.baz, mock.NonCallableMagicMock)
self.assertIsInstance(foo.Baz.ban, mock.MagicMock)
# see gh-91803
self.assertIsInstance(foo.bar2(), mock.MagicMock)
self.assertEqual(foo.bar1(), 'a')
foo.bar1.return_value = 'new_a'
self.assertEqual(foo.bar1(), 'new_a')
@@ -212,7 +212,7 @@ class TestSealable(unittest.TestCase):
with self.assertRaises(AttributeError):
foo.bar = 1
with self.assertRaises(AttributeError):
foo.bar2()
foo.bar2().x
foo.bar2.return_value = 'bar2'
self.assertEqual(foo.bar2(), 'bar2')
File renamed without changes.
+201
View File
@@ -0,0 +1,201 @@
import time
import unittest
import concurrent.futures
from test.support import threading_helper
from unittest.mock import patch, ThreadingMock
threading_helper.requires_working_threading(module=True)
VERY_SHORT_TIMEOUT = 0.1
class Something:
def method_1(self):
pass # pragma: no cover
def method_2(self):
pass # pragma: no cover
class TestThreadingMock(unittest.TestCase):
def _call_after_delay(self, func, /, *args, **kwargs):
time.sleep(kwargs.pop("delay"))
func(*args, **kwargs)
def setUp(self):
self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
def tearDown(self):
self._executor.shutdown()
def run_async(self, func, /, *args, delay=0, **kwargs):
self._executor.submit(
self._call_after_delay, func, *args, **kwargs, delay=delay
)
def _make_mock(self, *args, **kwargs):
return ThreadingMock(*args, **kwargs)
def test_spec(self):
waitable_mock = self._make_mock(spec=Something)
with patch(f"{__name__}.Something", waitable_mock) as m:
something = m()
self.assertIsInstance(something.method_1, ThreadingMock)
self.assertIsInstance(something.method_1().method_2(), ThreadingMock)
with self.assertRaises(AttributeError):
m.test
def test_side_effect(self):
waitable_mock = self._make_mock()
with patch(f"{__name__}.Something", waitable_mock):
something = Something()
something.method_1.side_effect = [1]
self.assertEqual(something.method_1(), 1)
def test_instance_check(self):
waitable_mock = self._make_mock()
with patch(f"{__name__}.Something", waitable_mock):
something = Something()
self.assertIsInstance(something.method_1, ThreadingMock)
self.assertIsInstance(something.method_1().method_2(), ThreadingMock)
def test_dynamic_child_mocks_are_threading_mocks(self):
waitable_mock = self._make_mock()
self.assertIsInstance(waitable_mock.child, ThreadingMock)
def test_dynamic_child_mocks_inherit_timeout(self):
mock1 = self._make_mock()
self.assertIs(mock1._mock_wait_timeout, None)
mock2 = self._make_mock(timeout=2)
self.assertEqual(mock2._mock_wait_timeout, 2)
mock3 = self._make_mock(timeout=3)
self.assertEqual(mock3._mock_wait_timeout, 3)
self.assertIs(mock1.child._mock_wait_timeout, None)
self.assertEqual(mock2.child._mock_wait_timeout, 2)
self.assertEqual(mock3.child._mock_wait_timeout, 3)
self.assertEqual(mock2.really().__mul__().complex._mock_wait_timeout, 2)
def test_no_name_clash(self):
waitable_mock = self._make_mock()
waitable_mock._event = "myevent"
waitable_mock.event = "myevent"
waitable_mock.timeout = "mytimeout"
waitable_mock("works")
waitable_mock.wait_until_called()
waitable_mock.wait_until_any_call_with("works")
def test_patch(self):
waitable_mock = self._make_mock(spec=Something)
with patch(f"{__name__}.Something", waitable_mock):
something = Something()
something.method_1()
something.method_1.wait_until_called()
def test_wait_already_called_success(self):
waitable_mock = self._make_mock(spec=Something)
waitable_mock.method_1()
waitable_mock.method_1.wait_until_called()
waitable_mock.method_1.wait_until_any_call_with()
waitable_mock.method_1.assert_called()
def test_wait_until_called_success(self):
waitable_mock = self._make_mock(spec=Something)
self.run_async(waitable_mock.method_1, delay=VERY_SHORT_TIMEOUT)
waitable_mock.method_1.wait_until_called()
def test_wait_until_called_method_timeout(self):
waitable_mock = self._make_mock(spec=Something)
with self.assertRaises(AssertionError):
waitable_mock.method_1.wait_until_called(timeout=VERY_SHORT_TIMEOUT)
def test_wait_until_called_instance_timeout(self):
waitable_mock = self._make_mock(spec=Something, timeout=VERY_SHORT_TIMEOUT)
with self.assertRaises(AssertionError):
waitable_mock.method_1.wait_until_called()
def test_wait_until_called_global_timeout(self):
with patch.object(ThreadingMock, "DEFAULT_TIMEOUT"):
ThreadingMock.DEFAULT_TIMEOUT = VERY_SHORT_TIMEOUT
waitable_mock = self._make_mock(spec=Something)
with self.assertRaises(AssertionError):
waitable_mock.method_1.wait_until_called()
def test_wait_until_any_call_with_success(self):
waitable_mock = self._make_mock()
self.run_async(waitable_mock, delay=VERY_SHORT_TIMEOUT)
waitable_mock.wait_until_any_call_with()
def test_wait_until_any_call_with_instance_timeout(self):
waitable_mock = self._make_mock(timeout=VERY_SHORT_TIMEOUT)
with self.assertRaises(AssertionError):
waitable_mock.wait_until_any_call_with()
def test_wait_until_any_call_global_timeout(self):
with patch.object(ThreadingMock, "DEFAULT_TIMEOUT"):
ThreadingMock.DEFAULT_TIMEOUT = VERY_SHORT_TIMEOUT
waitable_mock = self._make_mock()
with self.assertRaises(AssertionError):
waitable_mock.wait_until_any_call_with()
def test_wait_until_any_call_positional(self):
waitable_mock = self._make_mock(timeout=VERY_SHORT_TIMEOUT)
waitable_mock.method_1(1, 2, 3)
waitable_mock.method_1.wait_until_any_call_with(1, 2, 3)
with self.assertRaises(AssertionError):
waitable_mock.method_1.wait_until_any_call_with(2, 3, 1)
with self.assertRaises(AssertionError):
waitable_mock.method_1.wait_until_any_call_with()
def test_wait_until_any_call_kw(self):
waitable_mock = self._make_mock(timeout=VERY_SHORT_TIMEOUT)
waitable_mock.method_1(a=1, b=2)
waitable_mock.method_1.wait_until_any_call_with(a=1, b=2)
with self.assertRaises(AssertionError):
waitable_mock.method_1.wait_until_any_call_with(a=2, b=1)
with self.assertRaises(AssertionError):
waitable_mock.method_1.wait_until_any_call_with()
def test_magic_methods_success(self):
waitable_mock = self._make_mock()
str(waitable_mock)
waitable_mock.__str__.wait_until_called()
waitable_mock.__str__.assert_called()
def test_reset_mock_resets_wait(self):
m = self._make_mock(timeout=VERY_SHORT_TIMEOUT)
with self.assertRaises(AssertionError):
m.wait_until_called()
with self.assertRaises(AssertionError):
m.wait_until_any_call_with()
m()
m.wait_until_called()
m.wait_until_any_call_with()
m.assert_called_once()
m.reset_mock()
with self.assertRaises(AssertionError):
m.wait_until_called()
with self.assertRaises(AssertionError):
m.wait_until_any_call_with()
m()
m.wait_until_called()
m.wait_until_any_call_with()
m.assert_called_once()
if __name__ == "__main__":
unittest.main()
@@ -1,7 +1,7 @@
import unittest
from warnings import catch_warnings
from unittest.test.testmock.support import is_instance
from test.test_unittest.testmock.support import is_instance
from unittest.mock import MagicMock, Mock, patch, sentinel, mock_open, call
@@ -158,7 +158,7 @@ class TestMockOpen(unittest.TestCase):
f.read()
expected_calls = [call('foo'), call().__enter__(), call().read(),
call().__exit__(None, None, None)]
call().__exit__(None, None, None), call().close()]
self.assertEqual(mock.mock_calls, expected_calls)
self.assertIs(f, handle)
@@ -172,9 +172,9 @@ class TestMockOpen(unittest.TestCase):
expected_calls = [
call('foo'), call().__enter__(), call().read(),
call().__exit__(None, None, None),
call().__exit__(None, None, None), call().close(),
call('bar'), call().__enter__(), call().read(),
call().__exit__(None, None, None)]
call().__exit__(None, None, None), call().close()]
self.assertEqual(mock.mock_calls, expected_calls)
def test_explicit_mock(self):
+57 -58
View File
@@ -9,6 +9,7 @@ import warnings
import collections
import contextlib
import traceback
import time
import types
from . import result
@@ -331,6 +332,23 @@ class _AssertWarnsContext(_AssertRaisesBaseContext):
self._raiseFailure("{} not triggered".format(exc_name))
class _AssertNotWarnsContext(_AssertWarnsContext):
def __exit__(self, exc_type, exc_value, tb):
self.warnings_manager.__exit__(exc_type, exc_value, tb)
if exc_type is not None:
# let unexpected exceptions pass through
return
try:
exc_name = self.expected.__name__
except AttributeError:
exc_name = str(self.expected)
for m in self.warnings:
w = m.message
if isinstance(w, self.expected):
self._raiseFailure(f"{exc_name} triggered")
class _OrderedChainMap(collections.ChainMap):
def __iter__(self):
seen = set()
@@ -572,6 +590,15 @@ class TestCase(object):
else:
addUnexpectedSuccess(self)
def _addDuration(self, result, elapsed):
try:
addDuration = result.addDuration
except AttributeError:
warnings.warn("TestResult has no addDuration method",
RuntimeWarning)
else:
addDuration(self, elapsed)
def _callSetUp(self):
self.setUp()
@@ -612,6 +639,7 @@ class TestCase(object):
getattr(testMethod, "__unittest_expecting_failure__", False)
)
outcome = _Outcome(result)
start_time = time.perf_counter()
try:
self._outcome = outcome
@@ -625,6 +653,7 @@ class TestCase(object):
with outcome.testPartExecutor(self):
self._callTearDown()
self.doCleanups()
self._addDuration(result, (time.perf_counter() - start_time))
if outcome.success:
if expecting_failure:
@@ -799,6 +828,11 @@ class TestCase(object):
context = _AssertWarnsContext(expected_warning, self)
return context.handle('assertWarns', args, kwargs)
def _assertNotWarns(self, expected_warning, *args, **kwargs):
"""The opposite of assertWarns. Private due to low demand."""
context = _AssertNotWarnsContext(expected_warning, self)
return context.handle('_assertNotWarns', args, kwargs)
def assertLogs(self, logger=None, level=None):
"""Fail unless a log message of level *level* or higher is emitted
on *logger_name* or its children. If omitted, *level* defaults to
@@ -1171,35 +1205,6 @@ class TestCase(object):
standardMsg = self._truncateMessage(standardMsg, diff)
self.fail(self._formatMessage(msg, standardMsg))
def assertDictContainsSubset(self, subset, dictionary, msg=None):
"""Checks whether dictionary is a superset of subset."""
warnings.warn('assertDictContainsSubset is deprecated',
DeprecationWarning)
missing = []
mismatched = []
for key, value in subset.items():
if key not in dictionary:
missing.append(key)
elif value != dictionary[key]:
mismatched.append('%s, expected: %s, actual: %s' %
(safe_repr(key), safe_repr(value),
safe_repr(dictionary[key])))
if not (missing or mismatched):
return
standardMsg = ''
if missing:
standardMsg = 'Missing: %s' % ','.join(safe_repr(m) for m in
missing)
if mismatched:
if standardMsg:
standardMsg += '; '
standardMsg += 'Mismatched values: %s' % ','.join(mismatched)
self.fail(self._formatMessage(msg, standardMsg))
def assertCountEqual(self, first, second, msg=None):
"""Asserts that two iterables have the same elements, the same number of
times, without regard to order.
@@ -1234,19 +1239,34 @@ class TestCase(object):
def assertMultiLineEqual(self, first, second, msg=None):
"""Assert that two multi-line strings are equal."""
self.assertIsInstance(first, str, 'First argument is not a string')
self.assertIsInstance(second, str, 'Second argument is not a string')
self.assertIsInstance(first, str, "First argument is not a string")
self.assertIsInstance(second, str, "Second argument is not a string")
if first != second:
# don't use difflib if the strings are too long
# Don't use difflib if the strings are too long
if (len(first) > self._diffThreshold or
len(second) > self._diffThreshold):
self._baseAssertEqual(first, second, msg)
firstlines = first.splitlines(keepends=True)
secondlines = second.splitlines(keepends=True)
if len(firstlines) == 1 and first.strip('\r\n') == first:
firstlines = [first + '\n']
secondlines = [second + '\n']
# Append \n to both strings if either is missing the \n.
# This allows the final ndiff to show the \n difference. The
# exception here is if the string is empty, in which case no
# \n should be added
first_presplit = first
second_presplit = second
if first and second:
if first[-1] != '\n' or second[-1] != '\n':
first_presplit += '\n'
second_presplit += '\n'
elif second and second[-1] != '\n':
second_presplit += '\n'
elif first and first[-1] != '\n':
first_presplit += '\n'
firstlines = first_presplit.splitlines(keepends=True)
secondlines = second_presplit.splitlines(keepends=True)
# Generate the message and diff, then raise the exception
standardMsg = '%s != %s' % _common_shorten_repr(first, second)
diff = '\n' + ''.join(difflib.ndiff(firstlines, secondlines))
standardMsg = self._truncateMessage(standardMsg, diff)
@@ -1363,27 +1383,6 @@ class TestCase(object):
raise self.failureException(msg)
def _deprecate(original_func):
def deprecated_func(*args, **kwargs):
warnings.warn(
'Please use {0} instead.'.format(original_func.__name__),
DeprecationWarning, 2)
return original_func(*args, **kwargs)
return deprecated_func
# see #9424
failUnlessEqual = assertEquals = _deprecate(assertEqual)
failIfEqual = assertNotEquals = _deprecate(assertNotEqual)
failUnlessAlmostEqual = assertAlmostEquals = _deprecate(assertAlmostEqual)
failIfAlmostEqual = assertNotAlmostEquals = _deprecate(assertNotAlmostEqual)
failUnless = assert_ = _deprecate(assertTrue)
failUnlessRaises = _deprecate(assertRaises)
failIf = _deprecate(assertFalse)
assertRaisesRegexp = _deprecate(assertRaisesRegex)
assertRegexpMatches = _deprecate(assertRegex)
assertNotRegexpMatches = _deprecate(assertNotRegex)
class FunctionTestCase(TestCase):
"""A test case that wraps a test function.
+17 -15
View File
@@ -3,12 +3,12 @@
import sys
import argparse
import os
import warnings
from . import loader, runner
from .signals import installHandler
__unittest = True
_NO_TESTS_EXITCODE = 5
MAIN_EXAMPLES = """\
Examples:
@@ -66,7 +66,8 @@ class TestProgram(object):
def __init__(self, module='__main__', defaultTest=None, argv=None,
testRunner=None, testLoader=loader.defaultTestLoader,
exit=True, verbosity=1, failfast=None, catchbreak=None,
buffer=None, warnings=None, *, tb_locals=False):
buffer=None, warnings=None, *, tb_locals=False,
durations=None):
if isinstance(module, str):
self.module = __import__(module)
for part in module.split('.')[1:]:
@@ -82,6 +83,7 @@ class TestProgram(object):
self.verbosity = verbosity
self.buffer = buffer
self.tb_locals = tb_locals
self.durations = durations
if warnings is None and not sys.warnoptions:
# even if DeprecationWarnings are ignored by default
# print them anyway unless other warnings settings are
@@ -101,16 +103,6 @@ class TestProgram(object):
self.parseArgs(argv)
self.runTests()
def usageExit(self, msg=None):
warnings.warn("TestProgram.usageExit() is deprecated and will be"
" removed in Python 3.13", DeprecationWarning)
if msg:
print(msg)
if self._discovery_parser is None:
self._initArgParsers()
self._print_help()
sys.exit(2)
def _print_help(self, *args, **kwargs):
if self.module is None:
print(self._main_parser.format_help())
@@ -178,6 +170,9 @@ class TestProgram(object):
parser.add_argument('--locals', dest='tb_locals',
action='store_true',
help='Show local variables in tracebacks')
parser.add_argument('--durations', dest='durations', type=int,
default=None, metavar="N",
help='Show the N slowest test cases (N=0 for all)')
if self.failfast is None:
parser.add_argument('-f', '--failfast', dest='failfast',
action='store_true',
@@ -258,9 +253,10 @@ class TestProgram(object):
failfast=self.failfast,
buffer=self.buffer,
warnings=self.warnings,
tb_locals=self.tb_locals)
tb_locals=self.tb_locals,
durations=self.durations)
except TypeError:
# didn't accept the tb_locals argument
# didn't accept the tb_locals or durations argument
testRunner = self.testRunner(verbosity=self.verbosity,
failfast=self.failfast,
buffer=self.buffer,
@@ -273,6 +269,12 @@ class TestProgram(object):
testRunner = self.testRunner
self.result = testRunner.run(self.test)
if self.exit:
sys.exit(not self.result.wasSuccessful())
if self.result.testsRun == 0 and len(self.result.skipped) == 0:
sys.exit(_NO_TESTS_EXITCODE)
elif self.result.wasSuccessful():
sys.exit(0)
else:
sys.exit(1)
main = TestProgram
+250 -47
View File
@@ -14,6 +14,7 @@ __all__ = (
'call',
'create_autospec',
'AsyncMock',
'ThreadingMock',
'FILTER_DIR',
'NonCallableMock',
'NonCallableMagicMock',
@@ -32,6 +33,7 @@ import sys
import builtins
import pkgutil
from asyncio import iscoroutinefunction
import threading
from types import CodeType, ModuleType, MethodType
from unittest.util import safe_repr
from functools import wraps, partial
@@ -98,6 +100,12 @@ def _get_signature_object(func, as_instance, eat_self):
func = func.__init__
# Skip the `self` argument in __init__
eat_self = True
elif isinstance(func, (classmethod, staticmethod)):
if isinstance(func, classmethod):
# Skip the `cls` argument of a class method
eat_self = True
# Use the original decorated method to extract the correct function signature
func = func.__func__
elif not isinstance(func, FunctionTypes):
# If we really want to model an instance of the passed type,
# __call__ should be looked up, not __init__.
@@ -198,6 +206,28 @@ def _set_signature(mock, original, instance=False):
_setup_func(funcopy, mock, sig)
return funcopy
def _set_async_signature(mock, original, instance=False, is_async_mock=False):
# creates an async function with signature (*args, **kwargs) that delegates to a
# mock. It still does signature checking by calling a lambda with the same
# signature as the original.
skipfirst = isinstance(original, type)
func, sig = _get_signature_object(original, instance, skipfirst)
def checksig(*args, **kwargs):
sig.bind(*args, **kwargs)
_copy_func_details(func, checksig)
name = original.__name__
context = {'_checksig_': checksig, 'mock': mock}
src = """async def %s(*args, **kwargs):
_checksig_(*args, **kwargs)
return await mock(*args, **kwargs)""" % name
exec (src, context)
funcopy = context[name]
_setup_func(funcopy, mock, sig)
_setup_async_mock(funcopy)
return funcopy
def _setup_func(funcopy, mock, sig):
funcopy.mock = mock
@@ -411,15 +441,18 @@ class NonCallableMock(Base):
# necessary.
_lock = RLock()
def __new__(cls, /, *args, **kw):
def __new__(
cls, spec=None, wraps=None, name=None, spec_set=None,
parent=None, _spec_state=None, _new_name='', _new_parent=None,
_spec_as_instance=False, _eat_self=None, unsafe=False, **kwargs
):
# every instance has its own class
# so we can create magic methods on the
# class without stomping on other mocks
bases = (cls,)
if not issubclass(cls, AsyncMockMixin):
# Check if spec is an async object or function
bound_args = _MOCK_SIG.bind_partial(cls, *args, **kw).arguments
spec_arg = bound_args.get('spec_set', bound_args.get('spec'))
spec_arg = spec_set or spec
if spec_arg is not None and _is_async_obj(spec_arg):
bases = (AsyncMockMixin, cls)
new = type(cls.__name__, bases, {'__doc__': cls.__doc__})
@@ -505,10 +538,6 @@ class NonCallableMock(Base):
_spec_signature = None
_spec_asyncs = []
for attr in dir(spec):
if iscoroutinefunction(getattr(spec, attr, None)):
_spec_asyncs.append(attr)
if spec is not None and not _is_list(spec):
if isinstance(spec, type):
_spec_class = spec
@@ -518,7 +547,19 @@ class NonCallableMock(Base):
_spec_as_instance, _eat_self)
_spec_signature = res and res[1]
spec = dir(spec)
spec_list = dir(spec)
for attr in spec_list:
static_attr = inspect.getattr_static(spec, attr, None)
unwrapped_attr = static_attr
try:
unwrapped_attr = inspect.unwrap(unwrapped_attr)
except ValueError:
pass
if iscoroutinefunction(unwrapped_attr):
_spec_asyncs.append(attr)
spec = spec_list
__dict__ = self.__dict__
__dict__['_spec_class'] = _spec_class
@@ -532,7 +573,7 @@ class NonCallableMock(Base):
if self._mock_delegate is not None:
ret = self._mock_delegate.return_value
if ret is DEFAULT:
if ret is DEFAULT and self._mock_wraps is None:
ret = self._get_child_mock(
_new_parent=self, _new_name='()'
)
@@ -587,7 +628,9 @@ class NonCallableMock(Base):
side_effect = property(__get_side_effect, __set_side_effect)
def reset_mock(self, visited=None,*, return_value=False, side_effect=False):
def reset_mock(self, visited=None, *,
return_value: bool = False,
side_effect: bool = False):
"Restore the mock object to its initial state."
if visited is None:
visited = []
@@ -648,7 +691,7 @@ class NonCallableMock(Base):
elif _is_magic(name):
raise AttributeError(name)
if not self._mock_unsafe and (not self._mock_methods or name not in self._mock_methods):
if name.startswith(('assert', 'assret', 'asert', 'aseert', 'assrt')):
if name.startswith(('assert', 'assret', 'asert', 'aseert', 'assrt')) or name in _ATTRIB_DENY_LIST:
raise AttributeError(
f"{name!r} is not a valid assertion. Use a spec "
f"for the mock if {name!r} is meant to be an attribute.")
@@ -789,6 +832,9 @@ class NonCallableMock(Base):
mock_name = f'{self._extract_mock_name()}.{name}'
raise AttributeError(f'Cannot set {mock_name}')
if isinstance(value, PropertyMock):
self.__dict__[name] = value
return
return object.__setattr__(self, name, value)
@@ -816,7 +862,7 @@ class NonCallableMock(Base):
def _format_mock_failure_message(self, args, kwargs, action='call'):
message = 'expected %s not found.\nExpected: %s\nActual: %s'
message = 'expected %s not found.\nExpected: %s\n Actual: %s'
expected_string = self._format_mock_call_signature(args, kwargs)
call_args = self.call_args
actual_string = self._format_mock_call_signature(*call_args)
@@ -919,7 +965,7 @@ class NonCallableMock(Base):
if self.call_args is None:
expected = self._format_mock_call_signature(args, kwargs)
actual = 'not called.'
error_message = ('expected call not found.\nExpected: %s\nActual: %s'
error_message = ('expected call not found.\nExpected: %s\n Actual: %s'
% (expected, actual))
raise AssertionError(error_message)
@@ -969,8 +1015,8 @@ class NonCallableMock(Base):
for e in expected])
raise AssertionError(
f'{problem}\n'
f'Expected: {_CallList(calls)}'
f'{self._calls_repr(prefix="Actual").rstrip(".")}'
f'Expected: {_CallList(calls)}\n'
f' Actual: {safe_repr(self.mock_calls)}'
) from cause
return
@@ -1044,7 +1090,7 @@ class NonCallableMock(Base):
return klass(**kw)
def _calls_repr(self, prefix="Calls"):
def _calls_repr(self):
"""Renders self.mock_calls as a string.
Example: "\nCalls: [call(1), call(2)]."
@@ -1054,10 +1100,15 @@ class NonCallableMock(Base):
"""
if not self.mock_calls:
return ""
return f"\n{prefix}: {safe_repr(self.mock_calls)}."
return f"\nCalls: {safe_repr(self.mock_calls)}."
_MOCK_SIG = inspect.signature(NonCallableMock.__init__)
# Denylist for forbidden attribute names in safe mode
_ATTRIB_DENY_LIST = frozenset({
name.removeprefix("assert_")
for name in dir(NonCallableMock)
if name.startswith("assert_")
})
class _AnyComparer(list):
@@ -1188,6 +1239,9 @@ class CallableMixin(Base):
if self._mock_return_value is not DEFAULT:
return self.return_value
if self._mock_delegate and self._mock_delegate.return_value is not DEFAULT:
return self.return_value
if self._mock_wraps is not None:
return self._mock_wraps(*args, **kwargs)
@@ -1229,9 +1283,11 @@ class Mock(CallableMixin, NonCallableMock):
`return_value` attribute.
* `unsafe`: By default, accessing any attribute whose name starts with
*assert*, *assret*, *asert*, *aseert* or *assrt* will raise an
AttributeError. Passing `unsafe=True` will allow access to
these attributes.
*assert*, *assret*, *asert*, *aseert*, or *assrt* raises an AttributeError.
Additionally, an AttributeError is raised when accessing
attributes that match the name of an assertion method without the prefix
`assert_`, e.g. accessing `called_once` instead of `assert_called_once`.
Passing `unsafe=True` will allow access to these attributes.
* `wraps`: Item for the mock object to wrap. If `wraps` is not None then
calling the Mock will pass the call through to the wrapped object
@@ -1303,6 +1359,7 @@ class _patch(object):
self.autospec = autospec
self.kwargs = kwargs
self.additional_patchers = []
self.is_started = False
def copy(self):
@@ -1415,6 +1472,9 @@ class _patch(object):
def __enter__(self):
"""Perform the patch."""
if self.is_started:
raise RuntimeError("Patch is already started")
new, spec, spec_set = self.new, self.spec, self.spec_set
autospec, kwargs = self.autospec, self.kwargs
new_callable = self.new_callable
@@ -1457,13 +1517,12 @@ class _patch(object):
if isinstance(original, type):
# If we're patching out a class and there is a spec
inherit = True
if spec is None and _is_async_obj(original):
Klass = AsyncMock
else:
Klass = MagicMock
_kwargs = {}
# Determine the Klass to use
if new_callable is not None:
Klass = new_callable
elif spec is None and _is_async_obj(original):
Klass = AsyncMock
elif spec is not None or spec_set is not None:
this_spec = spec
if spec_set is not None:
@@ -1476,7 +1535,12 @@ class _patch(object):
Klass = AsyncMock
elif not_callable:
Klass = NonCallableMagicMock
else:
Klass = MagicMock
else:
Klass = MagicMock
_kwargs = {}
if spec is not None:
_kwargs['spec'] = spec
if spec_set is not None:
@@ -1542,6 +1606,7 @@ class _patch(object):
self.temp_original = original
self.is_local = local
self._exit_stack = contextlib.ExitStack()
self.is_started = True
try:
setattr(self.target, self.attribute, new_attr)
if self.attribute_name is not None:
@@ -1561,6 +1626,9 @@ class _patch(object):
def __exit__(self, *exc_info):
"""Undo the patch."""
if not self.is_started:
return
if self.is_local and self.temp_original is not DEFAULT:
setattr(self.target, self.attribute, self.temp_original)
else:
@@ -1577,6 +1645,7 @@ class _patch(object):
del self.target
exit_stack = self._exit_stack
del self._exit_stack
self.is_started = False
return exit_stack.__exit__(*exc_info)
@@ -2138,10 +2207,8 @@ class NonCallableMagicMock(MagicMixin, NonCallableMock):
class AsyncMagicMixin(MagicMixin):
def __init__(self, /, *args, **kw):
self._mock_set_magics() # make magic work for kwargs in init
_safe_super(AsyncMagicMixin, self).__init__(*args, **kw)
self._mock_set_magics() # fix magic broken by upper level init
pass
class MagicMock(MagicMixin, Mock):
"""
@@ -2163,6 +2230,17 @@ class MagicMock(MagicMixin, Mock):
self._mock_add_spec(spec, spec_set)
self._mock_set_magics()
def reset_mock(self, /, *args, return_value: bool = False, **kwargs):
if (
return_value
and self._mock_name
and _is_magic(self._mock_name)
):
# Don't reset return values for magic methods,
# otherwise `m.__str__` will start
# to return `MagicMock` instances, instead of `str` instances.
return_value = False
super().reset_mock(*args, return_value=return_value, **kwargs)
class MagicProxy(Base):
@@ -2183,6 +2261,13 @@ class MagicProxy(Base):
return self.create_mock()
try:
_CODE_SIG = inspect.signature(partial(CodeType.__init__, None))
_CODE_ATTRS = dir(CodeType)
except ValueError:
_CODE_SIG = None
class AsyncMockMixin(Base):
await_count = _delegating_property('await_count')
await_args = _delegating_property('await_args')
@@ -2200,8 +2285,21 @@ class AsyncMockMixin(Base):
self.__dict__['_mock_await_count'] = 0
self.__dict__['_mock_await_args'] = None
self.__dict__['_mock_await_args_list'] = _CallList()
code_mock = NonCallableMock(spec_set=CodeType)
code_mock.co_flags = inspect.CO_COROUTINE
if _CODE_SIG:
code_mock = NonCallableMock(spec_set=_CODE_ATTRS)
code_mock.__dict__["_spec_class"] = CodeType
code_mock.__dict__["_spec_signature"] = _CODE_SIG
else:
code_mock = NonCallableMock(spec_set=CodeType)
code_mock.co_flags = (
inspect.CO_COROUTINE
+ inspect.CO_VARARGS
+ inspect.CO_VARKEYWORDS
)
code_mock.co_argcount = 0
code_mock.co_varnames = ('args', 'kwargs')
code_mock.co_posonlyargcount = 0
code_mock.co_kwonlyargcount = 0
self.__dict__['__code__'] = code_mock
self.__dict__['__name__'] = 'AsyncMock'
self.__dict__['__defaults__'] = tuple()
@@ -2679,6 +2777,12 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
if not unsafe:
_check_spec_arg_typos(kwargs)
_name = kwargs.pop('name', _name)
_new_name = _name
if _parent is None:
# for a top level object no _new_name should be set
_new_name = ''
_kwargs.update(kwargs)
Klass = MagicMock
@@ -2696,31 +2800,28 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
elif is_type and instance and not _instance_callable(spec):
Klass = NonCallableMagicMock
_name = _kwargs.pop('name', _name)
_new_name = _name
if _parent is None:
# for a top level object no _new_name should be set
_new_name = ''
mock = Klass(parent=_parent, _new_parent=_parent, _new_name=_new_name,
name=_name, **_kwargs)
if isinstance(spec, FunctionTypes):
# should only happen at the top level because we don't
# recurse for functions
mock = _set_signature(mock, spec)
if is_async_func:
_setup_async_mock(mock)
mock = _set_async_signature(mock, spec)
else:
mock = _set_signature(mock, spec)
else:
_check_signature(spec, mock, is_type, instance)
if _parent is not None and not instance:
_parent._mock_children[_name] = mock
# Pop wraps from kwargs because it must not be passed to configure_mock.
wrapped = kwargs.pop('wraps', None)
if is_type and not instance and 'return_value' not in kwargs:
mock.return_value = create_autospec(spec, spec_set, instance=True,
_name='()', _parent=mock)
_name='()', _parent=mock,
wraps=wrapped)
for entry in dir(spec):
if _is_magic(entry):
@@ -2741,9 +2842,12 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
except AttributeError:
continue
kwargs = {'spec': original}
child_kwargs = {'spec': original}
# Wrap child attributes also.
if wrapped and hasattr(wrapped, entry):
child_kwargs.update(wraps=original)
if spec_set:
kwargs = {'spec_set': original}
child_kwargs = {'spec_set': original}
if not isinstance(original, FunctionTypes):
new = _SpecState(original, spec_set, mock, entry, instance)
@@ -2754,15 +2858,15 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
parent = mock.mock
skipfirst = _must_skip(spec, entry, is_type)
kwargs['_eat_self'] = skipfirst
child_kwargs['_eat_self'] = skipfirst
if iscoroutinefunction(original):
child_klass = AsyncMock
else:
child_klass = MagicMock
new = child_klass(parent=parent, name=entry, _new_name=entry,
_new_parent=parent,
**kwargs)
_new_parent=parent, **child_kwargs)
mock._mock_children[entry] = new
new.return_value = child_klass()
_check_signature(original, new, skipfirst=skipfirst)
# so functions created with _set_signature become instance attributes,
@@ -2771,6 +2875,11 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
# setting as an instance attribute?
if isinstance(new, FunctionTypes):
setattr(mock, entry, new)
# kwargs are passed with respect to the parent mock so, they are not used
# for creating return_value of the parent mock. So, this condition
# should be true only for the parent mock if kwargs are given.
if _is_instance_mock(mock) and kwargs:
mock.configure_mock(**kwargs)
return mock
@@ -2876,6 +2985,9 @@ def mock_open(mock=None, read_data=''):
return handle.readline.return_value
return next(_state[0])
def _exit_side_effect(exctype, excinst, exctb):
handle.close()
global file_spec
if file_spec is None:
import _io
@@ -2902,6 +3014,7 @@ def mock_open(mock=None, read_data=''):
handle.readlines.side_effect = _readlines_side_effect
handle.__iter__.side_effect = _iter_side_effect
handle.__next__.side_effect = _next_side_effect
handle.__exit__.side_effect = _exit_side_effect
def reset_data(*args, **kwargs):
_state[0] = _to_stream(read_data)
@@ -2934,6 +3047,96 @@ class PropertyMock(Mock):
self(val)
_timeout_unset = sentinel.TIMEOUT_UNSET
class ThreadingMixin(Base):
DEFAULT_TIMEOUT = None
def _get_child_mock(self, /, **kw):
if isinstance(kw.get("parent"), ThreadingMixin):
kw["timeout"] = kw["parent"]._mock_wait_timeout
elif isinstance(kw.get("_new_parent"), ThreadingMixin):
kw["timeout"] = kw["_new_parent"]._mock_wait_timeout
return super()._get_child_mock(**kw)
def __init__(self, *args, timeout=_timeout_unset, **kwargs):
super().__init__(*args, **kwargs)
if timeout is _timeout_unset:
timeout = self.DEFAULT_TIMEOUT
self.__dict__["_mock_event"] = threading.Event() # Event for any call
self.__dict__["_mock_calls_events"] = [] # Events for each of the calls
self.__dict__["_mock_calls_events_lock"] = threading.Lock()
self.__dict__["_mock_wait_timeout"] = timeout
def reset_mock(self, /, *args, **kwargs):
"""
See :func:`.Mock.reset_mock()`
"""
super().reset_mock(*args, **kwargs)
self.__dict__["_mock_event"] = threading.Event()
self.__dict__["_mock_calls_events"] = []
def __get_event(self, expected_args, expected_kwargs):
with self._mock_calls_events_lock:
for args, kwargs, event in self._mock_calls_events:
if (args, kwargs) == (expected_args, expected_kwargs):
return event
new_event = threading.Event()
self._mock_calls_events.append((expected_args, expected_kwargs, new_event))
return new_event
def _mock_call(self, *args, **kwargs):
ret_value = super()._mock_call(*args, **kwargs)
call_event = self.__get_event(args, kwargs)
call_event.set()
self._mock_event.set()
return ret_value
def wait_until_called(self, *, timeout=_timeout_unset):
"""Wait until the mock object is called.
`timeout` - time to wait for in seconds, waits forever otherwise.
Defaults to the constructor provided timeout.
Use None to block undefinetively.
"""
if timeout is _timeout_unset:
timeout = self._mock_wait_timeout
if not self._mock_event.wait(timeout=timeout):
msg = (f"{self._mock_name or 'mock'} was not called before"
f" timeout({timeout}).")
raise AssertionError(msg)
def wait_until_any_call_with(self, *args, **kwargs):
"""Wait until the mock object is called with given args.
Waits for the timeout in seconds provided in the constructor.
"""
event = self.__get_event(args, kwargs)
if not event.wait(timeout=self._mock_wait_timeout):
expected_string = self._format_mock_call_signature(args, kwargs)
raise AssertionError(f'{expected_string} call not found')
class ThreadingMock(ThreadingMixin, MagicMixin, Mock):
"""
A mock that can be used to wait until on calls happening
in a different thread.
The constructor can take a `timeout` argument which
controls the timeout in seconds for all `wait` calls of the mock.
You can change the default timeout of all instances via the
`ThreadingMock.DEFAULT_TIMEOUT` attribute.
If no timeout is set, it will block undefinetively.
"""
pass
def seal(mock):
"""Disable the automatic generation of child mocks.
+12
View File
@@ -43,6 +43,7 @@ class TestResult(object):
self.skipped = []
self.expectedFailures = []
self.unexpectedSuccesses = []
self.collectedDurations = []
self.shouldStop = False
self.buffer = False
self.tb_locals = False
@@ -157,6 +158,17 @@ class TestResult(object):
"""Called when a test was expected to fail, but succeed."""
self.unexpectedSuccesses.append(test)
def addDuration(self, test, elapsed):
"""Called when a test finished to run, regardless of its outcome.
*test* is the test case corresponding to the test method.
*elapsed* is the time represented in seconds, and it includes the
execution of cleanup functions.
"""
# support for a TextTestRunner using an old TestResult class
if hasattr(self, "collectedDurations"):
# Pass test repr and not the test object itself to avoid resources leak
self.collectedDurations.append((str(test), elapsed))
def wasSuccessful(self):
"""Tells whether or not this result was a success."""
# The hasattr check is for test_result's OldResult test. That
+41 -12
View File
@@ -35,13 +35,16 @@ class TextTestResult(result.TestResult):
separator1 = '=' * 70
separator2 = '-' * 70
def __init__(self, stream, descriptions, verbosity):
def __init__(self, stream, descriptions, verbosity, *, durations=None):
"""Construct a TextTestResult. Subclasses should accept **kwargs
to ensure compatibility as the interface changes."""
super(TextTestResult, self).__init__(stream, descriptions, verbosity)
self.stream = stream
self.showAll = verbosity > 1
self.dots = verbosity == 1
self.descriptions = descriptions
self._newline = True
self.durations = durations
def getDescription(self, test):
doc_first_line = test.shortDescription()
@@ -168,7 +171,7 @@ class TextTestRunner(object):
def __init__(self, stream=None, descriptions=True, verbosity=1,
failfast=False, buffer=False, resultclass=None, warnings=None,
*, tb_locals=False):
*, tb_locals=False, durations=None):
"""Construct a TextTestRunner.
Subclasses should accept **kwargs to ensure compatibility as the
@@ -182,12 +185,41 @@ class TextTestRunner(object):
self.failfast = failfast
self.buffer = buffer
self.tb_locals = tb_locals
self.durations = durations
self.warnings = warnings
if resultclass is not None:
self.resultclass = resultclass
def _makeResult(self):
return self.resultclass(self.stream, self.descriptions, self.verbosity)
try:
return self.resultclass(self.stream, self.descriptions,
self.verbosity, durations=self.durations)
except TypeError:
# didn't accept the durations argument
return self.resultclass(self.stream, self.descriptions,
self.verbosity)
def _printDurations(self, result):
if not result.collectedDurations:
return
ls = sorted(result.collectedDurations, key=lambda x: x[1],
reverse=True)
if self.durations > 0:
ls = ls[:self.durations]
self.stream.writeln("Slowest test durations")
if hasattr(result, 'separator2'):
self.stream.writeln(result.separator2)
hidden = False
for test, elapsed in ls:
if self.verbosity < 2 and elapsed < 0.001:
hidden = True
continue
self.stream.writeln("%-10s %s" % ("%.3fs" % elapsed, test))
if hidden:
self.stream.writeln("\n(durations < 0.001s were hidden; "
"use -v to show these durations)")
else:
self.stream.writeln("")
def run(self, test):
"Run the given test case or test suite."
@@ -200,15 +232,6 @@ class TextTestRunner(object):
if self.warnings:
# if self.warnings is set, use it to filter all the warnings
warnings.simplefilter(self.warnings)
# if the filter is 'default' or 'always', special-case the
# warnings from the deprecated unittest methods to show them
# no more than once per module, because they can be fairly
# noisy. The -Wd and -Wa flags can be used to bypass this
# only when self.warnings is None.
if self.warnings in ['default', 'always']:
warnings.filterwarnings('module',
category=DeprecationWarning,
message=r'Please use assert\w+ instead.')
startTime = time.perf_counter()
startTestRun = getattr(result, 'startTestRun', None)
if startTestRun is not None:
@@ -222,8 +245,12 @@ class TextTestRunner(object):
stopTime = time.perf_counter()
timeTaken = stopTime - startTime
result.printErrors()
if self.durations is not None:
self._printDurations(result)
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))
@@ -247,6 +274,8 @@ class TextTestRunner(object):
infos.append("failures=%d" % failed)
if errored:
infos.append("errors=%d" % errored)
elif run == 0 and not skipped:
self.stream.write("NO TESTS RAN")
else:
self.stream.write("OK")
if skipped:
-25
View File
@@ -1,25 +0,0 @@
import os
import sys
import unittest
here = os.path.dirname(__file__)
loader = unittest.defaultTestLoader
def suite():
suite = unittest.TestSuite()
for fn in os.listdir(here):
if fn.startswith("test") and fn.endswith(".py"):
modname = "unittest.test." + fn[:-3]
try:
__import__(modname)
except unittest.SkipTest:
continue
module = sys.modules[modname]
suite.addTest(loader.loadTestsFromModule(module))
suite.addTest(loader.loadTestsFromName('unittest.test.testmock'))
return suite
if __name__ == "__main__":
unittest.main(defaultTest="suite")
-18
View File
@@ -1,18 +0,0 @@
import os
import unittest
def load_tests(loader, standard_tests, pattern):
# top level directory cached on loader instance
this_dir = os.path.dirname(__file__)
pattern = pattern or "test_*.py"
# We are inside unittest.test, so the top-level is two notches up
top_level_dir = os.path.dirname(os.path.dirname(this_dir))
package_tests = loader.discover(start_dir=this_dir, pattern=pattern,
top_level_dir=top_level_dir)
standard_tests.addTests(package_tests)
return standard_tests
if __name__ == '__main__':
unittest.main()
-19
View File
@@ -1,19 +0,0 @@
import os
import sys
import unittest
here = os.path.dirname(__file__)
loader = unittest.defaultTestLoader
def load_tests(*args):
suite = unittest.TestSuite()
# TODO: RUSTPYTHON; allow objects to be mocked better
return suite
for fn in os.listdir(here):
if fn.startswith("test") and fn.endswith(".py"):
modname = "unittest.test.testmock." + fn[:-3]
__import__(modname)
module = sys.modules[modname]
suite.addTest(loader.loadTestsFromModule(module))
return suite