# Copyright (c) Microsoft Corporation. # Licensed under the MIT License. """Tests for disk-performance benchmark.""" import unittest from unittest import mock from tests.helper import decorator from tests.helper.testcase import BenchmarkTestCase from superbench.benchmarks import BenchmarkRegistry, BenchmarkType, ReturnCode, Platform class DiskBenchmarkTest(BenchmarkTestCase, unittest.TestCase): """Test class for disk-performance benchmark.""" @classmethod def setUpClass(cls): """Hook method for setting up class fixture before running tests in the class.""" super().setUpClass() cls.createMockEnvs(cls) cls.createMockFiles(cls, ['bin/fio']) def test_disk_performance_empty_param(self): """Test disk-performance benchmark command generation with empty parameter.""" benchmark_name = 'disk-benchmark' (benchmark_class, predefine_params) = BenchmarkRegistry._BenchmarkRegistry__select_benchmark(benchmark_name, Platform.CPU) assert (benchmark_class) benchmark = benchmark_class(benchmark_name, parameters='') # Check basic information assert (benchmark) ret = benchmark._preprocess() assert (ret is True) assert (benchmark.return_code == ReturnCode.SUCCESS) assert (benchmark.name == 'disk-benchmark') assert (benchmark.type == BenchmarkType.MICRO) # Command list should be empty assert (0 == len(benchmark._commands)) @mock.patch('pathlib.Path.is_block_device') def test_disk_performance_invalid_block_device(self, mock_is_block_device): """Test disk-performance benchmark command generation with invalid block device.""" mock_is_block_device.return_value = False benchmark_name = 'disk-benchmark' (benchmark_class, predefine_params) = BenchmarkRegistry._BenchmarkRegistry__select_benchmark(benchmark_name, Platform.CPU) assert (benchmark_class) block_devices = ['mock_block_device_0'] block_device_option = '--block_devices ' + ' '.join(block_devices) benchmark = benchmark_class(benchmark_name, parameters=block_device_option) # Check basic information assert (benchmark) ret = benchmark._preprocess() assert (ret is False) assert (benchmark.return_code == ReturnCode.INVALID_ARGUMENT) assert (benchmark.name == 'disk-benchmark') assert (benchmark.type == BenchmarkType.MICRO) @mock.patch('pathlib.Path.is_block_device') def test_disk_performance_benchmark_disabled(self, mock_is_block_device): """Test disk-performance benchmark command generation with all benchmarks disabled.""" mock_is_block_device.return_value = True benchmark_name = 'disk-benchmark' (benchmark_class, predefine_params) = BenchmarkRegistry._BenchmarkRegistry__select_benchmark(benchmark_name, Platform.CPU) assert (benchmark_class) block_devices = ['/dev/nvme0n1', '/dev/nvme1n1'] block_device_option = '--block_devices ' + ' '.join(block_devices) param_str = block_device_option param_str += ' --rand_precond_time=0' param_str += ' --seq_read_runtime=0' param_str += ' --rand_read_runtime=0' benchmark = benchmark_class(benchmark_name, parameters=param_str) # Check basic information assert (benchmark) ret = benchmark._preprocess() assert (ret is True) assert (benchmark.return_code == ReturnCode.SUCCESS) assert (benchmark.name == 'disk-benchmark') assert (benchmark.type == BenchmarkType.MICRO) # Command list should be empty assert (0 == len(benchmark._commands)) @mock.patch('pathlib.Path.is_block_device') def test_disk_performance_benchmark_enabled(self, mock_is_block_device): """Test disk-performance benchmark command generation with all benchmarks enabled.""" mock_is_block_device.return_value = True benchmark_name = 'disk-benchmark' (benchmark_class, predefine_params) = BenchmarkRegistry._BenchmarkRegistry__select_benchmark(benchmark_name, Platform.CPU) assert (benchmark_class) block_devices = ['mock_block_device_0', 'mock_block_device_1'] block_device_option = '--block_devices ' + ' '.join(block_devices) init_test_magic = 45 curr_test_magic = init_test_magic param_str = block_device_option # Sequential precondition param_str += ' --enable_seq_precond' # Random precondition param_str += ' --rand_precond_time=%d' % curr_test_magic curr_test_magic += 1 # Seq/rand read/write for io_pattern in ['seq', 'rand']: for io_type in ['read', 'write', 'readwrite']: io_str = '%s_%s' % (io_pattern, io_type) param_str += ' --%s_ramp_time=%d' % (io_str, curr_test_magic) curr_test_magic += 1 param_str += ' --%s_runtime=%d' % (io_str, curr_test_magic) curr_test_magic += 1 param_str += ' --%s_iodepth=%d' % (io_str, curr_test_magic) curr_test_magic += 1 param_str += ' --%s_numjobs=%d' % (io_str, curr_test_magic) curr_test_magic += 1 benchmark = benchmark_class(benchmark_name, parameters=param_str) # Check basic information assert (benchmark) ret = benchmark._preprocess() assert (ret is True) assert (benchmark.return_code == ReturnCode.SUCCESS) assert (benchmark.name == 'disk-benchmark') assert (benchmark.type == BenchmarkType.MICRO) # Check command list # 2 files * (2 preconditions + 3 io_patterns * 2 io_types) = 16 commands assert (16 == len(benchmark._commands)) # Check parameter assignments command_idx = 0 default_rwmixread = 80 for block_device in block_devices: curr_test_magic = init_test_magic # Sequential precondition assert ('--filename=%s' % block_device in benchmark._commands[command_idx]) command_idx += 1 # Random precondition assert ('--filename=%s' % block_device in benchmark._commands[command_idx]) assert ('--runtime=%d' % curr_test_magic in benchmark._commands[command_idx]) curr_test_magic += 1 command_idx += 1 # Seq/rand read/write for io_pattern in ['seq', 'rand']: for io_type in ['read', 'write', 'rw']: assert ('--filename=%s' % block_device in benchmark._commands[command_idx]) fio_rw = '%s%s' % (io_pattern if io_pattern == 'rand' else '', io_type) assert ('--rw=%s' % fio_rw in benchmark._commands[command_idx]) assert ('--ramp_time=%d' % curr_test_magic in benchmark._commands[command_idx]) curr_test_magic += 1 assert ('--runtime=%d' % curr_test_magic in benchmark._commands[command_idx]) curr_test_magic += 1 assert ('--iodepth=%d' % curr_test_magic in benchmark._commands[command_idx]) curr_test_magic += 1 assert ('--numjobs=%d' % curr_test_magic in benchmark._commands[command_idx]) curr_test_magic += 1 if io_type == 'rw': assert ('--rwmixread=%d' % default_rwmixread in benchmark._commands[command_idx]) command_idx += 1 @decorator.load_data('tests/data/disk_performance.log') def test_disk_performance_result_parsing(self, test_raw_output): """Test disk-performance benchmark result parsing.""" benchmark_name = 'disk-benchmark' (benchmark_class, predefine_params) = BenchmarkRegistry._BenchmarkRegistry__select_benchmark(benchmark_name, Platform.CPU) assert (benchmark_class) benchmark = benchmark_class(benchmark_name, parameters='') assert (benchmark) ret = benchmark._preprocess() assert (ret is True) assert (benchmark.return_code == ReturnCode.SUCCESS) assert (benchmark.name == 'disk-benchmark') assert (benchmark.type == BenchmarkType.MICRO) # Positive case - valid raw output. jobname_prefix = 'nvme0n1_rand_read_write' assert (benchmark._process_raw_result(0, test_raw_output)) assert (benchmark.return_code == ReturnCode.SUCCESS) # bs + x assert (9 + benchmark.default_metric_count == len(benchmark.result.keys())) assert (1 == len(benchmark.result[jobname_prefix + '_bs'])) assert (4096 == benchmark.result[jobname_prefix + '_bs'][0]) assert (1 == len(benchmark.result[jobname_prefix + '_read_iops'])) assert (85138.890741 == benchmark.result[jobname_prefix + '_read_iops'][0]) assert (1 == len(benchmark.result[jobname_prefix + '_write_iops'])) assert (85066.128925 == benchmark.result[jobname_prefix + '_write_iops'][0]) assert (1 == len(benchmark.result[jobname_prefix + '_read_lat_ns_95.0'])) assert (1941504 == benchmark.result[jobname_prefix + '_read_lat_ns_95.0'][0]) assert (1 == len(benchmark.result[jobname_prefix + '_read_lat_ns_99.0'])) assert (2244608 == benchmark.result[jobname_prefix + '_read_lat_ns_99.0'][0]) assert (1 == len(benchmark.result[jobname_prefix + '_read_lat_ns_99.9'])) assert (3620864 == benchmark.result[jobname_prefix + '_read_lat_ns_99.9'][0]) assert (1 == len(benchmark.result[jobname_prefix + '_write_lat_ns_95.0'])) assert (1908736 == benchmark.result[jobname_prefix + '_write_lat_ns_95.0'][0]) assert (1 == len(benchmark.result[jobname_prefix + '_write_lat_ns_99.0'])) assert (2072576 == benchmark.result[jobname_prefix + '_write_lat_ns_99.0'][0]) assert (1 == len(benchmark.result[jobname_prefix + '_write_lat_ns_99.9'])) assert (2605056 == benchmark.result[jobname_prefix + '_write_lat_ns_99.9'][0]) # Negative case - invalid raw output. assert (benchmark._process_raw_result(1, 'Invalid raw output') is False) assert (benchmark.return_code == ReturnCode.MICROBENCHMARK_RESULT_PARSING_FAILURE)