modelverse/unit/test_all.py

import unittest

import sys
from utils import *

sys.path.append("wrappers")
from modelverse import *

model_hierarchy = \
            {"formalisms/": {"SimpleClassDiagrams": {},
                             "TypeMapping": {},
                             "Tracability": {},
                             "ProcessModel": {},
                             "ActionLanguage": {},
                             "ManualOperation": {},
                             "Bottom": {},
                            },
            "models/": {"conformance_mv": {},
                        },
            "administration/": {"core": {},
                                "CoreFormalism": {},
                               },
            "users/": {"admin/": {}
                      },
            "type mappings/": {"formalisms/": {"SimpleClassDiagrams": {},
                                               "TypeMapping": {},
                                               "Tracability": {},
                                               "ProcessModel": {},
                                               "ActionLanguage": {},
                                               "ManualOperation": {},
                                               "Bottom": {},
                                              },
                               "models/": {"conformance_mv": {},
                                          },
                               "administration/": {"core": {},
                                                   "CoreFormalism": {},
                                                  },
                               "users/": {"admin/": {}
                                         },
                              },
            }

def verify_clean():
    compare_locations("", set())
    compare_locations("models", set())
    compare_locations("formalisms", set())
    compare_locations("administration", set())
    compare_locations("type mappings", set())
    compare_locations("type mappings/models", set())
    compare_locations("type mappings/formalisms", set())
    compare_locations("type mappings/administration", set())

def get_model_list(location):
    try:
        location_parts = location.split("/")
        current = model_hierarchy
        while location_parts:
            l = location_parts.pop(0)
            if l != "":
                current = current[l + "/"]
        return set(current.keys())
    except:
        return set([])

def compare_locations(location, extra_to_default):
    assert model_list(location) == get_model_list(location) | set(extra_to_default)

def compare_unordered_lists(got, expected):
    assert len(got) == len(expected)
    for i in got:
        assert i in expected
    for i in expected:
        assert i in got

