Coverage for Applications / PyCharm.app / Contents / plugins / python-ce / helpers / pycharm / teamcity / messages.py: 59%

1# coding=utf-8 

2import sys 

3import time 

4 

5from teamcity.output import TeamCityMessagesPrinter 

6 

7if sys.version_info < (3, ): 

8 # Python 2 

9 # flake8: noqa 

10 text_type = unicode 

11else: 

12 # Python 3 

13 text_type = str 

14 

15# Capture some time functions to allow monkeypatching them in tests 

16_time = time.time 

17_localtime = time.localtime 

18_strftime = time.strftime 

19 

20_quote = {"'": "|'", "|": "||", "\n": "|n", "\r": "|r", '[': '|[', ']': '|]'} 

21 

22 

23def escape_value(value): 

24 return "".join(_quote.get(x, x) for x in value) 

25 

26 

27class TeamcityServiceMessages(object): 

28 def __init__(self, output=None, now=_time, encoding='auto', output_handler=None): 

29 if output_handler is None: 

30 output_handler = TeamCityMessagesPrinter(output) 

31 self.output_handler = output_handler 

32 self.now = now 

33 

34 if encoding and encoding != 'auto': 

35 self.encoding = encoding 

36 elif getattr(output, 'encoding', None) or encoding == 'auto': 

37 # Default encoding to 'utf-8' because it sucks if we fail with a 

38 # `UnicodeEncodeError` simply because LANG didn't get propagated to 

39 # a subprocess or something and sys.stdout.encoding is None 

40 self.encoding = getattr(output, 'encoding', None) or 'utf-8' 

41 else: 

42 self.encoding = None 

43 

44 def encode(self, value): 

45 if self.encoding and isinstance(value, text_type): 

46 value = value.encode(self.encoding, errors='backslashreplace') 

47 return value 

48 

49 def decode(self, value): 

50 if self.encoding and not isinstance(value, text_type): 

51 value = value.decode(self.encoding) 

52 return value 

53 

54 if sys.version_info < (3, ): 

55 def escapeValue(self, value): 

56 return escape_value(self.encode(value)) 

57 else: 

58 def escapeValue(self, value): 

59 return escape_value(self.decode(value)) 

60 

61 def message(self, messageName, **properties): 

62 current_time = self.now() 

63 (current_time_int, current_time_fraction) = divmod(current_time, 1) 

64 current_time_struct = _localtime(current_time_int) 

65 

66 timestamp = _strftime("%Y-%m-%dT%H:%M:%S.", current_time_struct) + "%03d" % (int(current_time_fraction * 1000)) 

67 message = ("##teamcity[%s timestamp='%s'" % (messageName, timestamp)) 

68 

69 for k in sorted(properties.keys()): 

70 value = properties[k] 

71 if value is None: 

72 continue 

73 

74 message += (" %s='%s'" % (k, self.escapeValue(value))) 

75 

76 message += ("]\n") 

77 

78 self.output_handler.send_message(self.encode(message)) 

79 

80 def _single_value_message(self, messageName, value): 

81 message = ("##teamcity[%s '%s']\n" % (messageName, self.escapeValue(value))) 

82 

83 self.output_handler.send_message(self.encode(message)) 

84 

85 def blockOpened(self, name, flowId=None): 

86 self.message('blockOpened', name=name, flowId=flowId) 

87 

88 def blockClosed(self, name, flowId=None): 

89 self.message('blockClosed', name=name, flowId=flowId) 

90 

91 # Special PyCharm-specific extension to track subtests, additional property is ignored by TeamCity 

92 def subTestBlockOpened(self, name, subTestResult, flowId=None): 

93 self.message('blockOpened', name=name, subTestResult=subTestResult, flowId=flowId) 

94 

95 def block(self, name, flowId=None): 

96 import teamcity.context_managers as cm 

97 return cm.block(self, name=name, flowId=flowId) 

98 

99 def compilationStarted(self, compiler): 

100 self.message('compilationStarted', compiler=compiler) 

101 

102 def compilationFinished(self, compiler): 

103 self.message('compilationFinished', compiler=compiler) 

104 

105 def compilation(self, compiler): 

106 import teamcity.context_managers as cm 

107 return cm.compilation(self, compiler=compiler) 

108 

109 def testSuiteStarted(self, suiteName, flowId=None): 

110 self.message('testSuiteStarted', name=suiteName, flowId=flowId) 

111 

112 def testSuiteFinished(self, suiteName, flowId=None): 

113 self.message('testSuiteFinished', name=suiteName, flowId=flowId) 

114 

115 def testSuite(self, name): 

116 import teamcity.context_managers as cm 

117 return cm.testSuite(self, name=name) 

118 

119 def testStarted(self, testName, captureStandardOutput=None, flowId=None, metainfo=None, **properties): 

