iv/linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
207e03b00b
linux/tools/hv/vmbus_testing
Ani Sinha ed0cf84e9c vmbus_testing: fix wrong python syntax for integer value comparison 
It is incorrect in python to compare integer values using the "is" keyword.
The "is" keyword in python is used to compare references to two objects,
not their values. Newer version of python3 (version 3.8) throws a warning
when such incorrect comparison is made. For value comparison, "==" should
be used.

Fix this in the code and suppress the following warning:

/usr/sbin/vmbus_testing:167: SyntaxWarning: "is" with a literal. Did you mean "=="?

Signed-off-by: Ani Sinha <anisinha@redhat.com>
Link: https://lore.kernel.org/r/20230705134408.6302-1-anisinha@redhat.com
Signed-off-by: Wei Liu <wei.liu@kernel.org>
2023-07-23 23:19:30 +00:00

377 lines
16 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
#
# Program to allow users to fuzz test Hyper-V drivers
# by interfacing with Hyper-V debugfs attributes.
# Current test methods available:
# 1. delay testing
#
# Current file/directory structure of hyper-V debugfs:
# /sys/kernel/debug/hyperv/UUID
# /sys/kernel/debug/hyperv/UUID/<test-state filename>
# /sys/kernel/debug/hyperv/UUID/<test-method sub-directory>
#
# author: Branden Bonaby <brandonbonaby94@gmail.com>
import os
import cmd
import argparse
import glob
from argparse import RawDescriptionHelpFormatter
from argparse import RawTextHelpFormatter
from enum import Enum
# Do not change unless, you change the debugfs attributes
# in /drivers/hv/debugfs.c. All fuzz testing
# attributes will start with "fuzz_test".
# debugfs path for hyperv must exist before proceeding
debugfs_hyperv_path = "/sys/kernel/debug/hyperv"
if not os.path.isdir(debugfs_hyperv_path):
print("{} doesn't exist/check permissions".format(debugfs_hyperv_path))
exit(-1)
class dev_state(Enum):
off = 0
on = 1
# File names, that correspond to the files created in
# /drivers/hv/debugfs.c
class f_names(Enum):
state_f = "fuzz_test_state"
buff_f = "fuzz_test_buffer_interrupt_delay"
mess_f = "fuzz_test_message_delay"
# Both single_actions and all_actions are used
# for error checking and to allow for some subparser
# names to be abbreviated. Do not abbreviate the
# test method names, as it will become less intuitive
# as to what the user can do. If you do decide to
# abbreviate the test method name, make sure the main
# function reflects this change.
all_actions = [
"disable_all",
"D",
"enable_all",
"view_all",
"V"
]
single_actions = [
"disable_single",
"d",
"enable_single",
"view_single",
"v"
]
def main():
file_map = recursive_file_lookup(debugfs_hyperv_path, dict())
args = parse_args()
if (not args.action):
print ("Error, no options selected...exiting")
exit(-1)
arg_set = { k for (k,v) in vars(args).items() if v and k != "action" }
arg_set.add(args.action)
path = args.path if "path" in arg_set else None
if (path and path[-1] == "/"):
path = path[:-1]
validate_args_path(path, arg_set, file_map)
if (path and "enable_single" in arg_set):
state_path = locate_state(path, file_map)
set_test_state(state_path, dev_state.on.value, args.quiet)
# Use subparsers as the key for different actions
if ("delay" in arg_set):
validate_delay_values(args.delay_time)
if (args.enable_all):
set_delay_all_devices(file_map, args.delay_time,
args.quiet)
else:
set_delay_values(path, file_map, args.delay_time,
args.quiet)
elif ("disable_all" in arg_set or "D" in arg_set):
disable_all_testing(file_map)
elif ("disable_single" in arg_set or "d" in arg_set):
disable_testing_single_device(path, file_map)
elif ("view_all" in arg_set or "V" in arg_set):
get_all_devices_test_status(file_map)
elif ("view_single" in arg_set or "v" in arg_set):
get_device_test_values(path, file_map)
# Get the state location
def locate_state(device, file_map):
return file_map[device][f_names.state_f.value]
# Validate delay values to make sure they are acceptable to
# enable delays on a device
def validate_delay_values(delay):
if (delay[0] == -1 and delay[1] == -1):
print("\nError, At least 1 value must be greater than 0")
exit(-1)
for i in delay:
if (i < -1 or i == 0 or i > 1000):
print("\nError, Values must be equal to -1 "
"or be > 0 and <= 1000")
exit(-1)
# Validate argument path
def validate_args_path(path, arg_set, file_map):
if (not path and any(element in arg_set for element in single_actions)):
print("Error, path (-p) REQUIRED for the specified option. "
"Use (-h) to check usage.")
exit(-1)
elif (path and any(item in arg_set for item in all_actions)):
print("Error, path (-p) NOT REQUIRED for the specified option. "
"Use (-h) to check usage." )
exit(-1)
elif (path not in file_map and any(item in arg_set
for item in single_actions)):
print("Error, path '{}' not a valid vmbus device".format(path))
exit(-1)
# display Testing status of single device
def get_device_test_values(path, file_map):
for name in file_map[path]:
file_location = file_map[path][name]
print( name + " = " + str(read_test_files(file_location)))
# Create a map of the vmbus devices and their associated files
# [key=device, value = [key = filename, value = file path]]
def recursive_file_lookup(path, file_map):
for f_path in glob.iglob(path + '**/*'):
if (os.path.isfile(f_path)):
if (f_path.rsplit("/",2)[0] == debugfs_hyperv_path):
directory = f_path.rsplit("/",1)[0]
else:
directory = f_path.rsplit("/",2)[0]
f_name = f_path.split("/")[-1]
if (file_map.get(directory)):
file_map[directory].update({f_name:f_path})
else:
file_map[directory] = {f_name:f_path}
elif (os.path.isdir(f_path)):
recursive_file_lookup(f_path,file_map)
return file_map
# display Testing state of devices
def get_all_devices_test_status(file_map):
for device in file_map:
if (get_test_state(locate_state(device, file_map)) == 1):
print("Testing = ON for: {}"
.format(device.split("/")[5]))
else:
print("Testing = OFF for: {}"
.format(device.split("/")[5]))
# read the vmbus device files, path must be absolute path before calling
def read_test_files(path):
try:
with open(path,"r") as f:
file_value = f.readline().strip()
return int(file_value)
except IOError as e:
errno, strerror = e.args
print("I/O error({0}): {1} on file {2}"
.format(errno, strerror, path))
exit(-1)
except ValueError:
print ("Element to int conversion error in: \n{}".format(path))
exit(-1)
# writing to vmbus device files, path must be absolute path before calling
def write_test_files(path, value):
try:
with open(path,"w") as f:
f.write("{}".format(value))
except IOError as e:
errno, strerror = e.args
print("I/O error({0}): {1} on file {2}"
.format(errno, strerror, path))
exit(-1)
# set testing state of device
def set_test_state(state_path, state_value, quiet):
write_test_files(state_path, state_value)
if (get_test_state(state_path) == 1):
if (not quiet):
print("Testing = ON for device: {}"
.format(state_path.split("/")[5]))
else:
if (not quiet):
print("Testing = OFF for device: {}"
.format(state_path.split("/")[5]))
# get testing state of device
def get_test_state(state_path):
#state == 1 - test = ON
#state == 0 - test = OFF
return read_test_files(state_path)
# write 1 - 1000 microseconds, into a single device using the
# fuzz_test_buffer_interrupt_delay and fuzz_test_message_delay
# debugfs attributes
def set_delay_values(device, file_map, delay_length, quiet):
try:
interrupt = file_map[device][f_names.buff_f.value]
message = file_map[device][f_names.mess_f.value]
# delay[0]- buffer interrupt delay, delay[1]- message delay
if (delay_length[0] >= 0 and delay_length[0] <= 1000):
write_test_files(interrupt, delay_length[0])
if (delay_length[1] >= 0 and delay_length[1] <= 1000):
write_test_files(message, delay_length[1])
if (not quiet):
print("Buffer delay testing = {} for: {}"
.format(read_test_files(interrupt),
interrupt.split("/")[5]))
print("Message delay testing = {} for: {}"
.format(read_test_files(message),
message.split("/")[5]))
except IOError as e:
errno, strerror = e.args
print("I/O error({0}): {1} on files {2}{3}"
.format(errno, strerror, interrupt, message))
exit(-1)
# enabling delay testing on all devices
def set_delay_all_devices(file_map, delay, quiet):
for device in (file_map):
set_test_state(locate_state(device, file_map),
dev_state.on.value,
quiet)
set_delay_values(device, file_map, delay, quiet)
# disable all testing on a SINGLE device.
def disable_testing_single_device(device, file_map):
for name in file_map[device]:
file_location = file_map[device][name]
write_test_files(file_location, dev_state.off.value)
print("ALL testing now OFF for {}".format(device.split("/")[-1]))
# disable all testing on ALL devices
def disable_all_testing(file_map):
for device in file_map:
disable_testing_single_device(device, file_map)
def parse_args():
parser = argparse.ArgumentParser(prog = "vmbus_testing",usage ="\n"
"%(prog)s [delay] [-h] [-e|-E] -t [-p]\n"
"%(prog)s [view_all | V] [-h]\n"
"%(prog)s [disable_all | D] [-h]\n"
"%(prog)s [disable_single | d] [-h|-p]\n"
"%(prog)s [view_single | v] [-h|-p]\n"
"%(prog)s --version\n",
description = "\nUse lsvmbus to get vmbus device type "
"information.\n" "\nThe debugfs root path is "
"/sys/kernel/debug/hyperv",
formatter_class = RawDescriptionHelpFormatter)
subparsers = parser.add_subparsers(dest = "action")
parser.add_argument("--version", action = "version",
version = '%(prog)s 0.1.0')
parser.add_argument("-q","--quiet", action = "store_true",
help = "silence none important test messages."
" This will only work when enabling testing"
" on a device.")
# Use the path parser to hold the --path attribute so it can
# be shared between subparsers. Also do the same for the state
# parser, as all testing methods will use --enable_all and
# enable_single.
path_parser = argparse.ArgumentParser(add_help=False)
path_parser.add_argument("-p","--path", metavar = "",
help = "Debugfs path to a vmbus device. The path "
"must be the absolute path to the device.")
state_parser = argparse.ArgumentParser(add_help=False)
state_group = state_parser.add_mutually_exclusive_group(required = True)
state_group.add_argument("-E", "--enable_all", action = "store_const",
const = "enable_all",
help = "Enable the specified test type "
"on ALL vmbus devices.")
state_group.add_argument("-e", "--enable_single",
action = "store_const",
const = "enable_single",
help = "Enable the specified test type on a "
"SINGLE vmbus device.")
parser_delay = subparsers.add_parser("delay",
parents = [state_parser, path_parser],
help = "Delay the ring buffer interrupt or the "
"ring buffer message reads in microseconds.",
prog = "vmbus_testing",
usage = "%(prog)s [-h]\n"
"%(prog)s -E -t [value] [value]\n"
"%(prog)s -e -t [value] [value] -p",
description = "Delay the ring buffer interrupt for "
"vmbus devices, or delay the ring buffer message "
"reads for vmbus devices (both in microseconds). This "
"is only on the host to guest channel.")
parser_delay.add_argument("-t", "--delay_time", metavar = "", nargs = 2,
type = check_range, default =[0,0], required = (True),
help = "Set [buffer] & [message] delay time. "
"Value constraints: -1 == value "
"or 0 < value <= 1000.\n"
"Use -1 to keep the previous value for that delay "
"type, or a value > 0 <= 1000 to change the delay "
"time.")
parser_dis_all = subparsers.add_parser("disable_all",
aliases = ['D'], prog = "vmbus_testing",
usage = "%(prog)s [disable_all | D] -h\n"
"%(prog)s [disable_all | D]\n",
help = "Disable ALL testing on ALL vmbus devices.",
description = "Disable ALL testing on ALL vmbus "
"devices.")
parser_dis_single = subparsers.add_parser("disable_single",
aliases = ['d'],
parents = [path_parser], prog = "vmbus_testing",
usage = "%(prog)s [disable_single | d] -h\n"
"%(prog)s [disable_single | d] -p\n",
help = "Disable ALL testing on a SINGLE vmbus device.",
description = "Disable ALL testing on a SINGLE vmbus "
"device.")
parser_view_all = subparsers.add_parser("view_all", aliases = ['V'],
help = "View the test state for ALL vmbus devices.",
prog = "vmbus_testing",
usage = "%(prog)s [view_all | V] -h\n"
"%(prog)s [view_all | V]\n",
description = "This shows the test state for ALL the "
"vmbus devices.")
parser_view_single = subparsers.add_parser("view_single",
aliases = ['v'],parents = [path_parser],
help = "View the test values for a SINGLE vmbus "
"device.",
description = "This shows the test values for a SINGLE "
"vmbus device.", prog = "vmbus_testing",
usage = "%(prog)s [view_single | v] -h\n"
"%(prog)s [view_single | v] -p")
return parser.parse_args()
# value checking for range checking input in parser
def check_range(arg1):
try:
val = int(arg1)
except ValueError as err:
raise argparse.ArgumentTypeError(str(err))
if val < -1 or val > 1000:
message = ("\n\nvalue must be -1 or 0 < value <= 1000. "
"Value program received: {}\n").format(val)
raise argparse.ArgumentTypeError(message)
return val
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink