Coverage for pycotap/__init__.py: 99%

1#!/usr/bin/env python 

2 

3# Copyright (c) 2015 Remko Tronçon (https://el-tramo.be) 

4# Released under the MIT license 

5# See COPYING for details 

6 

7import base64 

8import sys 

9import unittest 

10 

11if sys.hexversion >= 0x03000000: 

12 from io import StringIO 

13else: 

14 from StringIO import StringIO # pylint: disable=import-error 

15 

16 

17# Log modes 

18class LogMode(object): 

19 LogToError, LogToDiagnostics, LogToYAML, LogToAttachment = range(4) 

20 

21 

22class TAPTestResult(unittest.TestResult): 

23 def __init__(self, output_stream, error_stream, message_log, test_output_log): 

24 super(TAPTestResult, self).__init__(self, output_stream) 

25 self.have_fail = False 

26 self.output_stream = output_stream 

27 self.error_stream = error_stream 

28 self.orig_stdout = None 

29 self.orig_stderr = None 

30 self.message = None 

31 self.test_output = None 

32 self.message_log = message_log 

33 self.test_output_log = test_output_log 

34 self.output_stream.write("TAP version 13\n") 

35 self._set_streams() 

36 

37 def printErrors(self): 

38 self.print_raw("1..%d\n" % self.testsRun) 

39 self._reset_streams() 

40 

41 def _set_streams(self): 

42 self.orig_stdout = sys.stdout 

43 self.orig_stderr = sys.stderr 

44 if self.message_log == LogMode.LogToError: 

45 self.message = self.error_stream 

46 else: 

47 self.message = StringIO() 

48 if self.test_output_log == LogMode.LogToError: 

49 self.test_output = self.error_stream 

50 else: 

51 self.test_output = StringIO() 

52 

53 if self.message_log == self.test_output_log: 

54 self.test_output = self.message 

55 sys.stdout = sys.stderr = self.test_output 

56 

57 def _reset_streams(self): 

58 sys.stdout = self.orig_stdout 

59 sys.stderr = self.orig_stderr 

60 

61 def print_raw(self, text): 

62 self.output_stream.write(text) 

63 self.output_stream.flush() 

64 

65 def print_result(self, result, test, directive = None): 

66 self.output_stream.write("%s %d %s" % (result, self.testsRun, test.id())) 

67 if directive: 

68 self.output_stream.write(" # " + directive) 

69 self.output_stream.write("\n") 

70 self.output_stream.flush() 

71 

72 def ok(self, test, directive = None): 

73 self.print_result("ok", test, directive) 

74 

75 def not_ok(self, test): 

76 self.have_fail = True 

77 self.print_result("not ok", test) 

78 

79 # def startTest(self, test): 

80 # super(TAPTestResult, self).startTest(test) 

81 

82 def stopTest(self, test): 

83 super(TAPTestResult, self).stopTest(test) 

84 if self.message_log == self.test_output_log: 

85 logs = [(self.message_log, self.message, "output")] 

86 else: 

87 logs = [ 

88 (self.test_output_log, self.test_output, "test_output"), 

89 (self.message_log, self.message, "message") 

90 ] 

91 for log_mode, log, log_name in logs: 

92 if log_mode != LogMode.LogToError: 

93 output = log.getvalue() 

94 if len(output): 

95 if log_mode == LogMode.LogToYAML: 

96 self.print_raw(" ---\n") 

97 self.print_raw(" " + log_name + ": |\n") 

98 self.print_raw(" " + output.rstrip().replace("\n", "\n ") + "\n") 

99 self.print_raw(" ...\n") 

100 elif log_mode == LogMode.LogToAttachment: 

101 content = base64.b64encode(output.encode("utf-8")).decode("utf-8") 

102 self.print_raw(" ---\n") 

103 self.print_raw(" " + log_name + ":\n") 

104 self.print_raw(" File-Name: " + log_name + ".txt\n") 

105 self.print_raw(" File-Type: text/plain\n") 

106 self.print_raw(" File-Content: " + content + "\n") 

107 self.print_raw(" ...\n") 

108 else: 

109 self.print_raw("# " + output.rstrip().replace("\n", "\n# ") + "\n") 

110 log.seek(0) 

111 log.truncate(0) 

112 

113 def addSuccess(self, test): 

114 super(TAPTestResult, self).addSuccess(test) 

115 self.ok(test) 

116 

117 def addError(self, test, err): 

118 super(TAPTestResult, self).addError(test, err) 

119 self.message.write(self.errors[-1][1] + "\n") 

120 self.not_ok(test) 

121 

122 def addFailure(self, test, err): 

123 super(TAPTestResult, self).addFailure(test, err) 

124 self.message.write(self.failures[-1][1] + "\n") 

125 self.not_ok(test) 

126 

127 def addSkip(self, test, reason): 

128 super(TAPTestResult, self).addSkip(test, reason) 

129 self.ok(test, "SKIP " + reason) 

130 

131 def addExpectedFailure(self, test, err): 

132 super(TAPTestResult, self).addExpectedFailure(test, err) 

133 self.ok(test) 

134 

135 def addUnexpectedSuccess(self, test): 

136 super(TAPTestResult, self).addUnexpectedSuccess(test) 

137 self.message.write("Unexpected success" + "\n") 

138 self.not_ok(test) 

139 

140 

141class TAPTestRunner(object): 

142 def __init__( 

143 self, 

144 message_log = LogMode.LogToYAML, 

145 test_output_log = LogMode.LogToDiagnostics, 

146 output_stream = sys.stdout, 

147 error_stream = sys.stderr 

148 ): 

149 self.output_stream = output_stream 

150 self.error_stream = error_stream 

151 self.message_log = message_log 

152 self.test_output_log = test_output_log 

153 

154 def run(self, test): 

155 result = TAPTestResult( 

156 self.output_stream, self.error_stream, self.message_log, self.test_output_log 

157 ) 

158 test(result) 

159 result.printErrors() 

160 

161 return not result.have_fail