common.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270
  1. #
  2. # Metrix++, Copyright 2009-2019, Metrix++ Project
  3. # Link: https://github.com/metrixplusplus/metrixplusplus
  4. #
  5. # This file is a part of Metrix++ Tool.
  6. #
  7. import inspect
  8. import os.path
  9. import subprocess
  10. import logging
  11. import difflib
  12. import unittest
  13. import shutil
  14. import ast
  15. class ToolRunner(object):
  16. def __init__(self,
  17. tool_name,
  18. opts_list = [],
  19. dirs_list = None,
  20. cwd='sources',
  21. prefix = "default",
  22. exit_code = 0,
  23. save_prev = False,
  24. use_prev = False,
  25. check_stderr = None,
  26. remove_exiting_dbfile = None,
  27. remove_exiting_dbfile_prev = False):
  28. self.message = ""
  29. # identify gold_file_location
  30. curframe = inspect.currentframe()
  31. calframe = inspect.getouterframes(curframe, 2)
  32. test_name = calframe[1][3]
  33. suite_name = os.path.splitext(os.path.basename(calframe[1][1]))[0]
  34. group_name = os.path.basename(os.path.dirname(calframe[1][1]))
  35. self.suite_location = os.path.join('tests', group_name, suite_name)
  36. self.test_location = os.path.join(self.suite_location, test_name + "_" + tool_name + "_" + str(prefix))
  37. db_file = os.path.join(os.environ['METRIXPLUSPLUS_INSTALL_DIR'], self.suite_location, test_name + ".db")
  38. self.dbfile = db_file
  39. if (remove_exiting_dbfile == True or (remove_exiting_dbfile == None and tool_name == 'collect')) and os.path.exists(db_file):
  40. os.unlink(db_file)
  41. db_file_prev = os.path.join(os.environ['METRIXPLUSPLUS_INSTALL_DIR'], self.suite_location, test_name + ".prev.db")
  42. self.dbfile_prev = db_file_prev
  43. if (remove_exiting_dbfile_prev == True or (remove_exiting_dbfile_prev == None and tool_name == 'collect')) and os.path.exists(db_file_prev):
  44. os.unlink(db_file_prev)
  45. self.cwd = cwd
  46. db_opts = ['--db-file=' + db_file]
  47. if use_prev == True:
  48. db_opts.append('--db-file-prev=' + db_file_prev)
  49. self.dbopts = db_opts
  50. self.dirs_list = []
  51. if dirs_list != None:
  52. for each in dirs_list:
  53. self.dirs_list.append(each)
  54. self.call_args = ['python', os.path.join(os.environ['METRIXPLUSPLUS_INSTALL_DIR'], "metrix++.py"), tool_name] \
  55. + db_opts + opts_list + ['--'] + self.dirs_list
  56. self.cmd = " ".join(self.call_args)
  57. self.exit_code_expected = exit_code
  58. self.stderr_lines = check_stderr
  59. self.save_prev = save_prev
  60. def run(self):
  61. logging.debug(self.get_description())
  62. child = subprocess.Popen(self.call_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
  63. cwd=os.path.join(self.suite_location, self.cwd))
  64. (child_stdout, child_stderr) = child.communicate()
  65. self.exit_code = child.returncode
  66. gold_file_stdout = self.test_location + "_stdout.gold.txt"
  67. real_file_stdout = self.test_location + "_stdout.real.txt"
  68. diff_file_stdout = self.test_location + "_stdout.diff.html"
  69. gold_file_stderr = self.test_location + "_stderr.gold.txt"
  70. real_file_stderr = self.test_location + "_stderr.real.txt"
  71. diff_file_stderr = self.test_location + "_stderr.diff.html"
  72. # Regenerate gold files if it was requested
  73. if os.environ['METRIXPLUSPLUS_TEST_GENERATE_GOLDS'] == "True":
  74. f = open(gold_file_stdout, 'wb');
  75. f.write(child_stdout);
  76. f.close()
  77. if self.stderr_lines != None:
  78. f = open(gold_file_stderr, 'wb');
  79. f.write(child_stderr);
  80. f.close()
  81. # Match with gold
  82. self.is_stdout_matched = self.inetrnal_compare_with_gold(child_stdout, gold_file_stdout, real_file_stdout, diff_file_stdout)
  83. if self.stderr_lines != None:
  84. self.is_stderr_matched = self.inetrnal_compare_with_gold(child_stderr, gold_file_stderr, real_file_stderr, diff_file_stderr, self.stderr_lines)
  85. else:
  86. self.is_stderr_matched = None
  87. if self.is_stdout_matched == False:
  88. f = open(real_file_stderr, 'wb');
  89. f.write(child_stderr);
  90. f.close()
  91. else:
  92. if os.path.exists(real_file_stderr):
  93. os.unlink(real_file_stderr)
  94. if self.save_prev == True:
  95. shutil.copy2(self.dbfile, self.dbfile_prev)
  96. return self
  97. def inetrnal_compare_with_gold(self, text, gold_file, real_file, diff_file, lines = None):
  98. if os.path.exists(gold_file) == False:
  99. self.message += "\nGold file does not exist: " + gold_file
  100. return False
  101. f = open(gold_file, 'r');
  102. gold_text = f.read();
  103. f.close()
  104. # don't compare dictionaries as string - they are not in order... (test case failes sometimes)
  105. try:
  106. textDict = ast.literal_eval(str(text.decode('ascii')))
  107. goldDict = ast.literal_eval(str(gold_text))
  108. except:
  109. textDict = text
  110. goldDict = gold_text
  111. if isinstance(textDict, dict) and isinstance(goldDict, dict):
  112. result = (textDict == goldDict)
  113. else:
  114. gold_to_compare = gold_text
  115. text_to_compare = str(text.decode('ascii'))
  116. if lines != None:
  117. gold_to_compare = ""
  118. text_to_compare = ""
  119. gold_lines = gold_text.splitlines(True)
  120. text_lines = str(text.decode('ascii')).splitlines(True)
  121. for each in lines:
  122. gold_to_compare += "".join(gold_lines[each[0] : each[1]])
  123. text_to_compare += "".join(text_lines[each[0] : each[1]])
  124. gold_to_compare = gold_to_compare.replace('\n', ' ').replace('\r', '').replace(" ", "")
  125. text_to_compare = text_to_compare.replace('\n', ' ').replace('\r', '').replace(" ", "")
  126. result = (gold_to_compare == text_to_compare)
  127. if result == False:
  128. f = open(real_file, 'wb')
  129. f.write(text)
  130. f.close()
  131. diff_text = difflib.HtmlDiff().make_file(gold_text.splitlines(), text.decode('ascii').splitlines(), "Gold file", "Real output")
  132. f = open(diff_file, 'w')
  133. f.write(diff_text)
  134. f.close()
  135. else:
  136. if os.path.exists(real_file):
  137. os.unlink(real_file)
  138. if os.path.exists(diff_file):
  139. os.unlink(diff_file)
  140. return result
  141. def check_exit_code(self):
  142. return self.exit_code == self.exit_code_expected
  143. def check_stdout(self):
  144. return self.is_stdout_matched
  145. def check_stderr(self):
  146. if self.is_stderr_matched == None:
  147. return True
  148. return self.is_stderr_matched
  149. def check_all(self):
  150. result = self.check_exit_code() and self.check_stdout() and self.check_stderr()
  151. if result == False:
  152. self.message += "\nCheck for exit code: " + str(self.check_exit_code()) \
  153. + ", gold: " + str(self.exit_code_expected) + ", real: " + str(self.exit_code) + \
  154. "\nCheck for stdout: " + str(self.check_stdout()) + "\nCheck for stderr: " + str(self.check_stderr())
  155. return result
  156. def get_message(self):
  157. return self.message
  158. def get_cmd(self):
  159. return self.cmd
  160. def get_description(self):
  161. return self.get_message() + "\nProcess: " + self.get_cmd() + "\nCWD: " + os.path.join(self.suite_location, self.cwd)
  162. def get_dbfile(self):
  163. return self.dbfile
  164. def get_dbfile_prev(self):
  165. return self.dbfile_prev
  166. class TestCase(unittest.TestCase):
  167. def __init__(self, methodName='runTest'):
  168. unittest.TestCase.__init__(self, methodName=methodName)
  169. if 'METRIXPLUSPLUS_LOG_LEVEL' not in list(os.environ.keys()):
  170. # launch of individual unit test
  171. os.environ['METRIXPLUSPLUS_LOG_LEVEL'] = 'ERROR'
  172. os.environ['METRIXPLUSPLUS_INSTALL_DIR'] = os.path.dirname(os.path.dirname(__file__))
  173. os.environ['METRIXPLUSPLUS_TEST_MODE'] = str("True")
  174. if 'METRIXPLUSPLUS_TEST_GENERATE_GOLDS' not in list(os.environ.keys()):
  175. os.environ['METRIXPLUSPLUS_TEST_GENERATE_GOLDS'] = str("False")
  176. os.chdir(os.environ['METRIXPLUSPLUS_INSTALL_DIR'])
  177. def get_content_paths(self, cwd='sources'):
  178. curframe = inspect.currentframe()
  179. calframe = inspect.getouterframes(curframe, 2)
  180. test_name = calframe[1][3]
  181. suite_name = os.path.splitext(os.path.basename(calframe[1][1]))[0]
  182. group_name = os.path.basename(os.path.dirname(calframe[1][1]))
  183. class ContentPaths(object):
  184. def __init__(self, cwd, dbfile, dbfile_prev):
  185. self.cwd = cwd
  186. self.dbfile = dbfile
  187. self.dbfile_prev = dbfile_prev
  188. return ContentPaths(os.path.join('tests', group_name, suite_name, cwd),
  189. os.path.join(os.environ['METRIXPLUSPLUS_INSTALL_DIR'], 'tests', group_name, suite_name, test_name + ".db"),
  190. os.path.join(os.environ['METRIXPLUSPLUS_INSTALL_DIR'], 'tests', group_name, suite_name, test_name + ".prev.db"))
  191. def setUp(self):
  192. unittest.TestCase.setUp(self)
  193. logging.basicConfig(format="[TEST-LOG]: %(levelname)s:\t%(message)s", level=logging.WARN)
  194. log_level = os.environ['METRIXPLUSPLUS_LOG_LEVEL']
  195. if log_level == 'ERROR':
  196. log_level = logging.ERROR
  197. elif log_level == 'WARNING':
  198. log_level = logging.WARNING
  199. elif log_level == 'INFO':
  200. log_level = logging.INFO
  201. elif log_level == 'DEBUG':
  202. log_level = logging.DEBUG
  203. else:
  204. raise AssertionError("Unhandled choice of log level")
  205. logging.getLogger().setLevel(log_level)
  206. self.runners = []
  207. def assertExec(self, runner):
  208. # keep reference, so files are not removed during test case time
  209. self.runners.append(runner)
  210. self.assertTrue(runner.check_all(), runner.get_description())
  211. def run(self, result=None):
  212. self.current_result = result # remember result for use in tearDown
  213. unittest.TestCase.run(self, result)
  214. def tearDown(self):
  215. unittest.TestCase.tearDown(self)
  216. if self.current_result.wasSuccessful() == True:
  217. for each in self.runners:
  218. if each.check_all() == True:
  219. if os.path.exists(each.get_dbfile()):
  220. os.unlink(each.get_dbfile())
  221. if os.path.exists(each.get_dbfile_prev()):
  222. os.unlink(each.get_dbfile_prev())