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", [])
        self.mvk = ModelverseKernel(self.root)

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

        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", [])])
        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"]), task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"]), task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])]), "init")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), func)
        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"]), evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"]), inst)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [evalstack, "phase"])]), "call")

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

        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", [])])
        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"]), task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"]), task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])]), "init")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), func)
        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"]), evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"]), inst)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "phase"]), params)

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

        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", [])])
        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"]), task_root)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])]), "finish")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), inst)
        new_frame = self.mvs.execute("RD", [task_root, "frame"])
        self.assertNotEquals(new_frame, None)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"]), new_frame)
        new_evalstack = self.mvs.execute("RD", [new_frame, "evalstack"])
        self.assertNotEquals(new_evalstack, None)
        new_symbols = self.mvs.execute("RD", [new_frame, "symbols"])
        self.assertNotEquals(new_symbols, None)
        new_returnvalue = self.mvs.execute("RD", [new_frame, "returnvalue"])
        self.assertNotEquals(new_returnvalue, None)
        new_caller = self.mvs.execute("RD", [new_frame, "caller"])
        self.assertNotEquals(new_caller, None)
        new_phase = self.mvs.execute("RD", [new_frame, "phase"])
        self.assertNotEquals(new_phase, None)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [new_frame, "phase"])]), "init")
        new_IP = self.mvs.execute("RD", [new_frame, "IP"])
        self.assertNotEquals(new_IP, None)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "IP"]), body)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "caller"]), inst)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "prev"]), task_frame)

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

        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])
        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"]), task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"]), new_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])]), "finish")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), inst)
        new_frame = self.mvs.execute("RD", [task_root, "frame"])
        self.assertNotEquals(new_frame, None)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "IP"]), body)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "caller"]), inst)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "prev"]), task_frame)

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

        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", [])])
        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"]), task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"]), task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])]), "init")
        new_frame = self.mvs.execute("RRD", [task_frame, "prev"])
        self.assertNotEquals(new_frame, None)
        new_frame = new_frame[0]
        new_evalstack = self.mvs.execute("RD", [new_frame, "evalstack"])
        self.assertNotEquals(new_evalstack, None)
        new_symbols = self.mvs.execute("RD", [new_frame, "symbols"])
        self.assertNotEquals(new_symbols, None)
        new_returnvalue = self.mvs.execute("RD", [new_frame, "returnvalue"])
        self.assertNotEquals(new_returnvalue, None)
        new_caller = self.mvs.execute("RD", [new_frame, "caller"])
        self.assertNotEquals(new_caller, None)
        new_phase = self.mvs.execute("RD", [new_frame, "phase"])
        self.assertNotEquals(new_phase, None)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [new_frame, "phase"])]), "init")
        new_IP = self.mvs.execute("RD", [new_frame, "IP"])
        self.assertNotEquals(new_IP, None)
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), param_value)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "caller"]), inst)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "prev"]), task_frame)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"]), inst)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "phase"]), next_param)

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

        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"]), task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"]), task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])]), "init")
        new_frame = self.mvs.execute("RRD", [task_frame, "prev"])
        self.assertNotEquals(new_frame, None)
        new_frame = new_frame[0]
        new_evalstack = self.mvs.execute("RD", [new_frame, "evalstack"])
        self.assertNotEquals(new_evalstack, None)
        new_symbols = self.mvs.execute("RD", [new_frame, "symbols"])
        self.assertNotEquals(new_symbols, None)
        new_returnvalue = self.mvs.execute("RD", [new_frame, "returnvalue"])
        self.assertNotEquals(new_returnvalue, None)
        new_caller = self.mvs.execute("RD", [new_frame, "caller"])
        self.assertNotEquals(new_caller, None)
        new_phase = self.mvs.execute("RD", [new_frame, "phase"])
        self.assertNotEquals(new_phase, None)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [new_frame, "phase"])]), "init")
        new_IP = self.mvs.execute("RD", [new_frame, "IP"])
        self.assertNotEquals(new_IP, None)
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), param_value)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "caller"]), inst)
        self.assertEquals(self.mvs.execute("RD", [new_frame, "prev"]), task_frame)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"]), inst)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [evalstack, "phase"])]), "call")

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

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

        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])
        self.mvs.execute("CE", [t1, formal_parameter_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"]), task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"]), task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])]), "init")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), param_value)
        var = self.mvs.execute("RDN", [symbols, formal_parameter])
        self.assertNotEquals(var, None)
        self.assertEquals(self.mvs.execute("RD", [var, "value"]), returnvalue)
        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"]), evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"]), inst)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [evalstack, "phase"])]), "call")

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

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

        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", [])])
        self.mvs.execute("CD", [inst, "last_param", self.mvs.execute("CN", [])])
        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])
        self.mvs.execute("CE", [t1, formal_parameter_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"]), task_root)
        self.assertEquals(self.mvs.execute("RD", [task_root, "frame"]), task_frame)
        self.assertEquals(self.mvs.execute("RV", [self.mvs.execute("RD", [task_frame, "phase"])]), "init")
        self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), param_value)
        new_evalstack = self.mvs.execute("RD", [task_frame, "evalstack"])
        self.assertNotEquals(evalstack, new_evalstack)
        self.assertEquals(self.mvs.execute("RD", [new_evalstack, "prev"]), evalstack)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "inst"]), inst)
        self.assertEquals(self.mvs.execute("RD", [evalstack, "phase"]), param_c)
        var = self.mvs.execute("RDN", [symbols, formal_parameter])
        self.assertNotEquals(var, None)
        self.assertEquals(self.mvs.execute("RD", [var, "value"]), returnvalue)