1  from arti_rpc_tests import arti_test
 2  from arti_rpc import ArtiRpcError, ArtiRpcResponseKind, ArtiRpcErrorStatus
 3  
 4  import json
 5  
 6  
 7  @arti_test
 8  def test_trivial(context):
 9      connection = context.open_rpc_connection()
10  
11      # Run a method that doesn't actually require anything major to be working.
12      #
13      # TODO: Pick a better method once we have more of the RPC system
14      # working.
15      result = connection.session().invoke("arti:get_rpc_proxy_info")
16      assert len(result["proxies"]) > 0
17  
18  
19  @arti_test
20  def test_execute(context):
21      connection = context.open_rpc_connection()
22  
23      req = {
24          "obj": connection.session().id(),
25          "method": "arti:get_rpc_proxy_info",
26          "params": {},
27      }
28      result = connection.execute(req)
29      assert len(result["proxies"]) > 0
30  
31      result = connection.execute(json.dumps(req))
32      assert len(result["proxies"]) > 0
33  
34  
35  @arti_test
36  def test_execute_with_handle(context):
37      connection = context.open_rpc_connection()
38      handle = connection.execute_with_handle(
39          {
40              "obj": connection.session().id(),
41              "method": "arti:get_rpc_proxy_info",
42              "params": {},
43          }
44      )
45  
46      response = handle.wait()
47      assert response.kind() == ArtiRpcResponseKind.RESULT
48      assert len(response.result()["proxies"]) > 1
49      assert len(response["result"]["proxies"]) > 1
50  
51      try:
52          response = handle.wait()
53          assert False
54      except ArtiRpcError as e:
55          assert e.status_code() == ArtiRpcErrorStatus.REQUEST_COMPLETED
56          assert str(e) == "Request has already completed (or failed)"