Use cleaner test doubles in test_test_protocol.
[third_party/subunit] / python / subunit / tests / test_test_protocol.py
1 #
2 #  subunit: extensions to Python unittest to get test results from subprocesses.
3 #  Copyright (C) 2005  Robert Collins <robertc@robertcollins.net>
4 #
5 #  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
6 #  license at the users choice. A copy of both licenses are available in the
7 #  project source as Apache-2.0 and BSD. You may not use this file except in
8 #  compliance with one of these two licences.
9 #  
10 #  Unless required by applicable law or agreed to in writing, software
11 #  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
12 #  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
13 #  license you chose for the specific language governing permissions and
14 #  limitations under that license.
15 #
16
17 import datetime
18 import unittest
19 from StringIO import StringIO
20 import os
21 import sys
22
23 import subunit
24 from subunit.content import Content, TracebackContent
25 from subunit.content_type import ContentType
26 import subunit.iso8601 as iso8601
27 from subunit.tests.test_test_results import (
28     ExtendedTestResult,
29     Python26TestResult,
30     Python27TestResult,
31     )
32
33
34 class TestTestImports(unittest.TestCase):
35
36     def test_imports(self):
37         from subunit import DiscardStream
38         from subunit import TestProtocolServer
39         from subunit import RemotedTestCase
40         from subunit import RemoteError
41         from subunit import ExecTestCase
42         from subunit import IsolatedTestCase
43         from subunit import TestProtocolClient
44
45
46 class TestDiscardStream(unittest.TestCase):
47
48     def test_write(self):
49         subunit.DiscardStream().write("content")
50
51
52 class TestTestProtocolServerPipe(unittest.TestCase):
53
54     def test_story(self):
55         client = unittest.TestResult()
56         protocol = subunit.TestProtocolServer(client)
57         pipe = StringIO("test old mcdonald\n"
58                         "success old mcdonald\n"
59                         "test bing crosby\n"
60                         "failure bing crosby [\n"
61                         "foo.c:53:ERROR invalid state\n"
62                         "]\n"
63                         "test an error\n"
64                         "error an error\n")
65         protocol.readFrom(pipe)
66         mcdonald = subunit.RemotedTestCase("old mcdonald")
67         bing = subunit.RemotedTestCase("bing crosby")
68         an_error = subunit.RemotedTestCase("an error")
69         self.assertEqual(client.errors,
70                          [(an_error, 'RemoteException: \n\n')])
71         self.assertEqual(
72             client.failures,
73             [(bing, "RemoteException: foo.c:53:ERROR invalid state\n\n")])
74         self.assertEqual(client.testsRun, 3)
75
76
77 class TestTestProtocolServerStartTest(unittest.TestCase):
78
79     def setUp(self):
80         self.client = Python26TestResult()
81         self.protocol = subunit.TestProtocolServer(self.client)
82
83     def test_start_test(self):
84         self.protocol.lineReceived("test old mcdonald\n")
85         self.assertEqual(self.client._calls,
86             [('startTest', subunit.RemotedTestCase("old mcdonald"))])
87
88     def test_start_testing(self):
89         self.protocol.lineReceived("testing old mcdonald\n")
90         self.assertEqual(self.client._calls,
91             [('startTest', subunit.RemotedTestCase("old mcdonald"))])
92
93     def test_start_test_colon(self):
94         self.protocol.lineReceived("test: old mcdonald\n")
95         self.assertEqual(self.client._calls,
96             [('startTest', subunit.RemotedTestCase("old mcdonald"))])
97
98     def test_start_testing_colon(self):
99         self.protocol.lineReceived("testing: old mcdonald\n")
100         self.assertEqual(self.client._calls,
101             [('startTest', subunit.RemotedTestCase("old mcdonald"))])
102
103
104 class TestTestProtocolServerPassThrough(unittest.TestCase):
105
106     def setUp(self):
107         self.stdout = StringIO()
108         self.test = subunit.RemotedTestCase("old mcdonald")
109         self.client = Python26TestResult()
110         self.protocol = subunit.TestProtocolServer(self.client, self.stdout)
111
112     def keywords_before_test(self):
113         self.protocol.lineReceived("failure a\n")
114         self.protocol.lineReceived("failure: a\n")
115         self.protocol.lineReceived("error a\n")
116         self.protocol.lineReceived("error: a\n")
117         self.protocol.lineReceived("success a\n")
118         self.protocol.lineReceived("success: a\n")
119         self.protocol.lineReceived("successful a\n")
120         self.protocol.lineReceived("successful: a\n")
121         self.protocol.lineReceived("]\n")
122         self.assertEqual(self.stdout.getvalue(), "failure a\n"
123                                                  "failure: a\n"
124                                                  "error a\n"
125                                                  "error: a\n"
126                                                  "success a\n"
127                                                  "success: a\n"
128                                                  "successful a\n"
129                                                  "successful: a\n"
130                                                  "]\n")
131
132     def test_keywords_before_test(self):
133         self.keywords_before_test()
134         self.assertEqual(self.client._calls, [])
135
136     def test_keywords_after_error(self):
137         self.protocol.lineReceived("test old mcdonald\n")
138         self.protocol.lineReceived("error old mcdonald\n")
139         self.keywords_before_test()
140         self.assertEqual([
141             ('startTest', self.test),
142             ('addError', self.test, subunit.RemoteError("")),
143             ('stopTest', self.test),
144             ], self.client._calls)
145
146     def test_keywords_after_failure(self):
147         self.protocol.lineReceived("test old mcdonald\n")
148         self.protocol.lineReceived("failure old mcdonald\n")
149         self.keywords_before_test()
150         self.assertEqual(self.client._calls, [
151             ('startTest', self.test),
152             ('addFailure', self.test, subunit.RemoteError("")),
153             ('stopTest', self.test),
154             ])
155
156     def test_keywords_after_success(self):
157         self.protocol.lineReceived("test old mcdonald\n")
158         self.protocol.lineReceived("success old mcdonald\n")
159         self.keywords_before_test()
160         self.assertEqual([
161             ('startTest', self.test),
162             ('addSuccess', self.test),
163             ('stopTest', self.test),
164             ], self.client._calls)
165
166     def test_keywords_after_test(self):
167         self.protocol.lineReceived("test old mcdonald\n")
168         self.protocol.lineReceived("test old mcdonald\n")
169         self.protocol.lineReceived("failure a\n")
170         self.protocol.lineReceived("failure: a\n")
171         self.protocol.lineReceived("error a\n")
172         self.protocol.lineReceived("error: a\n")
173         self.protocol.lineReceived("success a\n")
174         self.protocol.lineReceived("success: a\n")
175         self.protocol.lineReceived("successful a\n")
176         self.protocol.lineReceived("successful: a\n")
177         self.protocol.lineReceived("]\n")
178         self.protocol.lineReceived("failure old mcdonald\n")
179         self.assertEqual(self.stdout.getvalue(), "test old mcdonald\n"
180                                                  "failure a\n"
181                                                  "failure: a\n"
182                                                  "error a\n"
183                                                  "error: a\n"
184                                                  "success a\n"
185                                                  "success: a\n"
186                                                  "successful a\n"
187                                                  "successful: a\n"
188                                                  "]\n")
189         self.assertEqual(self.client._calls, [
190             ('startTest', self.test),
191             ('addFailure', self.test, subunit.RemoteError("")),
192             ('stopTest', self.test),
193             ])
194
195     def test_keywords_during_failure(self):
196         self.protocol.lineReceived("test old mcdonald\n")
197         self.protocol.lineReceived("failure: old mcdonald [\n")
198         self.protocol.lineReceived("test old mcdonald\n")
199         self.protocol.lineReceived("failure a\n")
200         self.protocol.lineReceived("failure: a\n")
201         self.protocol.lineReceived("error a\n")
202         self.protocol.lineReceived("error: a\n")
203         self.protocol.lineReceived("success a\n")
204         self.protocol.lineReceived("success: a\n")
205         self.protocol.lineReceived("successful a\n")
206         self.protocol.lineReceived("successful: a\n")
207         self.protocol.lineReceived(" ]\n")
208         self.protocol.lineReceived("]\n")
209         self.assertEqual(self.stdout.getvalue(), "")
210         failure = subunit.RemoteError("test old mcdonald\n"
211             "failure a\n"
212             "failure: a\n"
213             "error a\n"
214             "error: a\n"
215             "success a\n"
216             "success: a\n"
217             "successful a\n"
218             "successful: a\n"
219             "]\n")
220         self.assertEqual(self.client._calls, [
221             ('startTest', self.test),
222             ('addFailure', self.test, failure),
223             ('stopTest', self.test),
224             ])
225
226     def test_stdout_passthrough(self):
227         """Lines received which cannot be interpreted as any protocol action
228         should be passed through to sys.stdout.
229         """
230         bytes = "randombytes\n"
231         self.protocol.lineReceived(bytes)
232         self.assertEqual(self.stdout.getvalue(), bytes)
233
234
235 class TestTestProtocolServerLostConnection(unittest.TestCase):
236
237     def setUp(self):
238         self.client = Python26TestResult()
239         self.protocol = subunit.TestProtocolServer(self.client)
240         self.test = subunit.RemotedTestCase("old mcdonald")
241
242     def test_lost_connection_no_input(self):
243         self.protocol.lostConnection()
244         self.assertEqual([], self.client._calls)
245
246     def test_lost_connection_after_start(self):
247         self.protocol.lineReceived("test old mcdonald\n")
248         self.protocol.lostConnection()
249         failure = subunit.RemoteError(
250             "lost connection during test 'old mcdonald'")
251         self.assertEqual([
252             ('startTest', self.test),
253             ('addError', self.test, failure),
254             ('stopTest', self.test),
255             ], self.client._calls)
256
257     def test_lost_connected_after_error(self):
258         self.protocol.lineReceived("test old mcdonald\n")
259         self.protocol.lineReceived("error old mcdonald\n")
260         self.protocol.lostConnection()
261         self.assertEqual([
262             ('startTest', self.test),
263             ('addError', self.test, subunit.RemoteError("")),
264             ('stopTest', self.test),
265             ], self.client._calls)
266
267     def do_connection_lost(self, outcome, opening):
268         self.protocol.lineReceived("test old mcdonald\n")
269         self.protocol.lineReceived("%s old mcdonald %s" % (outcome, opening))
270         self.protocol.lostConnection()
271         failure = subunit.RemoteError(
272             "lost connection during %s report of test 'old mcdonald'" % 
273             outcome)
274         self.assertEqual([
275             ('startTest', self.test),
276             ('addError', self.test, failure),
277             ('stopTest', self.test),
278             ], self.client._calls)
279
280     def test_lost_connection_during_error(self):
281         self.do_connection_lost("error", "[\n")
282
283     def test_lost_connection_during_error_details(self):
284         self.do_connection_lost("error", "[ multipart\n")
285
286     def test_lost_connected_after_failure(self):
287         self.protocol.lineReceived("test old mcdonald\n")
288         self.protocol.lineReceived("failure old mcdonald\n")
289         self.protocol.lostConnection()
290         self.assertEqual([
291             ('startTest', self.test),
292             ('addFailure', self.test, subunit.RemoteError("")),
293             ('stopTest', self.test),
294             ], self.client._calls)
295
296     def test_lost_connection_during_failure(self):
297         self.do_connection_lost("failure", "[\n")
298
299     def test_lost_connection_during_failure_details(self):
300         self.do_connection_lost("failure", "[ multipart\n")
301
302     def test_lost_connection_after_success(self):
303         self.protocol.lineReceived("test old mcdonald\n")
304         self.protocol.lineReceived("success old mcdonald\n")
305         self.protocol.lostConnection()
306         self.assertEqual([
307             ('startTest', self.test),
308             ('addSuccess', self.test),
309             ('stopTest', self.test),
310             ], self.client._calls)
311
312     def test_lost_connection_during_success(self):
313         self.do_connection_lost("success", "[\n")
314
315     def test_lost_connection_during_success_details(self):
316         self.do_connection_lost("success", "[ multipart\n")
317
318     def test_lost_connection_during_skip(self):
319         self.do_connection_lost("skip", "[\n")
320
321     def test_lost_connection_during_skip_details(self):
322         self.do_connection_lost("skip", "[ multipart\n")
323
324     def test_lost_connection_during_xfail(self):
325         self.do_connection_lost("xfail", "[\n")
326
327     def test_lost_connection_during_xfail_details(self):
328         self.do_connection_lost("xfail", "[ multipart\n")
329
330
331 class TestInTestMultipart(unittest.TestCase):
332
333     def setUp(self):
334         self.client = ExtendedTestResult()
335         self.protocol = subunit.TestProtocolServer(self.client)
336         self.protocol.lineReceived("test mcdonalds farm\n")
337         self.test = subunit.RemotedTestCase("mcdonalds farm")
338
339     def test__outcome_sets_details_parser(self):
340         self.protocol._reading_success_details.details_parser = None
341         self.protocol._state._outcome(0, "mcdonalds farm [ multipart\n",
342             None, self.protocol._reading_success_details)
343         parser = self.protocol._reading_success_details.details_parser
344         self.assertNotEqual(None, parser)
345         self.assertTrue(isinstance(parser,
346             subunit.details.MultipartDetailsParser))
347
348
349 class TestTestProtocolServerAddError(unittest.TestCase):
350
351     def setUp(self):
352         self.client = Python26TestResult()
353         self.protocol = subunit.TestProtocolServer(self.client)
354         self.protocol.lineReceived("test mcdonalds farm\n")
355         self.test = subunit.RemotedTestCase("mcdonalds farm")
356
357     def simple_error_keyword(self, keyword):
358         self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
359         failure = subunit.RemoteError("")
360         self.assertEqual([
361             ('startTest', self.test),
362             ('addError', self.test, failure),
363             ('stopTest', self.test),
364             ], self.client._calls)
365
366     def test_simple_error(self):
367         self.simple_error_keyword("error")
368
369     def test_simple_error_colon(self):
370         self.simple_error_keyword("error:")
371
372     def test_error_empty_message(self):
373         self.protocol.lineReceived("error mcdonalds farm [\n")
374         self.protocol.lineReceived("]\n")
375         failure = subunit.RemoteError("")
376         self.assertEqual([
377             ('startTest', self.test),
378             ('addError', self.test, failure),
379             ('stopTest', self.test),
380             ], self.client._calls)
381
382     def error_quoted_bracket(self, keyword):
383         self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
384         self.protocol.lineReceived(" ]\n")
385         self.protocol.lineReceived("]\n")
386         failure = subunit.RemoteError("]\n")
387         self.assertEqual([
388             ('startTest', self.test),
389             ('addError', self.test, failure),
390             ('stopTest', self.test),
391             ], self.client._calls)
392
393     def test_error_quoted_bracket(self):
394         self.error_quoted_bracket("error")
395
396     def test_error_colon_quoted_bracket(self):
397         self.error_quoted_bracket("error:")
398
399
400 class TestTestProtocolServerAddFailure(unittest.TestCase):
401
402     def setUp(self):
403         self.client = Python26TestResult()
404         self.protocol = subunit.TestProtocolServer(self.client)
405         self.protocol.lineReceived("test mcdonalds farm\n")
406         self.test = subunit.RemotedTestCase("mcdonalds farm")
407
408     def simple_failure_keyword(self, keyword):
409         self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
410         failure = subunit.RemoteError("")
411         self.assertEqual([
412             ('startTest', self.test),
413             ('addFailure', self.test, failure),
414             ('stopTest', self.test),
415             ], self.client._calls)
416
417     def test_simple_failure(self):
418         self.simple_failure_keyword("failure")
419
420     def test_simple_failure_colon(self):
421         self.simple_failure_keyword("failure:")
422
423     def test_failure_empty_message(self):
424         self.protocol.lineReceived("failure mcdonalds farm [\n")
425         self.protocol.lineReceived("]\n")
426         failure = subunit.RemoteError("")
427         self.assertEqual([
428             ('startTest', self.test),
429             ('addFailure', self.test, failure),
430             ('stopTest', self.test),
431             ], self.client._calls)
432
433     def failure_quoted_bracket(self, keyword):
434         self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
435         self.protocol.lineReceived(" ]\n")
436         self.protocol.lineReceived("]\n")
437         failure = subunit.RemoteError("]\n")
438         self.assertEqual([
439             ('startTest', self.test),
440             ('addFailure', self.test, failure),
441             ('stopTest', self.test),
442             ], self.client._calls)
443
444     def test_failure_quoted_bracket(self):
445         self.failure_quoted_bracket("failure")
446
447     def test_failure_colon_quoted_bracket(self):
448         self.failure_quoted_bracket("failure:")
449
450
451 class TestTestProtocolServerAddxFail(unittest.TestCase):
452     """Tests for the xfail keyword.
453
454     In Python this can thunk through to Success due to stdlib limitations (see
455     README).
456     """
457
458     def capture_expected_failure(self, test, err):
459         self._calls.append((test, err))
460
461     def setup_python26(self):
462         """Setup a test object ready to be xfailed and thunk to success."""
463         self.client = Python26TestResult()
464         self.setup_protocol()
465
466     def setup_python27(self):
467         """Setup a test object ready to be xfailed and thunk to success."""
468         self.client = Python27TestResult()
469         self.setup_protocol()
470
471     def setup_protocol(self):
472         """Setup the protocol based on self.client."""
473         self.protocol = subunit.TestProtocolServer(self.client)
474         self.protocol.lineReceived("test mcdonalds farm\n")
475         self.test = self.client._calls[-1][-1]
476
477     def simple_xfail_keyword(self, keyword, as_success):
478         self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
479         self.check_success_or_xfail(as_success)
480
481     def check_success_or_xfail(self, as_success, error_message=""):
482         if as_success:
483             self.assertEqual([
484                 ('startTest', self.test),
485                 ('addSuccess', self.test),
486                 ('stopTest', self.test),
487                 ], self.client._calls)
488         else:
489             error = subunit.RemoteError(error_message)
490             self.assertEqual([
491                 ('startTest', self.test),
492                 ('addExpectedFailure', self.test, error),
493                 ('stopTest', self.test),
494                 ], self.client._calls)
495
496     def test_simple_xfail(self):
497         self.setup_python26()
498         self.simple_xfail_keyword("xfail", True)
499         self.setup_python27()
500         self.simple_xfail_keyword("xfail",  False)
501
502     def test_simple_xfail_colon(self):
503         self.setup_python26()
504         self.simple_xfail_keyword("xfail:", True)
505         self.setup_python27()
506         self.simple_xfail_keyword("xfail:", False)
507
508     def test_xfail_empty_message(self):
509         self.setup_python26()
510         self.empty_message(True)
511         self.setup_python27()
512         self.empty_message(False)
513
514     def empty_message(self, as_success):
515         self.protocol.lineReceived("xfail mcdonalds farm [\n")
516         self.protocol.lineReceived("]\n")
517         self.check_success_or_xfail(as_success)
518
519     def xfail_quoted_bracket(self, keyword, as_success):
520         # This tests it is accepted, but cannot test it is used today, because
521         # of not having a way to expose it in Python so far.
522         self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
523         self.protocol.lineReceived(" ]\n")
524         self.protocol.lineReceived("]\n")
525         self.check_success_or_xfail(as_success, "]\n")
526
527     def test_xfail_quoted_bracket(self):
528         self.setup_python26()
529         self.xfail_quoted_bracket("xfail", True)
530         self.setup_python27()
531         self.xfail_quoted_bracket("xfail", False)
532
533     def test_xfail_colon_quoted_bracket(self):
534         self.setup_python26()
535         self.xfail_quoted_bracket("xfail:", True)
536         self.setup_python27()
537         self.xfail_quoted_bracket("xfail:", False)
538
539
540 class TestTestProtocolServerAddSkip(unittest.TestCase):
541     """Tests for the skip keyword.
542
543     In Python this meets the testtools extended TestResult contract.
544     (See https://launchpad.net/testtools).
545     """
546
547     def setUp(self):
548         """Setup a test object ready to be skipped."""
549         self.client = Python27TestResult()
550         self.protocol = subunit.TestProtocolServer(self.client)
551         self.protocol.lineReceived("test mcdonalds farm\n")
552         self.test = self.client._calls[-1][-1]
553
554     def simple_skip_keyword(self, keyword):
555         self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
556         self.assertEqual([
557             ('startTest', self.test),
558             ('addSkip', self.test, "No reason given"),
559             ('stopTest', self.test),
560             ], self.client._calls)
561
562     def test_simple_skip(self):
563         self.simple_skip_keyword("skip")
564
565     def test_simple_skip_colon(self):
566         self.simple_skip_keyword("skip:")
567
568     def test_skip_empty_message(self):
569         self.protocol.lineReceived("skip mcdonalds farm [\n")
570         self.protocol.lineReceived("]\n")
571         self.assertEqual([
572             ('startTest', self.test),
573             ('addSkip', self.test, "No reason given"),
574             ('stopTest', self.test),
575             ], self.client._calls)
576
577     def skip_quoted_bracket(self, keyword):
578         # This tests it is accepted, but cannot test it is used today, because
579         # of not having a way to expose it in Python so far.
580         self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
581         self.protocol.lineReceived(" ]\n")
582         self.protocol.lineReceived("]\n")
583         self.assertEqual([
584             ('startTest', self.test),
585             ('addSkip', self.test, "]\n"),
586             ('stopTest', self.test),
587             ], self.client._calls)
588
589     def test_skip_quoted_bracket(self):
590         self.skip_quoted_bracket("skip")
591
592     def test_skip_colon_quoted_bracket(self):
593         self.skip_quoted_bracket("skip:")
594
595
596 class TestTestProtocolServerAddSuccess(unittest.TestCase):
597
598     def setUp(self):
599         self.client = Python26TestResult()
600         self.protocol = subunit.TestProtocolServer(self.client)
601         self.protocol.lineReceived("test mcdonalds farm\n")
602         self.test = subunit.RemotedTestCase("mcdonalds farm")
603
604     def simple_success_keyword(self, keyword):
605         self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
606         self.assertEqual([
607             ('startTest', self.test),
608             ('addSuccess', self.test),
609             ('stopTest', self.test),
610             ], self.client._calls)
611
612     def test_simple_success(self):
613         self.simple_success_keyword("failure")
614
615     def test_simple_success_colon(self):
616         self.simple_success_keyword("failure:")
617
618     def test_simple_success(self):
619         self.simple_success_keyword("successful")
620
621     def test_simple_success_colon(self):
622         self.simple_success_keyword("successful:")
623
624     def test_success_empty_message(self):
625         self.protocol.lineReceived("success mcdonalds farm [\n")
626         self.protocol.lineReceived("]\n")
627         self.assertEqual([
628             ('startTest', self.test),
629             ('addSuccess', self.test),
630             ('stopTest', self.test),
631             ], self.client._calls)
632
633     def success_quoted_bracket(self, keyword):
634         # This tests it is accepted, but cannot test it is used today, because
635         # of not having a way to expose it in Python so far.
636         self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
637         self.protocol.lineReceived(" ]\n")
638         self.protocol.lineReceived("]\n")
639         self.assertEqual([
640             ('startTest', self.test),
641             ('addSuccess', self.test),
642             ('stopTest', self.test),
643             ], self.client._calls)
644
645     def test_success_quoted_bracket(self):
646         self.success_quoted_bracket("success")
647
648     def test_success_colon_quoted_bracket(self):
649         self.success_quoted_bracket("success:")
650
651
652 class TestTestProtocolServerProgress(unittest.TestCase):
653     """Test receipt of progress: directives."""
654
655     def test_progress_accepted_stdlib(self):
656         self.result = Python26TestResult()
657         self.stream = StringIO()
658         self.protocol = subunit.TestProtocolServer(self.result,
659             stream=self.stream)
660         self.protocol.lineReceived("progress: 23")
661         self.protocol.lineReceived("progress: -2")
662         self.protocol.lineReceived("progress: +4")
663         self.assertEqual("", self.stream.getvalue())
664
665     def test_progress_accepted_extended(self):
666         # With a progress capable TestResult, progress events are emitted.
667         self.result = ExtendedTestResult()
668         self.stream = StringIO()
669         self.protocol = subunit.TestProtocolServer(self.result,
670             stream=self.stream)
671         self.protocol.lineReceived("progress: 23")
672         self.protocol.lineReceived("progress: push")
673         self.protocol.lineReceived("progress: -2")
674         self.protocol.lineReceived("progress: pop")
675         self.protocol.lineReceived("progress: +4")
676         self.assertEqual("", self.stream.getvalue())
677         self.assertEqual([
678             ('progress', 23, subunit.PROGRESS_SET),
679             ('progress', None, subunit.PROGRESS_PUSH),
680             ('progress', -2, subunit.PROGRESS_CUR),
681             ('progress', None, subunit.PROGRESS_POP),
682             ('progress', 4, subunit.PROGRESS_CUR),
683             ], self.result._calls)
684
685
686 class TestTestProtocolServerStreamTags(unittest.TestCase):
687     """Test managing tags on the protocol level."""
688
689     def setUp(self):
690         self.client = ExtendedTestResult()
691         self.protocol = subunit.TestProtocolServer(self.client)
692
693     def test_initial_tags(self):
694         self.protocol.lineReceived("tags: foo bar:baz  quux\n")
695         self.assertEqual([
696             ('tags', set(["foo", "bar:baz", "quux"]), set()),
697             ], self.client._calls)
698
699     def test_minus_removes_tags(self):
700         self.protocol.lineReceived("tags: -bar quux\n")
701         self.assertEqual([
702             ('tags', set(["quux"]), set(["bar"])),
703             ], self.client._calls)
704
705     def test_tags_do_not_get_set_on_test(self):
706         self.protocol.lineReceived("test mcdonalds farm\n")
707         test = self.client._calls[0][-1]
708         self.assertEqual(None, getattr(test, 'tags', None))
709
710     def test_tags_do_not_get_set_on_global_tags(self):
711         self.protocol.lineReceived("tags: foo bar\n")
712         self.protocol.lineReceived("test mcdonalds farm\n")
713         test = self.client._calls[-1][-1]
714         self.assertEqual(None, getattr(test, 'tags', None))
715
716     def test_tags_get_set_on_test_tags(self):
717         self.protocol.lineReceived("test mcdonalds farm\n")
718         test = self.client._calls[-1][-1]
719         self.protocol.lineReceived("tags: foo bar\n")
720         self.protocol.lineReceived("success mcdonalds farm\n")
721         self.assertEqual(None, getattr(test, 'tags', None))
722
723
724 class TestTestProtocolServerStreamTime(unittest.TestCase):
725     """Test managing time information at the protocol level."""
726
727     def test_time_accepted_stdlib(self):
728         self.result = Python26TestResult()
729         self.stream = StringIO()
730         self.protocol = subunit.TestProtocolServer(self.result,
731             stream=self.stream)
732         self.protocol.lineReceived("time: 2001-12-12 12:59:59Z\n")
733         self.assertEqual("", self.stream.getvalue())
734
735     def test_time_accepted_extended(self):
736         self.result = ExtendedTestResult()
737         self.stream = StringIO()
738         self.protocol = subunit.TestProtocolServer(self.result,
739             stream=self.stream)
740         self.protocol.lineReceived("time: 2001-12-12 12:59:59Z\n")
741         self.assertEqual("", self.stream.getvalue())
742         self.assertEqual([
743             ('time', datetime.datetime(2001, 12, 12, 12, 59, 59, 0,
744             iso8601.Utc()))
745             ], self.result._calls)
746
747
748 class TestRemotedTestCase(unittest.TestCase):
749
750     def test_simple(self):
751         test = subunit.RemotedTestCase("A test description")
752         self.assertRaises(NotImplementedError, test.setUp)
753         self.assertRaises(NotImplementedError, test.tearDown)
754         self.assertEqual("A test description",
755                          test.shortDescription())
756         self.assertEqual("A test description",
757                          test.id())
758         self.assertEqual("A test description (subunit.RemotedTestCase)", "%s" % test)
759         self.assertEqual("<subunit.RemotedTestCase description="
760                          "'A test description'>", "%r" % test)
761         result = unittest.TestResult()
762         test.run(result)
763         self.assertEqual([(test, "RemoteException: "
764                                  "Cannot run RemotedTestCases.\n\n")],
765                          result.errors)
766         self.assertEqual(1, result.testsRun)
767         another_test = subunit.RemotedTestCase("A test description")
768         self.assertEqual(test, another_test)
769         different_test = subunit.RemotedTestCase("ofo")
770         self.assertNotEqual(test, different_test)
771         self.assertNotEqual(another_test, different_test)
772
773
774 class TestRemoteError(unittest.TestCase):
775
776     def test_eq(self):
777         error = subunit.RemoteError("Something went wrong")
778         another_error = subunit.RemoteError("Something went wrong")
779         different_error = subunit.RemoteError("boo!")
780         self.assertEqual(error, another_error)
781         self.assertNotEqual(error, different_error)
782         self.assertNotEqual(different_error, another_error)
783
784     def test_empty_constructor(self):
785         self.assertEqual(subunit.RemoteError(), subunit.RemoteError(""))
786
787
788 class TestExecTestCase(unittest.TestCase):
789
790     class SampleExecTestCase(subunit.ExecTestCase):
791
792         def test_sample_method(self):
793             """sample-script.py"""
794             # the sample script runs three tests, one each
795             # that fails, errors and succeeds
796
797         def test_sample_method_args(self):
798             """sample-script.py foo"""
799             # sample that will run just one test.
800
801     def test_construct(self):
802         test = self.SampleExecTestCase("test_sample_method")
803         self.assertEqual(test.script,
804                          subunit.join_dir(__file__, 'sample-script.py'))
805
806     def test_args(self):
807         result = unittest.TestResult()
808         test = self.SampleExecTestCase("test_sample_method_args")
809         test.run(result)
810         self.assertEqual(1, result.testsRun)
811
812     def test_run(self):
813         result = Python26TestResult()
814         test = self.SampleExecTestCase("test_sample_method")
815         test.run(result)
816         mcdonald = subunit.RemotedTestCase("old mcdonald")
817         bing = subunit.RemotedTestCase("bing crosby")
818         an_error = subunit.RemotedTestCase("an error")
819         error_error = subunit.RemoteError()
820         bing_failure = subunit.RemoteError(
821             "foo.c:53:ERROR invalid state\n")
822         self.assertEqual([
823             ('startTest', mcdonald),
824             ('addSuccess', mcdonald),
825             ('stopTest', mcdonald),
826             ('startTest', bing),
827             ('addFailure', bing, bing_failure),
828             ('stopTest', bing),
829             ('startTest', an_error),
830             ('addError', an_error, error_error),
831             ('stopTest', an_error),
832             ], result._calls)
833
834     def test_debug(self):
835         test = self.SampleExecTestCase("test_sample_method")
836         test.debug()
837
838     def test_count_test_cases(self):
839         """TODO run the child process and count responses to determine the count."""
840
841     def test_join_dir(self):
842         sibling = subunit.join_dir(__file__, 'foo')
843         expected = '%s/foo' % (os.path.split(__file__)[0],)
844         self.assertEqual(sibling, expected)
845
846
847 class DoExecTestCase(subunit.ExecTestCase):
848
849     def test_working_script(self):
850         """sample-two-script.py"""
851
852
853 class TestIsolatedTestCase(unittest.TestCase):
854
855     class SampleIsolatedTestCase(subunit.IsolatedTestCase):
856
857         SETUP = False
858         TEARDOWN = False
859         TEST = False
860
861         def setUp(self):
862             TestIsolatedTestCase.SampleIsolatedTestCase.SETUP = True
863
864         def tearDown(self):
865             TestIsolatedTestCase.SampleIsolatedTestCase.TEARDOWN = True
866
867         def test_sets_global_state(self):
868             TestIsolatedTestCase.SampleIsolatedTestCase.TEST = True
869
870
871     def test_construct(self):
872         test = self.SampleIsolatedTestCase("test_sets_global_state")
873
874     def test_run(self):
875         result = unittest.TestResult()
876         test = self.SampleIsolatedTestCase("test_sets_global_state")
877         test.run(result)
878         self.assertEqual(result.testsRun, 1)
879         self.assertEqual(self.SampleIsolatedTestCase.SETUP, False)
880         self.assertEqual(self.SampleIsolatedTestCase.TEARDOWN, False)
881         self.assertEqual(self.SampleIsolatedTestCase.TEST, False)
882
883     def test_debug(self):
884         pass
885         #test = self.SampleExecTestCase("test_sample_method")
886         #test.debug()
887
888
889 class TestIsolatedTestSuite(unittest.TestCase):
890
891     class SampleTestToIsolate(unittest.TestCase):
892
893         SETUP = False
894         TEARDOWN = False
895         TEST = False
896
897         def setUp(self):
898             TestIsolatedTestSuite.SampleTestToIsolate.SETUP = True
899
900         def tearDown(self):
901             TestIsolatedTestSuite.SampleTestToIsolate.TEARDOWN = True
902
903         def test_sets_global_state(self):
904             TestIsolatedTestSuite.SampleTestToIsolate.TEST = True
905
906
907     def test_construct(self):
908         suite = subunit.IsolatedTestSuite()
909
910     def test_run(self):
911         result = unittest.TestResult()
912         suite = subunit.IsolatedTestSuite()
913         sub_suite = unittest.TestSuite()
914         sub_suite.addTest(self.SampleTestToIsolate("test_sets_global_state"))
915         sub_suite.addTest(self.SampleTestToIsolate("test_sets_global_state"))
916         suite.addTest(sub_suite)
917         suite.addTest(self.SampleTestToIsolate("test_sets_global_state"))
918         suite.run(result)
919         self.assertEqual(result.testsRun, 3)
920         self.assertEqual(self.SampleTestToIsolate.SETUP, False)
921         self.assertEqual(self.SampleTestToIsolate.TEARDOWN, False)
922         self.assertEqual(self.SampleTestToIsolate.TEST, False)
923
924
925 class TestTestProtocolClient(unittest.TestCase):
926
927     def setUp(self):
928         self.io = StringIO()
929         self.protocol = subunit.TestProtocolClient(self.io)
930         self.test = TestTestProtocolClient("test_start_test")
931         self.sample_details = {'something':Content(
932             ContentType('text', 'plain'), lambda:['serialised\nform'])}
933         self.sample_tb_details = dict(self.sample_details)
934         self.sample_tb_details['traceback'] = TracebackContent(
935             subunit.RemoteError("boo qux"))
936
937     def test_start_test(self):
938         """Test startTest on a TestProtocolClient."""
939         self.protocol.startTest(self.test)
940         self.assertEqual(self.io.getvalue(), "test: %s\n" % self.test.id())
941
942     def test_stop_test(self):
943         # stopTest doesn't output anything.
944         self.protocol.stopTest(self.test)
945         self.assertEqual(self.io.getvalue(), "")
946
947     def test_add_success(self):
948         """Test addSuccess on a TestProtocolClient."""
949         self.protocol.addSuccess(self.test)
950         self.assertEqual(
951             self.io.getvalue(), "successful: %s\n" % self.test.id())
952
953     def test_add_success_details(self):
954         """Test addSuccess on a TestProtocolClient with details."""
955         self.protocol.addSuccess(self.test, details=self.sample_details)
956         self.assertEqual(
957             self.io.getvalue(), "successful: %s [ multipart\n"
958                 "Content-Type: text/plain\n"
959                 "something\n"
960                 "F\r\nserialised\nform0\r\n]\n" % self.test.id())
961
962     def test_add_failure(self):
963         """Test addFailure on a TestProtocolClient."""
964         self.protocol.addFailure(
965             self.test, subunit.RemoteError("boo qux"))
966         self.assertEqual(
967             self.io.getvalue(),
968             'failure: %s [\nRemoteException: boo qux\n]\n' % self.test.id())
969
970     def test_add_failure_details(self):
971         """Test addFailure on a TestProtocolClient with details."""
972         self.protocol.addFailure(
973             self.test, details=self.sample_tb_details)
974         self.assertEqual(
975             self.io.getvalue(),
976             "failure: %s [ multipart\n"
977             "Content-Type: text/plain\n"
978             "something\n"
979             "F\r\nserialised\nform0\r\n"
980             "Content-Type: text/x-traceback;language=python\n"
981             "traceback\n"
982             "19\r\nRemoteException: boo qux\n0\r\n"
983             "]\n" % self.test.id())
984
985     def test_add_error(self):
986         """Test stopTest on a TestProtocolClient."""
987         self.protocol.addError(
988             self.test, subunit.RemoteError("phwoar crikey"))
989         self.assertEqual(
990             self.io.getvalue(),
991             'error: %s [\n'
992             "RemoteException: phwoar crikey\n"
993             "]\n" % self.test.id())
994
995     def test_add_error_details(self):
996         """Test stopTest on a TestProtocolClient with details."""
997         self.protocol.addError(
998             self.test, details=self.sample_tb_details)
999         self.assertEqual(
1000             self.io.getvalue(),
1001             "error: %s [ multipart\n"
1002             "Content-Type: text/plain\n"
1003             "something\n"
1004             "F\r\nserialised\nform0\r\n"
1005             "Content-Type: text/x-traceback;language=python\n"
1006             "traceback\n"
1007             "19\r\nRemoteException: boo qux\n0\r\n"
1008             "]\n" % self.test.id())
1009
1010     def test_add_expected_failure(self):
1011         """Test addExpectedFailure on a TestProtocolClient."""
1012         self.protocol.addExpectedFailure(
1013             self.test, subunit.RemoteError("phwoar crikey"))
1014         self.assertEqual(
1015             self.io.getvalue(),
1016             'xfail: %s [\n'
1017             "RemoteException: phwoar crikey\n"
1018             "]\n" % self.test.id())
1019
1020     def test_add_expected_failure_details(self):
1021         """Test addExpectedFailure on a TestProtocolClient with details."""
1022         self.protocol.addExpectedFailure(
1023             self.test, details=self.sample_tb_details)
1024         self.assertEqual(
1025             self.io.getvalue(),
1026             "xfail: %s [ multipart\n"
1027             "Content-Type: text/plain\n"
1028             "something\n"
1029             "F\r\nserialised\nform0\r\n"
1030             "Content-Type: text/x-traceback;language=python\n"
1031             "traceback\n"
1032             "19\r\nRemoteException: boo qux\n0\r\n"
1033             "]\n" % self.test.id())
1034
1035     def test_add_skip(self):
1036         """Test addSkip on a TestProtocolClient."""
1037         self.protocol.addSkip(
1038             self.test, "Has it really?")
1039         self.assertEqual(
1040             self.io.getvalue(),
1041             'skip: %s [\nHas it really?\n]\n' % self.test.id())
1042     
1043     def test_add_skip_details(self):
1044         """Test addSkip on a TestProtocolClient with details."""
1045         details = {'reason':Content(
1046             ContentType('text', 'plain'), lambda:['Has it really?'])}
1047         self.protocol.addSkip(
1048             self.test, details=details)
1049         self.assertEqual(
1050             self.io.getvalue(),
1051             "skip: %s [ multipart\n"
1052             "Content-Type: text/plain\n"
1053             "reason\n"
1054             "E\r\nHas it really?0\r\n"
1055             "]\n" % self.test.id())
1056
1057     def test_progress_set(self):
1058         self.protocol.progress(23, subunit.PROGRESS_SET)
1059         self.assertEqual(self.io.getvalue(), 'progress: 23\n')
1060
1061     def test_progress_neg_cur(self):
1062         self.protocol.progress(-23, subunit.PROGRESS_CUR)
1063         self.assertEqual(self.io.getvalue(), 'progress: -23\n')
1064
1065     def test_progress_pos_cur(self):
1066         self.protocol.progress(23, subunit.PROGRESS_CUR)
1067         self.assertEqual(self.io.getvalue(), 'progress: +23\n')
1068
1069     def test_progress_pop(self):
1070         self.protocol.progress(1234, subunit.PROGRESS_POP)
1071         self.assertEqual(self.io.getvalue(), 'progress: pop\n')
1072
1073     def test_progress_push(self):
1074         self.protocol.progress(1234, subunit.PROGRESS_PUSH)
1075         self.assertEqual(self.io.getvalue(), 'progress: push\n')
1076
1077     def test_time(self):
1078         # Calling time() outputs a time signal immediately.
1079         self.protocol.time(
1080             datetime.datetime(2009,10,11,12,13,14,15, iso8601.Utc()))
1081         self.assertEqual(
1082             "time: 2009-10-11 12:13:14.000015Z\n",
1083             self.io.getvalue())
1084
1085     def test_add_unexpected_success(self):
1086         """Test addUnexpectedSuccess on a TestProtocolClient."""
1087         self.protocol.addUnexpectedSuccess(self.test)
1088         self.assertEqual(
1089             self.io.getvalue(), "successful: %s\n" % self.test.id())
1090
1091     def test_add_unexpected_success_details(self):
1092         """Test addUnexpectedSuccess on a TestProtocolClient with details."""
1093         self.protocol.addUnexpectedSuccess(self.test, details=self.sample_details)
1094         self.assertEqual(
1095             self.io.getvalue(), "successful: %s [ multipart\n"
1096                 "Content-Type: text/plain\n"
1097                 "something\n"
1098                 "F\r\nserialised\nform0\r\n]\n" % self.test.id())
1099
1100
1101 def test_suite():
1102     loader = subunit.tests.TestUtil.TestLoader()
1103     result = loader.loadTestsFromName(__name__)
1104     return result