333
343
def test_scenarios(self):
334
344
# check that constructor parameters are passed through to the adapted
336
from bzrlib.tests.per_workingtree import make_scenarios
346
from .per_workingtree import make_scenarios
339
formats = [workingtree.WorkingTreeFormat2(),
340
workingtree.WorkingTreeFormat3(),]
341
scenarios = make_scenarios(server1, server2, formats)
349
formats = [workingtree_4.WorkingTreeFormat4(),
350
workingtree_3.WorkingTreeFormat3(),
351
workingtree_4.WorkingTreeFormat6()]
352
scenarios = make_scenarios(server1, server2, formats,
353
remote_server='c', remote_readonly_server='d',
354
remote_backing_server='e')
342
355
self.assertEqual([
343
('WorkingTreeFormat2',
344
{'bzrdir_format': formats[0]._matchingbzrdir,
356
('WorkingTreeFormat4',
357
{'bzrdir_format': formats[0]._matchingcontroldir,
345
358
'transport_readonly_server': 'b',
346
359
'transport_server': 'a',
347
360
'workingtree_format': formats[0]}),
348
361
('WorkingTreeFormat3',
349
{'bzrdir_format': formats[1]._matchingbzrdir,
350
'transport_readonly_server': 'b',
351
'transport_server': 'a',
352
'workingtree_format': formats[1]})],
362
{'bzrdir_format': formats[1]._matchingcontroldir,
363
'transport_readonly_server': 'b',
364
'transport_server': 'a',
365
'workingtree_format': formats[1]}),
366
('WorkingTreeFormat6',
367
{'bzrdir_format': formats[2]._matchingcontroldir,
368
'transport_readonly_server': 'b',
369
'transport_server': 'a',
370
'workingtree_format': formats[2]}),
371
('WorkingTreeFormat6,remote',
372
{'bzrdir_format': formats[2]._matchingcontroldir,
373
'repo_is_remote': True,
374
'transport_readonly_server': 'd',
375
'transport_server': 'c',
376
'vfs_transport_factory': 'e',
377
'workingtree_format': formats[2]}),
356
381
class TestTreeScenarios(tests.TestCase):
358
383
def test_scenarios(self):
359
384
# the tree implementation scenario generator is meant to setup one
360
# instance for each working tree format, and one additional instance
385
# instance for each working tree format, one additional instance
361
386
# that will use the default wt format, but create a revision tree for
362
# the tests. this means that the wt ones should have the
363
# workingtree_to_test_tree attribute set to 'return_parameter' and the
364
# revision one set to revision_tree_from_workingtree.
387
# the tests, and one more that uses the default wt format as a
388
# lightweight checkout of a remote repository. This means that the wt
389
# ones should have the workingtree_to_test_tree attribute set to
390
# 'return_parameter' and the revision one set to
391
# revision_tree_from_workingtree.
366
from bzrlib.tests.per_tree import (
393
from .per_tree import (
367
394
_dirstate_tree_from_workingtree,
369
396
preview_tree_pre,
376
formats = [workingtree.WorkingTreeFormat2(),
377
workingtree.WorkingTreeFormat3(),]
403
smart_server = test_server.SmartTCPServer_for_testing
404
smart_readonly_server = test_server.ReadonlySmartTCPServer_for_testing
405
mem_server = memory.MemoryServer
406
formats = [workingtree_4.WorkingTreeFormat4(),
407
workingtree_3.WorkingTreeFormat3(), ]
378
408
scenarios = make_scenarios(server1, server2, formats)
379
self.assertEqual(7, len(scenarios))
380
default_wt_format = workingtree.WorkingTreeFormat4._default_format
381
wt4_format = workingtree.WorkingTreeFormat4()
382
wt5_format = workingtree.WorkingTreeFormat5()
409
self.assertEqual(9, len(scenarios))
410
default_wt_format = workingtree.format_registry.get_default()
411
wt4_format = workingtree_4.WorkingTreeFormat4()
412
wt5_format = workingtree_4.WorkingTreeFormat5()
413
wt6_format = workingtree_4.WorkingTreeFormat6()
414
git_wt_format = git_workingtree.GitWorkingTreeFormat()
383
415
expected_scenarios = [
384
('WorkingTreeFormat2',
385
{'bzrdir_format': formats[0]._matchingbzrdir,
416
('WorkingTreeFormat4',
417
{'bzrdir_format': formats[0]._matchingcontroldir,
386
418
'transport_readonly_server': 'b',
387
419
'transport_server': 'a',
388
420
'workingtree_format': formats[0],
389
421
'_workingtree_to_test_tree': return_parameter,
391
423
('WorkingTreeFormat3',
392
{'bzrdir_format': formats[1]._matchingbzrdir,
424
{'bzrdir_format': formats[1]._matchingcontroldir,
393
425
'transport_readonly_server': 'b',
394
426
'transport_server': 'a',
395
427
'workingtree_format': formats[1],
396
428
'_workingtree_to_test_tree': return_parameter,
430
('WorkingTreeFormat6,remote',
431
{'bzrdir_format': wt6_format._matchingcontroldir,
432
'repo_is_remote': True,
433
'transport_readonly_server': smart_readonly_server,
434
'transport_server': smart_server,
435
'vfs_transport_factory': mem_server,
436
'workingtree_format': wt6_format,
437
'_workingtree_to_test_tree': return_parameter,
399
440
{'_workingtree_to_test_tree': revision_tree_from_workingtree,
400
'bzrdir_format': default_wt_format._matchingbzrdir,
441
'bzrdir_format': default_wt_format._matchingcontroldir,
401
442
'transport_readonly_server': 'b',
402
443
'transport_server': 'a',
403
444
'workingtree_format': default_wt_format,
447
{'_workingtree_to_test_tree': revision_tree_from_workingtree,
448
'bzrdir_format': git_wt_format._matchingcontroldir,
449
'transport_readonly_server': 'b',
450
'transport_server': 'a',
451
'workingtree_format': git_wt_format,
405
454
('DirStateRevisionTree,WT4',
406
455
{'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
407
'bzrdir_format': wt4_format._matchingbzrdir,
456
'bzrdir_format': wt4_format._matchingcontroldir,
408
457
'transport_readonly_server': 'b',
409
458
'transport_server': 'a',
410
459
'workingtree_format': wt4_format,
412
461
('DirStateRevisionTree,WT5',
413
462
{'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
414
'bzrdir_format': wt5_format._matchingbzrdir,
463
'bzrdir_format': wt5_format._matchingcontroldir,
415
464
'transport_readonly_server': 'b',
416
465
'transport_server': 'a',
417
466
'workingtree_format': wt5_format,
420
469
{'_workingtree_to_test_tree': preview_tree_pre,
421
'bzrdir_format': default_wt_format._matchingbzrdir,
470
'bzrdir_format': default_wt_format._matchingcontroldir,
422
471
'transport_readonly_server': 'b',
423
472
'transport_server': 'a',
424
473
'workingtree_format': default_wt_format}),
425
474
('PreviewTreePost',
426
475
{'_workingtree_to_test_tree': preview_tree_post,
427
'bzrdir_format': default_wt_format._matchingbzrdir,
476
'bzrdir_format': default_wt_format._matchingcontroldir,
428
477
'transport_readonly_server': 'b',
429
478
'transport_server': 'a',
430
479
'workingtree_format': default_wt_format}),
432
481
self.assertEqual(expected_scenarios, scenarios)
548
603
tree = self.make_branch_and_memory_tree('dir')
549
604
# Guard against regression into MemoryTransport leaking
550
605
# files to disk instead of keeping them in memory.
551
self.failIf(osutils.lexists('dir'))
606
self.assertFalse(osutils.lexists('dir'))
552
607
self.assertIsInstance(tree, memorytree.MemoryTree)
554
609
def test_make_branch_and_memory_tree_with_format(self):
555
610
"""make_branch_and_memory_tree should accept a format option."""
556
611
format = bzrdir.BzrDirMetaFormat1()
557
format.repository_format = weaverepo.RepositoryFormat7()
612
format.repository_format = repository.format_registry.get_default()
558
613
tree = self.make_branch_and_memory_tree('dir', format=format)
559
614
# Guard against regression into MemoryTransport leaking
560
615
# files to disk instead of keeping them in memory.
561
self.failIf(osutils.lexists('dir'))
616
self.assertFalse(osutils.lexists('dir'))
562
617
self.assertIsInstance(tree, memorytree.MemoryTree)
563
618
self.assertEqual(format.repository_format.__class__,
564
tree.branch.repository._format.__class__)
619
tree.branch.repository._format.__class__)
566
621
def test_make_branch_builder(self):
567
622
builder = self.make_branch_builder('dir')
568
623
self.assertIsInstance(builder, branchbuilder.BranchBuilder)
569
624
# Guard against regression into MemoryTransport leaking
570
625
# files to disk instead of keeping them in memory.
571
self.failIf(osutils.lexists('dir'))
626
self.assertFalse(osutils.lexists('dir'))
573
628
def test_make_branch_builder_with_format(self):
574
629
# Use a repo layout that doesn't conform to a 'named' layout, to ensure
575
630
# that the format objects are used.
576
631
format = bzrdir.BzrDirMetaFormat1()
577
repo_format = weaverepo.RepositoryFormat7()
632
repo_format = repository.format_registry.get_default()
578
633
format.repository_format = repo_format
579
634
builder = self.make_branch_builder('dir', format=format)
580
635
the_branch = builder.get_branch()
581
636
# Guard against regression into MemoryTransport leaking
582
637
# files to disk instead of keeping them in memory.
583
self.failIf(osutils.lexists('dir'))
638
self.assertFalse(osutils.lexists('dir'))
584
639
self.assertEqual(format.repository_format.__class__,
585
640
the_branch.repository._format.__class__)
586
641
self.assertEqual(repo_format.get_format_string(),
587
642
self.get_transport().get_bytes(
588
'dir/.bzr/repository/format'))
643
'dir/.bzr/repository/format'))
590
645
def test_make_branch_builder_with_format_name(self):
591
646
builder = self.make_branch_builder('dir', format='knit')
592
647
the_branch = builder.get_branch()
593
648
# Guard against regression into MemoryTransport leaking
594
649
# files to disk instead of keeping them in memory.
595
self.failIf(osutils.lexists('dir'))
596
dir_format = bzrdir.format_registry.make_bzrdir('knit')
650
self.assertFalse(osutils.lexists('dir'))
651
dir_format = controldir.format_registry.make_controldir('knit')
597
652
self.assertEqual(dir_format.repository_format.__class__,
598
653
the_branch.repository._format.__class__)
599
self.assertEqual('Bazaar-NG Knit Repository Format 1',
654
self.assertEqual(b'Bazaar-NG Knit Repository Format 1',
600
655
self.get_transport().get_bytes(
601
'dir/.bzr/repository/format'))
656
'dir/.bzr/repository/format'))
603
658
def test_dangling_locks_cause_failures(self):
604
659
class TestDanglingLock(tests.TestCaseWithMemoryTransport):
605
660
def test_function(self):
606
t = self.get_transport('.')
661
t = self.get_transport_from_path('.')
607
662
l = lockdir.LockDir(t, 'lock')
610
665
test = TestDanglingLock('test_function')
611
666
result = test.run()
667
total_failures = result.errors + result.failures
612
668
if self._lock_check_thorough:
613
self.assertEqual(1, len(result.errors))
669
self.assertEqual(1, len(total_failures))
615
671
# When _lock_check_thorough is disabled, then we don't trigger a
617
self.assertEqual(0, len(result.errors))
673
self.assertEqual(0, len(total_failures))
620
676
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
621
677
"""Tests for the convenience functions TestCaseWithTransport introduces."""
623
679
def test_get_readonly_url_none(self):
624
from bzrlib.transport import get_transport
625
from bzrlib.transport.readonly import ReadonlyTransportDecorator
680
from ..transport.readonly import ReadonlyTransportDecorator
626
681
self.vfs_transport_factory = memory.MemoryServer
627
682
self.transport_readonly_server = None
628
683
# calling get_readonly_transport() constructs a decorator on the url
630
685
url = self.get_readonly_url()
631
686
url2 = self.get_readonly_url('foo/bar')
632
t = get_transport(url)
633
t2 = get_transport(url2)
634
self.failUnless(isinstance(t, ReadonlyTransportDecorator))
635
self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
687
t = transport.get_transport_from_url(url)
688
t2 = transport.get_transport_from_url(url2)
689
self.assertIsInstance(t, ReadonlyTransportDecorator)
690
self.assertIsInstance(t2, ReadonlyTransportDecorator)
636
691
self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
638
693
def test_get_readonly_url_http(self):
639
from bzrlib.tests.http_server import HttpServer
640
from bzrlib.transport import get_transport
641
from bzrlib.transport.http import HttpTransportBase
694
from .http_server import HttpServer
695
from ..transport.http import HttpTransport
642
696
self.transport_server = test_server.LocalURLServer
643
697
self.transport_readonly_server = HttpServer
644
698
# calling get_readonly_transport() gives us a HTTP server instance.
645
699
url = self.get_readonly_url()
646
700
url2 = self.get_readonly_url('foo/bar')
647
701
# the transport returned may be any HttpTransportBase subclass
648
t = get_transport(url)
649
t2 = get_transport(url2)
650
self.failUnless(isinstance(t, HttpTransportBase))
651
self.failUnless(isinstance(t2, HttpTransportBase))
702
t = transport.get_transport_from_url(url)
703
t2 = transport.get_transport_from_url(url2)
704
self.assertIsInstance(t, HttpTransport)
705
self.assertIsInstance(t2, HttpTransport)
652
706
self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
654
708
def test_is_directory(self):
743
797
r"^ +[0-9]+ms\*$")
745
799
def test_unittest_reporting_unittest_class(self):
746
# getting the time from a non-bzrlib test works ok
800
# getting the time from a non-breezy test works ok
747
801
class ShortDelayTestCase(unittest.TestCase):
748
802
def test_short_delay(self):
749
803
time.sleep(0.003)
750
804
self.check_timing(ShortDelayTestCase('test_short_delay'),
753
def _patch_get_bzr_source_tree(self):
754
# Reading from the actual source tree breaks isolation, but we don't
755
# want to assume that thats *all* that would happen.
756
self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', lambda: None)
758
def test_assigned_benchmark_file_stores_date(self):
759
self._patch_get_bzr_source_tree()
761
result = bzrlib.tests.TextTestResult(self._log_file,
766
output_string = output.getvalue()
767
# if you are wondering about the regexp please read the comment in
768
# test_bench_history (bzrlib.tests.test_selftest.TestRunner)
769
# XXX: what comment? -- Andrew Bennetts
770
self.assertContainsRe(output_string, "--date [0-9.]+")
772
def test_benchhistory_records_test_times(self):
773
self._patch_get_bzr_source_tree()
774
result_stream = StringIO()
775
result = bzrlib.tests.TextTestResult(
779
bench_history=result_stream
782
# we want profile a call and check that its test duration is recorded
783
# make a new test instance that when run will generate a benchmark
784
example_test_case = TestTestResult("_time_hello_world_encoding")
785
# execute the test, which should succeed and record times
786
example_test_case.run(result)
787
lines = result_stream.getvalue().splitlines()
788
self.assertEqual(2, len(lines))
789
self.assertContainsRe(lines[1],
790
" *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
791
"._time_hello_world_encoding")
793
807
def _time_hello_world_encoding(self):
794
808
"""Profile two sleep calls
796
810
This is used to exercise the test framework.
798
self.time(unicode, 'hello', errors='replace')
799
self.time(unicode, 'world', errors='replace')
812
self.time(text_type, b'hello', errors='replace')
813
self.time(text_type, b'world', errors='replace')
801
815
def test_lsprofiling(self):
802
816
"""Verbose test result prints lsprof statistics from test cases."""
803
self.requireFeature(test_lsprof.LSProfFeature)
817
self.requireFeature(features.lsprof_feature)
804
818
result_stream = StringIO()
805
result = bzrlib.tests.VerboseTestResult(
806
unittest._WritelnDecorator(result_stream),
819
result = breezy.tests.VerboseTestResult(
826
840
# and then repeated but with 'world', rather than 'hello'.
827
841
# this should appear in the output stream of our test result.
828
842
output = result_stream.getvalue()
829
self.assertContainsRe(output,
830
r"LSProf output for <type 'unicode'>\(\('hello',\), {'errors': 'replace'}\)")
831
self.assertContainsRe(output,
832
r" *CallCount *Recursive *Total\(ms\) *Inline\(ms\) *module:lineno\(function\)\n")
833
self.assertContainsRe(output,
834
r"( +1 +0 +0\.\d+ +0\.\d+ +<method 'disable' of '_lsprof\.Profiler' objects>\n)?")
835
self.assertContainsRe(output,
836
r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
844
self.assertContainsRe(output,
845
r"LSProf output for <class 'str'>\(\(b'hello',\), {'errors': 'replace'}\)")
846
self.assertContainsRe(output,
847
r"LSProf output for <class 'str'>\(\(b'world',\), {'errors': 'replace'}\)")
849
self.assertContainsRe(output,
850
r"LSProf output for <type 'unicode'>\(\('hello',\), {'errors': 'replace'}\)")
851
self.assertContainsRe(output,
852
r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
853
self.assertContainsRe(output,
854
r" *CallCount *Recursive *Total\(ms\) *Inline\(ms\) *module:lineno\(function\)\n")
855
self.assertContainsRe(output,
856
r"( +1 +0 +0\.\d+ +0\.\d+ +<method 'disable' of '_lsprof\.Profiler' objects>\n)?")
858
def test_uses_time_from_testtools(self):
859
"""Test case timings in verbose results should use testtools times"""
862
class TimeAddedVerboseTestResult(tests.VerboseTestResult):
863
def startTest(self, test):
864
self.time(datetime.datetime.utcfromtimestamp(1.145))
865
super(TimeAddedVerboseTestResult, self).startTest(test)
867
def addSuccess(self, test):
868
self.time(datetime.datetime.utcfromtimestamp(51.147))
869
super(TimeAddedVerboseTestResult, self).addSuccess(test)
871
def report_tests_starting(self): pass
873
self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
874
self.assertEndsWith(sio.getvalue(), "OK 50002ms\n")
838
876
def test_known_failure(self):
839
"""A KnownFailure being raised should trigger several result actions."""
877
"""Using knownFailure should trigger several result actions."""
840
878
class InstrumentedTestResult(tests.ExtendedTestResult):
841
879
def stopTestRun(self): pass
842
def startTests(self): pass
843
def report_test_start(self, test): pass
881
def report_tests_starting(self): pass
844
883
def report_known_failure(self, test, err=None, details=None):
845
884
self._call = test, 'known failure'
846
885
result = InstrumentedTestResult(None, None, None, None)
847
887
class Test(tests.TestCase):
848
888
def test_function(self):
849
raise tests.KnownFailure('failed!')
889
self.knownFailure('failed!')
850
890
test = Test("test_function")
852
892
# it should invoke 'report_known_failure'.
1193
def _patch_get_bzr_source_tree(self):
1194
# Reading from the actual source tree breaks isolation, but we don't
1195
# want to assume that thats *all* that would happen.
1196
self._get_source_tree_calls = []
1198
self._get_source_tree_calls.append("called")
1200
self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', new_get)
1202
def test_bench_history(self):
1203
# tests that the running the benchmark passes bench_history into
1204
# the test result object. We can tell that happens if
1205
# _get_bzr_source_tree is called.
1206
self._patch_get_bzr_source_tree()
1207
test = TestRunner('dummy_test')
1209
runner = tests.TextTestRunner(stream=self._log_file,
1210
bench_history=output)
1211
result = self.run_test_runner(runner, test)
1212
output_string = output.getvalue()
1213
self.assertContainsRe(output_string, "--date [0-9.]+")
1214
self.assertLength(1, self._get_source_tree_calls)
1260
def test_verbose_test_count(self):
1261
"""A verbose test run reports the right test count at the start"""
1262
suite = TestUtil.TestSuite([
1263
unittest.FunctionTestCase(lambda:None),
1264
unittest.FunctionTestCase(lambda:None)])
1265
self.assertEqual(suite.countTestCases(), 2)
1267
runner = tests.TextTestRunner(stream=stream, verbosity=2)
1268
# Need to use the CountingDecorator as that's what sets num_tests
1269
self.run_test_runner(runner, tests.CountingDecorator(suite))
1270
self.assertStartsWith(stream.getvalue(), "running 2 tests")
1216
1272
def test_startTestRun(self):
1217
1273
"""run should call result.startTestRun()"""
1219
class LoggingDecorator(tests.ForwardingResult):
1276
class LoggingDecorator(ExtendedToOriginalDecorator):
1220
1277
def startTestRun(self):
1221
tests.ForwardingResult.startTestRun(self)
1278
ExtendedToOriginalDecorator.startTestRun(self)
1222
1279
calls.append('startTestRun')
1223
test = unittest.FunctionTestCase(lambda:None)
1280
test = unittest.FunctionTestCase(lambda: None)
1224
1281
stream = StringIO()
1225
1282
runner = tests.TextTestRunner(stream=stream,
1226
result_decorators=[LoggingDecorator])
1227
result = self.run_test_runner(runner, test)
1283
result_decorators=[LoggingDecorator])
1284
self.run_test_runner(runner, test)
1228
1285
self.assertLength(1, calls)
1230
1287
def test_stopTestRun(self):
1231
1288
"""run should call result.stopTestRun()"""
1233
class LoggingDecorator(tests.ForwardingResult):
1291
class LoggingDecorator(ExtendedToOriginalDecorator):
1234
1292
def stopTestRun(self):
1235
tests.ForwardingResult.stopTestRun(self)
1293
ExtendedToOriginalDecorator.stopTestRun(self)
1236
1294
calls.append('stopTestRun')
1237
test = unittest.FunctionTestCase(lambda:None)
1295
test = unittest.FunctionTestCase(lambda: None)
1238
1296
stream = StringIO()
1239
1297
runner = tests.TextTestRunner(stream=stream,
1240
result_decorators=[LoggingDecorator])
1241
result = self.run_test_runner(runner, test)
1298
result_decorators=[LoggingDecorator])
1299
self.run_test_runner(runner, test)
1242
1300
self.assertLength(1, calls)
1302
def test_unicode_test_output_on_ascii_stream(self):
1303
"""Showing results should always succeed even on an ascii console"""
1304
class FailureWithUnicode(tests.TestCase):
1305
def test_log_unicode(self):
1307
self.fail("Now print that log!")
1310
out = TextIOWrapper(bio, 'ascii', 'backslashreplace')
1312
bio = out = StringIO()
1313
self.overrideAttr(osutils, "get_terminal_encoding",
1314
lambda trace=False: "ascii")
1315
self.run_test_runner(
1316
tests.TextTestRunner(stream=out),
1317
FailureWithUnicode("test_log_unicode"))
1319
self.assertContainsRe(bio.getvalue(),
1320
b"(?:Text attachment: )?log"
1322
b"\\d+\\.\\d+ \\\\u2606"
1323
b"(?:\n-+\n|}}}\n)")
1245
1326
class SampleTestCase(tests.TestCase):
1247
1328
def _test_pass(self):
1250
1332
class _TestException(Exception):
1254
1336
class TestTestCase(tests.TestCase):
1255
"""Tests that test the core bzrlib TestCase."""
1337
"""Tests that test the core breezy TestCase."""
1257
1339
def test_assertLength_matches_empty(self):
1633
1740
obj.test_attr = 'modified'
1634
1741
self.assertEqual('modified', obj.test_attr)
1636
test = Test('test_value')
1637
test.run(unittest.TestResult())
1743
self._run_successful_test(Test('test_value'))
1638
1744
self.assertEqual('original', obj.test_attr)
1640
1746
def test_overrideAttr_with_value(self):
1641
self.test_attr = 'original' # Define a test attribute
1642
obj = self # Make 'obj' visible to the embedded test
1747
self.test_attr = 'original' # Define a test attribute
1748
obj = self # Make 'obj' visible to the embedded test
1643
1750
class Test(tests.TestCase):
1645
1752
def setUp(self):
1646
tests.TestCase.setUp(self)
1753
super(Test, self).setUp()
1647
1754
self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1649
1756
def test_value(self):
1650
1757
self.assertEqual('original', self.orig)
1651
1758
self.assertEqual('modified', obj.test_attr)
1653
test = Test('test_value')
1654
test.run(unittest.TestResult())
1760
self._run_successful_test(Test('test_value'))
1655
1761
self.assertEqual('original', obj.test_attr)
1763
def test_overrideAttr_with_no_existing_value_and_value(self):
1764
# Do not define the test_attribute
1765
obj = self # Make 'obj' visible to the embedded test
1767
class Test(tests.TestCase):
1770
tests.TestCase.setUp(self)
1771
self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1773
def test_value(self):
1774
self.assertEqual(tests._unitialized_attr, self.orig)
1775
self.assertEqual('modified', obj.test_attr)
1777
self._run_successful_test(Test('test_value'))
1778
self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1780
def test_overrideAttr_with_no_existing_value_and_no_value(self):
1781
# Do not define the test_attribute
1782
obj = self # Make 'obj' visible to the embedded test
1784
class Test(tests.TestCase):
1787
tests.TestCase.setUp(self)
1788
self.orig = self.overrideAttr(obj, 'test_attr')
1790
def test_value(self):
1791
self.assertEqual(tests._unitialized_attr, self.orig)
1792
self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1794
self._run_successful_test(Test('test_value'))
1795
self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1797
def test_recordCalls(self):
1798
from breezy.tests import test_selftest
1799
calls = self.recordCalls(
1800
test_selftest, '_add_numbers')
1801
self.assertEqual(test_selftest._add_numbers(2, 10),
1803
self.assertEqual(calls, [((2, 10), {})])
1806
def _add_numbers(a, b):
1810
class _MissingFeature(features.Feature):
1815
missing_feature = _MissingFeature()
1818
def _get_test(name):
1819
"""Get an instance of a specific example test.
1821
We protect this in a function so that they don't auto-run in the test
1825
class ExampleTests(tests.TestCase):
1827
def test_fail(self):
1828
mutter('this was a failing test')
1829
self.fail('this test will fail')
1831
def test_error(self):
1832
mutter('this test errored')
1833
raise RuntimeError('gotcha')
1835
def test_missing_feature(self):
1836
mutter('missing the feature')
1837
self.requireFeature(missing_feature)
1839
def test_skip(self):
1840
mutter('this test will be skipped')
1841
raise tests.TestSkipped('reason')
1843
def test_success(self):
1844
mutter('this test succeeds')
1846
def test_xfail(self):
1847
mutter('test with expected failure')
1848
self.knownFailure('this_fails')
1850
def test_unexpected_success(self):
1851
mutter('test with unexpected success')
1852
self.expectFailure('should_fail', lambda: None)
1854
return ExampleTests(name)
1857
class TestTestCaseLogDetails(tests.TestCase):
1859
def _run_test(self, test_name):
1860
test = _get_test(test_name)
1861
result = testtools.TestResult()
1865
def test_fail_has_log(self):
1866
result = self._run_test('test_fail')
1867
self.assertEqual(1, len(result.failures))
1868
result_content = result.failures[0][1]
1869
self.assertContainsRe(result_content,
1870
'(?m)^(?:Text attachment: )?log(?:$|: )')
1871
self.assertContainsRe(result_content, 'this was a failing test')
1873
def test_error_has_log(self):
1874
result = self._run_test('test_error')
1875
self.assertEqual(1, len(result.errors))
1876
result_content = result.errors[0][1]
1877
self.assertContainsRe(result_content,
1878
'(?m)^(?:Text attachment: )?log(?:$|: )')
1879
self.assertContainsRe(result_content, 'this test errored')
1881
def test_skip_has_no_log(self):
1882
result = self._run_test('test_skip')
1883
reasons = result.skip_reasons
1884
self.assertEqual({'reason'}, set(reasons))
1885
skips = reasons['reason']
1886
self.assertEqual(1, len(skips))
1888
self.assertFalse('log' in test.getDetails())
1890
def test_missing_feature_has_no_log(self):
1891
# testtools doesn't know about addNotSupported, so it just gets
1892
# considered as a skip
1893
result = self._run_test('test_missing_feature')
1894
reasons = result.skip_reasons
1895
self.assertEqual({str(missing_feature)}, set(reasons))
1896
skips = reasons[str(missing_feature)]
1897
self.assertEqual(1, len(skips))
1899
self.assertFalse('log' in test.getDetails())
1901
def test_xfail_has_no_log(self):
1902
result = self._run_test('test_xfail')
1903
self.assertEqual(1, len(result.expectedFailures))
1904
result_content = result.expectedFailures[0][1]
1905
self.assertNotContainsRe(result_content,
1906
'(?m)^(?:Text attachment: )?log(?:$|: )')
1907
self.assertNotContainsRe(result_content, 'test with expected failure')
1909
def test_unexpected_success_has_log(self):
1910
result = self._run_test('test_unexpected_success')
1911
self.assertEqual(1, len(result.unexpectedSuccesses))
1912
# Inconsistency, unexpectedSuccesses is a list of tests,
1913
# expectedFailures is a list of reasons?
1914
test = result.unexpectedSuccesses[0]
1915
details = test.getDetails()
1916
self.assertTrue('log' in details)
1919
class TestTestCloning(tests.TestCase):
1920
"""Tests that test cloning of TestCases (as used by multiply_tests)."""
1922
def test_cloned_testcase_does_not_share_details(self):
1923
"""A TestCase cloned with clone_test does not share mutable attributes
1924
such as details or cleanups.
1926
class Test(tests.TestCase):
1928
self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1929
orig_test = Test('test_foo')
1930
cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1931
orig_test.run(unittest.TestResult())
1932
self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1933
self.assertEqual(None, cloned_test.getDetails().get('foo'))
1935
def test_double_apply_scenario_preserves_first_scenario(self):
1936
"""Applying two levels of scenarios to a test preserves the attributes
1937
added by both scenarios.
1939
class Test(tests.TestCase):
1942
test = Test('test_foo')
1943
scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1944
scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1945
suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1946
suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1947
all_tests = list(tests.iter_suite_tests(suite))
1948
self.assertLength(4, all_tests)
1949
all_xys = sorted((t.x, t.y) for t in all_tests)
1950
self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1658
1953
# NB: Don't delete this; it's not actually from 0.11!
1659
1954
@deprecated_function(deprecated_in((0, 11, 0)))
1721
2028
sample_object = ApplyDeprecatedHelper()
1722
2029
# calling an undeprecated callable raises an assertion
1723
2030
self.assertRaises(AssertionError, self.applyDeprecated,
1724
deprecated_in((0, 11, 0)),
1725
sample_object.sample_normal_method)
2031
deprecated_in((0, 11, 0)),
2032
sample_object.sample_normal_method)
1726
2033
self.assertRaises(AssertionError, self.applyDeprecated,
1727
deprecated_in((0, 11, 0)),
1728
sample_undeprecated_function, "a param value")
2034
deprecated_in((0, 11, 0)),
2035
sample_undeprecated_function, "a param value")
1729
2036
# calling a deprecated callable (function or method) with the wrong
1730
2037
# expected deprecation fails.
1731
2038
self.assertRaises(AssertionError, self.applyDeprecated,
1732
deprecated_in((0, 10, 0)),
1733
sample_object.sample_deprecated_method, "a param value")
2039
deprecated_in((0, 10, 0)),
2040
sample_object.sample_deprecated_method,
1734
2042
self.assertRaises(AssertionError, self.applyDeprecated,
1735
deprecated_in((0, 10, 0)),
1736
sample_deprecated_function)
2043
deprecated_in((0, 10, 0)),
2044
sample_deprecated_function)
1737
2045
# calling a deprecated callable (function or method) with the right
1738
2046
# expected deprecation returns the functions result.
1739
self.assertEqual("a param value",
1740
self.applyDeprecated(deprecated_in((0, 11, 0)),
1741
sample_object.sample_deprecated_method, "a param value"))
2049
self.applyDeprecated(
2050
deprecated_in((0, 11, 0)),
2051
sample_object.sample_deprecated_method, "a param value"))
1742
2052
self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 11, 0)),
1743
sample_deprecated_function))
2053
sample_deprecated_function))
1744
2054
# calling a nested deprecation with the wrong deprecation version
1745
2055
# fails even if a deeper nested function was deprecated with the
1746
2056
# supplied version.
1747
self.assertRaises(AssertionError, self.applyDeprecated,
2058
AssertionError, self.applyDeprecated,
1748
2059
deprecated_in((0, 11, 0)), sample_object.sample_nested_deprecation)
1749
2060
# calling a nested deprecation with the right deprecation value
1750
2061
# returns the calls result.
1751
self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 10, 0)),
1752
sample_object.sample_nested_deprecation))
2063
2, self.applyDeprecated(
2064
deprecated_in((0, 10, 0)),
2065
sample_object.sample_nested_deprecation))
1754
2067
def test_callDeprecated(self):
1755
2068
def testfunc(be_deprecated, result=None):
1806
2117
self.transport_server = test_server.FakeVFATServer
1807
2118
self.assertFalse(self.get_url('t1').startswith('file://'))
1808
2119
tree = self.make_branch_and_tree('t1')
1809
base = tree.bzrdir.root_transport.base
2120
base = tree.controldir.root_transport.base
1810
2121
self.assertStartsWith(base, 'file://')
1811
self.assertEquals(tree.bzrdir.root_transport,
1812
tree.branch.bzrdir.root_transport)
1813
self.assertEquals(tree.bzrdir.root_transport,
1814
tree.branch.repository.bzrdir.root_transport)
1817
class SelfTestHelper:
2122
self.assertEqual(tree.controldir.root_transport,
2123
tree.branch.controldir.root_transport)
2124
self.assertEqual(tree.controldir.root_transport,
2125
tree.branch.repository.controldir.root_transport)
2128
class SelfTestHelper(object):
1819
2130
def run_selftest(self, **kwargs):
1820
2131
"""Run selftest returning its output."""
1822
old_transport = bzrlib.tests.default_transport
2134
output = TextIOWrapper(bio, 'utf-8')
2136
bio = output = StringIO()
2137
old_transport = breezy.tests.default_transport
1823
2138
old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
1824
2139
tests.TestCaseWithMemoryTransport.TEST_ROOT = None
1826
2141
self.assertEqual(True, tests.selftest(stream=output, **kwargs))
1828
bzrlib.tests.default_transport = old_transport
2143
breezy.tests.default_transport = old_transport
1829
2144
tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
1834
2152
class TestSelftest(tests.TestCase, SelfTestHelper):
1835
"""Tests of bzrlib.tests.selftest."""
2153
"""Tests of breezy.tests.selftest."""
1837
def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
2155
def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(
1838
2157
factory_called = []
1840
2160
factory_called.append(True)
1841
2161
return TestUtil.TestSuite()
1842
2162
out = StringIO()
1843
2163
err = StringIO()
1844
self.apply_redirected(out, err, None, bzrlib.tests.selftest,
1845
test_suite_factory=factory)
2164
self.apply_redirected(out, err, None, breezy.tests.selftest,
2165
test_suite_factory=factory)
1846
2166
self.assertEqual([True], factory_called)
1848
2168
def factory(self):
1849
2169
"""A test suite factory."""
1850
2170
class Test(tests.TestCase):
2172
return __name__ + ".Test." + self._testMethodName
1857
2182
return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
1859
2184
def test_list_only(self):
1860
2185
output = self.run_selftest(test_suite_factory=self.factory,
1862
2187
self.assertEqual(3, len(output.readlines()))
1864
2189
def test_list_only_filtered(self):
1865
2190
output = self.run_selftest(test_suite_factory=self.factory,
1866
list_only=True, pattern="Test.b")
1867
self.assertEndsWith(output.getvalue(), "Test.b\n")
2191
list_only=True, pattern="Test.b")
2192
self.assertEndsWith(output.getvalue(), b"Test.b\n")
1868
2193
self.assertLength(1, output.readlines())
1870
2195
def test_list_only_excludes(self):
1871
2196
output = self.run_selftest(test_suite_factory=self.factory,
1872
list_only=True, exclude_pattern="Test.b")
1873
self.assertNotContainsRe("Test.b", output.getvalue())
2197
list_only=True, exclude_pattern="Test.b")
2198
self.assertNotContainsRe(b"Test.b", output.getvalue())
1874
2199
self.assertLength(2, output.readlines())
1876
2201
def test_lsprof_tests(self):
1877
self.requireFeature(test_lsprof.LSProfFeature)
2202
self.requireFeature(features.lsprof_feature)
1879
2205
class Test(object):
1880
2206
def __call__(test, result):
1881
2207
test.run(result)
1882
2209
def run(test, result):
1883
self.assertIsInstance(result, tests.ForwardingResult)
1884
calls.append("called")
2210
results.append(result)
1885
2212
def countTestCases(self):
1887
2214
self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
1888
self.assertLength(1, calls)
2215
self.assertLength(1, results)
2216
self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
1890
2218
def test_random(self):
1891
2219
# test randomising by listing a number of tests.
1892
2220
output_123 = self.run_selftest(test_suite_factory=self.factory,
1893
list_only=True, random_seed="123")
2221
list_only=True, random_seed="123")
1894
2222
output_234 = self.run_selftest(test_suite_factory=self.factory,
1895
list_only=True, random_seed="234")
2223
list_only=True, random_seed="234")
1896
2224
self.assertNotEqual(output_123, output_234)
1897
2225
# "Randominzing test order..\n\n
1898
2226
self.assertLength(5, output_123.readlines())
1967
2302
def test_load_unknown(self):
1968
2303
# Provide a list with one test - this test.
1969
2304
# And generate a list of the tests in the suite.
1970
err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
1971
load_list='missing file name', list_only=True)
2305
self.assertRaises(errors.NoSuchFile, self.run_selftest,
2306
load_list='missing file name', list_only=True)
2309
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2311
_test_needs_features = [features.subunit]
2313
def run_subunit_stream(self, test_name):
2314
from subunit import ProtocolTestCase
2317
return TestUtil.TestSuite([_get_test(test_name)])
2318
stream = self.run_selftest(
2319
runner_class=tests.SubUnitBzrRunnerv1,
2320
test_suite_factory=factory)
2321
test = ProtocolTestCase(stream)
2322
result = testtools.TestResult()
2324
content = stream.getvalue()
2325
return content, result
2327
def test_fail_has_log(self):
2328
content, result = self.run_subunit_stream('test_fail')
2329
self.assertEqual(1, len(result.failures))
2330
self.assertContainsRe(content, b'(?m)^log$')
2331
self.assertContainsRe(content, b'this test will fail')
2333
def test_error_has_log(self):
2334
content, result = self.run_subunit_stream('test_error')
2335
self.assertContainsRe(content, b'(?m)^log$')
2336
self.assertContainsRe(content, b'this test errored')
2338
def test_skip_has_no_log(self):
2339
content, result = self.run_subunit_stream('test_skip')
2340
self.assertNotContainsRe(content, b'(?m)^log$')
2341
self.assertNotContainsRe(content, b'this test will be skipped')
2342
reasons = result.skip_reasons
2343
self.assertEqual({'reason'}, set(reasons))
2344
skips = reasons['reason']
2345
self.assertEqual(1, len(skips))
2347
# RemotedTestCase doesn't preserve the "details"
2348
# self.assertFalse('log' in test.getDetails())
2350
def test_missing_feature_has_no_log(self):
2351
content, result = self.run_subunit_stream('test_missing_feature')
2352
self.assertNotContainsRe(content, b'(?m)^log$')
2353
self.assertNotContainsRe(content, b'missing the feature')
2354
reasons = result.skip_reasons
2355
self.assertEqual({'_MissingFeature\n'}, set(reasons))
2356
skips = reasons['_MissingFeature\n']
2357
self.assertEqual(1, len(skips))
2359
# RemotedTestCase doesn't preserve the "details"
2360
# self.assertFalse('log' in test.getDetails())
2362
def test_xfail_has_no_log(self):
2363
content, result = self.run_subunit_stream('test_xfail')
2364
self.assertNotContainsRe(content, b'(?m)^log$')
2365
self.assertNotContainsRe(content, b'test with expected failure')
2366
self.assertEqual(1, len(result.expectedFailures))
2367
result_content = result.expectedFailures[0][1]
2368
self.assertNotContainsRe(result_content,
2369
'(?m)^(?:Text attachment: )?log(?:$|: )')
2370
self.assertNotContainsRe(result_content, 'test with expected failure')
2372
def test_unexpected_success_has_log(self):
2373
content, result = self.run_subunit_stream('test_unexpected_success')
2374
self.assertContainsRe(content, b'(?m)^log$')
2375
self.assertContainsRe(content, b'test with unexpected success')
2376
self.assertEqual(1, len(result.unexpectedSuccesses))
2377
# test = result.unexpectedSuccesses[0]
2378
# RemotedTestCase doesn't preserve the "details"
2379
# self.assertTrue('log' in test.getDetails())
2381
def test_success_has_no_log(self):
2382
content, result = self.run_subunit_stream('test_success')
2383
self.assertEqual(1, result.testsRun)
2384
self.assertNotContainsRe(content, b'(?m)^log$')
2385
self.assertNotContainsRe(content, b'this test succeeds')
1974
2388
class TestRunBzr(tests.TestCase):
1979
def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
2394
def _run_bzr_core(self, argv, encoding=None, stdin=None,
2395
stdout=None, stderr=None, working_dir=None):
1981
2396
"""Override _run_bzr_core to test how it is invoked by run_bzr.
1983
2398
Attempts to run bzr from inside this class don't actually run it.
2173
2572
self.next_subprocess = process
2175
2574
result = self.run_bzr_subprocess(*args, **kwargs)
2575
except BaseException:
2177
2576
self.next_subprocess = None
2178
for key, expected in expected_args.iteritems():
2577
for key, expected in expected_args.items():
2179
2578
self.assertEqual(expected, self.subprocess_calls[-1][key])
2182
2581
self.next_subprocess = None
2183
for key, expected in expected_args.iteritems():
2582
for key, expected in expected_args.items():
2184
2583
self.assertEqual(expected, self.subprocess_calls[-1][key])
2187
2586
def test_run_bzr_subprocess(self):
2188
2587
"""The run_bzr_helper_external command behaves nicely."""
2189
self.assertRunBzrSubprocess({'process_args':['--version']},
2190
StubProcess(), '--version')
2191
self.assertRunBzrSubprocess({'process_args':['--version']},
2192
StubProcess(), ['--version'])
2588
self.assertRunBzrSubprocess({'process_args': ['--version']},
2589
StubProcess(), '--version')
2590
self.assertRunBzrSubprocess({'process_args': ['--version']},
2591
StubProcess(), ['--version'])
2193
2592
# retcode=None disables retcode checking
2194
result = self.assertRunBzrSubprocess({},
2195
StubProcess(retcode=3), '--version', retcode=None)
2196
result = self.assertRunBzrSubprocess({},
2197
StubProcess(out="is free software"), '--version')
2593
result = self.assertRunBzrSubprocess(
2594
{}, StubProcess(retcode=3), '--version', retcode=None)
2595
result = self.assertRunBzrSubprocess(
2596
{}, StubProcess(out="is free software"), '--version')
2198
2597
self.assertContainsRe(result[0], 'is free software')
2199
2598
# Running a subcommand that is missing errors
2200
2599
self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2201
{'process_args':['--versionn']}, StubProcess(retcode=3),
2600
{'process_args': ['--versionn']
2601
}, StubProcess(retcode=3),
2203
2603
# Unless it is told to expect the error from the subprocess
2204
result = self.assertRunBzrSubprocess({},
2205
StubProcess(retcode=3), '--versionn', retcode=3)
2604
result = self.assertRunBzrSubprocess(
2605
{}, StubProcess(retcode=3), '--versionn', retcode=3)
2206
2606
# Or to ignore retcode checking
2207
result = self.assertRunBzrSubprocess({},
2208
StubProcess(err="unknown command", retcode=3), '--versionn',
2607
result = self.assertRunBzrSubprocess(
2608
{}, StubProcess(err="unknown command", retcode=3),
2609
'--versionn', retcode=None)
2210
2610
self.assertContainsRe(result[1], 'unknown command')
2212
2612
def test_env_change_passes_through(self):
2213
2613
self.assertRunBzrSubprocess(
2214
{'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2614
{'env_changes': {'new': 'value', 'changed': 'newvalue', 'deleted': None}},
2215
2615
StubProcess(), '',
2216
env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2616
env_changes={'new': 'value', 'changed': 'newvalue', 'deleted': None})
2218
2618
def test_no_working_dir_passed_as_None(self):
2219
2619
self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2221
2621
def test_no_working_dir_passed_through(self):
2222
2622
self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2225
2625
def test_run_bzr_subprocess_no_plugins(self):
2226
2626
self.assertRunBzrSubprocess({'allow_plugins': False},
2229
2629
def test_allow_plugins(self):
2230
2630
self.assertRunBzrSubprocess({'allow_plugins': True},
2231
StubProcess(), '', allow_plugins=True)
2631
StubProcess(), '', allow_plugins=True)
2234
2634
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2264
2664
class TestStartBzrSubProcess(tests.TestCase):
2665
"""Stub test start_bzr_subprocess."""
2266
def check_popen_state(self):
2267
"""Replace to make assertions when popen is called."""
2667
def _subprocess_log_cleanup(self):
2668
"""Inhibits the base version as we don't produce a log file."""
2269
2670
def _popen(self, *args, **kwargs):
2270
"""Record the command that is run, so that we can ensure it is correct"""
2671
"""Override the base version to record the command that is run.
2673
From there we can ensure it is correct without spawning a real process.
2271
2675
self.check_popen_state()
2272
2676
self._popen_args = args
2273
2677
self._popen_kwargs = kwargs
2274
2678
raise _DontSpawnProcess()
2680
def check_popen_state(self):
2681
"""Replace to make assertions when popen is called."""
2276
2683
def test_run_bzr_subprocess_no_plugins(self):
2277
2684
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2278
2685
command = self._popen_args[0]
2279
2686
self.assertEqual(sys.executable, command[0])
2280
self.assertEqual(self.get_bzr_path(), command[1])
2687
self.assertEqual(self.get_brz_path(), command[1])
2281
2688
self.assertEqual(['--no-plugins'], command[2:])
2283
2690
def test_allow_plugins(self):
2284
2691
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2286
2693
command = self._popen_args[0]
2287
2694
self.assertEqual([], command[2:])
2289
2696
def test_set_env(self):
2290
self.failIf('EXISTANT_ENV_VAR' in os.environ)
2697
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2291
2698
# set in the child
2292
2700
def check_environment():
2293
2701
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2294
2702
self.check_popen_state = check_environment
2295
2703
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2296
env_changes={'EXISTANT_ENV_VAR':'set variable'})
2704
env_changes={'EXISTANT_ENV_VAR': 'set variable'})
2297
2705
# not set in theparent
2298
2706
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2300
2708
def test_run_bzr_subprocess_env_del(self):
2301
2709
"""run_bzr_subprocess can remove environment variables too."""
2302
self.failIf('EXISTANT_ENV_VAR' in os.environ)
2710
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2303
2712
def check_environment():
2304
2713
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2305
2714
os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2306
2715
self.check_popen_state = check_environment
2307
2716
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2308
env_changes={'EXISTANT_ENV_VAR':None})
2717
env_changes={'EXISTANT_ENV_VAR': None})
2309
2718
# Still set in parent
2310
2719
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2311
2720
del os.environ['EXISTANT_ENV_VAR']
2313
2722
def test_env_del_missing(self):
2314
self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2723
self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2315
2725
def check_environment():
2316
2726
self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2317
2727
self.check_popen_state = check_environment
2318
2728
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2319
env_changes={'NON_EXISTANT_ENV_VAR':None})
2729
env_changes={'NON_EXISTANT_ENV_VAR': None})
2321
2731
def test_working_dir(self):
2322
2732
"""Test that we can specify the working dir for the child"""
2323
orig_getcwd = osutils.getcwd
2324
orig_chdir = os.chdir
2326
2735
def chdir(path):
2327
2736
chdirs.append(path)
2332
osutils.getcwd = getcwd
2334
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2337
osutils.getcwd = orig_getcwd
2339
os.chdir = orig_chdir
2737
self.overrideAttr(os, 'chdir', chdir)
2741
self.overrideAttr(osutils, 'getcwd', getcwd)
2742
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2340
2744
self.assertEqual(['foo', 'current'], chdirs)
2746
def test_get_brz_path_with_cwd_breezy(self):
2747
self.get_source_path = lambda: ""
2748
self.overrideAttr(os.path, "isfile", lambda path: True)
2749
self.assertEqual(self.get_brz_path(), "brz")
2343
2752
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2344
2753
"""Tests that really need to do things with an external bzr."""
2350
2759
self.disable_missing_extensions_warning()
2351
2760
process = self.start_bzr_subprocess(['wait-until-signalled'],
2352
2761
skip_if_plan_to_signal=True)
2353
self.assertEqual('running\n', process.stdout.readline())
2762
self.assertEqual(b'running\n', process.stdout.readline())
2354
2763
result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2356
self.assertEqual('', result[0])
2357
self.assertEqual('bzr: interrupted\n', result[1])
2360
class TestFeature(tests.TestCase):
2362
def test_caching(self):
2363
"""Feature._probe is called by the feature at most once."""
2364
class InstrumentedFeature(tests.Feature):
2366
super(InstrumentedFeature, self).__init__()
2369
self.calls.append('_probe')
2371
feature = InstrumentedFeature()
2373
self.assertEqual(['_probe'], feature.calls)
2375
self.assertEqual(['_probe'], feature.calls)
2377
def test_named_str(self):
2378
"""Feature.__str__ should thunk to feature_name()."""
2379
class NamedFeature(tests.Feature):
2380
def feature_name(self):
2382
feature = NamedFeature()
2383
self.assertEqual('symlinks', str(feature))
2385
def test_default_str(self):
2386
"""Feature.__str__ should default to __class__.__name__."""
2387
class NamedFeature(tests.Feature):
2389
feature = NamedFeature()
2390
self.assertEqual('NamedFeature', str(feature))
2393
class TestUnavailableFeature(tests.TestCase):
2395
def test_access_feature(self):
2396
feature = tests.Feature()
2397
exception = tests.UnavailableFeature(feature)
2398
self.assertIs(feature, exception.args[0])
2401
simple_thunk_feature = tests._CompatabilityThunkFeature(
2402
deprecated_in((2, 1, 0)),
2403
'bzrlib.tests.test_selftest',
2404
'simple_thunk_feature','UnicodeFilename',
2405
replacement_module='bzrlib.tests'
2408
class Test_CompatibilityFeature(tests.TestCase):
2410
def test_does_thunk(self):
2411
res = self.callDeprecated(
2412
['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
2413
' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
2414
simple_thunk_feature.available)
2415
self.assertEqual(tests.UnicodeFilename.available(), res)
2418
class TestModuleAvailableFeature(tests.TestCase):
2420
def test_available_module(self):
2421
feature = tests.ModuleAvailableFeature('bzrlib.tests')
2422
self.assertEqual('bzrlib.tests', feature.module_name)
2423
self.assertEqual('bzrlib.tests', str(feature))
2424
self.assertTrue(feature.available())
2425
self.assertIs(tests, feature.module)
2427
def test_unavailable_module(self):
2428
feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
2429
self.assertEqual('bzrlib.no_such_module_exists', str(feature))
2430
self.assertFalse(feature.available())
2431
self.assertIs(None, feature.module)
2765
self.assertEqual(b'', result[0])
2766
self.assertEqual(b'brz: interrupted\n', result[1])
2434
2769
class TestSelftestFiltering(tests.TestCase):
2436
2771
def setUp(self):
2437
tests.TestCase.setUp(self)
2772
super(TestSelftestFiltering, self).setUp()
2438
2773
self.suite = TestUtil.TestSuite()
2439
2774
self.loader = TestUtil.TestLoader()
2440
2775
self.suite.addTest(self.loader.loadTestsFromModule(
2441
sys.modules['bzrlib.tests.test_selftest']))
2776
sys.modules['breezy.tests.test_selftest']))
2442
2777
self.all_names = _test_ids(self.suite)
2444
2779
def test_condition_id_re(self):
2445
test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2446
'test_condition_id_re')
2780
test_name = ('breezy.tests.test_selftest.TestSelftestFiltering.'
2781
'test_condition_id_re')
2447
2782
filtered_suite = tests.filter_suite_by_condition(
2448
2783
self.suite, tests.condition_id_re('test_condition_id_re'))
2449
2784
self.assertEqual([test_name], _test_ids(filtered_suite))
2451
2786
def test_condition_id_in_list(self):
2452
test_names = ['bzrlib.tests.test_selftest.TestSelftestFiltering.'
2787
test_names = ['breezy.tests.test_selftest.TestSelftestFiltering.'
2453
2788
'test_condition_id_in_list']
2454
2789
id_list = tests.TestIdList(test_names)
2455
2790
filtered_suite = tests.filter_suite_by_condition(
2492
2827
self.all_names = _test_ids(self.suite)
2493
2828
filtered_suite = tests.exclude_tests_by_re(self.suite,
2494
2829
'exclude_tests_by_re')
2495
excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2496
'test_exclude_tests_by_re')
2830
excluded_name = ('breezy.tests.test_selftest.TestSelftestFiltering.'
2831
'test_exclude_tests_by_re')
2497
2832
self.assertEqual(len(self.all_names) - 1,
2498
filtered_suite.countTestCases())
2833
filtered_suite.countTestCases())
2499
2834
self.assertFalse(excluded_name in _test_ids(filtered_suite))
2500
2835
remaining_names = list(self.all_names)
2501
2836
remaining_names.remove(excluded_name)
2502
2837
self.assertEqual(remaining_names, _test_ids(filtered_suite))
2504
2839
def test_filter_suite_by_condition(self):
2505
test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2506
'test_filter_suite_by_condition')
2507
filtered_suite = tests.filter_suite_by_condition(self.suite,
2508
lambda x:x.id() == test_name)
2840
test_name = ('breezy.tests.test_selftest.TestSelftestFiltering.'
2841
'test_filter_suite_by_condition')
2842
filtered_suite = tests.filter_suite_by_condition(
2843
self.suite, lambda x: x.id() == test_name)
2509
2844
self.assertEqual([test_name], _test_ids(filtered_suite))
2511
2846
def test_filter_suite_by_re(self):
2512
2847
filtered_suite = tests.filter_suite_by_re(self.suite,
2513
2848
'test_filter_suite_by_r')
2514
2849
filtered_names = _test_ids(filtered_suite)
2515
self.assertEqual(filtered_names, ['bzrlib.tests.test_selftest.'
2516
'TestSelftestFiltering.test_filter_suite_by_re'])
2851
filtered_names, ['breezy.tests.test_selftest.'
2852
'TestSelftestFiltering.test_filter_suite_by_re'])
2518
2854
def test_filter_suite_by_id_list(self):
2519
test_list = ['bzrlib.tests.test_selftest.'
2855
test_list = ['breezy.tests.test_selftest.'
2520
2856
'TestSelftestFiltering.test_filter_suite_by_id_list']
2521
2857
filtered_suite = tests.filter_suite_by_id_list(
2522
2858
self.suite, tests.TestIdList(test_list))
2523
2859
filtered_names = _test_ids(filtered_suite)
2524
2860
self.assertEqual(
2525
2861
filtered_names,
2526
['bzrlib.tests.test_selftest.'
2862
['breezy.tests.test_selftest.'
2527
2863
'TestSelftestFiltering.test_filter_suite_by_id_list'])
2529
2865
def test_filter_suite_by_id_startswith(self):
2530
2866
# By design this test may fail if another test is added whose name also
2531
2867
# begins with one of the start value used.
2532
klass = 'bzrlib.tests.test_selftest.TestSelftestFiltering.'
2868
klass = 'breezy.tests.test_selftest.TestSelftestFiltering.'
2533
2869
start1 = klass + 'test_filter_suite_by_id_starts'
2534
2870
start2 = klass + 'test_filter_suite_by_id_li'
2535
2871
test_list = [klass + 'test_filter_suite_by_id_list',
2778
3119
self.assertEqual([], test_list)
2780
3121
self.assertSubset([
2785
3126
def test_test_suite(self):
2786
3127
# test_suite() loads the entire test suite to operate. To avoid this
2787
3128
# overhead, and yet still be confident that things are happening,
2788
# we temporarily replace two functions used by test_suite with
3129
# we temporarily replace two functions used by test_suite with
2789
3130
# test doubles that supply a few sample tests to load, and check they
2792
3134
def testmod_names():
2793
3135
calls.append("testmod_names")
2795
'bzrlib.tests.blackbox.test_branch',
2796
'bzrlib.tests.per_transport',
2797
'bzrlib.tests.test_selftest',
3137
'breezy.tests.blackbox.test_branch',
3138
'breezy.tests.per_transport',
3139
'breezy.tests.test_selftest',
2799
3141
self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
2800
3143
def doctests():
2801
3144
calls.append("modules_to_doctest")
2802
3145
if __doc__ is None:
2804
return ['bzrlib.timestamp']
3147
return ['breezy.timestamp']
2805
3148
self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
2806
3149
expected_test_list = [
2807
3150
# testmod_names
2808
'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
2809
('bzrlib.tests.per_transport.TransportTests'
3151
'breezy.tests.blackbox.test_branch.TestBranch.test_branch',
3152
('breezy.tests.per_transport.TransportTests'
2810
3153
'.test_abspath(LocalTransport,LocalURLServer)'),
2811
'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
3154
'breezy.tests.test_selftest.TestTestSuite.test_test_suite',
2812
3155
# plugins can't be tested that way since selftest may be run with
2815
if __doc__ is not None:
3158
if __doc__ is not None and not PY3:
2816
3159
expected_test_list.extend([
2817
3160
# modules_to_doctest
2818
'bzrlib.timestamp.format_highres_date',
3161
'breezy.timestamp.format_highres_date',
2820
3163
suite = tests.test_suite()
2821
self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
3165
self.assertEqual({"testmod_names"}, set(calls))
3167
self.assertEqual({"testmod_names", "modules_to_doctest"},
2823
3169
self.assertSubset(expected_test_list, _test_ids(suite))
2825
3171
def test_test_suite_list_and_start(self):
2826
# We cannot test this at the same time as the main load, because we want
2827
# to know that starting_with == None works. So a second load is
2828
# incurred - note that the starting_with parameter causes a partial load
2829
# rather than a full load so this test should be pretty quick.
2830
test_list = ['bzrlib.tests.test_selftest.TestTestSuite.test_test_suite']
3172
# We cannot test this at the same time as the main load, because we
3173
# want to know that starting_with == None works. So a second load is
3174
# incurred - note that the starting_with parameter causes a partial
3175
# load rather than a full load so this test should be pretty quick.
3177
'breezy.tests.test_selftest.TestTestSuite.test_test_suite']
2831
3178
suite = tests.test_suite(test_list,
2832
['bzrlib.tests.test_selftest.TestTestSuite'])
2833
# test_test_suite_list_and_start is not included
2834
self.assertEquals(test_list, _test_ids(suite))
3179
['breezy.tests.test_selftest.TestTestSuite'])
3180
# test_test_suite_list_and_start is not included
3181
self.assertEqual(test_list, _test_ids(suite))
2837
3184
class TestLoadTestIdList(tests.TestCaseInTempDir):
2953
3300
def test_predefined_prefixes(self):
2954
3301
tpr = tests.test_prefix_alias_registry
2955
self.assertEquals('bzrlib', tpr.resolve_alias('bzrlib'))
2956
self.assertEquals('bzrlib.doc', tpr.resolve_alias('bd'))
2957
self.assertEquals('bzrlib.utils', tpr.resolve_alias('bu'))
2958
self.assertEquals('bzrlib.tests', tpr.resolve_alias('bt'))
2959
self.assertEquals('bzrlib.tests.blackbox', tpr.resolve_alias('bb'))
2960
self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3302
self.assertEqual('breezy', tpr.resolve_alias('breezy'))
3303
self.assertEqual('breezy.doc', tpr.resolve_alias('bd'))
3304
self.assertEqual('breezy.utils', tpr.resolve_alias('bu'))
3305
self.assertEqual('breezy.tests', tpr.resolve_alias('bt'))
3306
self.assertEqual('breezy.tests.blackbox', tpr.resolve_alias('bb'))
3307
self.assertEqual('breezy.plugins', tpr.resolve_alias('bp'))
3310
class TestThreadLeakDetection(tests.TestCase):
3311
"""Ensure when tests leak threads we detect and report it"""
3313
class LeakRecordingResult(tests.ExtendedTestResult):
3315
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3318
def _report_thread_leak(self, test, leaks, alive):
3319
self.leaks.append((test, leaks))
3321
def test_testcase_without_addCleanups(self):
3322
"""Check old TestCase instances don't break with leak detection"""
3323
class Test(unittest.TestCase):
3326
result = self.LeakRecordingResult()
3328
result.startTestRun()
3330
result.stopTestRun()
3331
self.assertEqual(result._tests_leaking_threads_count, 0)
3332
self.assertEqual(result.leaks, [])
3334
def test_thread_leak(self):
3335
"""Ensure a thread that outlives the running of a test is reported
3337
Uses a thread that blocks on an event, and is started by the inner
3338
test case. As the thread outlives the inner case's run, it should be
3339
detected as a leak, but the event is then set so that the thread can
3340
be safely joined in cleanup so it's not leaked for real.
3342
event = threading.Event()
3343
thread = threading.Thread(name="Leaker", target=event.wait)
3345
class Test(tests.TestCase):
3346
def test_leak(self):
3348
result = self.LeakRecordingResult()
3349
test = Test("test_leak")
3350
self.addCleanup(thread.join)
3351
self.addCleanup(event.set)
3352
result.startTestRun()
3354
result.stopTestRun()
3355
self.assertEqual(result._tests_leaking_threads_count, 1)
3356
self.assertEqual(result._first_thread_leaker_id, test.id())
3357
self.assertEqual(result.leaks, [(test, {thread})])
3358
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3360
def test_multiple_leaks(self):
3361
"""Check multiple leaks are blamed on the test cases at fault
3363
Same concept as the previous test, but has one inner test method that
3364
leaks two threads, and one that doesn't leak at all.
3366
event = threading.Event()
3367
thread_a = threading.Thread(name="LeakerA", target=event.wait)
3368
thread_b = threading.Thread(name="LeakerB", target=event.wait)
3369
thread_c = threading.Thread(name="LeakerC", target=event.wait)
3371
class Test(tests.TestCase):
3372
def test_first_leak(self):
3375
def test_second_no_leak(self):
3378
def test_third_leak(self):
3381
result = self.LeakRecordingResult()
3382
first_test = Test("test_first_leak")
3383
third_test = Test("test_third_leak")
3384
self.addCleanup(thread_a.join)
3385
self.addCleanup(thread_b.join)
3386
self.addCleanup(thread_c.join)
3387
self.addCleanup(event.set)
3388
result.startTestRun()
3390
[first_test, Test("test_second_no_leak"), third_test]
3392
result.stopTestRun()
3393
self.assertEqual(result._tests_leaking_threads_count, 2)
3394
self.assertEqual(result._first_thread_leaker_id, first_test.id())
3395
self.assertEqual(result.leaks, [
3396
(first_test, {thread_b}),
3397
(third_test, {thread_a, thread_c})])
3398
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3401
class TestPostMortemDebugging(tests.TestCase):
3402
"""Check post mortem debugging works when tests fail or error"""
3404
class TracebackRecordingResult(tests.ExtendedTestResult):
3406
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3407
self.postcode = None
3409
def _post_mortem(self, tb=None):
3410
"""Record the code object at the end of the current traceback"""
3411
tb = tb or sys.exc_info()[2]
3414
while next is not None:
3417
self.postcode = tb.tb_frame.f_code
3419
def report_error(self, test, err):
3422
def report_failure(self, test, err):
3425
def test_location_unittest_error(self):
3426
"""Needs right post mortem traceback with erroring unittest case"""
3427
class Test(unittest.TestCase):
3430
result = self.TracebackRecordingResult()
3432
self.assertEqual(result.postcode, Test.runTest.__code__)
3434
def test_location_unittest_failure(self):
3435
"""Needs right post mortem traceback with failing unittest case"""
3436
class Test(unittest.TestCase):
3438
raise self.failureException
3439
result = self.TracebackRecordingResult()
3441
self.assertEqual(result.postcode, Test.runTest.__code__)
3443
def test_location_bt_error(self):
3444
"""Needs right post mortem traceback with erroring breezy.tests case"""
3445
class Test(tests.TestCase):
3446
def test_error(self):
3448
result = self.TracebackRecordingResult()
3449
Test("test_error").run(result)
3450
self.assertEqual(result.postcode, Test.test_error.__code__)
3452
def test_location_bt_failure(self):
3453
"""Needs right post mortem traceback with failing breezy.tests case"""
3454
class Test(tests.TestCase):
3455
def test_failure(self):
3456
raise self.failureException
3457
result = self.TracebackRecordingResult()
3458
Test("test_failure").run(result)
3459
self.assertEqual(result.postcode, Test.test_failure.__code__)
3461
def test_env_var_triggers_post_mortem(self):
3462
"""Check pdb.post_mortem is called iff BRZ_TEST_PDB is set"""
3464
result = tests.ExtendedTestResult(StringIO(), 0, 1)
3465
post_mortem_calls = []
3466
self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3467
self.overrideEnv('BRZ_TEST_PDB', None)
3468
result._post_mortem(1)
3469
self.overrideEnv('BRZ_TEST_PDB', 'on')
3470
result._post_mortem(2)
3471
self.assertEqual([2], post_mortem_calls)
2963
3474
class TestRunSuite(tests.TestCase):
2976
3488
self.verbosity)
2977
3489
tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2978
3490
self.assertLength(1, calls)
3493
class _Selftest(object):
3494
"""Mixin for tests needing full selftest output"""
3496
def _inject_stream_into_subunit(self, stream):
3497
"""To be overridden by subclasses that run tests out of process"""
3499
def _run_selftest(self, **kwargs):
3502
sio = TextIOWrapper(bio, 'utf-8')
3504
sio = bio = StringIO()
3505
self._inject_stream_into_subunit(bio)
3506
tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
3508
return bio.getvalue()
3511
class _ForkedSelftest(_Selftest):
3512
"""Mixin for tests needing full selftest output with forked children"""
3514
_test_needs_features = [features.subunit]
3516
def _inject_stream_into_subunit(self, stream):
3517
"""Monkey-patch subunit so the extra output goes to stream not stdout
3519
Some APIs need rewriting so this kind of bogus hackery can be replaced
3520
by passing the stream param from run_tests down into ProtocolTestCase.
3522
from subunit import ProtocolTestCase
3523
_original_init = ProtocolTestCase.__init__
3525
def _init_with_passthrough(self, *args, **kwargs):
3526
_original_init(self, *args, **kwargs)
3527
self._passthrough = stream
3528
self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
3530
def _run_selftest(self, **kwargs):
3531
# GZ 2011-05-26: Add a PosixSystem feature so this check can go away
3532
if getattr(os, "fork", None) is None:
3533
raise tests.TestNotApplicable("Platform doesn't support forking")
3534
# Make sure the fork code is actually invoked by claiming two cores
3535
self.overrideAttr(osutils, "local_concurrency", lambda: 2)
3536
kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
3537
return super(_ForkedSelftest, self)._run_selftest(**kwargs)
3540
class TestParallelFork(_ForkedSelftest, tests.TestCase):
3541
"""Check operation of --parallel=fork selftest option"""
3543
def test_error_in_child_during_fork(self):
3544
"""Error in a forked child during test setup should get reported"""
3545
class Test(tests.TestCase):
3546
def testMethod(self):
3548
# We don't care what, just break something that a child will run
3549
self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
3550
out = self._run_selftest(test_suite_factory=Test)
3551
# Lines from the tracebacks of the two child processes may be mixed
3552
# together due to the way subunit parses and forwards the streams,
3553
# so permit extra lines between each part of the error output.
3554
self.assertContainsRe(out,
3557
b".+ in fork_for_tests\n"
3559
b"\\s*workaround_zealous_crypto_random\\(\\)\n"
3564
class TestUncollectedWarnings(_Selftest, tests.TestCase):
3565
"""Check a test case still alive after being run emits a warning"""
3567
class Test(tests.TestCase):
3568
def test_pass(self):
3571
def test_self_ref(self):
3572
self.also_self = self.test_self_ref
3574
def test_skip(self):
3575
self.skipTest("Don't need")
3577
def _get_suite(self):
3578
return TestUtil.TestSuite([
3579
self.Test("test_pass"),
3580
self.Test("test_self_ref"),
3581
self.Test("test_skip"),
3584
def _run_selftest_with_suite(self, **kwargs):
3585
old_flags = tests.selftest_debug_flags
3586
tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
3587
gc_on = gc.isenabled()
3591
output = self._run_selftest(test_suite_factory=self._get_suite,
3596
tests.selftest_debug_flags = old_flags
3597
self.assertNotContainsRe(output, b"Uncollected test case.*test_pass")
3598
self.assertContainsRe(output, b"Uncollected test case.*test_self_ref")
3601
def test_testsuite(self):
3602
self._run_selftest_with_suite()
3604
def test_pattern(self):
3605
out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
3606
self.assertNotContainsRe(out, b"test_skip")
3608
def test_exclude_pattern(self):
3609
out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
3610
self.assertNotContainsRe(out, b"test_skip")
3612
def test_random_seed(self):
3613
self._run_selftest_with_suite(random_seed="now")
3615
def test_matching_tests_first(self):
3616
self._run_selftest_with_suite(matching_tests_first=True,
3617
pattern="test_self_ref$")
3619
def test_starting_with_and_exclude(self):
3620
out = self._run_selftest_with_suite(starting_with=["bt."],
3621
exclude_pattern="test_skip$")
3622
self.assertNotContainsRe(out, b"test_skip")
3624
def test_additonal_decorator(self):
3625
self._run_selftest_with_suite(suite_decorators=[tests.TestDecorator])
3628
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
3629
"""Check warnings from tests staying alive are emitted with subunit"""
3631
_test_needs_features = [features.subunit]
3633
def _run_selftest_with_suite(self, **kwargs):
3634
return TestUncollectedWarnings._run_selftest_with_suite(
3635
self, runner_class=tests.SubUnitBzrRunnerv1, **kwargs)
3638
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
3639
"""Check warnings from tests staying alive are emitted when forking"""
3642
class TestEnvironHandling(tests.TestCase):
3644
def test_overrideEnv_None_called_twice_doesnt_leak(self):
3645
self.assertFalse('MYVAR' in os.environ)
3646
self.overrideEnv('MYVAR', '42')
3647
# We use an embedded test to make sure we fix the _captureVar bug
3649
class Test(tests.TestCase):
3651
# The first call save the 42 value
3652
self.overrideEnv('MYVAR', None)
3653
self.assertEqual(None, os.environ.get('MYVAR'))
3654
# Make sure we can call it twice
3655
self.overrideEnv('MYVAR', None)
3656
self.assertEqual(None, os.environ.get('MYVAR'))
3658
result = tests.TextTestResult(output, 0, 1)
3659
Test('test_me').run(result)
3660
if not result.wasStrictlySuccessful():
3661
self.fail(output.getvalue())
3662
# We get our value back
3663
self.assertEqual('42', os.environ.get('MYVAR'))
3666
class TestIsolatedEnv(tests.TestCase):
3667
"""Test isolating tests from os.environ.
3669
Since we use tests that are already isolated from os.environ a bit of care
3670
should be taken when designing the tests to avoid bootstrap side-effects.
3671
The tests start an already clean os.environ which allow doing valid
3672
assertions about which variables are present or not and design tests around
3676
class ScratchMonkey(tests.TestCase):
3681
def test_basics(self):
3682
# Make sure we know the definition of BRZ_HOME: not part of os.environ
3683
# for tests.TestCase.
3684
self.assertTrue('BRZ_HOME' in tests.isolated_environ)
3685
self.assertEqual(None, tests.isolated_environ['BRZ_HOME'])
3686
# Being part of isolated_environ, BRZ_HOME should not appear here
3687
self.assertFalse('BRZ_HOME' in os.environ)
3688
# Make sure we know the definition of LINES: part of os.environ for
3690
self.assertTrue('LINES' in tests.isolated_environ)
3691
self.assertEqual('25', tests.isolated_environ['LINES'])
3692
self.assertEqual('25', os.environ['LINES'])
3694
def test_injecting_unknown_variable(self):
3695
# BRZ_HOME is known to be absent from os.environ
3696
test = self.ScratchMonkey('test_me')
3697
tests.override_os_environ(test, {'BRZ_HOME': 'foo'})
3698
self.assertEqual('foo', os.environ['BRZ_HOME'])
3699
tests.restore_os_environ(test)
3700
self.assertFalse('BRZ_HOME' in os.environ)
3702
def test_injecting_known_variable(self):
3703
test = self.ScratchMonkey('test_me')
3704
# LINES is known to be present in os.environ
3705
tests.override_os_environ(test, {'LINES': '42'})
3706
self.assertEqual('42', os.environ['LINES'])
3707
tests.restore_os_environ(test)
3708
self.assertEqual('25', os.environ['LINES'])
3710
def test_deleting_variable(self):
3711
test = self.ScratchMonkey('test_me')
3712
# LINES is known to be present in os.environ
3713
tests.override_os_environ(test, {'LINES': None})
3714
self.assertTrue('LINES' not in os.environ)
3715
tests.restore_os_environ(test)
3716
self.assertEqual('25', os.environ['LINES'])
3719
class TestDocTestSuiteIsolation(tests.TestCase):
3720
"""Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3722
Since tests.TestCase alreay provides an isolation from os.environ, we use
3723
the clean environment as a base for testing. To precisely capture the
3724
isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3727
We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3728
not `os.environ` so each test overrides it to suit its needs.
3732
def get_doctest_suite_for_string(self, klass, string):
3733
class Finder(doctest.DocTestFinder):
3735
def find(*args, **kwargs):
3736
test = doctest.DocTestParser().get_doctest(
3737
string, {}, 'foo', 'foo.py', 0)
3740
suite = klass(test_finder=Finder())
3743
def run_doctest_suite_for_string(self, klass, string):
3744
suite = self.get_doctest_suite_for_string(klass, string)
3746
result = tests.TextTestResult(output, 0, 1)
3748
return result, output
3750
def assertDocTestStringSucceds(self, klass, string):
3751
result, output = self.run_doctest_suite_for_string(klass, string)
3752
if not result.wasStrictlySuccessful():
3753
self.fail(output.getvalue())
3755
def assertDocTestStringFails(self, klass, string):
3756
result, output = self.run_doctest_suite_for_string(klass, string)
3757
if result.wasStrictlySuccessful():
3758
self.fail(output.getvalue())
3760
def test_injected_variable(self):
3761
self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3764
>>> os.environ['LINES']
3767
# doctest.DocTestSuite fails as it sees '25'
3768
self.assertDocTestStringFails(doctest.DocTestSuite, test)
3769
# tests.DocTestSuite sees '42'
3770
self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3772
def test_deleted_variable(self):
3773
self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3776
>>> os.environ.get('LINES')
3778
# doctest.DocTestSuite fails as it sees '25'
3779
self.assertDocTestStringFails(doctest.DocTestSuite, test)
3780
# tests.DocTestSuite sees None
3781
self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3784
class TestSelftestExcludePatterns(tests.TestCase):
3787
super(TestSelftestExcludePatterns, self).setUp()
3788
self.overrideAttr(tests, 'test_suite', self.suite_factory)
3790
def suite_factory(self, keep_only=None, starting_with=None):
3791
"""A test suite factory with only a few tests."""
3792
class Test(tests.TestCase):
3794
# We don't need the full class path
3795
return self._testMethodName
3805
return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3807
def assertTestList(self, expected, *selftest_args):
3808
# We rely on setUp installing the right test suite factory so we can
3809
# test at the command level without loading the whole test suite
3810
out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3811
actual = out.splitlines()
3812
self.assertEqual(expected, actual)
3814
def test_full_list(self):
3815
self.assertTestList(['a', 'b', 'c'])
3817
def test_single_exclude(self):
3818
self.assertTestList(['b', 'c'], '-x', 'a')
3820
def test_mutiple_excludes(self):
3821
self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3824
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3826
_test_needs_features = [features.subunit]
3829
super(TestCounterHooks, self).setUp()
3831
class Test(tests.TestCase):
3834
super(Test, self).setUp()
3835
self.hooks = hooks.Hooks()
3836
self.hooks.add_hook('myhook', 'Foo bar blah', (2, 4))
3837
self.install_counter_hook(self.hooks, 'myhook')
3842
def run_hook_once(self):
3843
for hook in self.hooks['myhook']:
3846
self.test_class = Test
3848
def assertHookCalls(self, expected_calls, test_name):
3849
test = self.test_class(test_name)
3850
result = unittest.TestResult()
3852
self.assertTrue(hasattr(test, '_counters'))
3853
self.assertTrue('myhook' in test._counters)
3854
self.assertEqual(expected_calls, test._counters['myhook'])
3856
def test_no_hook(self):
3857
self.assertHookCalls(0, 'no_hook')
3859
def test_run_hook_once(self):
3860
tt = features.testtools
3861
if tt.module.__version__ < (0, 9, 8):
3862
raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3863
self.assertHookCalls(1, 'run_hook_once')