# coding: utf-8 import collections import functools import math import os import re import sys import yatest_lib.tools class Subtest(object): def __init__(self, name, test_name, status, comment, elapsed, result=None, test_type=None, logs=None, cwd=None, metrics=None): self._name = name self._test_name = test_name self.status = status self.elapsed = elapsed self.comment = comment self.result = result self.test_type = test_type self.logs = logs or {} self.cwd = cwd self.metrics = metrics def __eq__(self, other): if not isinstance(other, Subtest): return False return self.name == other.name and self.test_name == other.test_name def __str__(self): return yatest_lib.tools.to_utf8(unicode(self)) def __unicode__(self): return u"{}::{}".format(self.test_name, self.test_name) @property def name(self): return yatest_lib.tools.to_utf8(self._name) @property def test_name(self): return yatest_lib.tools.to_utf8(self._test_name) def __repr__(self): return "Subtest [{}::{} - {}[{}]: {}]".format(self.name, self.test_name, self.status, self.elapsed, self.comment) def __hash__(self): return hash(str(self)) class SubtestInfo(object): skipped_prefix = '[SKIPPED] ' @classmethod def from_str(cls, s): if s.startswith(SubtestInfo.skipped_prefix): s = s[len(SubtestInfo.skipped_prefix):] skipped = True else: skipped = False return SubtestInfo(*s.rsplit(TEST_SUBTEST_SEPARATOR, 1), skipped=skipped) def __init__(self, test, subtest="", skipped=False, **kwargs): self.test = test self.subtest = subtest self.skipped = skipped for key, value in kwargs.iteritems(): setattr(self, key, value) def __str__(self): s = '' if self.skipped: s += SubtestInfo.skipped_prefix return s + TEST_SUBTEST_SEPARATOR.join([self.test, self.subtest]) def __repr__(self): return str(self) class Status(object): GOOD, XFAIL, FAIL, XPASS, MISSING, CRASHED, TIMEOUT = range(7) SKIPPED = -100 NOT_LAUNCHED = -200 CANON_DIFF = -300 FLAKY = -1 BY_NAME = {'good': GOOD, 'fail': FAIL, 'xfail': XFAIL, 'xpass': XPASS, 'missing': MISSING, 'crashed': CRASHED, 'skipped': SKIPPED, 'flaky': FLAKY, 'not_launched': NOT_LAUNCHED, 'timeout': TIMEOUT, 'diff': CANON_DIFF} TO_STR = {GOOD: 'good', FAIL: 'fail', XFAIL: 'xfail', XPASS: 'xpass', MISSING: 'missing', CRASHED: 'crashed', SKIPPED: 'skipped', FLAKY: 'flaky', NOT_LAUNCHED: 'not_launched', TIMEOUT: 'timeout', CANON_DIFF: 'diff'} class Test(object): def __init__(self, name, path, status=None, comment=None, subtests=None): self.name = name self.path = path self.status = status self.comment = comment self.subtests = subtests or [] def __eq__(self, other): if not isinstance(other, Test): return False return self.name == other.name and self.path == other.path def __str__(self): return "Test [{} {}] - {} - {}".format(self.name, self.path, self.status, self.comment) def __repr__(self): return str(self) def add_subtest(self, subtest): self.subtests.append(subtest) def setup_status(self, status, comment): self.status = Status.BY_NAME[status or 'good'] if len(self.subtests) != 0: self.status = max(self.status, max(s.status for s in self.subtests)) self.comment = comment def subtests_by_status(self, status): return [x.status for x in self.subtests].count(status) class NoMd5FileException(Exception): pass TEST_SUBTEST_SEPARATOR = '::' # TODO: extract color theme logic from ya COLOR_THEME = { 'test_name': 'light-blue', 'test_project_path': 'dark-blue', 'test_dir_desc': 'dark-magenta', 'test_binary_path': 'light-gray', } # XXX: remove me class YaCtx(object): pass ya_ctx = YaCtx() TRACE_FILE_NAME = "ytest.report.trace" def lazy(func): mem = {} @functools.wraps(func) def wrapper(): if "results" not in mem: mem["results"] = func() return mem["results"] return wrapper @lazy def _get_mtab(): if os.path.exists("/etc/mtab"): with open("/etc/mtab") as afile: data = afile.read() return [line.split(" ") for line in data.split("\n") if line] return [] def get_max_filename_length(dirname): """ Return maximum filename length for the filesystem :return: """ if sys.platform.startswith("linux"): # Linux user's may work on mounted ecryptfs filesystem # which has filename length limitations for entry in _get_mtab(): mounted_dir, filesystem = entry[1], entry[2] # http://unix.stackexchange.com/questions/32795/what-is-the-maximum-allowed-filename-and-folder-size-with-ecryptfs if filesystem == "ecryptfs" and dirname and dirname.startswith(mounted_dir): return 140 # default maximum filename length for most filesystems return 255 def get_unique_file_path(dir_path, filename, cache=collections.defaultdict(set)): """ Get unique filename in dir with proper filename length, using given filename/dir. File/dir won't be created (thread nonsafe) :param dir_path: path to dir :param filename: original filename :return: unique filename """ max_suffix = 10000 # + 1 symbol for dot before suffix tail_length = int(round(math.log(max_suffix, 10))) + 1 # truncate filename length in accordance with filesystem limitations filename, extension = os.path.splitext(filename) # XXX if sys.platform.startswith("win"): # Trying to fit into MAX_PATH if it's possible. # Remove after DEVTOOLS-1646 max_path = 260 filename_len = len(dir_path) + len(extension) + tail_length + len(os.sep) if filename_len < max_path: filename = yatest_lib.tools.trim_string(filename, max_path - filename_len) filename = yatest_lib.tools.trim_string(filename, get_max_filename_length(dir_path) - tail_length - len(extension)) + extension candidate = os.path.join(dir_path, filename) key = dir_path + filename counter = sorted(cache.get(key, {0, }))[-1] while os.path.exists(candidate): cache[key].add(counter) counter += 1 assert counter < max_suffix candidate = os.path.join(dir_path, filename + ".{}".format(counter)) return candidate def escape_for_fnmatch(s): return s.replace("[", "[").replace("]", "]") def get_python_cmd(opts=None, use_huge=True, suite=None): if opts and getattr(opts, 'flags', {}).get("USE_ARCADIA_PYTHON") == "no": return ["python"] if suite and not suite._use_arcadia_python: return ["python"] if use_huge: return ["$(PYTHON)/python"] ymake_path = opts.ymake_bin if opts and getattr(opts, 'ymake_bin', None) else "$(YMAKE)/ymake" return [ymake_path, "--python"] def normalize_name(name): replacements = [ ("\\", "\\\\"), ("\n", "\\n"), ("\t", "\\t"), ("\r", "\\r"), ] for l, r in replacements: name = name.replace(l, r) return name def normalize_filename(filename): """ Replace invalid for file names characters with string equivalents :param some_string: string to be converted to a valid file name :return: valid file name """ not_allowed_pattern = r"[\[\]\/:*?\"\'<>|+\0\\\s\x0b\x0c]" filename = re.sub(not_allowed_pattern, ".", filename) return re.sub(r"\.{2,}", ".", filename) def get_test_log_file_path(output_dir, class_name, test_name, extension="log"): """ get test log file path, platform dependant :param output_dir: dir where log file should be placed :param class_name: test class name :param test_name: test name :return: test log file name """ if os.name == "nt": # don't add class name to the log's filename # to reduce it's length on windows filename = test_name else: filename = "{}.{}".format(class_name, test_name) if not filename: filename = "test" filename += "." + extension filename = normalize_filename(filename) return get_unique_file_path(output_dir, filename) def split_node_id(nodeid, test_suffix=None): path, possible_open_bracket, params = nodeid.partition('[') separator = "::" if separator in path: path, test_name = path.split(separator, 1) else: test_name = os.path.basename(path) if test_suffix: test_name += "::" + test_suffix class_name = os.path.basename(path.strip()) if separator in test_name: klass_name, test_name = test_name.split(separator, 1) if not test_suffix: # test suffix is used for flakes and pep8, no need to add class_name as it's === class_name class_name += separator + klass_name if separator in test_name: test_name = test_name.split(separator)[-1] test_name += possible_open_bracket + params return yatest_lib.tools.to_utf8(class_name), yatest_lib.tools.to_utf8(test_name)