forked from M-Labs/artiq
pc_rpc: test #1186.
Test pc_rpc can correctly process an annotated function. Also check pyon can encode & decode the annotated function. Signed-off-by: Drew Risinger <drewrisinger@users.noreply.github.com>
This commit is contained in:
parent
b173f4f5b9
commit
7074615ebc
|
@ -1,13 +1,13 @@
|
||||||
import unittest
|
|
||||||
import sys
|
|
||||||
import subprocess
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import inspect
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
import time
|
import time
|
||||||
|
import unittest
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
|
||||||
from artiq.protocols import pc_rpc, fire_and_forget
|
from artiq.protocols import fire_and_forget, pc_rpc, pyon
|
||||||
|
|
||||||
|
|
||||||
test_address = "::1"
|
test_address = "::1"
|
||||||
test_port = 7777
|
test_port = 7777
|
||||||
|
@ -92,6 +92,31 @@ class RPCCase(unittest.TestCase):
|
||||||
def test_asyncio_echo_autotarget(self):
|
def test_asyncio_echo_autotarget(self):
|
||||||
self._run_server_and_test(self._loop_asyncio_echo, pc_rpc.AutoTarget)
|
self._run_server_and_test(self._loop_asyncio_echo, pc_rpc.AutoTarget)
|
||||||
|
|
||||||
|
def test_rpc_encode_function(self):
|
||||||
|
"""Test that `pc_rpc` can encode a function properly.
|
||||||
|
|
||||||
|
Used in `get_rpc_method_list` part of
|
||||||
|
:meth:`artiq.protocols.pc_rpc.Server._process_action`
|
||||||
|
"""
|
||||||
|
|
||||||
|
def _annotated_function(
|
||||||
|
arg1: str, arg2: np.ndarray = np.array([1,])
|
||||||
|
) -> np.ndarray:
|
||||||
|
"""Sample docstring."""
|
||||||
|
return arg1
|
||||||
|
|
||||||
|
argspec, docstring = pc_rpc.Server._document_function(_annotated_function)
|
||||||
|
print(argspec) # for debug
|
||||||
|
self.assertEqual(docstring, "Sample docstring.")
|
||||||
|
|
||||||
|
# purposefully ignore how argspec["annotations"] is treated.
|
||||||
|
# allows option to change PYON later to encode annotations.
|
||||||
|
argspec_master = dict(inspect.getfullargspec(_annotated_function)._asdict())
|
||||||
|
argspec_without_annotation = argspec_master.copy()
|
||||||
|
del argspec_without_annotation["annotations"]
|
||||||
|
# check if all items (excluding annotations) are same in both dictionaries
|
||||||
|
self.assertLessEqual(argspec_without_annotation.items(), argspec.items())
|
||||||
|
self.assertDictEqual(argspec, pyon.decode(pyon.encode(argspec)))
|
||||||
|
|
||||||
class FireAndForgetCase(unittest.TestCase):
|
class FireAndForgetCase(unittest.TestCase):
|
||||||
def _set_ok(self):
|
def _set_ok(self):
|
||||||
|
@ -130,5 +155,6 @@ def run_server():
|
||||||
finally:
|
finally:
|
||||||
loop.close()
|
loop.close()
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
run_server()
|
run_server()
|
||||||
|
|
Loading…
Reference in New Issue