120 """ 

121 

122 :param metainfo: Used to pass any payload from test runner to Intellij. See IDEA-176950 

123 """ 

124 self.message('testStarted', name=testName, captureStandardOutput=captureStandardOutput, flowId=flowId, metainfo=metainfo, **properties) 

125 

126 def testFinished(self, testName, testDuration=None, flowId=None): 

127 if testDuration is not None: 

128 duration_ms = testDuration.days * 86400000 + \ 

129 testDuration.seconds * 1000 + \ 

130 int(testDuration.microseconds / 1000) 

131 self.message('testFinished', name=testName, duration=str(duration_ms), flowId=flowId) 

132 else: 

133 self.message('testFinished', name=testName, flowId=flowId) 

134 

135 def test(self, testName, captureStandardOutput=None, testDuration=None, flowId=None): 

136 import teamcity.context_managers as cm 

137 return cm.test(self, testName=testName, captureStandardOutput=captureStandardOutput, testDuration=testDuration, flowId=flowId) 

138 

139 # Unsupported in TeamCity, used in IntellIJ-based IDEs to predict number of tests to be run in the test session 

140 def testCount(self, count, flowId=None): 

141 self.message('testCount', count=str(count), flowId=flowId) 

142 

143 def testIgnored(self, testName, message='', flowId=None): 

144 self.message('testIgnored', name=testName, message=message, flowId=flowId) 

145 

146 def testStopped(self, testName, message='', flowId=None): 

147 self.message('testIgnored', name=testName, message=message, flowId=flowId, stopped="true") 

148 

149 def testFailed(self, testName, message='', details='', flowId=None, comparison_failure=None): 

150 if not comparison_failure: 

151 self.message('testFailed', name=testName, message=message, details=details, flowId=flowId) 

152 else: 

153 diff_message = u"\n{0} != {1}\n".format(comparison_failure.actual, comparison_failure.expected) 

154 self.message('testFailed', 

155 name=testName, 

156 message=text_type(message) + text_type(diff_message), 

157 details=details, 

158 flowId=flowId, 

159 type="comparisonFailure", 

160 actual=comparison_failure.actual, 

161 expected=comparison_failure.expected) 

162 

163 def testStdOut(self, testName, out, flowId=None): 

164 self.message('testStdOut', name=testName, out=out, flowId=flowId) 

165 

166 def testStdErr(self, testName, out, flowId=None): 

167 self.message('testStdErr', name=testName, out=out, flowId=flowId) 

168 

169 def publishArtifacts(self, path, flowId=None): 

170 self._single_value_message('publishArtifacts', path) 

171 

172 def progressMessage(self, message): 

173 self._single_value_message('progressMessage', message) 

174 

175 def progressStart(self, message): 

176 self._single_value_message('progressStart', message) 

177 

178 def progressFinish(self, message): 

179 self._single_value_message('progressFinish', message) 

180 

181 def progress(self, message): 

182 import teamcity.context_managers as cm 

183 return cm.progress(self, message=message) 

184 

185 def buildProblem(self, description, identity): 

186 self.message('buildProblem', description=description, identity=identity) 

187 

188 def buildStatus(self, status, text): 

189 self.message('buildStatus', status=status, text=text) 

190 

191 def setParameter(self, name, value): 

192 self.message('setParameter', name=name, value=value) 

193 

194 def buildStatisticLinesCovered(self, linesCovered): 

195 self.message('buildStatisticValue', key='CodeCoverageAbsLCovered', value=str(linesCovered)) 

196 

197 def buildStatisticTotalLines(self, totalLines): 

198 self.message('buildStatisticValue', key='CodeCoverageAbsLTotal', value=str(totalLines)) 

199 

200 def buildStatisticLinesUncovered(self, linesUncovered): 

201 self.message('buildStatisticValue', key='CodeCoverageAbsLUncovered', value=str(linesUncovered)) 

202 

203 def enableServiceMessages(self, flowId=None): 

204 self.message('enableServiceMessages', flowId=flowId) 

205 

206 def disableServiceMessages(self, flowId=None): 

207 self.message('disableServiceMessages', flowId=flowId) 

208 

209 def serviceMessagesDisabled(self, flowId=None): 

210 import teamcity.context_managers as cm 

211 return cm.serviceMessagesDisabled(self, flowId=flowId) 

212 

213 def serviceMessagesEnabled(self, flowId=None): 

214 import teamcity.context_managers as cm 

215 return cm.serviceMessagesEnabled(self, flowId=flowId) 

216 

217 def importData(self, typeID, pathToXMLFile): 

218 self.message('importData', type=typeID, path=pathToXMLFile) 

219 

220 def customMessage(self, text, status, errorDetails='', flowId=None): 

221 self.message('message', text=text, status=status, errorDetails=errorDetails, flowId=flowId)