modelverse/kernel/test/rules/test_rules_call.py

import unittest
from modelverse_kernel.main import ModelverseKernel
from utils import execute_until_finished, MvSWrapper

class Testcall(unittest.TestCase):
    def setUp(self):
        self.mvs = MvSWrapper()
        self.root = self.mvs.execute("RR", [])[0]
        self.mvk = ModelverseKernel(self.root)

    def test_call_resolve_none(self):
        root = self.root
        task_root = self.mvs.execute("CN", [])[0]
        task_frame = self.mvs.execute("CN", [])[0]
        phase = self.mvs.execute("CNV", ["init"])[0]
        inst = self.mvs.execute("CNV", [{"value":"call"}])[0]
        evalstack = self.mvs.execute("CN", [])[0]
        func = self.mvs.execute("CN", [])[0]

        self.mvs.execute("CD", [root, "task_1", task_root])
        self.mvs.execute("CD", [task_root, "frame", task_frame])
        self.mvs.execute("CD", [task_frame, "phase", phase])
        self.mvs.execute("CD", [task_frame, "symbols", self.mvs.execute("CN", [])[0]])
        self.mvs.execute("CD", [task_frame, "IP", inst])
        self.mvs.execute("CD", [inst, "func", func])
        self.mvs.execute("CD", [task_frame, "evalstack", evalstack])

        # Do the invocation until it returns None
        execute_until_finished(self.mvk, self.mvs)

        # Execution of the command finished, so the MvS should be in the correct new state right now
        self.assertEquals(self.mvs.execute("RD", [root, "task_1"])[0], task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"])[0], task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])[0]])[0], "init")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"])[0], func)
        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])[0]
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"])[0], evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"])[0], inst)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [evalstack, "phase"])[0]])[0], "call")

    def test_call_resolve_params(self):
        root = self.root
        task_root = self.mvs.execute("CN", [])[0]
        task_frame = self.mvs.execute("CN", [])[0]
        phase = self.mvs.execute("CNV", ["init"])[0]
        inst = self.mvs.execute("CNV", [{"value":"call"}])[0]
        evalstack = self.mvs.execute("CN", [])[0]
        func = self.mvs.execute("CN", [])[0]
        params = self.mvs.execute("CN", [])[0]

        self.mvs.execute("CD", [root, "task_1", task_root])
        self.mvs.execute("CD", [task_root, "frame", task_frame])
        self.mvs.execute("CD", [task_frame, "phase", phase])
        self.mvs.execute("CD", [task_frame, "IP", inst])
        self.mvs.execute("CD", [task_frame, "symbols", self.mvs.execute("CN", [])[0]])
        self.mvs.execute("CD", [inst, "func", func])
        self.mvs.execute("CD", [task_frame, "evalstack", evalstack])
        self.mvs.execute("CD", [inst, "params", params])

        # Do the invocation until it returns None
        execute_until_finished(self.mvk, self.mvs)

        # Execution of the command finished, so the MvS should be in the correct new state right now
        self.assertEquals(self.mvs.execute("RD", [root, "task_1"])[0], task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"])[0], task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])[0]])[0], "init")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"])[0], func)
        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])[0]
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"])[0], evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"])[0], inst)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "phase"])[0], params)

    def test_call_execute_none(self):
        root = self.root
        task_root = self.mvs.execute("CN", [])[0]
        task_frame = self.mvs.execute("CN", [])[0]
        phase = self.mvs.execute("CNV", ["call"])[0]
        inst = self.mvs.execute("CNV", [{"value":"call"}])[0]
        returnvalue = self.mvs.execute("CN", [])[0]
        body = self.mvs.execute("CN", [])[0]

        self.mvs.execute("CD", [root, "task_1", task_root])
        self.mvs.execute("CD", [task_root, "frame", task_frame])
        self.mvs.execute("CD", [task_frame, "phase", phase])
        self.mvs.execute("CD", [task_frame, "IP", inst])
        self.mvs.execute("CD", [task_frame, "symbols", self.mvs.execute("CN", [])[0]])
        self.mvs.execute("CD", [task_frame, "returnvalue", returnvalue])
        self.mvs.execute("CD", [returnvalue, "body", body])

        # Do the invocation until it returns None
        execute_until_finished(self.mvk, self.mvs)

        # Execution of the command finished, so the MvS should be in the correct new state right now
        self.assertEquals(self.mvs.execute("RD", [root, "task_1"])[0], task_root)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])[0]])[0], "finish")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"])[0], inst)
        new_frame = self.mvs.execute("RD", [task_root, "frame"])[0]
        self.assertNotEquals(new_frame, None)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"])[0], new_frame)
        new_evalstack = self.mvs.execute("RD", [new_frame, "evalstack"])[0]
        self.assertNotEquals(new_evalstack, None)
        new_symbols = self.mvs.execute("RD", [new_frame, "symbols"])[0]
        self.assertNotEquals(new_symbols, None)
        new_returnvalue = self.mvs.execute("RD", [new_frame, "returnvalue"])[0]
        self.assertNotEquals(new_returnvalue, None)
        new_caller = self.mvs.execute("RD", [new_frame, "caller"])[0]
        self.assertNotEquals(new_caller, None)
        new_phase = self.mvs.execute("RD", [new_frame, "phase"])[0]
        self.assertNotEquals(new_phase, None)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [new_frame, "phase"])[0]])[0], "init")
        new_IP = self.mvs.execute("RD", [new_frame, "IP"])[0]
        self.assertNotEquals(new_IP, None)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "IP"])[0], body)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "caller"])[0], inst)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "prev"])[0], task_frame)

    def test_call_execute_params(self):
        root = self.root
        task_root = self.mvs.execute("CN", [])[0]
        task_frame = self.mvs.execute("CN", [])[0]
        phase = self.mvs.execute("CNV", ["call"])[0]
        inst = self.mvs.execute("CNV", [{"value":"call"}])[0]
        returnvalue = self.mvs.execute("CN", [])[0]
        body = self.mvs.execute("CN", [])[0]
        new_frame = self.mvs.execute("CN", [])[0]
        symbols = self.mvs.execute("CN", [])[0]
        signature = self.mvs.execute("CN", [])[0]
        params = self.mvs.execute("CN", [])[0]
        name = self.mvs.execute("CNV", ["param_a"])[0]
        last_param = self.mvs.execute("CN", [])[0]
        formal_parameter = self.mvs.execute("CN", [])[0]

        self.mvs.execute("CD", [root, "task_1", task_root])
        self.mvs.execute("CD", [task_root, "frame", task_frame])
        self.mvs.execute("CD", [task_frame, "phase", phase])
        self.mvs.execute("CD", [task_frame, "IP", inst])
        self.mvs.execute("CD", [task_frame, "returnvalue", returnvalue])
        self.mvs.execute("CD", [new_frame, "caller", inst])
        self.mvs.execute("CD", [new_frame, "prev", task_frame])
        self.mvs.execute("CD", [new_frame, "symbols", symbols])
        self.mvs.execute("CD", [signature, "body", body])
        self.mvs.execute("CD", [signature, "params", params])
        self.mvs.execute("CD", [new_frame, "IP", body])
        self.mvs.execute("CD", [inst, "last_param", last_param])
        self.mvs.execute("CD", [last_param, "name", name])

        t1 = self.mvs.execute("CE", [params, formal_parameter])[0]
        self.mvs.execute("CE", [t1, name])

        # Do the invocation until it returns None
        execute_until_finished(self.mvk, self.mvs)

        # Execution of the command finished, so the MvS should be in the correct new state right now
        self.assertEquals(self.mvs.execute("RD", [root, "task_1"])[0], task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"])[0], new_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])[0]])[0], "finish")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"])[0], inst)
        new_frame = self.mvs.execute("RD", [task_root, "frame"])[0]
        self.assertNotEquals(new_frame, None)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "IP"])[0], body)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "caller"])[0], inst)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "prev"])[0], task_frame)

    def test_call_param_first_multi(self):
        root = self.root
        task_root = self.mvs.execute("CN", [])[0]
        task_frame = self.mvs.execute("CN", [])[0]
        inst = self.mvs.execute("CNV", [{"value":"call"}])[0]
        returnvalue = self.mvs.execute("CN", [])[0]
        body = self.mvs.execute("CN", [])[0]
        new_frame = self.mvs.execute("CN", [])[0]
        symbols = self.mvs.execute("CN", [])[0]
        signature = self.mvs.execute("CN", [])[0]
        params = self.mvs.execute("CN", [])[0]
        param_value = self.mvs.execute("CN", [])[0]
        next_param = self.mvs.execute("CN", [])[0]
        evalstack = self.mvs.execute("CN", [])[0]

        self.mvs.execute("CD", [root, "task_1", task_root])
        self.mvs.execute("CD", [task_root, "frame", task_frame])
        self.mvs.execute("CD", [task_frame, "phase", params])
        self.mvs.execute("CD", [task_frame, "IP", inst])
        self.mvs.execute("CD", [task_frame, "returnvalue", returnvalue])
        self.mvs.execute("CD", [signature, "body", body])
        self.mvs.execute("CD", [signature, "params", params])
        self.mvs.execute("CD", [inst, "params", params])
        self.mvs.execute("CD", [inst, "last_param", self.mvs.execute("CN", [])[0]])
        self.mvs.execute("CD", [params, "next_param", next_param])
        self.mvs.execute("CD", [params, "value", param_value])
        self.mvs.execute("CD", [returnvalue, "body", body])
        self.mvs.execute("CD", [task_frame, "evalstack", evalstack])

        # Do the invocation until it returns None
        execute_until_finished(self.mvk, self.mvs)

        # Execution of the command finished, so the MvS should be in the correct new state right now
        self.assertEquals(self.mvs.execute("RD", [root, "task_1"])[0], task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"])[0], task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])[0]])[0], "init")
        new_frame = self.mvs.execute("RRD", [task_frame, "prev"])[0][0]
        self.assertNotEquals(new_frame, None)
        new_evalstack = self.mvs.execute("RD", [new_frame, "evalstack"])[0]
        self.assertNotEquals(new_evalstack, None)
        new_symbols = self.mvs.execute("RD", [new_frame, "symbols"])[0]
        self.assertNotEquals(new_symbols, None)
        new_returnvalue = self.mvs.execute("RD", [new_frame, "returnvalue"])[0]
        self.assertNotEquals(new_returnvalue, None)
        new_caller = self.mvs.execute("RD", [new_frame, "caller"])[0]
        self.assertNotEquals(new_caller, None)
        new_phase = self.mvs.execute("RD", [new_frame, "phase"])[0]
        self.assertNotEquals(new_phase, None)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [new_frame, "phase"])[0]])[0], "init")
        new_IP = self.mvs.execute("RD", [new_frame, "IP"])[0]
        self.assertNotEquals(new_IP, None)
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"])[0], param_value)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "caller"])[0], inst)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "prev"])[0], task_frame)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"])[0], inst)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "phase"])[0], next_param)

    def test_call_param_first_single(self):
        root = self.root
        task_root = self.mvs.execute("CN", [])[0]
        task_frame = self.mvs.execute("CN", [])[0]
        inst = self.mvs.execute("CNV", [{"value":"call"}])[0]
        returnvalue = self.mvs.execute("CN", [])[0]
        body = self.mvs.execute("CN", [])[0]
        new_frame = self.mvs.execute("CN", [])[0]
        symbols = self.mvs.execute("CN", [])[0]
        signature = self.mvs.execute("CN", [])[0]
        params = self.mvs.execute("CN", [])[0]
        param_value = self.mvs.execute("CN", [])[0]
        evalstack = self.mvs.execute("CN", [])[0]

        self.mvs.execute("CD", [root, "task_1", task_root])
        self.mvs.execute("CD", [task_root, "frame", task_frame])
        self.mvs.execute("CD", [task_frame, "phase", params])
        self.mvs.execute("CD", [task_frame, "IP", inst])
        self.mvs.execute("CD", [task_frame, "returnvalue", returnvalue])
        self.mvs.execute("CD", [signature, "body", body])
        self.mvs.execute("CD", [inst, "params", params])
        self.mvs.execute("CD", [inst, "last_param", params])
        self.mvs.execute("CD", [params, "value", param_value])
        self.mvs.execute("CD", [returnvalue, "body", body])
        self.mvs.execute("CD", [task_frame, "evalstack", evalstack])

        # Do the invocation until it returns None
        execute_until_finished(self.mvk, self.mvs)

        # Execution of the command finished, so the MvS should be in the correct new state right now
        self.assertEquals(self.mvs.execute("RD", [root, "task_1"])[0], task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"])[0], task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])[0]])[0], "init")
        new_frame = self.mvs.execute("RRD", [task_frame, "prev"])[0][0]
        self.assertNotEquals(new_frame, None)
        new_evalstack = self.mvs.execute("RD", [new_frame, "evalstack"])[0]
        self.assertNotEquals(new_evalstack, None)
        new_symbols = self.mvs.execute("RD", [new_frame, "symbols"])[0]
        self.assertNotEquals(new_symbols, None)
        new_returnvalue = self.mvs.execute("RD", [new_frame, "returnvalue"])[0]
        self.assertNotEquals(new_returnvalue, None)
        new_caller = self.mvs.execute("RD", [new_frame, "caller"])[0]
        self.assertNotEquals(new_caller, None)
        new_phase = self.mvs.execute("RD", [new_frame, "phase"])[0]
        self.assertNotEquals(new_phase, None)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [new_frame, "phase"])[0]])[0], "init")
        new_IP = self.mvs.execute("RD", [new_frame, "IP"])[0]
        self.assertNotEquals(new_IP, None)
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"])[0], param_value)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "caller"])[0], inst)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "prev"])[0], task_frame)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"])[0], inst)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [evalstack, "phase"])[0]])[0], "call")

        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])[0]
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"])[0], evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"])[0], inst)

    def test_params_last(self):
        root = self.root
        task_root = self.mvs.execute("CN", [])[0]
        task_frame = self.mvs.execute("CN", [])[0]
        inst = self.mvs.execute("CNV", [{"value":"call"}])[0]
        returnvalue = self.mvs.execute("CN", [])[0]
        body = self.mvs.execute("CN", [])[0]
        new_frame = self.mvs.execute("CN", [])[0]
        symbols = self.mvs.execute("CN", [])[0]
        signature = self.mvs.execute("CN", [])[0]
        params = self.mvs.execute("CN", [])[0]
        param_value = self.mvs.execute("CN", [])[0]
        evalstack = self.mvs.execute("CN", [])[0]
        last_param = self.mvs.execute("CN", [])[0]
        prev_param = self.mvs.execute("CN", [])[0]
        formal_parameter = self.mvs.execute("CN", [])[0]
        formal_parameter_name = self.mvs.execute("CNV", ["param_a"])[0]

        self.mvs.execute("CD", [root, "task_1", task_root])
        self.mvs.execute("CD", [task_root, "frame", task_frame])
        self.mvs.execute("CD", [task_frame, "phase", last_param])
        self.mvs.execute("CD", [task_frame, "IP", inst])
        self.mvs.execute("CD", [task_frame, "returnvalue", returnvalue])
        self.mvs.execute("CD", [inst, "last_param", last_param])
        self.mvs.execute("CD", [inst, "params", prev_param])
        self.mvs.execute("CD", [params, "value", param_value])
        self.mvs.execute("CD", [task_frame, "evalstack", evalstack])
        self.mvs.execute("CD", [new_frame, "caller", inst])
        self.mvs.execute("CD", [new_frame, "prev", task_frame])
        self.mvs.execute("CD", [new_frame, "symbols", symbols])
        self.mvs.execute("CD", [new_frame, "IP", body])
        self.mvs.execute("CD", [signature, "body", body])
        self.mvs.execute("CD", [signature, "params", params])
        self.mvs.execute("CD", [prev_param, "next_param", last_param])
        self.mvs.execute("CD", [prev_param, "name", formal_parameter_name])
        self.mvs.execute("CD", [last_param, "value", param_value])
        t1 = self.mvs.execute("CE", [params, formal_parameter])[0]
        self.mvs.execute("CE", [t1, formal_parameter_name])[0]

        # Do the invocation until it returns None
        execute_until_finished(self.mvk, self.mvs)

        # Execution of the command finished, so the MvS should be in the correct new state right now
        self.assertEquals(self.mvs.execute("RD", [root, "task_1"])[0], task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"])[0], task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])[0]])[0], "init")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"])[0], param_value)
        var = self.mvs.execute("RDN", [symbols, formal_parameter])[0]
        self.assertNotEquals(var, None)
        self.assertEquals(self.mvs.execute("RD", [var, "value"])[0], returnvalue)
        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])[0]
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"])[0], evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"])[0], inst)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [evalstack, "phase"])[0]])[0], "call")

    def test_params_next(self):
        root = self.root
        task_root = self.mvs.execute("CN", [])[0]
        task_frame = self.mvs.execute("CN", [])[0]
        inst = self.mvs.execute("CNV", [{"value":"call"}])[0]
        returnvalue = self.mvs.execute("CN", [])[0]
        body = self.mvs.execute("CN", [])[0]
        new_frame = self.mvs.execute("CN", [])[0]
        symbols = self.mvs.execute("CN", [])[0]
        signature = self.mvs.execute("CN", [])[0]
        params = self.mvs.execute("CN", [])[0]
        param_value = self.mvs.execute("CN", [])[0]
        evalstack = self.mvs.execute("CN", [])[0]
        formal_parameter = self.mvs.execute("CN", [])[0]
        formal_parameter_name = self.mvs.execute("CNV", ["param_a"])[0]

        param_a = self.mvs.execute("CN", [])[0]
        param_b = self.mvs.execute("CN", [])[0]
        param_c = self.mvs.execute("CN", [])[0]

        self.mvs.execute("CD", [root, "task_1", task_root])
        self.mvs.execute("CD", [task_root, "frame", task_frame])
        self.mvs.execute("CD", [task_frame, "phase", param_b])
        self.mvs.execute("CD", [task_frame, "IP", inst])
        self.mvs.execute("CD", [inst, "params", self.mvs.execute("CN", [])[0]])
        self.mvs.execute("CD", [inst, "last_param", self.mvs.execute("CN", [])[0]])
        self.mvs.execute("CD", [task_frame, "returnvalue", returnvalue])
        self.mvs.execute("CD", [params, "value", param_value])
        self.mvs.execute("CD", [task_frame, "evalstack", evalstack])
        self.mvs.execute("CD", [new_frame, "caller", inst])
        self.mvs.execute("CD", [new_frame, "prev", task_frame])
        self.mvs.execute("CD", [new_frame, "symbols", symbols])
        self.mvs.execute("CD", [new_frame, "IP", body])
        self.mvs.execute("CD", [signature, "body", body])
        self.mvs.execute("CD", [signature, "params", params])
        self.mvs.execute("CD", [param_a, "name", formal_parameter_name])
        self.mvs.execute("CD", [param_b, "value", param_value])
        self.mvs.execute("CD", [param_a, "next_param", param_b])
        self.mvs.execute("CD", [param_b, "next_param", param_c])
        t1 = self.mvs.execute("CE", [params, formal_parameter])[0]
        self.mvs.execute("CE", [t1, formal_parameter_name])[0]

        # Do the invocation until it returns None
        execute_until_finished(self.mvk, self.mvs)

        # Execution of the command finished, so the MvS should be in the correct new state right now
        self.assertEquals(self.mvs.execute("RD", [root, "task_1"])[0], task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"])[0], task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])[0]])[0], "init")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"])[0], param_value)
        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])[0]
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"])[0], evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"])[0], inst)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "phase"])[0], param_c)
        var = self.mvs.execute("RDN", [symbols, formal_parameter])[0]
        self.assertNotEquals(var, None)
        self.assertEquals(self.mvs.execute("RD", [var, "value"])[0], returnvalue)