legacy_test.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105
  1. import unittest
  2. import argparse
  3. import threading
  4. import queue
  5. from lib.os_tools import *
  6. from sccd.legacy.xml_loader import load_model
  7. from sccd.controller.controller import *
  8. class PyTestCase(unittest.TestCase):
  9. def __init__(self, src_file):
  10. unittest.TestCase.__init__(self)
  11. self.src_file = src_file
  12. def __str__(self):
  13. return self.src_file
  14. def runTest(self):
  15. # Build & load
  16. model, test = load_model(self.src_file)
  17. inputs = test.input_events
  18. expected = test.expected_events
  19. controller = Controller(model)
  20. # generate input
  21. for i in inputs:
  22. controller.add_input(i)
  23. pipe = queue.Queue()
  24. def model_thread():
  25. try:
  26. # Run as-fast-as-possible, always advancing time to the next item in event queue, no sleeping.
  27. # The call returns when the event queue is empty and therefore the simulation is finished.
  28. controller.run_until(None, pipe)
  29. except Exception as e:
  30. pipe.put(e, block=True, timeout=None)
  31. return
  32. pipe.put(None, block=True, timeout=None)
  33. # start the controller
  34. thread = threading.Thread(target=model_thread)
  35. thread.start()
  36. # check output
  37. slot_index = 0
  38. while True:
  39. output = pipe.get(block=True, timeout=None)
  40. if isinstance(output, Exception):
  41. thread.join()
  42. raise output # Exception was caught in Controller thread, throw it here instead.
  43. elif output is None:
  44. self.assertEqual(slot_index, len(expected), "Less output was received than expected.")
  45. thread.join()
  46. return
  47. else:
  48. self.assertLess(slot_index, len(expected), "More output was received than expected.")
  49. exp_slot = expected[slot_index]
  50. # print("slot:", slot_index, ", events: ", output)
  51. self.assertEqual(len(exp_slot), len(output), "Slot %d length differs: Expected %s, but got %s instead." % (slot_index, exp_slot, output))
  52. # sort both expected and actual lists of events before comparing,
  53. # in theory the set of events at the end of a big step is unordered
  54. key_f = lambda e: "%s.%s"%(e.port, e.name)
  55. exp_slot.sort(key=key_f)
  56. output.sort(key=key_f)
  57. for (exp_event, event) in zip(exp_slot, output):
  58. matches = True
  59. if exp_event.name != event.name :
  60. matches = False
  61. if exp_event.port != event.port :
  62. matches = False
  63. if len(exp_event.parameters) != len(event.parameters) :
  64. matches = False
  65. for index in range(len(exp_event.parameters)) :
  66. if exp_event.parameters[index] != event.parameters[index]:
  67. matches = False
  68. self.assertTrue(matches, "Slot %d entry differs: Expected %s, but got %s instead." % (slot_index, exp_slot, output))
  69. slot_index += 1
  70. if __name__ == '__main__':
  71. parser = argparse.ArgumentParser(
  72. description="Run SCCD tests.",
  73. epilog="Set environment variable SCCDDEBUG=1 to display debug information about the inner workings of the runtime.")
  74. parser.add_argument('path', metavar='PATH', type=str, nargs='*', help="Tests to run. Can be a XML file or a directory. If a directory, it will be recursively scanned for XML files.")
  75. parser.add_argument('--build-dir', metavar='BUILD_DIR', type=str, default='build', help="Directory for built tests. Defaults to 'build'")
  76. args = parser.parse_args()
  77. src_files = get_files(args.path, filter=filter_xml)
  78. suite = unittest.TestSuite()
  79. for src_file in src_files:
  80. suite.addTest(PyTestCase(src_file))
  81. if len(src_files) == 0:
  82. print("No input files specified.")
  83. print()
  84. parser.print_usage()
  85. else:
  86. unittest.TextTestRunner(verbosity=2).run(suite)