12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273 |
- import unittest
- from modelverse_kernel.generated import ModelverseKernel
- from modelverse_kernel.primitives import SleepKernel
- from test_generated.utils import execute_until_finished, MvSWrapper
- class TestInput(unittest.TestCase):
- def setUp(self):
- self.mvs = MvSWrapper()
- self.root = self.mvs.execute("RR", [])
- self.mvk = ModelverseKernel(self.root)
- def test_input(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":"input"}])
- old_retval = self.mvs.execute("CN", [])
- input_queue_first = self.mvs.execute("CN", [])
- input_queue_next = self.mvs.execute("CN", [])
- value = 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", old_retval])
- self.mvs.execute("CD", [task_root, "input", input_queue_first])
- self.mvs.execute("CD", [input_queue_first, "value", value])
- self.mvs.execute("CD", [input_queue_first, "next", input_queue_next])
- # 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"])]), "finish")
- self.assertEquals(self.mvs.execute("RD", [task_frame, "IP"]), inst)
- self.assertEquals(self.mvs.execute("RD", [task_frame, "returnvalue"]), value)
- self.assertEquals(self.mvs.execute("RD", [task_root, "input"]), input_queue_next)
- def test_input_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":"input"}])
- old_retval = self.mvs.execute("CN", [])
- input_queue_first = 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", old_retval])
- self.mvs.execute("CD", [task_root, "input", input_queue_first])
- try:
- # Do the invocation until it returns None
- execute_until_finished(self.mvk, self.mvs)
- self.fail()
- except SleepKernel:
- pass
- # 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"]), inst)
- self.assertEquals(self.mvs.execute("RD", [task_frame, "returnvalue"]), old_retval)
- self.assertEquals(self.mvs.execute("RD", [task_root, "input"]), input_queue_first)
|