xref: /spdk/test/bdev/bdevio/tests.py (revision 1966f1eef3f99d9ff26320cff449c6c03017ea66) 
1#!/usr/bin/env python3
2
3from rpc.client import print_dict, JSONRPCException
4
5import logging
6import argparse
7import rpc
8import sys
9import shlex
10
11try:
12    from shlex import quote
13except ImportError:
14    from pipes import quote
15
16
17def print_array(a):
18    print(" ".join((quote(v) for v in a)))
19
20
21def 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
36if __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)
85