pytest: Add helper to get a grpc stub and test decode

Author: cdecker

contrib/pyln-testing/pyln/testing/utils.py

@@ -851,6 +851,39 @@ def _create_jsonrpc_rpc(self, jsonschemas):
             jsonschemas=jsonschemas
         )
 
+    @property
+    def grpc(self):
+        """Tiny helper to return a grpc stub if grpc was configured.
+        """
+        # Before doing anything let's see if we have a grpc-port at all
+        try:
+            grpc_port = int(filter(
+                lambda v: v[0] == 'grpc-port',
+                self.daemon.opts.items()
+            ).__next__()[1])
+        except Exception as e:
+            raise ValueError("grpc-port is not specified, can't connect over grpc")
+
+        import grpc
+        p = Path(self.daemon.lightning_dir) / TEST_NETWORK
+        cert, key, ca = [f.open('rb').read() for f in [
+            p / 'client.pem',
+            p / 'client-key.pem',
+            p / "ca.pem"]]
+        creds = grpc.ssl_channel_credentials(
+            root_certificates=ca,
+            private_key=key,
+            certificate_chain=cert,
+        )
+
+        channel = grpc.secure_channel(
+            f"localhost:{grpc_port}",
+            creds,
+            options=(('grpc.ssl_target_name_override', 'cln'),)
+        )
+        from pyln.testing import node_pb2_grpc as nodegrpc
+        return nodegrpc.NodeStub(channel)
+
     def connect(self, remote_node):
         self.rpc.connect(remote_node.info['id'], '127.0.0.1', remote_node.port)
 

tests/test_cln_rs.py

@@ -303,27 +303,7 @@ def test_grpc_keysend_routehint(bitcoind, node_factory):
     bitcoind.generate_block(3)
     sync_blockheight(bitcoind, [l1, l2, l3])
 
-    def connect(node):
-        p = Path(node.daemon.lightning_dir) / TEST_NETWORK
-        cert, key, ca = [f.open('rb').read() for f in [
-            p / 'client.pem',
-            p / 'client-key.pem',
-            p / "ca.pem"]]
-
-        creds = grpc.ssl_channel_credentials(
-            root_certificates=ca,
-            private_key=key,
-            certificate_chain=cert,
-        )
-
-        channel = grpc.secure_channel(
-            f"localhost:{grpc_port}",
-            creds,
-            options=(('grpc.ssl_target_name_override', 'cln'),)
-        )
-        return nodegrpc.NodeStub(channel)
-
-    stub = connect(l1)
+    stub = l1.grpc
     chan = l2.rpc.listpeerchannels(l3.info['id'])
 
     routehint = primitivespb.RoutehintList(hints=[
@@ -362,26 +342,7 @@ def test_grpc_listpeerchannels(bitcoind, node_factory):
         announce_channels=True,  # Do not enforce scid-alias
     )
 
-    def connect(node):
-        p = Path(node.daemon.lightning_dir) / TEST_NETWORK
-        cert, key, ca = [f.open('rb').read() for f in [
-            p / 'client.pem',
-            p / 'client-key.pem',
-            p / "ca.pem"]]
-
-        creds = grpc.ssl_channel_credentials(
-            root_certificates=ca,
-            private_key=key,
-            certificate_chain=cert,
-        )
-
-        channel = grpc.secure_channel(
-            f"localhost:{grpc_port}",
-            creds,
-            options=(('grpc.ssl_target_name_override', 'cln'),)
-        )
-        return nodegrpc.NodeStub(channel)
-    stub = connect(l1)
+    stub = l1.grpc
     res = stub.ListPeerChannels(nodepb.ListpeerchannelsRequest(id=None))
 
     # Way too many fields to check, so just do a couple
@@ -399,3 +360,24 @@ def connect(node):
     l1.daemon.wait_for_log(r'onchaind complete, forgetting peer')
 
     stub.ListClosedChannels(nodepb.ListclosedchannelsRequest())
+
+
+def test_grpc_decode(node_factory):
+    grpc_port = reserve()
+    l1 = node_factory.get_node(options={'grpc-port': str(grpc_port)})
+    inv = l1.grpc.Invoice(nodepb.InvoiceRequest(
+        amount_msat=primitivespb.AmountOrAny(any=True),
+        description="desc",
+        label="label",
+    ))
+
+    res = l1.grpc.DecodePay(nodepb.DecodepayRequest(
+        bolt11=inv.bolt11
+    ))
+    # If we get here we're good, conversions work
+    print(res)
+
+    res = l1.grpc.Decode(nodepb.DecodeRequest(
+        string=inv.bolt11
+    ))
+    print(res)