tests/topotests/lib/grpc-query.py

   1 #!/usr/bin/env python3
   2 # -*- coding: utf-8 eval: (blacken-mode 1) -*-
   3 #
   4 # February 22 2022, Christian Hopps <chopps@labn.net>
   5 #
   6 # Copyright (c) 2022, LabN Consulting, L.L.C.
   7 #
   8 # Permission is hereby granted, free of charge, to any person obtaining a copy
   9 # of this software and associated documentation files (the "Software"), to deal
  10 # in the Software without restriction, including without limitation the rights
  11 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  12 # copies of the Software, and to permit persons to whom the Software is
  13 # furnished to do so, subject to the following conditions:
  14 #
  15 # The above copyright notice and this permission notice shall be included in all
  16 # copies or substantial portions of the Software.
  17 #
  18 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  19 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  20 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  21 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  22 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  23 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  24 # SOFTWARE.
  25
  26 import argparse
  27 import logging
  28 import os
  29 import sys
  30
  31 import pytest
  32
  33 CWD = os.path.dirname(os.path.realpath(__file__))
  34
  35 # This is painful but works if you have installed grpc and grpc_tools would be *way*
  36 # better if we actually built and installed these but ... python packaging.
  37 try:
  38     import grpc
  39     import grpc_tools
  40
  41     from micronet import commander
  42
  43     commander.cmd_raises(f"cp {CWD}/../../../grpc/frr-northbound.proto .")
  44     commander.cmd_raises(
  45         f"python -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I . frr-northbound.proto"
  46     )
  47 except Exception as error:
  48     logging.error("can't create proto definition modules %s", error)
  49     raise
  50
  51 try:
  52     sys.path[0:0] = "."
  53     import frr_northbound_pb2
  54     import frr_northbound_pb2_grpc
  55
  56     # Would be nice if compiling the modules internally from the source worked
  57     # # import grpc_tools.protoc
  58     # # proto_include = pkg_resources.resource_filename("grpc_tools", "_proto")
  59     # from grpc_tools.protoc import _proto_file_to_module_name, _protos_and_services
  60     # try:
  61     #     frr_northbound_pb2, frr_northbound_pb2_grpc = _protos_and_services(
  62     #         "frr_northbound.proto"
  63     #     )
  64     # finally:
  65     #     os.chdir(CWD)
  66 except Exception as error:
  67     logging.error("can't import proto definition modules %s", error)
  68     raise
  69
  70
  71 class GRPCClient:
  72     def __init__(self, server, port):
  73         self.channel = grpc.insecure_channel("{}:{}".format(server, port))
  74         self.stub = frr_northbound_pb2_grpc.NorthboundStub(self.channel)
  75
  76     def get_capabilities(self):
  77         request = frr_northbound_pb2.GetCapabilitiesRequest()
  78         response = "NONE"
  79         try:
  80             response = self.stub.GetCapabilities(request)
  81         except Exception as error:
  82             logging.error("Got exception from stub: %s", error)
  83
  84         logging.debug("GRPC Capabilities: %s", response)
  85         return response
  86
  87     def get(self, xpath):
  88         request = frr_northbound_pb2.GetRequest()
  89         request.path.append(xpath)
  90         request.type = frr_northbound_pb2.GetRequest.ALL
  91         request.encoding = frr_northbound_pb2.XML
  92         xml = ""
  93         for r in self.stub.Get(request):
  94             logging.info('GRPC Get path: "%s" value: %s', request.path, r)
  95             xml += str(r.data.data)
  96         return xml
  97
  98
  99 def next_action(action_list=None):
 100     "Get next action from list or STDIN"
 101     if action_list:
 102         for action in action_list:
 103             yield action
 104     else:
 105         while True:
 106             try:
 107                 action = input("")
 108                 if not action:
 109                     break
 110                 yield action.strip()
 111             except EOFError:
 112                 break
 113
 114
 115 def main(*args):
 116     parser = argparse.ArgumentParser(description="gRPC Client")
 117     parser.add_argument(
 118         "-s", "--server", default="localhost", help="gRPC Server Address"
 119     )
 120     parser.add_argument(
 121         "-p", "--port", type=int, default=50051, help="gRPC Server TCP Port"
 122     )
 123     parser.add_argument("-v", "--verbose", action="store_true", help="be verbose")
 124     parser.add_argument("--check", action="store_true", help="check runable")
 125     parser.add_argument("actions", nargs="*", help="GETCAP|GET,xpath")
 126     args = parser.parse_args(*args)
 127
 128     level = logging.DEBUG if args.verbose else logging.INFO
 129     logging.basicConfig(
 130         level=level,
 131         format="%(asctime)s %(levelname)s: GRPC-CLI-CLIENT: %(name)s %(message)s",
 132     )
 133
 134     if args.check:
 135         sys.exit(0)
 136
 137     c = GRPCClient(args.server, args.port)
 138
 139     for action in next_action(args.actions):
 140         action = action.casefold()
 141         logging.info("GOT ACTION: %s", action)
 142         if action == "getcap":
 143             caps = c.get_capabilities()
 144             print("Capabilities:", caps)
 145         elif action.startswith("get,"):
 146             # Print Interface State and Config
 147             _, xpath = action.split(",", 1)
 148             print("Get XPath: ", xpath)
 149             xml = c.get(xpath)
 150             print("{}: {}".format(xpath, xml))
 151             # for _ in range(0, 1):
 152
 153
 154 if __name__ == "__main__":
 155     main()