diff --git a/src/frostfs_testlib/utils/__init__.py b/src/frostfs_testlib/utils/__init__.py index 01cf462..0ac903a 100644 --- a/src/frostfs_testlib/utils/__init__.py +++ b/src/frostfs_testlib/utils/__init__.py @@ -3,4 +3,6 @@ import frostfs_testlib.utils.datetime_utils import frostfs_testlib.utils.json_utils import frostfs_testlib.utils.string_utils import frostfs_testlib.utils.wallet_utils + +# TODO: Circullar dependency FileKeeper -> NodeBase -> Utils -> FileKeeper -> NodeBase from frostfs_testlib.utils.file_keeper import FileKeeper diff --git a/tests/test_load_config.py b/tests/test_load_config.py new file mode 100644 index 0000000..a9b6de1 --- /dev/null +++ b/tests/test_load_config.py @@ -0,0 +1,541 @@ +from dataclasses import Field, dataclass, fields, is_dataclass +from typing import Any, get_args + +import pytest + +from frostfs_testlib.load.load_config import LoadParams, LoadScenario, LoadType, Preset + + +@dataclass +class MetaTestField: + field: Field + field_type: type + instance: Any + + +class TestLoadConfig: + @pytest.fixture + def set_empty(self, request: pytest.FixtureRequest): + # Workaround for verify + if "param" in request.__dict__ and request.param: + return request.param + + return False + + @pytest.fixture + def load_type(self, request: pytest.FixtureRequest): + # Workaround for verify + if "param" in request.__dict__ and request.param: + return request.param + + return None + + @pytest.fixture + def load_params(self, load_type: LoadType, set_empty: bool, request: pytest.FixtureRequest): + load_scenario = request.param + return self._get_filled_load_params(load_type, load_scenario, set_empty) + + def test_load_params_only_load_type_required(self): + LoadParams(load_type=LoadType.S3) + + def test_load_params_initially_have_all_values_none(self): + load_params = LoadParams(load_type=LoadType.S3) + self._check_all_values_none(load_params, ["load_type"]) + + def test_preset_initially_have_all_values_none(self): + preset = Preset() + self._check_all_values_none(preset) + + def test_load_set_id_changes_fields(self): + load_params = LoadParams(load_type=LoadType.S3) + load_params.preset = Preset() + load_params.working_dir = "/tmp" + load_params.set_id("test_id") + + assert load_params.registry_file == "/tmp/test_id_registry.bolt" + assert load_params.preset.pregen_json == "/tmp/test_id_prepare.json" + assert load_params.load_id == "test_id" + + # No other values should be changed + self._check_all_values_none( + load_params, ["load_type", "working_dir", "load_id", "registry_file", "preset"] + ) + self._check_all_values_none(load_params.preset, ["pregen_json"]) + + @pytest.mark.parametrize("load_params", [LoadScenario.gRPC], indirect=True) + def test_argument_parsing_for_grpc_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '11'", + "--preload_obj '13'", + "--out 'pregen_json'", + "--workers '7'", + "--containers '16'", + "--policy 'container_placement_policy'", + ] + expected_env_vars = { + "DURATION": 9, + "WRITE_OBJ_SIZE": 11, + "REGISTRY_FILE": "registry_file", + "K6_MIN_ITERATION_DURATION": "min_iteration_duration", + "K6_SETUP_TIMEOUT": "setup_timeout", + "WRITERS": 7, + "READERS": 7, + "DELETERS": 8, + "PREGEN_JSON": "pregen_json", + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params", [LoadScenario.gRPC_CAR], indirect=True) + def test_argument_parsing_for_grpc_car_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '11'", + "--preload_obj '13'", + "--out 'pregen_json'", + "--workers '7'", + "--containers '16'", + "--policy 'container_placement_policy'", + ] + expected_env_vars = { + "DURATION": 9, + "WRITE_OBJ_SIZE": 11, + "REGISTRY_FILE": "registry_file", + "K6_MIN_ITERATION_DURATION": "min_iteration_duration", + "K6_SETUP_TIMEOUT": "setup_timeout", + "MAX_WRITERS": 11, + "MAX_READERS": 11, + "MAX_DELETERS": 12, + "PRE_ALLOC_DELETERS": 21, + "PRE_ALLOC_READERS": 20, + "PRE_ALLOC_WRITERS": 20, + "PREGEN_JSON": "pregen_json", + "TIME_UNIT": "time_unit", + "WRITE_RATE": 10, + "READ_RATE": 9, + "DELETE_RATE": 11, + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params", [LoadScenario.S3], indirect=True) + def test_argument_parsing_for_s3_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '11'", + "--preload_obj '13'", + "--no-verify-ssl", + "--out 'pregen_json'", + "--workers '7'", + "--buckets '13'", + "--location 's3_location'", + ] + expected_env_vars = { + "DURATION": 9, + "WRITE_OBJ_SIZE": 11, + "REGISTRY_FILE": "registry_file", + "K6_MIN_ITERATION_DURATION": "min_iteration_duration", + "K6_SETUP_TIMEOUT": "setup_timeout", + "WRITERS": 7, + "READERS": 7, + "DELETERS": 8, + "NO_VERIFY_SSL": True, + "PREGEN_JSON": "pregen_json", + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params", [LoadScenario.S3_CAR], indirect=True) + def test_argument_parsing_for_s3_car_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '11'", + "--preload_obj '13'", + "--no-verify-ssl", + "--out 'pregen_json'", + "--workers '7'", + "--buckets '13'", + "--location 's3_location'", + ] + expected_env_vars = { + "DURATION": 9, + "WRITE_OBJ_SIZE": 11, + "REGISTRY_FILE": "registry_file", + "K6_MIN_ITERATION_DURATION": "min_iteration_duration", + "K6_SETUP_TIMEOUT": "setup_timeout", + "NO_VERIFY_SSL": True, + "MAX_WRITERS": 11, + "MAX_READERS": 11, + "MAX_DELETERS": 12, + "PRE_ALLOC_DELETERS": 21, + "PRE_ALLOC_READERS": 20, + "PRE_ALLOC_WRITERS": 20, + "PREGEN_JSON": "pregen_json", + "TIME_UNIT": "time_unit", + "WRITE_RATE": 10, + "READ_RATE": 9, + "DELETE_RATE": 11, + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params", [LoadScenario.HTTP], indirect=True) + def test_argument_parsing_for_http_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--no-verify-ssl", + "--size '11'", + "--preload_obj '13'", + "--out 'pregen_json'", + "--workers '7'", + "--containers '16'", + "--policy 'container_placement_policy'", + ] + expected_env_vars = { + "DURATION": 9, + "WRITE_OBJ_SIZE": 11, + "NO_VERIFY_SSL": True, + "REGISTRY_FILE": "registry_file", + "K6_MIN_ITERATION_DURATION": "min_iteration_duration", + "K6_SETUP_TIMEOUT": "setup_timeout", + "WRITERS": 7, + "READERS": 7, + "DELETERS": 8, + "PREGEN_JSON": "pregen_json", + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params", [LoadScenario.LOCAL], indirect=True) + def test_argument_parsing_for_local_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '11'", + "--preload_obj '13'", + "--out 'pregen_json'", + "--workers '7'", + "--containers '16'", + "--policy 'container_placement_policy'", + ] + expected_env_vars = { + "CONFIG_FILE": "config_file", + "DURATION": 9, + "WRITE_OBJ_SIZE": 11, + "REGISTRY_FILE": "registry_file", + "K6_MIN_ITERATION_DURATION": "min_iteration_duration", + "K6_SETUP_TIMEOUT": "setup_timeout", + "WRITERS": 7, + "READERS": 7, + "DELETERS": 8, + "PREGEN_JSON": "pregen_json", + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize( + "load_params, load_type", [(LoadScenario.VERIFY, LoadType.S3)], indirect=True + ) + def test_argument_parsing_for_s3_verify_scenario(self, load_params: LoadParams): + expected_env_vars = { + "CLIENTS": 14, + "REGISTRY_FILE": "registry_file", + "K6_SETUP_TIMEOUT": "setup_timeout", + "NO_VERIFY_SSL": True, + "TIME_LIMIT": 11, + } + + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize( + "load_params, load_type", [(LoadScenario.VERIFY, LoadType.gRPC)], indirect=True + ) + def test_argument_parsing_for_grpc_verify_scenario(self, load_params: LoadParams): + expected_env_vars = { + "CLIENTS": 14, + "REGISTRY_FILE": "registry_file", + "K6_SETUP_TIMEOUT": "setup_timeout", + "NO_VERIFY_SSL": True, + "TIME_LIMIT": 11, + } + + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params, set_empty", [(LoadScenario.gRPC, True)], indirect=True) + def test_empty_argument_parsing_for_grpc_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '0'", + "--preload_obj '0'", + "--out ''", + "--workers '0'", + "--containers '0'", + "--policy ''", + ] + expected_env_vars = { + "DURATION": 0, + "WRITE_OBJ_SIZE": 0, + "REGISTRY_FILE": "", + "K6_MIN_ITERATION_DURATION": "", + "K6_SETUP_TIMEOUT": "", + "WRITERS": 0, + "READERS": 0, + "DELETERS": 0, + "PREGEN_JSON": "", + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize( + "load_params, set_empty", [(LoadScenario.gRPC_CAR, True)], indirect=True + ) + def test_empty_argument_parsing_for_grpc_car_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '0'", + "--preload_obj '0'", + "--out ''", + "--workers '0'", + "--containers '0'", + "--policy ''", + ] + expected_env_vars = { + "DURATION": 0, + "WRITE_OBJ_SIZE": 0, + "REGISTRY_FILE": "", + "K6_MIN_ITERATION_DURATION": "", + "K6_SETUP_TIMEOUT": "", + "MAX_WRITERS": 0, + "MAX_READERS": 0, + "MAX_DELETERS": 0, + "PRE_ALLOC_DELETERS": 0, + "PRE_ALLOC_READERS": 0, + "PRE_ALLOC_WRITERS": 0, + "PREGEN_JSON": "", + "TIME_UNIT": "", + "WRITE_RATE": 0, + "READ_RATE": 0, + "DELETE_RATE": 0, + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params, set_empty", [(LoadScenario.S3, True)], indirect=True) + def test_empty_argument_parsing_for_s3_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '0'", + "--preload_obj '0'", + "--out ''", + "--workers '0'", + "--buckets '0'", + "--location ''", + ] + expected_env_vars = { + "DURATION": 0, + "WRITE_OBJ_SIZE": 0, + "REGISTRY_FILE": "", + "K6_MIN_ITERATION_DURATION": "", + "K6_SETUP_TIMEOUT": "", + "WRITERS": 0, + "READERS": 0, + "DELETERS": 0, + "NO_VERIFY_SSL": False, + "PREGEN_JSON": "", + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params, set_empty", [(LoadScenario.S3_CAR, True)], indirect=True) + def test_empty_argument_parsing_for_s3_car_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '0'", + "--preload_obj '0'", + "--out ''", + "--workers '0'", + "--buckets '0'", + "--location ''", + ] + expected_env_vars = { + "DURATION": 0, + "WRITE_OBJ_SIZE": 0, + "REGISTRY_FILE": "", + "K6_MIN_ITERATION_DURATION": "", + "K6_SETUP_TIMEOUT": "", + "NO_VERIFY_SSL": False, + "MAX_WRITERS": 0, + "MAX_READERS": 0, + "MAX_DELETERS": 0, + "PRE_ALLOC_DELETERS": 0, + "PRE_ALLOC_READERS": 0, + "PRE_ALLOC_WRITERS": 0, + "PREGEN_JSON": "", + "TIME_UNIT": "", + "WRITE_RATE": 0, + "READ_RATE": 0, + "DELETE_RATE": 0, + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params, set_empty", [(LoadScenario.HTTP, True)], indirect=True) + def test_empty_argument_parsing_for_http_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '0'", + "--preload_obj '0'", + "--out ''", + "--workers '0'", + "--containers '0'", + "--policy ''", + ] + expected_env_vars = { + "DURATION": 0, + "WRITE_OBJ_SIZE": 0, + "NO_VERIFY_SSL": False, + "REGISTRY_FILE": "", + "K6_MIN_ITERATION_DURATION": "", + "K6_SETUP_TIMEOUT": "", + "WRITERS": 0, + "READERS": 0, + "DELETERS": 0, + "PREGEN_JSON": "", + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize("load_params, set_empty", [(LoadScenario.LOCAL, True)], indirect=True) + def test_empty_argument_parsing_for_local_scenario(self, load_params: LoadParams): + expected_preset_args = [ + "--size '0'", + "--preload_obj '0'", + "--out ''", + "--workers '0'", + "--containers '0'", + "--policy ''", + ] + expected_env_vars = { + "CONFIG_FILE": "", + "DURATION": 0, + "WRITE_OBJ_SIZE": 0, + "REGISTRY_FILE": "", + "K6_MIN_ITERATION_DURATION": "", + "K6_SETUP_TIMEOUT": "", + "WRITERS": 0, + "READERS": 0, + "DELETERS": 0, + "PREGEN_JSON": "", + } + + self._check_preset_params(load_params, expected_preset_args) + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize( + "load_params, load_type, set_empty", + [(LoadScenario.VERIFY, LoadType.S3, True)], + indirect=True, + ) + def test_empty_argument_parsing_for_s3_verify_scenario(self, load_params: LoadParams): + expected_env_vars = { + "CLIENTS": 0, + "REGISTRY_FILE": "", + "K6_SETUP_TIMEOUT": "", + "NO_VERIFY_SSL": False, + "TIME_LIMIT": 0, + } + + self._check_env_vars(load_params, expected_env_vars) + + @pytest.mark.parametrize( + "load_params, load_type, set_empty", + [(LoadScenario.VERIFY, LoadType.gRPC, True)], + indirect=True, + ) + def test_argument_parsing_for_grpc_verify_scenario(self, load_params: LoadParams): + expected_env_vars = { + "CLIENTS": 0, + "REGISTRY_FILE": "", + "K6_SETUP_TIMEOUT": "", + "NO_VERIFY_SSL": False, + "TIME_LIMIT": 0, + } + + self._check_env_vars(load_params, expected_env_vars) + + def _check_preset_params(self, load_params: LoadParams, expected_preset_args: list[str]): + preset_parameters = load_params.get_preset_arguments() + assert sorted(preset_parameters) == sorted(expected_preset_args) + + def _check_env_vars(self, load_params: LoadParams, expected_env_vars: dict[str, str]): + env_vars = load_params.get_env_vars() + assert env_vars == expected_env_vars + + def _check_all_values_none(self, dataclass, skip_fields=None): + if skip_fields is None: + skip_fields = [] + + dataclass_fields = [field for field in fields(dataclass) if field.name not in skip_fields] + for field in dataclass_fields: + value = getattr(dataclass, field.name) + assert value is None, f"{field.name} is not None" + + def _check_all_values_not_none(self, dataclass, skip_fields=None): + if skip_fields is None: + skip_fields = [] + + dataclass_fields = [field for field in fields(dataclass) if field.name not in skip_fields] + for field in dataclass_fields: + value = getattr(dataclass, field.name) + assert value is not None, f"{field.name} is not None" + + def _get_filled_load_params( + self, load_type: LoadType, load_scenario: LoadScenario, set_emtpy: bool = False + ) -> LoadParams: + load_type_map = { + LoadScenario.S3: LoadType.S3, + LoadScenario.S3_CAR: LoadType.S3, + LoadScenario.gRPC: LoadType.gRPC, + LoadScenario.gRPC_CAR: LoadType.gRPC, + LoadScenario.LOCAL: LoadType.gRPC, + LoadScenario.HTTP: LoadType.HTTP, + } + load_type = load_type_map[load_scenario] if not load_type else load_type + + load_params = LoadParams(load_type) + load_params.scenario = load_scenario + load_params.preset = Preset() + + meta_fields = self._get_meta_fields(load_params) + for field in meta_fields: + if ( + getattr(field.instance, field.field.name) is None + and load_params.scenario in field.field.metadata["applicable_scenarios"] + ): + value_to_set_map = { + int: 0 if set_emtpy else len(field.field.name), + str: "" if set_emtpy else field.field.name, + bool: False if set_emtpy else True, + } + value_to_set = value_to_set_map[field.field_type] + setattr(field.instance, field.field.name, value_to_set) + + return load_params + + def _get_actual_field_type(self, field: Field) -> type: + return get_args(field.type)[0] if len(get_args(field.type)) else get_args(field.type) + + def _get_meta_fields(self, instance): + data_fields = fields(instance) + fields_with_data = [ + MetaTestField(field, self._get_actual_field_type(field), instance) + for field in data_fields + if field.metadata + ] + + for field in data_fields: + actual_field_type = self._get_actual_field_type(field) + if is_dataclass(actual_field_type) and getattr(instance, field.name): + fields_with_data += self._get_meta_fields(getattr(instance, field.name)) + + return fields_with_data or []