ceph/src/spdk/test/bdev/bdevio/tests.py

   1 #!/usr/bin/env python3
   2
   3 from rpc.client import print_dict, JSONRPCException
   4
   5 import logging
   6 import argparse
   7 import rpc
   8 import sys
   9 import shlex
  10
  11 try:
  12     from shlex import quote
  13 except ImportError:
  14     from pipes import quote
  15
  16
  17 def print_array(a):
  18     print(" ".join((quote(v) for v in a)))
  19
  20
  21 def perform_tests_func(client, name=None):
  22     """
  23
  24     Args:
  25         name: bdev name to perform bdevio tests on (optional; if omitted, test all bdevs)
  26
  27     Returns:
  28         Number of failures in tests. 0 means no errors found.
  29     """
  30     params = {}
  31     if name:
  32         params['name'] = name
  33     return client.call('perform_tests', params)
  34
  35
  36 if __name__ == "__main__":
  37     parser = argparse.ArgumentParser(
  38         description='SPDK RPC command line interface. NOTE: spdk/scripts/ is expected in PYTHONPATH')
  39     parser.add_argument('-s', dest='server_addr',
  40                         help='RPC domain socket path or IP address', default='/var/tmp/spdk.sock')
  41     parser.add_argument('-p', dest='port',
  42                         help='RPC port number (if server_addr is IP address)',
  43                         default=5260, type=int)
  44     parser.add_argument('-t', dest='timeout',
  45                         help='Timeout as a floating point number expressed in seconds waiting for response. Default: 60.0',
  46                         default=60.0, type=float)
  47     parser.add_argument('-v', dest='verbose', action='store_const', const="INFO",
  48                         help='Set verbose mode to INFO', default="ERROR")
  49     parser.add_argument('--verbose', dest='verbose', choices=['DEBUG', 'INFO', 'ERROR'],
  50                         help="""Set verbose level. """)
  51     subparsers = parser.add_subparsers(help='RPC methods')
  52
  53     def perform_tests(args):
  54         print_dict(perform_tests_func(args.client, name=args.name))
  55
  56     p = subparsers.add_parser('perform_tests', help='Perform all bdevio tests on select bdev')
  57     p.add_argument('-b', '--name', help="Name of the Blockdev. Example: Nvme0n1")
  58     p.set_defaults(func=perform_tests)
  59
  60     def call_rpc_func(args):
  61         try:
  62             args.func(args)
  63         except JSONRPCException as ex:
  64             print(ex.message)
  65             exit(1)
  66
  67     def execute_script(parser, client, fd):
  68         for rpc_call in map(str.rstrip, fd):
  69             if not rpc_call.strip():
  70                 continue
  71             args = parser.parse_args(shlex.split(rpc_call))
  72             args.client = client
  73             call_rpc_func(args)
  74
  75     args = parser.parse_args()
  76     args.client = rpc.client.JSONRPCClient(args.server_addr, args.port, args.timeout, log_level=getattr(logging, args.verbose.upper()))
  77     if hasattr(args, 'func'):
  78         call_rpc_func(args)
  79     elif sys.stdin.isatty():
  80         # No arguments and no data piped through stdin
  81         parser.print_help()
  82         exit(1)
  83     else:
  84         execute_script(parser, args.client, sys.stdin)