class TestModelverse(unittest.TestCase):
    proc = None

    @classmethod
    def setUpClass(self):
        TestModelverse.proc, address = start_mvc()
        init(address)
        login("admin", "admin")

    @classmethod
    def tearDownClass(self):
        try:
            kill(TestModelverse.proc)
        except:
            print("Got exception during teardown.")

    def setUp(self):
        verify_clean()
        folder_create("test")
        folder_create("type mappings/test")

    def tearDown(self):
        model_delete("test")
        model_delete("type mappings/test")
        verify_clean()
        pass

    def test_list_full(self):
        assert model_list_full("") == set([("formalisms/", "admin", "admin", "221"),
                                           ("models/", "admin", "admin", "221"),
                                           ("administration/", "admin", "admin", "110"),
                                           ("type mappings/", "admin", "admin", "221"),
                                           ("users/", "admin", "admin", "221"),
                                           ("test/", "admin", "nobody", "200"),
                                          ])

    def test_modelling(self):
        # Add a model
        model_add("test/Empty", "formalisms/SimpleClassDiagrams")

        # Check that it exists
        compare_locations("test", set(["Empty"]))
        compare_locations("type mappings/test", set(["Empty"]))

        # Check that it conforms
        assert verify("test/Empty", "formalisms/SimpleClassDiagrams") == "OK"

        # Check for permissions
        assert model_list_full("test") == set([(name, "admin", "admin", "221") for name in get_model_list("test")]) | \
                                                set([("Empty", "admin", "nobody", "200")])
        assert model_list_full("type mappings/test") == set([(name, "admin", "admin", "221") for name in get_model_list("type mappings/test")]) | \
                                                              set([("Empty", "admin", "nobody", "200")])

        # Instantiate it further
        model_add("test/my_empty", "test/Empty")
        assert verify("test/my_empty", "test/Empty") == "OK"

        # Check that it exists
        compare_locations("test", set(["my_empty", "Empty"]))
        compare_locations("type mappings/test", set(["my_empty", "Empty"]))

        # Check that an instantiate of "A" fails
        try:
            res = instantiate("test/my_empty", "A")
            print("RESULT: " + str(res))
            assert False
        except UnknownIdentifier:
            assert verify("test/Empty", "formalisms/SimpleClassDiagrams") == "OK"

        # Create something in the formalism
        instantiate("test/Empty", "Class", ID="A")
        assert verify("test/Empty", "formalisms/SimpleClassDiagrams") == "OK"

        # Now instantiate that in the model as well, which now works
        instantiate("test/my_empty", "A")
        assert verify("test/my_empty", "test/Empty") == "OK"

    def test_overwrite(self):
        model_add("test/Empty", "formalisms/SimpleClassDiagrams")
        assert element_list("test/Empty") == set([])

        instantiate("test/Empty", "Class", ID="A")
        assert element_list("test/Empty") == set([("A", "Class")])

        model_overwrite("test/Empty", "")
        assert element_list("test/Empty") == set([])
        compare_locations("test", set(["Empty"]))
        compare_locations("type mappings/test", set(["Empty"]))
        
        assert element_list("test/Empty") == set([])

        instantiate("test/Empty", "Class", ID="B")
        compare_locations("test", set(["Empty"]))
        compare_locations("type mappings/test", set(["Empty"]))

    def test_operations(self):
        log = []

        def callback(value):
            log.append(value)

        def manual_callback():
            p1 = instantiate(None, "PetriNet_Runtime/Place")
            attr_assign(None, p1, "tokens", 1)
            attr_assign(None, p1, "name", "p1")
            p2 = instantiate(None, "PetriNet_Runtime/Place")
            attr_assign(None, p2, "tokens", 2)
            attr_assign(None, p2, "name", "p2")
            p3 = instantiate(None, "PetriNet_Runtime/Place")
            attr_assign(None, p3, "tokens", 3)
            attr_assign(None, p3, "name", "p3")
            t1 = instantiate(None, "PetriNet_Runtime/Transition")
            attr_assign(None, t1, "name", "t1")
            attr_assign(None, t1, "executing", False)
            p2t1 = instantiate(None, "PetriNet_Runtime/P2T", (p1, t1))
            attr_assign(None, p2t1, "weight", 1)
            p2t2 = instantiate(None, "PetriNet_Runtime/P2T", (p2, t1))
            attr_assign(None, p2t2, "weight", 1)
            t2p1 = instantiate(None, "PetriNet_Runtime/T2P", (t1, p3))
            attr_assign(None, t2p1, "weight", 2)

        model_add("test/PetriNet", "formalisms/SimpleClassDiagrams", open("integration/code/pn_design.mvc", "r").read())
        model_add("test/PetriNet_Runtime", "formalisms/SimpleClassDiagrams", open("integration/code/pn_runtime.mvc", "r").read())

        model_add("test/my_pn", "test/PetriNet", open("integration/code/pn_design_model.mvc", "r").read())

        def add_tracability_D2R():
            instantiate(None, "Association", ("PetriNet/Place", "PetriNet_Runtime/Place"), ID="D2R_PlaceLink")
            instantiate(None, "Association", ("PetriNet/Transition", "PetriNet_Runtime/Transition"), ID="D2R_TransitionLink")

        def add_tracability_R2D():
            instantiate(None, "Association", ("PetriNet_Runtime/Place", "PetriNet/Place"), ID="R2D_PlaceLink")
            instantiate(None, "Association", ("PetriNet_Runtime/Transition", "PetriNet/Transition"), ID="R2D_TransitionLink")

        transformation_add_MT({"PetriNet": "test/PetriNet"}, {}, "test/print_pn", open("integration/code/pn_print.mvc").read())
        transformation_add_MANUAL({"PetriNet": "test/PetriNet"}, {"PetriNet_Runtime": "test/PetriNet_Runtime"}, "test/pn_design_to_runtime", add_tracability_D2R)
        transformation_add_AL({"PetriNet_Runtime": "test/PetriNet_Runtime"}, {"PetriNet_Runtime": "test/PetriNet_Runtime"}, "test/pn_simulate", open("integration/code/pn_simulate.alc").read())
        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)

        import log_output
        log = []
        ctrl = log_output.Controller(log, keep_running=False)
        thrd = threading.Thread(target=ctrl.start)
        thrd.daemon = True
        thrd.start()

        assert transformation_execute_MT("test/print_pn", {"PetriNet": "test/my_pn"}, {}, (ctrl, "inp", "outp")) == None
        thrd.join()
        assert set(log) == set(['"p1" --> 1',
                                '"p2" --> 2',
                                '"p3" --> 3'])

        assert transformation_execute_MANUAL("test/pn_design_to_runtime", {"PetriNet": "test/my_pn"}, {"PetriNet_Runtime": "test/my_pn_RT"}, manual_callback) == True
        assert transformation_execute_AL("test/pn_simulate", {"PetriNet_Runtime": "test/my_pn_RT"}, {"PetriNet_Runtime": "test/my_pn_RT"}) == True
        assert transformation_execute_MT("test/pn_runtime_to_design", {"PetriNet_Runtime": "test/my_pn_RT"}, {"PetriNet": "test/my_pn"}) == True

        log = []
        ctrl = log_output.Controller(log, keep_running=False)
        thrd = threading.Thread(target=ctrl.start)
        thrd.daemon = True
        thrd.start()

        assert transformation_execute_MT("test/print_pn", {"PetriNet": "test/my_pn"}, {}, (ctrl, "inp", "outp")) == None
        thrd.join()
        assert set(log) == set(['"p1" --> 0',
                                '"p2" --> 1',
                                '"p3" --> 5'])

        model_delete("RAMified")
        model_delete("merged")
        model_delete("type mappings/RAMified")
        model_delete("type mappings/merged")

    def test_process_model_trivial_pn_subfunction(self):
        model_add("test/PetriNet", "formalisms/SimpleClassDiagrams", open("integration/code/pn_design.mvc", "r").read())
        model_add("test/ReachabilityGraph", "formalisms/SimpleClassDiagrams", open("integration/code/reachability_graph.mvc", "r").read())
        model_add("test/pn_reachability", "formalisms/ProcessModel", open("integration/code/pm_pn_reachability.mvc", "r").read())
        transformation_add_MT({}, {"PetriNet": "test/PetriNet"}, "test/initialize_PN", open("integration/code/initialize_PN.mvc", "r").read())
        transformation_add_MANUAL({"PetriNet": "test/PetriNet"}, {"PetriNet": "test/PetriNet"}, "test/refine_PN")
        transformation_add_AL({"PetriNet": "test/PetriNet"}, {"ReachabilityGraph": "test/ReachabilityGraph"}, "test/reachability", open("integration/code/reachability_subfunction.alc", "r").read())
        transformation_add_MT({"ReachabilityGraph": "test/ReachabilityGraph"}, {}, "test/reachability_print", open("integration/code/reachabilitygraph_print.mvc", 'r').read())

        def callback_refine_PN():
            p1 = instantiate(None, "PetriNet/Place")
            attr_assign(None, p1, "name", "p1")
            attr_assign(None, p1, "tokens", 1)

            t1 = instantiate(None, "PetriNet/Transition")
            attr_assign(None, t1, "name", "t1")

            p2t = instantiate(None, "PetriNet/P2T", (p1, t1))
            attr_assign(None, p2t, "weight", 1)

        log = set([])
        def callback_print(value):
            log.add(value)

        process_execute("test/pn_reachability", "", {"test/refine_PN": callback_refine_PN, "test/reachability_print": callback_print})

        assert log == set(['"0": {"p1": 1, }',
                           '"1": {"p1": 0, }',
                           '"0" --["t1"]--> "1"'])

        model_delete("RAMified")
        model_delete("merged")
        model_delete("type mappings/RAMified")
        model_delete("type mappings/merged")

    def test_render(self):
        model_add("test/CausalBlockDiagrams", "formalisms/SimpleClassDiagrams", open("integration/code/cbd_design.mvc", 'r').read())
        model_add("test/MM_rendered_graphical", "formalisms/SimpleClassDiagrams", open("models/MM_rendered_graphical.mvc", 'r').read())
        model_add("test/my_CBD", "test/CausalBlockDiagrams", open("integration/code/my_cbd.mvc", 'r').read())

        def add_tracability():
            instantiate(None, "Association", ("abstract/Block", "rendered/Group"), ID="TracabilityLink")

        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)
        result = model_render("test/my_CBD", "test/render_graphical_CBD")
        assert len(result) == 23

        model_delete("RAMified")
        model_delete("merged")
        model_delete("type mappings/RAMified")
        model_delete("type mappings/merged")
        model_delete("rendered")
        model_delete("tracability")
        model_delete("type mappings/rendered")
        model_delete("type mappings/tracability")

    def test_SCCD_basic(self):
        model_add("test/SCCD", "formalisms/SimpleClassDiagrams", open("models/SCCD.mvc", 'r').read())
        model_add("test/SCCD_Trace", "formalisms/SimpleClassDiagrams", open("models/SCCD_Trace.mvc", 'r').read())

        model_add("test/my_SCCD", "test/SCCD", open("integration/code/SCCD_all.mvc", 'r').read())

        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"))
        transformation_execute_AL("test/SCCD_execute_afap", {"SCCD": "test/my_SCCD"}, {"trace": "test/my_SCCD_trace"})

        alter_context("test/my_SCCD_trace", "test/SCCD_Trace")
        lst = element_list_nice("test/my_SCCD_trace")

        model_delete("merged")
        model_delete("type mappings/merged")

        lst.sort(key=lambda i: (i["timestamp"], i["name"]))
        result = [(i["timestamp"], str(i["name"])) for i in lst if i["name"] not in ["updateTimerValue", "updateTimerColour", "resetTimer"]]
        print(result)

        assert result == [(5.0, "displayRed"),
                          (20.0, "displayYellow"),
                          (20.5, "displayNone"),
                          (21.0, "displayYellow"),
                          (21.5, "displayNone"),
                          (22.0, "displayYellow"),
                          (22.5, "displayNone"),
                          (23.0, "displayYellow"),
                          (23.5, "displayNone"),
                          (24.0, "displayYellow"),
                          (24.5, "displayNone"),
                          (25.0, "displayYellow"),
                          (25.5, "displayNone"),
                          (26.0, "displayYellow"),
                          (26.5, "displayNone"),
                          (27.0, "displayYellow"),
                          (27.5, "displayNone"),
                          (28.0, "displayYellow"),
                          (28.5, "displayNone"),
                          (29.0, "displayYellow"),
                          (29.4, "displayNone"),
                          (29.4, "displayRed"),
                          (89.4, "displayGreen"),
                          (129.4, "displayNone"),
                          (129.4, "displayRed"),
                          (139.4, "displayYellow"),
                          (139.9, "displayNone"),
                          (140.4, "displayYellow"),
                          (140.9, "displayNone"),
                          (141.4, "displayYellow"),
                          (141.9, "displayNone"),
                          (142.4, "displayYellow"),
                          (142.9, "displayNone"),
                          (143.4, "displayYellow"),
                          (143.9, "displayNone"),
                          (144.4, "displayYellow"),
                          (144.9, "displayNone"),
                          (145.4, "displayYellow"),
                          (145.9, "displayNone"),
                          (146.4, "displayYellow"),
                          (146.9, "displayNone"),
                          (147.4, "displayYellow"),
                          (147.9, "displayNone"),
                          (148.4, "displayYellow"),
                         ]

    def test_switch_MM(self):
        model_add("test/PetriNet", "formalisms/SimpleClassDiagrams", open("integration/code/pn_design.mvc", "r").read())
        model_add("test/my_pn", "test/PetriNet", open("integration/code/pn_design_model.mvc", "r").read())

        alter_context("test/PetriNet", "formalisms/SimpleClassDiagrams")
        alter_context("test/my_pn", "test/PetriNet")
        
        got = element_list_nice("test/PetriNet")
        expected = \
            [{'id': 'Natural', 'type': 'SimpleAttribute', 'constraint': None},
             {'id': 'String', 'type': 'SimpleAttribute', 'constraint': None},
             {'id': 'Place', 'type': 'Class', 'lower_cardinality': None, 'upper_cardinality': None, 'constraint': None},
             {'id': 'Place_tokens', 'type': 'AttributeLink', '__source': 'Place', '__target': 'Natural', 'name': 'tokens', 'optional': False, 'constraint': None},
             {'id': 'Place_name', 'type': 'AttributeLink', '__source': 'Place', '__target': 'String', 'name': 'name', 'optional': False, 'constraint': None},
             {'id': 'Transition', 'type': 'Class', 'lower_cardinality': None, 'upper_cardinality': None, 'constraint': None},
             {'id': 'Transition_name', 'type': 'AttributeLink', '__source': 'Transition', '__target': 'String', 'name': 'name', 'optional': False, 'constraint': None},
             {'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},
             {'id': 'P2T_weight', 'type': 'AttributeLink', '__source': 'P2T', '__target': 'Natural', 'name': 'weight', 'optional': False, 'constraint': None},
             {'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},
             {'id': 'T2P_weight', 'type': 'AttributeLink', '__source': 'T2P', '__target': 'Natural', 'name': 'weight', 'optional': False, 'constraint': None}
            ]
        compare_unordered_lists(got, expected)

        got = element_list_nice("test/my_pn")
        expected = \
            [{'id': 'p1', 'type': 'Place', 'tokens': 1, 'name': 'p1'},
             {'id': 'p2', 'type': 'Place', 'tokens': 2, 'name': 'p2'},
             {'id': 'p3', 'type': 'Place', 'tokens': 3, 'name': 'p3'},
             {'id': 't1', 'type': 'Transition', 'name': 't1'},
             {'id': '__0', 'type': 'P2T', '__source': 'p1', '__target': 't1', 'weight': 1},
             {'id': '__1', 'type': 'P2T', '__source': 'p2', '__target': 't1', 'weight': 1},
             {'id': '__2', 'type': 'T2P', '__source': 't1', '__target': 'p3', 'weight': 2}
            ]
        compare_unordered_lists(got, expected)

        alter_context("test/PetriNet", "formalisms/Bottom")
        alter_context("test/my_pn", "formalisms/Bottom")

        count_nodes = 0
        count_edges = 0
        for entry in element_list_nice("test/PetriNet"):
            assert entry["type"] in ["Node", "Edge"]
            if entry["type"] == "Node":
                assert len(entry) == 2
                count_nodes += 1
            else:
                assert len(entry) == 4
                count_edges += 1
        assert count_nodes == 14
        assert count_edges == 17

        count_nodes = 0
        count_edges = 0
        for entry in element_list_nice("test/my_pn"):
            assert entry["type"] in ["Node", "Edge"]
            if entry["type"] == "Node":
                assert len(entry) == 2
                count_nodes += 1
            else:
                assert len(entry) == 4
                count_edges += 1
        assert count_nodes == 14
        assert count_edges == 13

        alter_context("test/PetriNet", "test/PetriNet")
        alter_context("test/my_pn", "formalisms/SimpleClassDiagrams")
        try:
            element_list_nice("test/PetriNet")
            self.fail()
        except:
            pass
        try:
            element_list_nice("test/my_pn")
            self.fail()
        except:
            pass