| import sys, os |
| import imp |
| |
| helpers_dir = os.getenv("PYCHARM_HELPERS_DIR", sys.path[0]) |
| if sys.path[0] != helpers_dir: |
| sys.path.insert(0, helpers_dir) |
| |
| from tcunittest import TeamcityTestResult |
| |
| from pycharm_run_utils import import_system_module |
| from pycharm_run_utils import adjust_sys_path |
| from pycharm_run_utils import debug, getModuleName |
| |
| adjust_sys_path() |
| |
| re = import_system_module("re") |
| inspect = import_system_module("inspect") |
| |
| try: |
| from attest.reporters import AbstractReporter |
| from attest.collectors import Tests |
| from attest import TestBase |
| except: |
| raise NameError("Please, install attests") |
| |
| class TeamCityReporter(AbstractReporter, TeamcityTestResult): |
| """Teamcity reporter for attests.""" |
| |
| def __init__(self, prefix): |
| TeamcityTestResult.__init__(self) |
| self.prefix = prefix |
| |
| def begin(self, tests): |
| """initialize suite stack and count tests""" |
| self.total = len(tests) |
| self.suite_stack = [] |
| self.messages.testCount(self.total) |
| |
| def success(self, result): |
| """called when test finished successfully""" |
| suite = self.get_suite_name(result.test) |
| self.start_suite(suite) |
| name = self.get_test_name(result) |
| self.start_test(result, name) |
| self.messages.testFinished(name) |
| |
| def failure(self, result): |
| """called when test failed""" |
| suite = self.get_suite_name(result.test) |
| self.start_suite(suite) |
| name = self.get_test_name(result) |
| self.start_test(result, name) |
| exctype, value, tb = result.exc_info |
| error_value = self.find_error_value(tb) |
| if (error_value.startswith("'") or error_value.startswith('"')) and\ |
| (error_value.endswith("'") or error_value.endswith('"')): |
| first = self._unescape(self.find_first(error_value)) |
| second = self._unescape(self.find_second(error_value)) |
| else: |
| first = second = "" |
| |
| err = self.formatErr(result.exc_info) |
| if isinstance(result.error, AssertionError): |
| self.messages.testFailed(name, message='Failure', |
| details=err, |
| expected=first, actual=second) |
| else: |
| self.messages.testError(name, message='Error', |
| details=err) |
| |
| def finished(self): |
| """called when all tests finished""" |
| self.end_last_suite() |
| for suite in self.suite_stack[::-1]: |
| self.messages.testSuiteFinished(suite) |
| |
| def get_test_name(self, result): |
| name = result.test_name |
| ind = name.find("%") #remove unique module prefix |
| if ind != -1: |
| name = name[:ind]+name[name.find(".", ind):] |
| return name |
| |
| def end_last_suite(self): |
| if self.current_suite: |
| self.messages.testSuiteFinished(self.current_suite) |
| self.current_suite = None |
| |
| def get_suite_name(self, test): |
| module = inspect.getmodule(test) |
| klass = getattr(test, "im_class", None) |
| file = module.__file__ |
| if file.endswith("pyc"): |
| file = file[:-1] |
| |
| suite = module.__name__ |
| if self.prefix: |
| tmp = file[:-3] |
| ind = tmp.split(self.prefix)[1] |
| suite = ind.replace("/", ".") |
| if klass: |
| suite += "." + klass.__name__ |
| lineno = inspect.getsourcelines(klass) |
| else: |
| lineno = ("", 1) |
| |
| return (suite, file+":"+str(lineno[1])) |
| |
| def start_suite(self, suite_info): |
| """finish previous suite and put current suite |
| to stack""" |
| suite, file = suite_info |
| if suite != self.current_suite: |
| if self.current_suite: |
| if suite.startswith(self.current_suite+"."): |
| self.suite_stack.append(self.current_suite) |
| else: |
| self.messages.testSuiteFinished(self.current_suite) |
| for s in self.suite_stack: |
| if not suite.startswith(s+"."): |
| self.current_suite = s |
| self.messages.testSuiteFinished(self.current_suite) |
| else: |
| break |
| self.current_suite = suite |
| self.messages.testSuiteStarted(self.current_suite, location="file://" + file) |
| |
| def start_test(self, result, name): |
| """trying to find test location """ |
| real_func = result.test.func_closure[0].cell_contents |
| lineno = inspect.getsourcelines(real_func) |
| file = inspect.getsourcefile(real_func) |
| self.messages.testStarted(name, "file://"+file+":"+str(lineno[1])) |
| |
| def get_subclasses(module, base_class=TestBase): |
| test_classes = [] |
| for name in dir(module): |
| obj = getattr(module, name) |
| try: |
| if issubclass(obj, base_class): |
| test_classes.append(obj) |
| except TypeError: # If 'obj' is not a class |
| pass |
| return test_classes |
| |
| def get_module(file_name): |
| baseName = os.path.splitext(os.path.basename(file_name))[0] |
| return imp.load_source(baseName, file_name) |
| |
| modules = {} |
| def getModuleName(prefix, cnt): |
| """ adds unique number to prevent name collisions""" |
| return prefix + "%" + str(cnt) |
| |
| def loadSource(fileName): |
| baseName = os.path.basename(fileName) |
| moduleName = os.path.splitext(baseName)[0] |
| |
| if moduleName in modules: |
| cnt = 2 |
| prefix = moduleName |
| while getModuleName(prefix, cnt) in modules: |
| cnt += 1 |
| moduleName = getModuleName(prefix, cnt) |
| debug("/ Loading " + fileName + " as " + moduleName) |
| module = imp.load_source(moduleName, fileName) |
| modules[moduleName] = module |
| return module |
| |
| |
| def register_tests_from_module(module, tests): |
| """add tests from module to main test suite""" |
| tests_to_register = [] |
| |
| for i in dir(module): |
| obj = getattr(module, i) |
| if isinstance(obj, Tests): |
| tests_to_register.append(i) |
| |
| for i in tests_to_register: |
| baseName = module.__name__+"."+i |
| tests.register(baseName) |
| test_subclasses = get_subclasses(module) |
| if test_subclasses: |
| for subclass in test_subclasses: |
| tests.register(subclass()) |
| |
| |
| def register_tests_from_folder(tests, folder, pattern=None): |
| """add tests from folder to main test suite""" |
| listing = os.listdir(folder) |
| files = listing |
| if pattern: #get files matched given pattern |
| prog_list = [re.compile(pat.strip()) for pat in pattern.split(',')] |
| files = [] |
| for fileName in listing: |
| if os.path.isdir(folder+fileName): |
| files.append(fileName) |
| for prog in prog_list: |
| if prog.match(fileName): |
| files.append(fileName) |
| |
| if not folder.endswith("/"): |
| folder += "/" |
| for fileName in files: |
| if os.path.isdir(folder+fileName): |
| register_tests_from_folder(tests, folder+fileName, pattern) |
| if not fileName.endswith("py"): |
| continue |
| |
| module = loadSource(folder+fileName) |
| register_tests_from_module(module, tests) |
| |
| def process_args(): |
| tests = Tests() |
| prefix = "" |
| if not sys.argv: |
| return |
| |
| arg = sys.argv[1].strip() |
| if not len(arg): |
| return |
| |
| argument_list = arg.split("::") |
| if len(argument_list) == 1: |
| # From module or folder |
| a_splitted = argument_list[0].split(";") |
| if len(a_splitted) != 1: |
| # means we have pattern to match against |
| if a_splitted[0].endswith("/"): |
| debug("/ from folder " + a_splitted[0] + ". Use pattern: " + a_splitted[1]) |
| prefix = a_splitted[0] |
| register_tests_from_folder(tests, a_splitted[0], a_splitted[1]) |
| else: |
| if argument_list[0].endswith("/"): |
| debug("/ from folder " + argument_list[0]) |
| prefix = a_splitted[0] |
| register_tests_from_folder(tests, argument_list[0]) |
| else: |
| debug("/ from file " + argument_list[0]) |
| module = get_module(argument_list[0]) |
| register_tests_from_module(module, tests) |
| |
| elif len(argument_list) == 2: |
| # From testcase |
| debug("/ from test class " + argument_list[1] + " in " + argument_list[0]) |
| module = get_module(argument_list[0]) |
| klass = getattr(module, argument_list[1]) |
| tests.register(klass()) |
| else: |
| # From method in class or from function |
| module = get_module(argument_list[0]) |
| if argument_list[1] == "": |
| debug("/ from function " + argument_list[2] + " in " + argument_list[0]) |
| # test function, not method |
| test = getattr(module, argument_list[2]) |
| else: |
| debug("/ from method " + argument_list[2] + " in class " + argument_list[1] + " in " + argument_list[0]) |
| klass = getattr(module, argument_list[1]) |
| test = getattr(klass(), argument_list[2]) |
| tests.register([test]) |
| |
| tests.run(reporter=TeamCityReporter(prefix)) |
| |
| if __name__ == "__main__": |
| process_args() |
