1 # Python module for parsing and generating the Subunit protocol
3 # Copyright (C) 2008-2009 Jelmer Vernooij <jelmer@samba.org>
5 # This program is free software; you can redistribute it and/or modify
6 # it under the terms of the GNU General Public License as published by
7 # the Free Software Foundation; either version 3 of the License, or
8 # (at your option) any later version.
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU General Public License for more details.
15 # You should have received a copy of the GNU General Public License
16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18 __all__
= ['parse_results']
23 from samba
import subunit
24 from samba
.subunit
.run
import TestProtocolClient
25 from samba
.subunit
import iso8601
28 VALID_RESULTS
= set(['success', 'successful', 'failure', 'fail', 'skip',
29 'knownfail', 'error', 'xfail', 'skip-testsuite',
30 'testsuite-failure', 'testsuite-xfail',
31 'testsuite-success', 'testsuite-error',
32 'uxsuccess', 'testsuite-uxsuccess'])
34 class TestsuiteEnabledTestResult(unittest
.TestResult
):
36 def start_testsuite(self
, name
):
37 raise NotImplementedError(self
.start_testsuite
)
40 def parse_results(msg_ops
, statistics
, fh
):
48 parts
= l
.split(None, 1)
49 if not len(parts
) == 2 or not l
.startswith(parts
[0]):
52 command
= parts
[0].rstrip(":")
54 if command
in ("test", "testing"):
55 msg_ops
.control_msg(l
)
57 test
= subunit
.RemotedTestCase(name
)
58 if name
in open_tests
:
59 msg_ops
.addError(open_tests
.pop(name
), subunit
.RemoteError(u
"Test already running"))
60 msg_ops
.startTest(test
)
61 open_tests
[name
] = test
62 elif command
== "time":
63 msg_ops
.control_msg(l
)
65 dt
= iso8601
.parse_date(arg
.rstrip("\n"))
67 print "Unable to parse time line: %s" % arg
.rstrip("\n")
70 elif command
in VALID_RESULTS
:
71 msg_ops
.control_msg(l
)
73 grp
= re
.match("(.*?)( \[)?([ \t]*)( multipart)?\n", arg
)
74 (testname
, hasreason
) = (grp
.group(1), grp
.group(2))
77 # reason may be specified in next lines
83 msg_ops
.control_msg(l
)
90 remote_error
= subunit
.RemoteError(reason
.decode("utf-8"))
93 statistics
['TESTS_ERROR']+=1
94 msg_ops
.addError(subunit
.RemotedTestCase(testname
), subunit
.RemoteError(u
"reason (%s) interrupted" % result
))
98 remote_error
= subunit
.RemoteError(u
"No reason specified")
99 if result
in ("success", "successful"):
101 test
= open_tests
.pop(testname
)
103 statistics
['TESTS_ERROR']+=1
105 msg_ops
.addError(subunit
.RemotedTestCase(testname
), subunit
.RemoteError(u
"Test was never started"))
107 statistics
['TESTS_EXPECTED_OK']+=1
108 msg_ops
.addSuccess(test
)
109 elif result
in ("xfail", "knownfail"):
111 test
= open_tests
.pop(testname
)
113 statistics
['TESTS_ERROR']+=1
115 msg_ops
.addError(subunit
.RemotedTestCase(testname
), subunit
.RemoteError(u
"Test was never started"))
117 statistics
['TESTS_EXPECTED_FAIL']+=1
118 msg_ops
.addExpectedFailure(test
, remote_error
)
119 elif result
in ("uxsuccess", ):
121 test
= open_tests
.pop(testname
)
123 statistics
['TESTS_ERROR']+=1
125 msg_ops
.addError(subunit
.RemotedTestCase(testname
), subunit
.RemoteError(u
"Test was never started"))
127 statistics
['TESTS_UNEXPECTED_OK']+=1
128 msg_ops
.addUnexpectedSuccess(test
)
130 elif result
in ("failure", "fail"):
132 test
= open_tests
.pop(testname
)
134 statistics
['TESTS_ERROR']+=1
136 msg_ops
.addError(subunit
.RemotedTestCase(testname
), subunit
.RemoteError(u
"Test was never started"))
138 statistics
['TESTS_UNEXPECTED_FAIL']+=1
140 msg_ops
.addFailure(test
, remote_error
)
141 elif result
== "skip":
142 statistics
['TESTS_SKIP']+=1
143 # Allow tests to be skipped without prior announcement of test
145 test
= open_tests
.pop(testname
)
147 test
= subunit
.RemotedTestCase(testname
)
148 msg_ops
.addSkip(test
, reason
)
149 elif result
== "error":
150 statistics
['TESTS_ERROR']+=1
153 test
= open_tests
.pop(testname
)
155 test
= subunit
.RemotedTestCase(testname
)
156 msg_ops
.addError(test
, remote_error
)
157 elif result
== "skip-testsuite":
158 msg_ops
.skip_testsuite(testname
)
159 elif result
== "testsuite-success":
160 msg_ops
.end_testsuite(testname
, "success", reason
)
161 elif result
== "testsuite-failure":
162 msg_ops
.end_testsuite(testname
, "failure", reason
)
164 elif result
== "testsuite-xfail":
165 msg_ops
.end_testsuite(testname
, "xfail", reason
)
166 elif result
== "testsuite-uxsuccess":
167 msg_ops
.end_testsuite(testname
, "uxsuccess", reason
)
169 elif result
== "testsuite-error":
170 msg_ops
.end_testsuite(testname
, "error", reason
)
173 raise AssertionError("Recognized but unhandled result %r" %
175 elif command
== "testsuite":
176 msg_ops
.start_testsuite(arg
.strip())
177 elif command
== "progress":
180 msg_ops
.progress(None, subunit
.PROGRESS_POP
)
182 msg_ops
.progress(None, subunit
.PROGRESS_PUSH
)
184 msg_ops
.progress(int(arg
), subunit
.PROGRESS_CUR
)
186 msg_ops
.progress(int(arg
), subunit
.PROGRESS_SET
)
188 msg_ops
.output_msg(l
)
191 test
= subunit
.RemotedTestCase(open_tests
.popitem()[1])
192 msg_ops
.addError(test
, subunit
.RemoteError(u
"was started but never finished!"))
193 statistics
['TESTS_ERROR']+=1
199 class SubunitOps(TestProtocolClient
,TestsuiteEnabledTestResult
):
201 def progress(self
, count
, whence
):
202 if whence
== subunit
.PROGRESS_POP
:
203 self
._stream
.write("progress: pop\n")
204 elif whence
== subunit
.PROGRESS_PUSH
:
205 self
._stream
.write("progress: push\n")
206 elif whence
== subunit
.PROGRESS_SET
:
207 self
._stream
.write("progress: %d\n" % count
)
208 elif whence
== subunit
.PROGRESS_CUR
:
209 raise NotImplementedError
211 # The following are Samba extensions:
212 def start_testsuite(self
, name
):
213 self
._stream
.write("testsuite: %s\n" % name
)
215 def skip_testsuite(self
, name
, reason
=None):
217 self
._stream
.write("skip-testsuite: %s [\n%s\n]\n" % (name
, reason
))
219 self
._stream
.write("skip-testsuite: %s\n" % name
)
221 def end_testsuite(self
, name
, result
, reason
=None):
223 self
._stream
.write("testsuite-%s: %s [\n%s\n]\n" % (result
, name
, reason
))
225 self
._stream
.write("testsuite-%s: %s\n" % (result
, name
))
227 def output_msg(self
, msg
):
228 self
._stream
.write(msg
)
231 def read_test_regexes(name
):
237 if l
== "" or l
[0] == "#":
240 (regex
, reason
) = l
.split("#", 1)
241 ret
[regex
.strip()] = reason
.strip()
249 def find_in_list(regexes
, fullname
):
250 for regex
, reason
in regexes
.iteritems():
251 if re
.match(regex
, fullname
):
258 class ImmediateFail(Exception):
259 """Raised to abort immediately."""
262 super(ImmediateFail
, self
).__init
__("test failed and fail_immediately set")
265 class FilterOps(unittest
.TestResult
):
267 def control_msg(self
, msg
):
268 pass # We regenerate control messages, so ignore this
270 def time(self
, time
):
273 def progress(self
, delta
, whence
):
274 self
._ops
.progress(delta
, whence
)
276 def output_msg(self
, msg
):
277 if self
.output
is None:
278 sys
.stdout
.write(msg
)
282 def startTest(self
, test
):
283 self
.seen_output
= True
284 test
= self
._add
_prefix
(test
)
285 if self
.strip_ok_output
:
288 self
._ops
.startTest(test
)
290 def _add_prefix(self
, test
):
291 return subunit
.RemotedTestCase(self
.prefix
+ test
.id() + self
.suffix
)
293 def addError(self
, test
, err
=None):
294 test
= self
._add
_prefix
(test
)
297 self
._ops
.addError(test
, err
)
299 if self
.fail_immediately
:
300 raise ImmediateFail()
302 def addSkip(self
, test
, reason
=None):
303 self
.seen_output
= True
304 test
= self
._add
_prefix
(test
)
305 self
._ops
.addSkip(test
, reason
)
308 def addExpectedFailure(self
, test
, err
=None):
309 test
= self
._add
_prefix
(test
)
310 self
._ops
.addExpectedFailure(test
, err
)
313 def addUnexpectedSuccess(self
, test
):
314 test
= self
._add
_prefix
(test
)
315 self
.uxsuccess_added
+=1
316 self
.total_uxsuccess
+=1
317 self
._ops
.addUnexpectedSuccess(test
)
319 self
._ops
.output_msg(self
.output
)
321 if self
.fail_immediately
:
322 raise ImmediateFail()
324 def addFailure(self
, test
, err
=None):
325 test
= self
._add
_prefix
(test
)
326 xfail_reason
= find_in_list(self
.expected_failures
, test
.id())
327 if xfail_reason
is None:
328 xfail_reason
= find_in_list(self
.flapping
, test
.id())
329 if xfail_reason
is not None:
332 self
._ops
.addExpectedFailure(test
, err
)
336 self
._ops
.addFailure(test
, err
)
338 self
._ops
.output_msg(self
.output
)
339 if self
.fail_immediately
:
340 raise ImmediateFail()
343 def addSuccess(self
, test
):
344 test
= self
._add
_prefix
(test
)
345 xfail_reason
= find_in_list(self
.expected_failures
, test
.id())
346 if xfail_reason
is not None:
347 self
.uxsuccess_added
+= 1
348 self
.total_uxsuccess
+= 1
349 self
._ops
.addUnexpectedSuccess(test
)
351 self
._ops
.output_msg(self
.output
)
352 if self
.fail_immediately
:
353 raise ImmediateFail()
355 self
._ops
.addSuccess(test
)
358 def skip_testsuite(self
, name
, reason
=None):
359 self
._ops
.skip_testsuite(name
, reason
)
361 def start_testsuite(self
, name
):
362 self
._ops
.start_testsuite(name
)
366 self
.uxsuccess_added
= 0
368 def end_testsuite(self
, name
, result
, reason
=None):
371 if self
.xfail_added
> 0:
373 if self
.fail_added
> 0 or self
.error_added
> 0 or self
.uxsuccess_added
> 0:
376 if xfail
and result
in ("fail", "failure"):
379 if self
.uxsuccess_added
> 0 and result
!= "uxsuccess":
382 reason
= "Subunit/Filter Reason"
383 reason
+= "\n uxsuccess[%d]" % self
.uxsuccess_added
385 if self
.fail_added
> 0 and result
!= "failure":
388 reason
= "Subunit/Filter Reason"
389 reason
+= "\n failures[%d]" % self
.fail_added
391 if self
.error_added
> 0 and result
!= "error":
394 reason
= "Subunit/Filter Reason"
395 reason
+= "\n errors[%d]" % self
.error_added
397 self
._ops
.end_testsuite(name
, result
, reason
)
398 if result
not in ("success", "xfail"):
400 self
._ops
.output_msg(self
.output
)
401 if self
.fail_immediately
:
402 raise ImmediateFail()
405 def __init__(self
, out
, prefix
=None, suffix
=None, expected_failures
=None,
406 strip_ok_output
=False, fail_immediately
=False,
409 self
.seen_output
= False
413 if expected_failures
is not None:
414 self
.expected_failures
= expected_failures
416 self
.expected_failures
= {}
417 if flapping
is not None:
418 self
.flapping
= flapping
421 self
.strip_ok_output
= strip_ok_output
424 self
.uxsuccess_added
= 0
428 self
.total_uxsuccess
= 0
430 self
.fail_immediately
= fail_immediately
433 class PerfFilterOps(unittest
.TestResult
):
435 def progress(self
, delta
, whence
):
438 def output_msg(self
, msg
):
441 def control_msg(self
, msg
):
444 def skip_testsuite(self
, name
, reason
=None):
445 self
._ops
.skip_testsuite(name
, reason
)
447 def start_testsuite(self
, name
):
448 self
.suite_has_time
= False
450 def end_testsuite(self
, name
, result
, reason
=None):
453 def _add_prefix(self
, test
):
454 return subunit
.RemotedTestCase(self
.prefix
+ test
.id() + self
.suffix
)
456 def time(self
, time
):
457 self
.latest_time
= time
458 #self._ops.output_msg("found time %s\n" % time)
459 self
.suite_has_time
= True
462 if self
.suite_has_time
:
463 return self
.latest_time
464 return datetime
.datetime
.utcnow()
466 def startTest(self
, test
):
467 self
.seen_output
= True
468 test
= self
._add
_prefix
(test
)
469 self
.starts
[test
.id()] = self
.get_time()
471 def addSuccess(self
, test
):
472 test
= self
._add
_prefix
(test
)
474 if tid
not in self
.starts
:
475 self
._ops
.addError(test
, "%s succeeded without ever starting!" % tid
)
476 delta
= self
.get_time() - self
.starts
[tid
]
477 self
._ops
.output_msg("elapsed-time: %s: %f\n" % (tid
, delta
.total_seconds()))
479 def addFailure(self
, test
, err
=''):
481 delta
= self
.get_time() - self
.starts
[tid
]
482 self
._ops
.output_msg("failure: %s failed after %f seconds (%s)\n" %
483 (tid
, delta
.total_seconds(), err
))
485 def addError(self
, test
, err
=''):
487 delta
= self
.get_time() - self
.starts
[tid
]
488 self
._ops
.output_msg("error: %s failed after %f seconds (%s)\n" %
489 (tid
, delta
.total_seconds(), err
))
491 def __init__(self
, out
, prefix
='', suffix
=''):
493 self
.prefix
= prefix
or ''
494 self
.suffix
= suffix
or ''
496 self
.seen_output
= False
497 self
.suite_has_time
= False
500 class PlainFormatter(TestsuiteEnabledTestResult
):
502 def __init__(self
, verbose
, immediate
, statistics
,
504 super(PlainFormatter
, self
).__init
__()
505 self
.verbose
= verbose
506 self
.immediate
= immediate
507 self
.statistics
= statistics
508 self
.start_time
= None
509 self
.test_output
= {}
510 self
.suitesfailed
= []
515 self
._progress
_level
= 0
516 self
.totalsuites
= totaltests
517 self
.last_time
= None
520 def _format_time(delta
):
521 minutes
, seconds
= divmod(delta
.seconds
, 60)
522 hours
, minutes
= divmod(minutes
, 60)
527 ret
+= "%dm" % minutes
528 ret
+= "%ds" % seconds
531 def progress(self
, offset
, whence
):
532 if whence
== subunit
.PROGRESS_POP
:
533 self
._progress
_level
-= 1
534 elif whence
== subunit
.PROGRESS_PUSH
:
535 self
._progress
_level
+= 1
536 elif whence
== subunit
.PROGRESS_SET
:
537 if self
._progress
_level
== 0:
538 self
.totalsuites
= offset
539 elif whence
== subunit
.PROGRESS_CUR
:
540 raise NotImplementedError
543 if self
.start_time
is None:
547 def start_testsuite(self
, name
):
552 self
.test_output
[name
] = ""
554 total_tests
= (self
.statistics
['TESTS_EXPECTED_OK'] +
555 self
.statistics
['TESTS_EXPECTED_FAIL'] +
556 self
.statistics
['TESTS_ERROR'] +
557 self
.statistics
['TESTS_UNEXPECTED_FAIL'] +
558 self
.statistics
['TESTS_UNEXPECTED_OK'])
560 out
= "[%d(%d)" % (self
.index
, total_tests
)
561 if self
.totalsuites
is not None:
562 out
+= "/%d" % self
.totalsuites
563 if self
.start_time
is not None:
564 out
+= " at " + self
._format
_time
(self
.last_time
- self
.start_time
)
565 if self
.suitesfailed
:
566 out
+= ", %d errors" % (len(self
.suitesfailed
),)
569 sys
.stdout
.write(out
+ "\n")
571 sys
.stdout
.write(out
+ ": ")
573 def output_msg(self
, output
):
575 sys
.stdout
.write(output
)
576 elif self
.name
is not None:
577 self
.test_output
[self
.name
] += output
579 sys
.stdout
.write(output
)
581 def control_msg(self
, output
):
584 def end_testsuite(self
, name
, result
, reason
):
588 if not name
in self
.test_output
:
589 print "no output for name[%s]" % name
591 if result
in ("success", "xfail"):
594 self
.output_msg("ERROR: Testsuite[%s]\n" % name
)
595 if reason
is not None:
596 self
.output_msg("REASON: %s\n" % (reason
,))
597 self
.suitesfailed
.append(name
)
598 if self
.immediate
and not self
.verbose
and name
in self
.test_output
:
599 out
+= self
.test_output
[name
]
602 if not self
.immediate
:
606 out
+= " " + result
.upper() + "\n"
608 sys
.stdout
.write(out
)
610 def startTest(self
, test
):
613 def addSuccess(self
, test
):
614 self
.end_test(test
.id(), "success", False)
616 def addError(self
, test
, err
=None):
617 self
.end_test(test
.id(), "error", True, err
)
619 def addFailure(self
, test
, err
=None):
620 self
.end_test(test
.id(), "failure", True, err
)
622 def addSkip(self
, test
, reason
=None):
623 self
.end_test(test
.id(), "skip", False, reason
)
625 def addExpectedFailure(self
, test
, err
=None):
626 self
.end_test(test
.id(), "xfail", False, err
)
628 def addUnexpectedSuccess(self
, test
):
629 self
.end_test(test
.id(), "uxsuccess", True)
631 def end_test(self
, testname
, result
, unexpected
, err
=None):
633 self
.test_output
[self
.name
] = ""
634 if not self
.immediate
:
639 'success': '.'}.get(result
, "?(%s)" % result
))
642 if not self
.name
in self
.test_output
:
643 self
.test_output
[self
.name
] = ""
645 self
.test_output
[self
.name
] += "UNEXPECTED(%s): %s\n" % (result
, testname
)
647 self
.test_output
[self
.name
] += "REASON: %s\n" % str(err
[1]).strip()
649 if self
.immediate
and not self
.verbose
:
650 sys
.stdout
.write(self
.test_output
[self
.name
])
651 self
.test_output
[self
.name
] = ""
653 if not self
.immediate
:
658 'success': 'S'}.get(result
, "?"))
660 def write_summary(self
, path
):
663 if self
.suitesfailed
:
664 f
.write("= Failed tests =\n")
666 for suite
in self
.suitesfailed
:
667 f
.write("== %s ==\n" % suite
)
668 if suite
in self
.test_output
:
669 f
.write(self
.test_output
[suite
]+"\n\n")
673 if not self
.immediate
and not self
.verbose
:
674 for suite
in self
.suitesfailed
:
676 print "FAIL: %s" % suite
677 if suite
in self
.test_output
:
678 print self
.test_output
[suite
]
681 f
.write("= Skipped tests =\n")
682 for reason
in self
.skips
.keys():
683 f
.write(reason
+ "\n")
684 for name
in self
.skips
[reason
]:
685 f
.write("\t%s\n" % name
)
689 if (not self
.suitesfailed
and
690 not self
.statistics
['TESTS_UNEXPECTED_FAIL'] and
691 not self
.statistics
['TESTS_UNEXPECTED_OK'] and
692 not self
.statistics
['TESTS_ERROR']):
693 ok
= (self
.statistics
['TESTS_EXPECTED_OK'] +
694 self
.statistics
['TESTS_EXPECTED_FAIL'])
695 print "\nALL OK (%d tests in %d testsuites)" % (ok
, self
.suites_ok
)
697 print "\nFAILED (%d failures, %d errors and %d unexpected successes in %d testsuites)" % (
698 self
.statistics
['TESTS_UNEXPECTED_FAIL'],
699 self
.statistics
['TESTS_ERROR'],
700 self
.statistics
['TESTS_UNEXPECTED_OK'],
701 len(self
.suitesfailed
))
703 def skip_testsuite(self
, name
, reason
="UNKNOWN"):
704 self
.skips
.setdefault(reason
, []).append(name
)