Updates for s3 k6

src/frostfs_testlib/load/load_config.py

@@ -1,7 +1,8 @@
 import os
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, fields, is_dataclass
 from enum import Enum
-from typing import Optional
+from types import MappingProxyType
+from typing import Any, Optional, get_args
 
 
 class LoadType(Enum):
@@ -42,6 +43,12 @@ grpc_preset_scenarios = [
 s3_preset_scenarios = [LoadScenario.S3, LoadScenario.S3_CAR]
 
 
+@dataclass
+class MetaField:
+    metadata: MappingProxyType
+    value: Any
+
+
 def metadata_field(
     applicable_scenarios: list[LoadScenario],
     preset_param: Optional[str] = None,
@@ -138,6 +145,12 @@ class LoadParams:
     preset: Optional[Preset] = None
     # K6 download url
     k6_url: Optional[str] = None
+    # No ssl verification flag
+    no_verify_ssl: Optional[bool] = metadata_field(
+        [LoadScenario.S3, LoadScenario.S3_CAR, LoadScenario.VERIFY, LoadScenario.HTTP],
+        "no-verify-ssl",
+        "NO_VERIFY_SSL",
+    )
 
     # ------- COMMON SCENARIO PARAMS -------
     # Load time is the maximum duration for k6 to give load. Default is the BACKGROUND_LOAD_DEFAULT_TIME value.
@@ -225,3 +238,53 @@ class LoadParams:
         self.registry_file = os.path.join(self.working_dir, f"{load_id}_registry.bolt")
         if self.preset:
             self.preset.pregen_json = os.path.join(self.working_dir, f"{load_id}_prepare.json")
+
+    def get_env_vars(self):
+        env_vars = {
+            meta_field.metadata["env_variable"]: meta_field.value
+            for meta_field in self._get_meta_fields(self)
+            if self.scenario in meta_field.metadata["applicable_scenarios"]
+            and meta_field.metadata["env_variable"]
+            and meta_field.value
+        }
+
+        return env_vars
+
+    def get_preset_arguments(self):
+        command_args = [
+            self._get_preset_argument(meta_field)
+            for meta_field in self._get_meta_fields(self)
+            if self.scenario in meta_field.metadata["applicable_scenarios"]
+            and meta_field.metadata["preset_argument"]
+            and meta_field.value
+            and self._get_preset_argument(meta_field)
+        ]
+
+        return command_args
+
+    @staticmethod
+    def _get_preset_argument(meta_field: MetaField) -> str:
+        if isinstance(meta_field.value, bool):
+            # For preset calls, bool values are passed with just --<argument_name> if the value is True
+            return f"--{meta_field.metadata['preset_argument']}" if meta_field.value else ""
+
+        return f"--{meta_field.metadata['preset_argument']} '{meta_field.value}'"
+
+    @staticmethod
+    def _get_meta_fields(instance) -> list[MetaField]:
+        data_fields = fields(instance)
+
+        fields_with_data = [
+            MetaField(field.metadata, getattr(instance, field.name))
+            for field in data_fields
+            if field.metadata and getattr(instance, field.name)
+        ]
+
+        for field in data_fields:
+            actual_field_type = (
+                get_args(field.type)[0] if len(get_args(field.type)) else get_args(field.type)
+            )
+            if is_dataclass(actual_field_type) and getattr(instance, field.name):
+                fields_with_data += LoadParams._get_meta_fields(getattr(instance, field.name))
+
+        return fields_with_data or []