test_all.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461
  1. import unittest
  2. import sys
  3. from utils import *
  4. sys.path.append("wrappers")
  5. from modelverse import *
  6. import threading
  7. model_hierarchy = \
  8. {"formalisms/": {"SimpleClassDiagrams": {},
  9. "TypeMapping": {},
  10. "Tracability": {},
  11. "ProcessModel": {},
  12. "ActionLanguage": {},
  13. "ManualOperation": {},
  14. "Bottom": {},
  15. },
  16. "models/": {"conformance_mv": {},
  17. },
  18. "administration/": {"core": {},
  19. "CoreFormalism": {},
  20. },
  21. "users/": {"admin/": {}
  22. },
  23. "type mappings/": {"formalisms/": {"SimpleClassDiagrams": {},
  24. "TypeMapping": {},
  25. "Tracability": {},
  26. "ProcessModel": {},
  27. "ActionLanguage": {},
  28. "ManualOperation": {},
  29. "Bottom": {},
  30. },
  31. "models/": {"conformance_mv": {},
  32. },
  33. "administration/": {"core": {},
  34. "CoreFormalism": {},
  35. },
  36. "users/": {"admin/": {}
  37. },
  38. },
  39. }
  40. def verify_clean():
  41. compare_locations("", set())
  42. compare_locations("models", set())
  43. compare_locations("formalisms", set())
  44. compare_locations("administration", set())
  45. compare_locations("type mappings", set())
  46. compare_locations("type mappings/models", set())
  47. compare_locations("type mappings/formalisms", set())
  48. compare_locations("type mappings/administration", set())
  49. def get_model_list(location):
  50. try:
  51. location_parts = location.split("/")
  52. current = model_hierarchy
  53. while location_parts:
  54. l = location_parts.pop(0)
  55. if l != "":
  56. current = current[l + "/"]
  57. return set(current.keys())
  58. except:
  59. return set([])
  60. def compare_locations(location, extra_to_default):
  61. assert model_list(location) == get_model_list(location) | set(extra_to_default)
  62. def compare_unordered_lists(got, expected):
  63. assert len(got) == len(expected)
  64. for i in got:
  65. assert i in expected
  66. for i in expected:
  67. assert i in got
  68. class TestModelverse(unittest.TestCase):
  69. proc = None
  70. @classmethod
  71. def setUpClass(self):
  72. TestModelverse.proc, address = start_mvc()
  73. init(address)
  74. login("admin", "admin")
  75. @classmethod
  76. def tearDownClass(self):
  77. try:
  78. kill(TestModelverse.proc)
  79. except:
  80. print("Got exception during teardown.")
  81. def setUp(self):
  82. verify_clean()
  83. folder_create("test")
  84. folder_create("type mappings/test")
  85. def tearDown(self):
  86. model_delete("test")
  87. model_delete("type mappings/test")
  88. verify_clean()
  89. pass
  90. def test_list_full(self):
  91. assert model_list_full("") == set([("formalisms/", "admin", "admin", "221"),
  92. ("models/", "admin", "admin", "221"),
  93. ("administration/", "admin", "admin", "110"),
  94. ("type mappings/", "admin", "admin", "221"),
  95. ("users/", "admin", "admin", "221"),
  96. ("test/", "admin", "nobody", "200"),
  97. ])
  98. def test_modelling(self):
  99. # Add a model
  100. model_add("test/Empty", "formalisms/SimpleClassDiagrams")
  101. # Check that it exists
  102. compare_locations("test", set(["Empty"]))
  103. compare_locations("type mappings/test", set(["Empty"]))
  104. # Check that it conforms
  105. assert verify("test/Empty", "formalisms/SimpleClassDiagrams") == "OK"
  106. # Check for permissions
  107. assert model_list_full("test") == set([(name, "admin", "admin", "221") for name in get_model_list("test")]) | \
  108. set([("Empty", "admin", "nobody", "200")])
  109. assert model_list_full("type mappings/test") == set([(name, "admin", "admin", "221") for name in get_model_list("type mappings/test")]) | \
  110. set([("Empty", "admin", "nobody", "200")])
  111. # Instantiate it further
  112. model_add("test/my_empty", "test/Empty")
  113. assert verify("test/my_empty", "test/Empty") == "OK"
  114. # Check that it exists
  115. compare_locations("test", set(["my_empty", "Empty"]))
  116. compare_locations("type mappings/test", set(["my_empty", "Empty"]))
  117. # Check that an instantiate of "A" fails
  118. try:
  119. res = instantiate("test/my_empty", "A")
  120. print("RESULT: " + str(res))
  121. assert False
  122. except UnknownIdentifier:
  123. assert verify("test/Empty", "formalisms/SimpleClassDiagrams") == "OK"
  124. # Create something in the formalism
  125. instantiate("test/Empty", "Class", ID="A")
  126. attr_assign("test/Empty", "A", "name", "A")
  127. assert verify("test/Empty", "formalisms/SimpleClassDiagrams") == "OK"
  128. # Now instantiate that in the model as well, which now works
  129. instantiate("test/my_empty", "A")
  130. assert verify("test/my_empty", "test/Empty") == "OK"
  131. def test_overwrite(self):
  132. model_add("test/Empty", "formalisms/SimpleClassDiagrams")
  133. assert element_list("test/Empty") == set([])
  134. instantiate("test/Empty", "Class", ID="A")
  135. assert element_list("test/Empty") == set([("A", "Class")])
  136. model_overwrite("test/Empty", "")
  137. assert element_list("test/Empty") == set([])
  138. compare_locations("test", set(["Empty"]))
  139. compare_locations("type mappings/test", set(["Empty"]))
  140. assert element_list("test/Empty") == set([])
  141. instantiate("test/Empty", "Class", ID="B")
  142. compare_locations("test", set(["Empty"]))
  143. compare_locations("type mappings/test", set(["Empty"]))
  144. def test_operations(self):
  145. log = []
  146. def manual_callback(model):
  147. print("Got in manual callback")
  148. p1 = instantiate(model, "PetriNet_Runtime/Place")
  149. p2 = instantiate(model, "PetriNet_Runtime/Place")
  150. p3 = instantiate(model, "PetriNet_Runtime/Place")
  151. t1 = instantiate(model, "PetriNet_Runtime/Transition")
  152. p2t1 = instantiate(model, "PetriNet_Runtime/P2T", (p1, t1))
  153. p2t2 = instantiate(model, "PetriNet_Runtime/P2T", (p2, t1))
  154. t2p1 = instantiate(model, "PetriNet_Runtime/T2P", (t1, p3))
  155. attr_assign(model, p1, "tokens", 1)
  156. attr_assign(model, p1, "name", "p1")
  157. attr_assign(model, p2, "tokens", 2)
  158. attr_assign(model, p2, "name", "p2")
  159. attr_assign(model, p3, "tokens", 3)
  160. attr_assign(model, p3, "name", "p3")
  161. attr_assign(model, t1, "name", "t1")
  162. attr_assign(model, t1, "executing", False)
  163. attr_assign(model, p2t1, "weight", 1)
  164. attr_assign(model, p2t2, "weight", 1)
  165. attr_assign(model, t2p1, "weight", 2)
  166. model_add("test/PetriNet", "formalisms/SimpleClassDiagrams", open("integration/code/pn_design.mvc", "r").read())
  167. model_add("test/PetriNet_Runtime", "formalisms/SimpleClassDiagrams", open("integration/code/pn_runtime.mvc", "r").read())
  168. model_add("test/my_pn", "test/PetriNet", open("integration/code/pn_design_model.mvc", "r").read())
  169. def add_tracability_D2R(model):
  170. instantiate(model, "Association", ("PetriNet/Place", "PetriNet_Runtime/Place"), ID="D2R_PlaceLink")
  171. instantiate(model, "Association", ("PetriNet/Transition", "PetriNet_Runtime/Transition"), ID="D2R_TransitionLink")
  172. def add_tracability_R2D(model):
  173. instantiate(model, "Association", ("PetriNet_Runtime/Place", "PetriNet/Place"), ID="R2D_PlaceLink")
  174. instantiate(model, "Association", ("PetriNet_Runtime/Transition", "PetriNet/Transition"), ID="R2D_TransitionLink")
  175. transformation_add_MT({"PetriNet": "test/PetriNet"}, {}, "test/print_pn", open("integration/code/pn_print.mvc").read())
  176. transformation_add_MANUAL({"PetriNet": "test/PetriNet"}, {"PetriNet_Runtime": "test/PetriNet_Runtime"}, "test/pn_design_to_runtime", add_tracability_D2R)
  177. transformation_add_AL({"PetriNet_Runtime": "test/PetriNet_Runtime"}, {"PetriNet_Runtime": "test/PetriNet_Runtime"}, "test/pn_simulate", open("integration/code/pn_simulate.alc").read())
  178. transformation_add_MT({"PetriNet_Runtime": "test/PetriNet_Runtime"}, {"PetriNet": "test/PetriNet"}, "test/pn_runtime_to_design", open("integration/code/pn_runtime_to_design.mvc").read(), add_tracability_R2D)
  179. import log_output
  180. log = []
  181. ctrl = log_output.Controller(log, keep_running=False)
  182. thrd = threading.Thread(target=ctrl.start)
  183. thrd.daemon = True
  184. thrd.start()
  185. assert transformation_execute_MT("test/print_pn", {"PetriNet": "test/my_pn"}, {}, (ctrl, "inp", "outp")) == True
  186. print("Joining")
  187. thrd.join()
  188. print("Joined")
  189. print("Got log: " + str(log))
  190. assert set(log) == set(['"p1" --> 1',
  191. '"p2" --> 2',
  192. '"p3" --> 3'])
  193. print("MAN")
  194. assert transformation_execute_MANUAL("test/pn_design_to_runtime", {"PetriNet": "test/my_pn"}, {"PetriNet_Runtime": "test/my_pn_RT"}, manual_callback) == True
  195. print("OK")
  196. assert transformation_execute_AL("test/pn_simulate", {"PetriNet_Runtime": "test/my_pn_RT"}, {"PetriNet_Runtime": "test/my_pn_RT"}) == True
  197. assert transformation_execute_MT("test/pn_runtime_to_design", {"PetriNet_Runtime": "test/my_pn_RT"}, {"PetriNet": "test/my_pn"}) == True
  198. print("MT OK")
  199. log = []
  200. ctrl = log_output.Controller(log, keep_running=False)
  201. thrd = threading.Thread(target=ctrl.start)
  202. thrd.daemon = True
  203. thrd.start()
  204. assert transformation_execute_MT("test/print_pn", {"PetriNet": "test/my_pn"}, {}, (ctrl, "inp", "outp")) == True
  205. thrd.join()
  206. assert set(log) == set(['"p1" --> 0',
  207. '"p2" --> 1',
  208. '"p3" --> 5'])
  209. model_delete("RAMified")
  210. model_delete("merged")
  211. model_delete(".tmp")
  212. model_delete("type mappings/RAMified")
  213. model_delete("type mappings/merged")
  214. model_delete("type mappings/.tmp")
  215. def test_process_model_trivial_pn_subfunction(self):
  216. model_add("test/PetriNet", "formalisms/SimpleClassDiagrams", open("integration/code/pn_design.mvc", "r").read())
  217. model_add("test/ReachabilityGraph", "formalisms/SimpleClassDiagrams", open("integration/code/reachability_graph.mvc", "r").read())
  218. model_add("test/pn_reachability", "formalisms/ProcessModel", open("integration/code/pm_pn_reachability.mvc", "r").read())
  219. transformation_add_MT({}, {"PetriNet": "test/PetriNet"}, "test/initialize_PN", open("integration/code/initialize_PN.mvc", "r").read())
  220. transformation_add_MANUAL({"PetriNet": "test/PetriNet"}, {"PetriNet": "test/PetriNet"}, "test/refine_PN")
  221. transformation_add_AL({"PetriNet": "test/PetriNet"}, {"ReachabilityGraph": "test/ReachabilityGraph"}, "test/reachability", open("integration/code/reachability_subfunction.alc", "r").read())
  222. transformation_add_MT({"ReachabilityGraph": "test/ReachabilityGraph"}, {}, "test/reachability_print", open("integration/code/reachabilitygraph_print.mvc", 'r').read())
  223. def callback_refine_PN(context):
  224. p1 = instantiate(None, "PetriNet/Place", context=context)
  225. attr_assign(None, p1, "name", "p1", context=context)
  226. attr_assign(None, p1, "tokens", 1, context=context)
  227. t1 = instantiate(None, "PetriNet/Transition", context=context)
  228. attr_assign(None, t1, "name", "t1", context=context)
  229. p2t = instantiate(None, "PetriNet/P2T", (p1, t1), context=context)
  230. attr_assign(None, p2t, "weight", 1, context=context)
  231. import log_output
  232. log = []
  233. ctrl = log_output.Controller(log, keep_running=False)
  234. thrd = threading.Thread(target=ctrl.start)
  235. thrd.daemon = True
  236. thrd.start()
  237. process_execute("test/pn_reachability", {}, {"test/refine_PN": callback_refine_PN, "test/reachability_print": (ctrl, "inp", "outp")})
  238. thrd.join()
  239. assert set(log) == set(['"0": {"p1": 1}',
  240. '"1": {"p1": 0}',
  241. '"0" --["t1"]--> "1"'])
  242. model_delete(".tmp")
  243. model_delete("RAMified")
  244. model_delete("merged")
  245. model_delete("type mappings/RAMified")
  246. model_delete("type mappings/merged")
  247. model_delete("type mappings/.tmp")
  248. def test_render(self):
  249. model_add("test/CausalBlockDiagrams", "formalisms/SimpleClassDiagrams", open("integration/code/cbd_design.mvc", 'r').read())
  250. model_add("test/MM_rendered_graphical", "formalisms/SimpleClassDiagrams", open("models/MM_rendered_graphical.mvc", 'r').read())
  251. model_add("test/my_CBD", "test/CausalBlockDiagrams", open("integration/code/my_cbd.mvc", 'r').read())
  252. def add_tracability(context):
  253. instantiate(None, "Association", ("abstract/Block", "rendered/Group"), ID="TracabilityLink", context=context)
  254. transformation_add_MT({"abstract": "test/CausalBlockDiagrams", "rendered": "test/MM_rendered_graphical"}, {"abstract": "test/CausalBlockDiagrams", "rendered": "test/MM_rendered_graphical"}, "test/render_graphical_CBD", open("models/CBD_mapper.mvc", 'r').read(), add_tracability)
  255. result = model_render("test/my_CBD", "test/render_graphical_CBD", "test/my_perceptualized_CBD")
  256. assert len(result) == 23
  257. model_delete("RAMified")
  258. model_delete("merged")
  259. model_delete("type mappings/RAMified")
  260. model_delete("type mappings/merged")
  261. model_delete("tracability")
  262. model_delete("type mappings/tracability")
  263. def test_SCCD_basic(self):
  264. model_add("test/SCCD", "formalisms/SimpleClassDiagrams", open("models/SCCD.mvc", 'r').read())
  265. model_add("test/SCCD_Trace", "formalisms/SimpleClassDiagrams", open("models/SCCD_Trace.mvc", 'r').read())
  266. model_add("test/my_SCCD", "test/SCCD", open("integration/code/SCCD_all.mvc", 'r').read())
  267. transformation_add_AL({"SCCD": "test/SCCD"}, {"trace": "test/SCCD_Trace"}, "test/SCCD_execute_afap", open("models/SCCD_execute.alc", 'r').read().replace("afap = False", "afap = True"))
  268. transformation_execute_AL("test/SCCD_execute_afap", {"SCCD": "test/my_SCCD"}, {"trace": "test/my_SCCD_trace"})
  269. print("Executed AL")
  270. alter_context("test/my_SCCD_trace", "test/SCCD_Trace")
  271. print("Altered context")
  272. lst = element_list_nice("test/my_SCCD_trace")
  273. model_delete("merged")
  274. model_delete("type mappings/merged")
  275. lst.sort(key=lambda i: (i["timestamp"], i["name"]))
  276. result = [(i["timestamp"], str(i["name"])) for i in lst if i["name"] not in ["updateTimerValue", "updateTimerColour", "resetTimer"]]
  277. print(result)
  278. assert result == [(5.0, "displayRed"),
  279. (20.0, "displayYellow"),
  280. (20.5, "displayNone"),
  281. (21.0, "displayYellow"),
  282. (21.5, "displayNone"),
  283. (22.0, "displayYellow"),
  284. (22.5, "displayNone"),
  285. (23.0, "displayYellow"),
  286. (23.5, "displayNone"),
  287. (24.0, "displayYellow"),
  288. (24.5, "displayNone"),
  289. (25.0, "displayYellow"),
  290. (25.5, "displayNone"),
  291. (26.0, "displayYellow"),
  292. (26.5, "displayNone"),
  293. (27.0, "displayYellow"),
  294. (27.5, "displayNone"),
  295. (28.0, "displayYellow"),
  296. (28.5, "displayNone"),
  297. (29.0, "displayYellow"),
  298. (29.4, "displayNone"),
  299. (29.4, "displayRed"),
  300. (89.4, "displayGreen"),
  301. (129.4, "displayNone"),
  302. (129.4, "displayRed"),
  303. (139.4, "displayYellow"),
  304. (139.9, "displayNone"),
  305. (140.4, "displayYellow"),
  306. (140.9, "displayNone"),
  307. (141.4, "displayYellow"),
  308. (141.9, "displayNone"),
  309. (142.4, "displayYellow"),
  310. (142.9, "displayNone"),
  311. (143.4, "displayYellow"),
  312. (143.9, "displayNone"),
  313. (144.4, "displayYellow"),
  314. (144.9, "displayNone"),
  315. (145.4, "displayYellow"),
  316. (145.9, "displayNone"),
  317. (146.4, "displayYellow"),
  318. (146.9, "displayNone"),
  319. (147.4, "displayYellow"),
  320. (147.9, "displayNone"),
  321. (148.4, "displayYellow"),
  322. ]
  323. def test_switch_MM(self):
  324. model_add("test/PetriNet", "formalisms/SimpleClassDiagrams", open("integration/code/pn_design.mvc", "r").read())
  325. model_add("test/my_pn", "test/PetriNet", open("integration/code/pn_design_model.mvc", "r").read())
  326. alter_context("test/PetriNet", "formalisms/SimpleClassDiagrams")
  327. alter_context("test/my_pn", "test/PetriNet")
  328. got = element_list_nice("test/PetriNet")
  329. expected = \
  330. [{'id': 'Natural', 'type': 'SimpleAttribute', 'constraint': None, 'name': 'Natural'},
  331. {'id': 'String', 'type': 'SimpleAttribute', 'constraint': None, 'name': 'String'},
  332. {'id': 'Place', 'type': 'Class', 'lower_cardinality': None, 'upper_cardinality': None, 'constraint': None, 'name': 'Place'},
  333. {'id': 'Place_tokens', 'type': 'AttributeLink', '__source': 'Place', '__target': 'Natural', 'name': 'tokens', 'optional': False, 'constraint': None},
  334. {'id': 'Place_name', 'type': 'AttributeLink', '__source': 'Place', '__target': 'String', 'name': 'name', 'optional': False, 'constraint': None},
  335. {'id': 'Transition', 'type': 'Class', 'lower_cardinality': None, 'upper_cardinality': None, 'constraint': None, 'name': 'Transition'},
  336. {'id': 'Transition_name', 'type': 'AttributeLink', '__source': 'Transition', '__target': 'String', 'name': 'name', 'optional': False, 'constraint': None},
  337. {'id': 'P2T', 'type': 'Association', '__source': 'Place', '__target': 'Transition', 'source_lower_cardinality': None, 'target_lower_cardinality': None, 'source_upper_cardinality': None, 'target_upper_cardinality': None, 'constraint': None, 'name': 'P2T'},
  338. {'id': 'P2T_weight', 'type': 'AttributeLink', '__source': 'P2T', '__target': 'Natural', 'name': 'weight', 'optional': False, 'constraint': None},
  339. {'id': 'T2P', 'type': 'Association', '__source': 'Transition', '__target': 'Place', 'source_lower_cardinality': None, 'target_lower_cardinality': None, 'source_upper_cardinality': None, 'target_upper_cardinality': None, 'constraint': None, 'name': 'T2P'},
  340. {'id': 'T2P_weight', 'type': 'AttributeLink', '__source': 'T2P', '__target': 'Natural', 'name': 'weight', 'optional': False, 'constraint': None}
  341. ]
  342. compare_unordered_lists(got, expected)
  343. got = element_list_nice("test/my_pn")
  344. expected = \
  345. [{'id': 'p1', 'type': 'Place', 'tokens': 1, 'name': 'p1'},
  346. {'id': 'p2', 'type': 'Place', 'tokens': 2, 'name': 'p2'},
  347. {'id': 'p3', 'type': 'Place', 'tokens': 3, 'name': 'p3'},
  348. {'id': 't1', 'type': 'Transition', 'name': 't1'},
  349. {'id': '__0', 'type': 'P2T', '__source': 'p1', '__target': 't1', 'weight': 1},
  350. {'id': '__1', 'type': 'P2T', '__source': 'p2', '__target': 't1', 'weight': 1},
  351. {'id': '__2', 'type': 'T2P', '__source': 't1', '__target': 'p3', 'weight': 2}
  352. ]
  353. compare_unordered_lists(got, expected)
  354. alter_context("test/PetriNet", "formalisms/Bottom")
  355. alter_context("test/my_pn", "formalisms/Bottom")
  356. count_nodes = 0
  357. count_edges = 0
  358. for entry in element_list_nice("test/PetriNet"):
  359. assert entry["type"] in ["Node", "Edge"]
  360. if entry["type"] == "Node":
  361. assert len(entry) == 2
  362. count_nodes += 1
  363. else:
  364. assert len(entry) == 4
  365. count_edges += 1
  366. assert count_nodes == 20
  367. assert count_edges == 23
  368. count_nodes = 0
  369. count_edges = 0
  370. for entry in element_list_nice("test/my_pn"):
  371. assert entry["type"] in ["Node", "Edge"]
  372. if entry["type"] == "Node":
  373. assert len(entry) == 2
  374. count_nodes += 1
  375. else:
  376. assert len(entry) == 4
  377. count_edges += 1
  378. assert count_nodes == 14
  379. assert count_edges == 13
  380. alter_context("test/PetriNet", "test/PetriNet")
  381. alter_context("test/my_pn", "formalisms/SimpleClassDiagrams")
  382. try:
  383. element_list_nice("test/PetriNet")
  384. self.fail()
  385. except:
  386. pass
  387. try:
  388. element_list_nice("test/my_pn")
  389. self.fail()
  390. except:
  391. pass