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