modelverse/integration/test_powerwindow.py

import unittest

from utils import *
import sys
import os
import shutil

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

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("formalisms/ReachabilityGraph", "formalisms/SimpleClassDiagrams", open("models/reachability_graph.mvc", "r").read())
        model_add("formalisms/PetriNet", "formalisms/SimpleClassDiagrams", open("integration/code/pn_design.mvc", 'r').read())
        model_add("formalisms/Encapsulated_PetriNet", "formalisms/SimpleClassDiagrams", open("models/petrinet_ports.mvc", 'r').read())
        model_add("formalisms/PW_Plant", "formalisms/SimpleClassDiagrams", open("models/plant_PW.mvc", 'r').read())
        model_add("formalisms/PW_Environment", "formalisms/SimpleClassDiagrams", open("models/environment_PW.mvc", 'r').read())
        model_add("formalisms/PW_Control", "formalisms/SimpleClassDiagrams", open("models/control_PW.mvc", 'r').read())
        model_add("formalisms/Requirements", "formalisms/SimpleClassDiagrams", open("models/requirements.mvc", 'r').read())
        model_add("formalisms/Query", "formalisms/SimpleClassDiagrams", open("models/query.mvc", 'r').read())
        model_add("formalisms/Architecture", "formalisms/SimpleClassDiagrams", open("models/architecture.mvc", 'r').read())

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

        transformation_add_MANUAL({"Requirements": "formalisms/Requirements"}, {"Requirements": "formalisms/Requirements"}, "models/revise_req")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "PW_Environment": "formalisms/PW_Environment"}, {"PW_Environment": "formalisms/PW_Environment"}, "models/revise_environment")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "PW_Plant": "formalisms/PW_Plant"}, {"PW_Plant": "formalisms/PW_Plant"}, "models/revise_plant")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "PW_Control": "formalisms/PW_Control"}, {"PW_Control": "formalisms/PW_Control"}, "models/revise_control")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "Query": "formalisms/Query"}, {"Query": "formalisms/Query"}, "models/revise_query")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "Architecture": "formalisms/Architecture"}, {"Architecture": "formalisms/Architecture"}, "models/revise_architecture")

        def tracability_CTRL2EPN(model):
            instantiate(model, "Association", ("PW_Control/State", "Encapsulated_PetriNet/Place"), ID="CTRL2EPN_link")
            instantiate(model, "Association", ("PW_Control/Transition", "Encapsulated_PetriNet/Transition"), ID="CTRL2EPN_tlink")

        def tracability_PLANT2EPN(model):
            instantiate(model, "Association", ("PW_Plant/State", "Encapsulated_PetriNet/Place"), ID="PLANT2EPN_link")
            instantiate(model, "Association", ("PW_Plant/Transition", "Encapsulated_PetriNet/Transition"), ID="PLANT2EPN_tlink")

        def tracability_ENV2EPN(model):
            instantiate(model, "Association", ("PW_Environment/Event", "Encapsulated_PetriNet/Place"), ID="ENV2EPN_link")

        def tracability_EPN2PN(model):
            instantiate(model, "Association", ("Encapsulated_PetriNet/Place", "PetriNet/Place"), ID="EPN2PN_transition_link")
            instantiate(model, "Association", ("Encapsulated_PetriNet/Transition", "PetriNet/Transition"), ID="EPN2PN_place_link")

        transformation_add_MT({}, {"PW_Plant": "formalisms/PW_Plant", "PW_Environment": "formalisms/PW_Environment", "PW_Control": "formalisms/PW_Control", "Query": "formalisms/Query", "Architecture": "formalisms/Architecture", "Requirements": "formalisms/Requirements"}, "models/make_initial_models", open("models/initialize.mvc", 'r').read())
        transformation_add_MT({"PW_Plant": "formalisms/PW_Plant"}, {"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet"}, "models/plant_to_EPN", open("models/plant_to_EPN.mvc", 'r').read(), tracability_PLANT2EPN)
        transformation_add_MT({"PW_Control": "formalisms/PW_Control"}, {"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet"}, "models/control_to_EPN", open("models/control_to_EPN.mvc", 'r').read(), tracability_CTRL2EPN)
        transformation_add_MT({"PW_Environment": "formalisms/PW_Environment"}, {"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet"}, "models/environment_to_EPN", open("models/environment_to_EPN.mvc", 'r').read(), tracability_ENV2EPN)
        transformation_add_MT({"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet", "Architecture": "formalisms/Architecture"}, {"PetriNet": "formalisms/PetriNet"}, "models/combine_EPN", open("models/combine_EPN.mvc", 'r').read(), tracability_EPN2PN)
        transformation_add_MT({"ReachabilityGraph": "formalisms/ReachabilityGraph", "Query": "formalisms/Query"}, {"ReachabilityGraph": "formalisms/ReachabilityGraph"}, "models/match", open("models/matches.mvc", 'r').read())

        transformation_add_AL({"PetriNet": "formalisms/PetriNet"}, {"ReachabilityGraph": "formalisms/ReachabilityGraph"}, "models/reachability", open("models/reachability.alc", 'r').read())
        transformation_add_AL({"ReachabilityGraph": "formalisms/ReachabilityGraph"}, {}, "models/bfs", open("models/bfs.alc", 'r').read())
        transformation_add_AL({"EPN_Plant": "formalisms/Encapsulated_PetriNet", "EPN_Control": "formalisms/Encapsulated_PetriNet", "EPN_Environment": "formalisms/Encapsulated_PetriNet"}, {"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet"}, "models/merge_EPN", open("models/merge_EPN.alc", 'r').read())

        nr_of_operations = 6

        try:
            shutil.rmtree(".TEST_POWER_WINDOW")
        except:
            pass
        os.makedirs(".TEST_POWER_WINDOW")

        def get_function(filename):
            def func(model):
                import uuid
                import os
                called = len(os.listdir(".TEST_POWER_WINDOW/"))
                if called > nr_of_operations:
                    raise Exception("Seemingly called some operation twice!")
                else:
                    f = open(".TEST_POWER_WINDOW/%s" % str(uuid.uuid4()), 'w')
                    f.write(filename)
                    f.close()
                model_overwrite(model, open(filename, "r").read())
            return func

        cb_req = get_function("models/requirements_model.mvc")
        cb_plant = get_function("models/plant_model.mvc")
        cb_env = get_function("models/environment_model.mvc")
        cb_ctrl = get_function("models/control_model.mvc")
        cb_query = get_function("models/query_model.mvc")
        cb_arch = get_function("models/architecture_model.mvc")

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

        try:
            process_execute("models/pm_powerwindow", {}, callbacks)
        except:
            import traceback
            print(traceback.format_exc())

        called = len(os.listdir(".TEST_POWER_WINDOW/"))
        shutil.rmtree(".TEST_POWER_WINDOW")
        if called != 6:
            print(called)
            raise Exception("Not executed sufficiently:" + str(called))

    @slow
    def test_process_powerwindow_debug(self):
        model_add("formalisms/ReachabilityGraph", "formalisms/SimpleClassDiagrams", open("models/reachability_graph.mvc", "r").read())
        model_add("formalisms/PetriNet", "formalisms/SimpleClassDiagrams", open("integration/code/pn_design.mvc", 'r').read())
        model_add("formalisms/Encapsulated_PetriNet", "formalisms/SimpleClassDiagrams", open("models/petrinet_ports.mvc", 'r').read())
        model_add("formalisms/PW_Plant", "formalisms/SimpleClassDiagrams", open("models/plant_PW.mvc", 'r').read())
        model_add("formalisms/PW_Environment", "formalisms/SimpleClassDiagrams", open("models/environment_PW.mvc", 'r').read())
        model_add("formalisms/PW_Control", "formalisms/SimpleClassDiagrams", open("models/control_PW.mvc", 'r').read())
        model_add("formalisms/Requirements", "formalisms/SimpleClassDiagrams", open("models/requirements.mvc", 'r').read())
        model_add("formalisms/Query", "formalisms/SimpleClassDiagrams", open("models/query.mvc", 'r').read())
        model_add("formalisms/Architecture", "formalisms/SimpleClassDiagrams", open("models/architecture.mvc", 'r').read())

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

        transformation_add_MANUAL({"Requirements": "formalisms/Requirements"}, {"Requirements": "formalisms/Requirements"}, "models/revise_req")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "PW_Environment": "formalisms/PW_Environment"}, {"PW_Environment": "formalisms/PW_Environment"}, "models/revise_environment")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "PW_Plant": "formalisms/PW_Plant"}, {"PW_Plant": "formalisms/PW_Plant"}, "models/revise_plant")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "PW_Control": "formalisms/PW_Control"}, {"PW_Control": "formalisms/PW_Control"}, "models/revise_control")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "Query": "formalisms/Query"}, {"Query": "formalisms/Query"}, "models/revise_query")
        transformation_add_MANUAL({"Requirements": "formalisms/Requirements", "Architecture": "formalisms/Architecture"}, {"Architecture": "formalisms/Architecture"}, "models/revise_architecture")

        def tracability_CTRL2EPN(model):
            instantiate(model, "Association", ("PW_Control/State", "Encapsulated_PetriNet/Place"), ID="CTRL2EPN_link")
            instantiate(model, "Association", ("PW_Control/Transition", "Encapsulated_PetriNet/Transition"), ID="CTRL2EPN_tlink")

        def tracability_PLANT2EPN(model):
            instantiate(model, "Association", ("PW_Plant/State", "Encapsulated_PetriNet/Place"), ID="PLANT2EPN_link")
            instantiate(model, "Association", ("PW_Plant/Transition", "Encapsulated_PetriNet/Transition"), ID="PLANT2EPN_tlink")

        def tracability_ENV2EPN(model):
            instantiate(model, "Association", ("PW_Environment/Event", "Encapsulated_PetriNet/Place"), ID="ENV2EPN_link")

        def tracability_EPN2PN(model):
            instantiate(model, "Association", ("Encapsulated_PetriNet/Place", "PetriNet/Place"), ID="EPN2PN_transition_link")
            instantiate(model, "Association", ("Encapsulated_PetriNet/Transition", "PetriNet/Transition"), ID="EPN2PN_place_link")

        transformation_add_MT({}, {"PW_Plant": "formalisms/PW_Plant", "PW_Environment": "formalisms/PW_Environment", "PW_Control": "formalisms/PW_Control", "Query": "formalisms/Query", "Architecture": "formalisms/Architecture", "Requirements": "formalisms/Requirements"}, "models/make_initial_models", open("models/initialize.mvc", 'r').read())
        transformation_add_MT({"PW_Plant": "formalisms/PW_Plant"}, {"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet"}, "models/plant_to_EPN", open("models/plant_to_EPN.mvc", 'r').read(), tracability_PLANT2EPN)
        transformation_add_MT({"PW_Control": "formalisms/PW_Control"}, {"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet"}, "models/control_to_EPN", open("models/control_to_EPN.mvc", 'r').read(), tracability_CTRL2EPN)
        transformation_add_MT({"PW_Environment": "formalisms/PW_Environment"}, {"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet"}, "models/environment_to_EPN", open("models/environment_to_EPN.mvc", 'r').read(), tracability_ENV2EPN)
        transformation_add_MT({"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet", "Architecture": "formalisms/Architecture"}, {"PetriNet": "formalisms/PetriNet"}, "models/combine_EPN", open("models/combine_EPN.mvc", 'r').read(), tracability_EPN2PN)
        transformation_add_MT({"ReachabilityGraph": "formalisms/ReachabilityGraph", "Query": "formalisms/Query"}, {"ReachabilityGraph": "formalisms/ReachabilityGraph"}, "models/match", open("models/matches.mvc", 'r').read())

        transformation_add_AL({"PetriNet": "formalisms/PetriNet"}, {"ReachabilityGraph": "formalisms/ReachabilityGraph"}, "models/reachability", open("models/reachability.alc", 'r').read())
        transformation_add_AL({"ReachabilityGraph": "formalisms/ReachabilityGraph"}, {}, "models/bfs", open("models/bfs.alc", 'r').read())
        transformation_add_AL({"EPN_Plant": "formalisms/Encapsulated_PetriNet", "EPN_Control": "formalisms/Encapsulated_PetriNet", "EPN_Environment": "formalisms/Encapsulated_PetriNet"}, {"Encapsulated_PetriNet": "formalisms/Encapsulated_PetriNet"}, "models/merge_EPN", open("models/merge_EPN.alc", 'r').read())

        #transformation_add_AL({"Encapsulated_PetriNet": "Encapsulated_PetriNet"}, {}, "epn_print", open("models/epn_print.alc").read())
        #transformation_add_AL({"PetriNet": "PetriNet"}, {}, "pn_print", open("models/pn_print.alc").read())

        try:
            shutil.rmtree(".TEST_POWER_WINDOW")
        except:
            pass
        os.makedirs(".TEST_POWER_WINDOW")

        executed = set([])

        def get_function(filename, fixed=None):
            def func(model):
                import uuid
                import os
                import glob
                nr_operations = 11
                called = len(os.listdir(".TEST_POWER_WINDOW/"))
                if (len(os.listdir(".TEST_POWER_WINDOW")) > nr_operations):
                    raise Exception("Seemingly called some operation twice!")
                elif (len(glob.glob(".TEST_POWER_WINDOW/%s_*" % str(filename.rsplit("/", 1)[-1]))) == 0) or fixed is None:
                    model_overwrite(model, open(filename, "r").read())
                else:
                    model_overwrite(model, open(fixed, "r").read())

                open(".TEST_POWER_WINDOW/%s_%s" % (filename.rsplit("/", 1)[-1], uuid.uuid4()), 'w').close()
            return func

        cb_req = get_function("models/requirements_model.mvc")
        cb_plant = get_function("models/plant_model.mvc")
        cb_env = get_function("models/environment_model.mvc")
        cb_ctrl = get_function("models/control_model_wrong.mvc", fixed="models/control_model.mvc")
        cb_query = get_function("models/query_model.mvc")
        cb_arch = get_function("models/architecture_model.mvc")

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

        callbacks = {
                "models/revise_req": cb_req,
                "models/revise_plant": cb_plant,
                "models/revise_environment": cb_env,
                "models/revise_control": cb_ctrl,
                "models/revise_query": cb_query,
                "models/revise_architecture": cb_arch,
                "models/bfs": (ctrl, "inp", "outp"),
            }

        process_execute("models/pm_powerwindow", {}, callbacks)

        thrd.join()

        called = len(os.listdir(".TEST_POWER_WINDOW"))
        shutil.rmtree(".TEST_POWER_WINDOW")

        if called != 11:
            print(called)
            raise Exception("Not executed sufficiently:" + str(called))

        assert len(error_path) == 10