1# coding=utf-8 2""" 3Tools to implement runners (https://confluence.jetbrains.com/display/~link/PyCharm+test+runners+protocol) 4""" 5import os 6import re 7import sys 8 9import _jb_utils 10from teamcity import teamcity_presence_env_var, messages 11 12# Some runners need it to "detect" TC and start protocol 13if teamcity_presence_env_var not in os.environ: 14 os.environ[teamcity_presence_env_var] = "LOCAL" 15 16# Providing this env variable disables output buffering. 17# anything sent to stdout/stderr goes to IDE directly, not after test is over like it is done by default. 18# out and err are not in sync, so output may go to wrong test 19JB_DISABLE_BUFFERING = "JB_DISABLE_BUFFERING" in os.environ 20# getcwd resolves symlinks, but PWD is not supported by some shells 21PROJECT_DIR = os.getenv('PWD', os.getcwd()) 22 23 24def _parse_parametrized(part): 25 """ 26 27 Support nose generators / pytest parameters and other functions that provides names like foo(1,2) 28 Until https://github.com/JetBrains/teamcity-messages/issues/121, all such tests are provided 29 with parentheses. 30 31 Tests with docstring are reported in similar way but they have space before parenthesis and should be ignored 32 by this function 33 34 """ 35 match = re.match("^([^\\s)(]+)(\\(.+\\))$", part) 36 if not match: 37 return [part] 38 else: 39 return [match.group(1), match.group(2)] 40 41 42class _TreeManagerHolder(object): 43 def __init__(self): 44 self.parallel = "JB_USE_PARALLEL_TREE_MANAGER" in os.environ 45 self.offset = 0 46 self._manager_imp = None 47 48 @property 49 def manager(self): 50 if not self._manager_imp: 51 self._fill_manager() 52 return self._manager_imp 53 54 def _fill_manager(self): 55 if self.parallel: 56 from _jb_parallel_tree_manager import ParallelTreeManager 57 self._manager_imp = ParallelTreeManager(self.offset) 58 else: 59 from _jb_serial_tree_manager import SerialTreeManager 60 self._manager_imp = SerialTreeManager(self.offset) 61 62 63_TREE_MANAGER_HOLDER = _TreeManagerHolder() 64 65 66def set_parallel_mode(): 67 _TREE_MANAGER_HOLDER.parallel = True 68 69 70def is_parallel_mode(): 71 return _TREE_MANAGER_HOLDER.parallel 72 73 74# Monkeypatching TC 75_old_service_messages = messages.TeamcityServiceMessages 76 77PARSE_FUNC = None 78 79 80class NewTeamcityServiceMessages(_old_service_messages): 81 _latest_subtest_result = None 82 83 def message(self, messageName, **properties): 84 if messageName in {"enteredTheMatrix", "testCount"}: 85 if "_jb_do_not_call_enter_matrix" not in os.environ: 86 _old_service_messages.message(self, messageName, **properties) 87 return 88 89 try: 90 # Report directory so Java site knows which folder to resolve names against 91 92 # tests with docstrings are reported in format "test.name (some test here)". 93 # text should be part of name, but not location. 94 possible_location = str(properties["name"]) 95 loc = possible_location.find("(") 96 if loc > 0: 97 possible_location = possible_location[:loc].strip() 98 properties["locationHint"] = "python<{0}>://{1}".format(PROJECT_DIR, possible_location) 99 except KeyError: 100 # If message does not have name, then it is not test 101 # Simply pass it 102 _old_service_messages.message(self, messageName, **properties) 103 return 104 105 current, parent = _TREE_MANAGER_HOLDER.manager.get_node_ids(properties["name"]) 106 if not current and not parent: 107 return 108 # Shortcut for name 109 try: 110 properties["name"] = str(properties["name"]).split(".")[-1] 111 except IndexError: 112 pass 113 114 properties["nodeId"] = str(current) 115 properties["parentNodeId"] = str(parent) 116 117 _old_service_messages.message(self, messageName, **properties) 118 119 def _test_to_list(self, test_name): 120 """ 121 Splits test name to parts to use it as list. 122 It most cases dot is used, but runner may provide custom function 123 """ 124 parts = test_name.split(".") 125 result = [] 126 for part in parts: 127 result += _parse_parametrized(part) 128 return result 129 130 def _fix_setup_teardown_name(self, test_name): 131 """ 132 133 Hack to rename setup and teardown methods to much real python signatures 134 """ 135 try: 136 return {"test setup": "setUpClass", "test teardown": "tearDownClass"}[test_name] 137 except KeyError: 138 return test_name 139 140 # Blocks are used for 2 cases now: 141 # 1) Unittest subtests (only closed, opened by subTestBlockOpened) 142 # 2) setup/teardown (does not work, see https://github.com/JetBrains/teamcity-messages/issues/114) 143 # def blockOpened(self, name, flowId=None): 144 # self.testStarted(".".join(TREE_MANAGER.current_branch + [self._fix_setup_teardown_name(name)])) 145 146 def blockClosed(self, name, flowId=None): 147 148 # If _latest_subtest_result is not set or does not exist we closing setup method, not a subtest 149 try: 150 if not self._latest_subtest_result: 151 return 152 except AttributeError: 153 return 154 155 # closing subtest 156 test_name = ".".join(_TREE_MANAGER_HOLDER.manager.current_branch) 157 if self._latest_subtest_result in {"Failure", "Error"}: 158 self.testFailed(test_name) 159 if self._latest_subtest_result == "Skip": 160 self.testIgnored(test_name) 161 162 self.testFinished(test_name) 163 self._latest_subtest_result = None 164 165 def subTestBlockOpened(self, name, subTestResult, flowId=None): 166 self.testStarted(".".join(_TREE_MANAGER_HOLDER.manager.current_branch + [name])) 167 self._latest_subtest_result = subTestResult 168 169 def testStarted(self, testName, captureStandardOutput=None, flowId=None, is_suite=False, metainfo=None): 170 test_name_as_list = self._test_to_list(testName) 171 testName = ".".join(test_name_as_list) 172 173 def _write_start_message(): 174 # testName, captureStandardOutput, flowId 175 args = {"name": testName, "captureStandardOutput": captureStandardOutput, "metainfo": metainfo} 176 if is_suite: 177 self.message("testSuiteStarted", **args) 178 else: 179 self.message("testStarted", **args) 180 181 commands = _TREE_MANAGER_HOLDER.manager.level_opened(self._test_to_list(testName), _write_start_message) 182 if commands: 183 self.do_commands(commands) 184 self.testStarted(testName, captureStandardOutput, metainfo=metainfo) 185 186 def testFailed(self, testName, message='', details='', flowId=None, comparison_failure=None): 187 testName = ".".join(self._test_to_list(testName)) 188 _old_service_messages.testFailed(self, testName, message, details, comparison_failure=comparison_failure) 189 190 def testFinished(self, testName, testDuration=None, flowId=None, is_suite=False): 191 testName = ".".join(self._test_to_list(testName)) 192 193 def _write_finished_message(): 194 # testName, captureStandardOutput, flowId 195 current, parent = _TREE_MANAGER_HOLDER.manager.get_node_ids(testName) 196 if not current and not parent: 197 return 198 args = {"nodeId": current, "parentNodeId": parent, "name": testName} 199 200 # TODO: Doc copy/paste with parent, extract 201 if testDuration is not None: 202 duration_ms = testDuration.days * 86400000 + \ 203 testDuration.seconds * 1000 + \ 204 int(testDuration.microseconds / 1000) 205 args["duration"] = str(duration_ms) 206 207 if is_suite: 208 if is_parallel_mode(): 209 del args["duration"] 210 self.message("testSuiteFinished", **args) 211 else: 212 self.message("testFinished", **args) 213 214 commands = _TREE_MANAGER_HOLDER.manager.level_closed(self._test_to_list(testName), _write_finished_message) 215 if commands: 216 self.do_commands(commands) 217 self.testFinished(testName, testDuration) 218 219 def do_commands(self, commands): 220 """ 221 222 Executes commands, returned by level_closed and level_opened 223 """ 224 for command, test in commands: 225 test_name = ".".join(test) 226 # By executing commands we open or close suites(branches) since tests(leaves) are always reported by runner 227 if command == "open": 228 self.testStarted(test_name, is_suite=True) 229 else: 230 self.testFinished(test_name, is_suite=True) 231 232 233messages.TeamcityServiceMessages = NewTeamcityServiceMessages 234 235 236# Monkeypatched 237 238def jb_patch_separator(targets, fs_glue, python_glue, fs_to_python_glue): 239 """ 240 Converts python target if format "/path/foo.py::parts.to.python" provided by Java to 241 python specific format 242 243 :param targets: list of dot-separated targets 244 :param fs_glue: how to glue fs parts of target. I.e.: module "eggs" in "spam" package is "spam[fs_glue]eggs" 245 :param python_glue: how to glue python parts (glue between class and function etc) 246 :param fs_to_python_glue: between last fs-part and first python part 247 :return: list of targets with patched separators 248 """ 249 if not targets: 250 return [] 251 252 def _patch_target(target): 253 # /path/foo.py::parts.to.python 254 match = re.match("^(:?(.+)[.]py::)?(.+)$", target) 255 assert match, "unexpected string: {0}".format(target) 256 fs_part = match.group(2) 257 python_part = match.group(3).replace(".", python_glue) 258 if fs_part: 259 return fs_part.replace("/", fs_glue) + fs_to_python_glue + python_part 260 else: 261 return python_part 262 263 return map(_patch_target, targets) 264 265 266def jb_start_tests(): 267 """ 268 Parses arguments, starts protocol and fixes syspath and returns tuple of arguments 269 """ 270 path, targets, additional_args = parse_arguments() 271 start_protocol() 272 return path, targets, additional_args 273 274 275def start_protocol(): 276 properties = {"durationStrategy": "manual"} if is_parallel_mode() else dict() 277 NewTeamcityServiceMessages().message('enteredTheMatrix', **properties) 278 279 280def parse_arguments(): 281 """ 282 Parses arguments, fixes syspath and returns tuple of arguments 283 284 :return: (string with path or None, list of targets or None, list of additional arguments) 285 It may return list with only one element (name itself) if name is the same or split names to several parts 286 """ 287 # Handle additional args after -- 288 additional_args = [] 289 try: 290 index = sys.argv.index("--") 291 additional_args = sys.argv[index + 1:] 292 del sys.argv[index:] 293 except ValueError: 294 pass 295 utils = _jb_utils.VersionAgnosticUtils() 296 namespace = utils.get_options( 297 _jb_utils.OptionDescription('--path', 'Path to file or folder to run'), 298 _jb_utils.OptionDescription('--offset', 'Root node offset'), 299 _jb_utils.OptionDescription('--target', 'Python target to run', "append")) 300 del sys.argv[1:] # Remove all args 301 302 # PyCharm helpers dir is first dir in sys.path because helper is launched. 303 # But sys.path should be same as when launched with test runner directly 304 try: 305 if os.path.abspath(sys.path[0]) == os.path.abspath(os.environ["PYCHARM_HELPERS_DIR"]): 306 path = sys.path.pop(0) 307 if path not in sys.path: 308 sys.path.append(path) 309 except KeyError: 310 pass 311 _TREE_MANAGER_HOLDER.offset = int(namespace.offset if namespace.offset else 0) 312 return namespace.path, namespace.target, additional_args 313 314 315def jb_doc_args(framework_name, args): 316 """ 317 Runner encouraged to report its arguments to user with aid of this function 318 319 """ 320 print("Launching {0} with arguments {1} in {2}\n".format(framework_name, " ".join(args), PROJECT_DIR)) 321