modelverse/integration/test_powerwindow.py

import unittest

from utils import *
import sys

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

class TestPowerWindow(unittest.TestCase):
    def setUp(self):
        self.proc, self.address = start_mvc()
        init(self.address)
        login("admin", "admin")

    def tearDown(self):
        kill(self.proc)

    def test_process_powerwindow_fast(self):
        model_add("ReachabilityGraph", "SimpleClassDiagrams", open("models/reachability_graph.mvc", "r").read())
        model_add("PetriNet", "SimpleClassDiagrams", open("integration/code/pn_design.mvc", 'r').read())
        model_add("Encapsulated_PetriNet", "SimpleClassDiagrams", open("models/petrinet_ports.mvc", 'r').read())

        model_add("PW_Plant", "SimpleClassDiagrams", open("models/plant_PW.mvc", 'r').read())
        model_add("PW_Environment", "SimpleClassDiagrams", open("models/environment_PW.mvc", 'r').read())
        model_add("PW_Control", "SimpleClassDiagrams", open("models/control_PW.mvc", 'r').read())
        model_add("Requirements", "SimpleClassDiagrams", open("models/requirements.mvc", 'r').read())
        model_add("Query", "SimpleClassDiagrams", open("models/query.mvc", 'r').read())
        model_add("Architecture", "SimpleClassDiagrams", open("models/architecture.mvc", 'r').read())

        model_add("pm_powerwindow", "ProcessModel", open("models/pm_req_analyse.mvc", 'r').read())

        assert model_list() == set([\
                ("ReachabilityGraph", "SimpleClassDiagrams"),
                ("PetriNet", "SimpleClassDiagrams"),
                ("Encapsulated_PetriNet", "SimpleClassDiagrams"),
                ("PW_Plant", "SimpleClassDiagrams"),
                ("PW_Environment", "SimpleClassDiagrams"),
                ("PW_Control", "SimpleClassDiagrams"),
                ("Requirements", "SimpleClassDiagrams"),
                ("Query", "SimpleClassDiagrams"),
                ("Architecture", "SimpleClassDiagrams"),
                ("pm_powerwindow", "ProcessModel"),
                ("SimpleClassDiagrams", "SimpleClassDiagrams"),
                ("CoreFormalism", "SimpleClassDiagrams"),
                ("ManualOperation", "SimpleClassDiagrams"),
                ("ActionLanguage", "SimpleClassDiagrams"),
                ("ProcessModel", "SimpleClassDiagrams"),
                ("Tracability", "SimpleClassDiagrams"),
                ("core", "CoreFormalism"),])

        transformation_add_MT_language(["PetriNet", "Encapsulated_PetriNet", "PW_Plant", "PW_Control", "PW_Environment", "Requirements", "ReachabilityGraph", "Query", "Architecture"], "All_RAM")
        instantiate("__merged_All_RAM", "Association", ("PW_Control/State", "Encapsulated_PetriNet/Place"), ID="CTRL2EPN_link")
        instantiate("__merged_All_RAM", "Association", ("PW_Control/Transition", "Encapsulated_PetriNet/Transition"), ID="CTRL2EPN_tlink")
        instantiate("__merged_All_RAM", "Association", ("PW_Plant/State", "Encapsulated_PetriNet/Place"), ID="PLANT2EPN_link")
        instantiate("__merged_All_RAM", "Association", ("PW_Plant/Transition", "Encapsulated_PetriNet/Transition"), ID="PLANT2EPN_tlink")
        instantiate("__merged_All_RAM", "Association", ("PW_Environment/Event", "Encapsulated_PetriNet/Place"), ID="ENV2EPN_link")
        instantiate("__merged_All_RAM", "Association", ("Encapsulated_PetriNet/Place", "PetriNet/Place"), ID="EPN2PN_transition_link")
        instantiate("__merged_All_RAM", "Association", ("Encapsulated_PetriNet/Transition", "PetriNet/Transition"), ID="EPN2PN_place_link")
        transformation_RAMify("__merged_All_RAM", "All_RAM")

        transformation_add_MANUAL(["Requirements"], ["Requirements"], "revise_req")
        transformation_add_MANUAL(["Requirements", "PW_Environment"], ["PW_Environment"], "revise_environment")
        transformation_add_MANUAL(["Requirements", "PW_Plant"], ["PW_Plant"], "revise_plant")
        transformation_add_MANUAL(["Requirements", "PW_Control"], ["PW_Control"], "revise_control")
        transformation_add_MANUAL(["Requirements", "Query"], ["Query"], "revise_query")
        transformation_add_MANUAL(["Requirements", "Architecture"], ["Architecture"], "revise_architecture")

        transformation_add_MT("All_RAM", [], ["PW_Plant", "PW_Environment", "PW_Control", "Query", "Architecture", "Requirements"], "make_initial_models", open("models/initialize.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["PW_Plant"], ["Encapsulated_PetriNet"], "plant_to_EPN", open("models/plant_to_EPN.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["PW_Control"], ["Encapsulated_PetriNet"], "control_to_EPN", open("models/control_to_EPN.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["PW_Environment"], ["Encapsulated_PetriNet"], "environment_to_EPN", open("models/environment_to_EPN.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["Encapsulated_PetriNet", "Architecture"], ["Encapsulated_PetriNet"], "combine_EPN", open("models/combine_EPN.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["ReachabilityGraph", "Query"], [], "match", open("models/matches.mvc", 'r').read())

        transformation_add_AL(["PetriNet"], ["ReachabilityGraph"], "reachability", open("models/reachability.alc", 'r').read())
        transformation_add_AL(["ReachabilityGraph"], [], "bfs", open("models/bfs.alc", 'r').read())

        global called
        called = 0

        def cb_req():
            global called
            if called > len(callbacks):
                raise Exception("Seemingly called some operation twice!")
            else:
                called += 1
            model_overwrite(None, open("models/requirements_model.mvc", 'r').read())

        def cb_plant():
            global called
            if called > len(callbacks):
                raise Exception("Seemingly called some operation twice!")
            else:
                called += 1
            model_overwrite(None, open("models/plant_model.mvc", 'r').read())

        def cb_env():
            global called
            if called > len(callbacks):
                raise Exception("Seemingly called some operation twice!")
            else:
                called += 1
            model_overwrite(None, open("models/environment_model.mvc", 'r').read())

        def cb_ctrl():
            global called
            if called > len(callbacks):
                raise Exception("Seemingly called some operation twice!")
            else:
                called += 1
            model_overwrite(None, open("models/control_model.mvc", 'r').read())

        def cb_query():
            global called
            if called > len(callbacks):
                raise Exception("Seemingly called some operation twice!")
            else:
                called += 1
            model_overwrite(None, open("models/query_model.mvc", 'r').read())

        def cb_arch():
            global called
            if called > len(callbacks):
                raise Exception("Seemingly called some operation twice!")
            else:
                called += 1
            model_overwrite(None, open("models/architecture_model.mvc", 'r').read())

        callbacks = {
                "revise_req": cb_req,
                "revise_plant": cb_plant,
                "revise_environment": cb_env,
                "revise_control": cb_ctrl,
                "revise_query": cb_query,
                "revise_architecture": cb_arch,
            }

        process_execute("pm_powerwindow", "pm_", callbacks)

    @slow
    def test_process_powerwindow_debug(self):
        model_add("ReachabilityGraph", "SimpleClassDiagrams", open("models/reachability_graph.mvc", "r").read())
        model_add("PetriNet", "SimpleClassDiagrams", open("integration/code/pn_design.mvc", 'r').read())
        model_add("Encapsulated_PetriNet", "SimpleClassDiagrams", open("models/petrinet_ports.mvc", 'r').read())

        model_add("PW_Plant", "SimpleClassDiagrams", open("models/plant_PW.mvc", 'r').read())
        model_add("PW_Environment", "SimpleClassDiagrams", open("models/environment_PW.mvc", 'r').read())
        model_add("PW_Control", "SimpleClassDiagrams", open("models/control_PW.mvc", 'r').read())
        model_add("Requirements", "SimpleClassDiagrams", open("models/requirements.mvc", 'r').read())
        model_add("Query", "SimpleClassDiagrams", open("models/query.mvc", 'r').read())
        model_add("Architecture", "SimpleClassDiagrams", open("models/architecture.mvc", 'r').read())

        model_add("pm_powerwindow", "ProcessModel", open("models/pm_req_analyse.mvc", 'r').read())

        assert model_list() == set([\
                ("ReachabilityGraph", "SimpleClassDiagrams"),
                ("PetriNet", "SimpleClassDiagrams"),
                ("Encapsulated_PetriNet", "SimpleClassDiagrams"),
                ("PW_Plant", "SimpleClassDiagrams"),
                ("PW_Environment", "SimpleClassDiagrams"),
                ("PW_Control", "SimpleClassDiagrams"),
                ("Requirements", "SimpleClassDiagrams"),
                ("Query", "SimpleClassDiagrams"),
                ("Architecture", "SimpleClassDiagrams"),
                ("pm_powerwindow", "ProcessModel"),
                ("SimpleClassDiagrams", "SimpleClassDiagrams"),
                ("CoreFormalism", "SimpleClassDiagrams"),
                ("ManualOperation", "SimpleClassDiagrams"),
                ("ActionLanguage", "SimpleClassDiagrams"),
                ("ProcessModel", "SimpleClassDiagrams"),
                ("Tracability", "SimpleClassDiagrams"),
                ("core", "CoreFormalism"),])

        transformation_add_MT_language(["PetriNet", "Encapsulated_PetriNet", "PW_Plant", "PW_Control", "PW_Environment", "Requirements", "ReachabilityGraph", "Query", "Architecture"], "All_RAM")
        instantiate("__merged_All_RAM", "Association", ("PW_Control/State", "Encapsulated_PetriNet/Place"), ID="CTRL2EPN_link")
        instantiate("__merged_All_RAM", "Association", ("PW_Control/Transition", "Encapsulated_PetriNet/Transition"), ID="CTRL2EPN_tlink")
        instantiate("__merged_All_RAM", "Association", ("PW_Plant/State", "Encapsulated_PetriNet/Place"), ID="PLANT2EPN_link")
        instantiate("__merged_All_RAM", "Association", ("PW_Plant/Transition", "Encapsulated_PetriNet/Transition"), ID="PLANT2EPN_tlink")
        instantiate("__merged_All_RAM", "Association", ("PW_Environment/Event", "Encapsulated_PetriNet/Place"), ID="ENV2EPN_link")
        instantiate("__merged_All_RAM", "Association", ("Encapsulated_PetriNet/Place", "PetriNet/Place"), ID="EPN2PN_transition_link")
        instantiate("__merged_All_RAM", "Association", ("Encapsulated_PetriNet/Transition", "PetriNet/Transition"), ID="EPN2PN_place_link")
        transformation_RAMify("__merged_All_RAM", "All_RAM")

        transformation_add_MANUAL(["Requirements"], ["Requirements"], "revise_req")
        transformation_add_MANUAL(["Requirements", "PW_Environment"], ["PW_Environment"], "revise_environment")
        transformation_add_MANUAL(["Requirements", "PW_Plant"], ["PW_Plant"], "revise_plant")
        transformation_add_MANUAL(["Requirements", "PW_Control"], ["PW_Control"], "revise_control")
        transformation_add_MANUAL(["Requirements", "Query"], ["Query"], "revise_query")
        transformation_add_MANUAL(["Requirements", "Architecture"], ["Architecture"], "revise_architecture")

        transformation_add_MT("All_RAM", [], ["PW_Plant", "PW_Environment", "PW_Control", "Query", "Architecture", "Requirements"], "make_initial_models", open("models/initialize.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["PW_Plant"], ["Encapsulated_PetriNet"], "plant_to_EPN", open("models/plant_to_EPN.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["PW_Control"], ["Encapsulated_PetriNet"], "control_to_EPN", open("models/control_to_EPN.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["PW_Environment"], ["Encapsulated_PetriNet"], "environment_to_EPN", open("models/environment_to_EPN.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["Encapsulated_PetriNet", "Architecture"], ["Encapsulated_PetriNet"], "combine_EPN", open("models/combine_EPN.mvc", 'r').read())
        transformation_add_MT("All_RAM", ["ReachabilityGraph", "Query"], [], "match", open("models/matches.mvc", 'r').read())

        transformation_add_AL(["PetriNet"], ["ReachabilityGraph"], "reachability", open("models/reachability.alc", 'r').read())
        transformation_add_AL(["ReachabilityGraph"], [], "bfs", open("models/bfs.alc", 'r').read())

        global called
        called = 0

        def cb_req():
            global called
            if called > len(callbacks) * 2:
                raise Exception("Seemingly called some operation twice!")
            if called < len(callbacks):
                model_overwrite(None, open("models/requirements_model.mvc", 'r').read())
            called += 1

        def cb_plant():
            global called
            if called > len(callbacks) * 2:
                raise Exception("Seemingly called some operation twice!")
            if called < len(callbacks):
                model_overwrite(None, open("models/plant_model.mvc", 'r').read())
            called += 1

        def cb_env():
            global called
            if called > len(callbacks) * 2:
                raise Exception("Seemingly called some operation twice!")
            if called < len(callbacks):
                model_overwrite(None, open("models/environment_model.mvc", 'r').read())
            called += 1

        def cb_ctrl():
            global called
            if called > len(callbacks) * 2:
                raise Exception("Seemingly called some operation twice!")
            if called < len(callbacks):
                model_overwrite(None, open("models/control_model_wrong.mvc", 'r').read())
            else:
                model_overwrite(None, open("models/control_model.mvc", 'r').read())
            called += 1

        def cb_query():
            global called
            if called > len(callbacks) * 2:
                raise Exception("Seemingly called some operation twice!")
            if called < len(callbacks):
                model_overwrite(None, open("models/query_model.mvc", 'r').read())
            called += 1

        def cb_arch():
            global called
            if called > len(callbacks) * 2:
                raise Exception("Seemingly called some operation twice!")
            if called < len(callbacks):
                model_overwrite(None, open("models/architecture_model.mvc", 'r').read())
            called += 1

        callbacks = {
                "revise_req": cb_req,
                "revise_plant": cb_plant,
                "revise_environment": cb_env,
                "revise_control": cb_ctrl,
                "revise_query": cb_query,
                "revise_architecture": cb_arch,
            }

        process_execute("pm_powerwindow", "pm_", callbacks)

        if called != 12:
            raise Exception("Not iterated twice!")