t/fiotestcommon.py

   1 #!/usr/bin/env python3
   2 """
   3 fiotestcommon.py
   4
   5 This contains constant definitions, helpers, and a Requirements class that can
   6 be used to help with running fio tests.
   7 """
   8
   9 import os
  10 import logging
  11 import platform
  12 import subprocess
  13 import multiprocessing
  14 from fiotestlib import FioJobTest
  15
  16
  17 SUCCESS_DEFAULT = {
  18     'zero_return': True,
  19     'stderr_empty': True,
  20     'timeout': 600,
  21     }
  22 SUCCESS_NONZERO = {
  23     'zero_return': False,
  24     'stderr_empty': False,
  25     'timeout': 600,
  26     }
  27 SUCCESS_STDERR = {
  28     'zero_return': True,
  29     'stderr_empty': False,
  30     'timeout': 600,
  31     }
  32
  33 class Requirements():
  34     """Requirements consists of multiple run environment characteristics.
  35     These are to determine if a particular test can be run"""
  36
  37     _linux = False
  38     _libaio = False
  39     _io_uring = False
  40     _zbd = False
  41     _root = False
  42     _zoned_nullb = False
  43     _not_macos = False
  44     _not_windows = False
  45     _unittests = False
  46     _cpucount4 = False
  47     _nvmecdev = False
  48
  49     def __init__(self, fio_root, args):
  50         Requirements._not_macos = platform.system() != "Darwin"
  51         Requirements._not_windows = platform.system() != "Windows"
  52         Requirements._linux = platform.system() == "Linux"
  53
  54         if Requirements._linux:
  55             config_file = os.path.join(fio_root, "config-host.h")
  56             contents, success = FioJobTest.get_file(config_file)
  57             if not success:
  58                 print(f"Unable to open {config_file} to check requirements")
  59                 Requirements._zbd = True
  60             else:
  61                 Requirements._zbd = "CONFIG_HAS_BLKZONED" in contents
  62                 Requirements._libaio = "CONFIG_LIBAIO" in contents
  63
  64             contents, success = FioJobTest.get_file("/proc/kallsyms")
  65             if not success:
  66                 print("Unable to open '/proc/kallsyms' to probe for io_uring support")
  67             else:
  68                 Requirements._io_uring = "io_uring_setup" in contents
  69
  70             Requirements._root = os.geteuid() == 0
  71             if Requirements._zbd and Requirements._root:
  72                 try:
  73                     subprocess.run(["modprobe", "null_blk"],
  74                                    stdout=subprocess.PIPE,
  75                                    stderr=subprocess.PIPE)
  76                     if os.path.exists("/sys/module/null_blk/parameters/zoned"):
  77                         Requirements._zoned_nullb = True
  78                 except Exception:
  79                     pass
  80
  81         if platform.system() == "Windows":
  82             utest_exe = "unittest.exe"
  83         else:
  84             utest_exe = "unittest"
  85         unittest_path = os.path.join(fio_root, "unittests", utest_exe)
  86         Requirements._unittests = os.path.exists(unittest_path)
  87
  88         Requirements._cpucount4 = multiprocessing.cpu_count() >= 4
  89         Requirements._nvmecdev = args.nvmecdev
  90
  91         req_list = [
  92                 Requirements.linux,
  93                 Requirements.libaio,
  94                 Requirements.io_uring,
  95                 Requirements.zbd,
  96                 Requirements.root,
  97                 Requirements.zoned_nullb,
  98                 Requirements.not_macos,
  99                 Requirements.not_windows,
 100                 Requirements.unittests,
 101                 Requirements.cpucount4,
 102                 Requirements.nvmecdev,
 103                     ]
 104         for req in req_list:
 105             value, desc = req()
 106             logging.debug("Requirements: Requirement '%s' met? %s", desc, value)
 107
 108     @classmethod
 109     def linux(cls):
 110         """Are we running on Linux?"""
 111         return Requirements._linux, "Linux required"
 112
 113     @classmethod
 114     def libaio(cls):
 115         """Is libaio available?"""
 116         return Requirements._libaio, "libaio required"
 117
 118     @classmethod
 119     def io_uring(cls):
 120         """Is io_uring available?"""
 121         return Requirements._io_uring, "io_uring required"
 122
 123     @classmethod
 124     def zbd(cls):
 125         """Is ZBD support available?"""
 126         return Requirements._zbd, "Zoned block device support required"
 127
 128     @classmethod
 129     def root(cls):
 130         """Are we running as root?"""
 131         return Requirements._root, "root required"
 132
 133     @classmethod
 134     def zoned_nullb(cls):
 135         """Are zoned null block devices available?"""
 136         return Requirements._zoned_nullb, "Zoned null block device support required"
 137
 138     @classmethod
 139     def not_macos(cls):
 140         """Are we running on a platform other than macOS?"""
 141         return Requirements._not_macos, "platform other than macOS required"
 142
 143     @classmethod
 144     def not_windows(cls):
 145         """Are we running on a platform other than Windws?"""
 146         return Requirements._not_windows, "platform other than Windows required"
 147
 148     @classmethod
 149     def unittests(cls):
 150         """Were unittests built?"""
 151         return Requirements._unittests, "Unittests support required"
 152
 153     @classmethod
 154     def cpucount4(cls):
 155         """Do we have at least 4 CPUs?"""
 156         return Requirements._cpucount4, "4+ CPUs required"
 157
 158     @classmethod
 159     def nvmecdev(cls):
 160         """Do we have an NVMe character device to test?"""
 161         return Requirements._nvmecdev, "NVMe character device test target required"