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