|
37 | 37 | from mxops.enums import LogGroupEnum, NetworkEnum, parse_network_enum |
38 | 38 | from mxops.execution import utils |
39 | 39 | from mxops.execution.account import AccountsManager |
40 | | -from mxops.smart_values import SmartInt, SmartPath, SmartValue |
| 40 | +from mxops.smart_values import SmartDict, SmartInt, SmartPath, SmartValue |
41 | 41 | from mxops.smart_values.mx_sdk import SmartAddress, SmartAddresses |
42 | 42 | from mxops.smart_values.native import SmartBool, SmartDatetime, SmartStr |
43 | 43 | from mxops.execution.steps.base import Step |
@@ -258,6 +258,60 @@ def _execute(self): |
258 | 258 | egld_transfer_step.execute() |
259 | 259 |
|
260 | 260 |
|
| 261 | +@dataclass |
| 262 | +class ChainSimulatorSetStateStep(Step): |
| 263 | + """ |
| 264 | + Represents a step to set specific storage key-value pairs for an address |
| 265 | + on the chain simulator using hex-encoded keys and values. |
| 266 | + """ |
| 267 | + |
| 268 | + address: SmartAddress |
| 269 | + keys: SmartDict |
| 270 | + ALLOWED_NETWORKS: ClassVar[set] = (NetworkEnum.CHAIN_SIMULATOR,) |
| 271 | + |
| 272 | + @staticmethod |
| 273 | + def _validate_hex_keys(keys: dict[str, str]): |
| 274 | + """Validate that all keys and values are valid hex strings.""" |
| 275 | + if not keys: |
| 276 | + raise errors.InvalidSceneDefinition( |
| 277 | + "ChainSimulatorSetState requires at least one key-value pair" |
| 278 | + ) |
| 279 | + for k, v in keys.items(): |
| 280 | + try: |
| 281 | + bytes.fromhex(k) |
| 282 | + except ValueError as exc: |
| 283 | + raise errors.InvalidSceneDefinition( |
| 284 | + f"ChainSimulatorSetState key must be hex-encoded, got: '{k}'" |
| 285 | + ) from exc |
| 286 | + try: |
| 287 | + bytes.fromhex(v) |
| 288 | + except ValueError as exc: |
| 289 | + raise errors.InvalidSceneDefinition( |
| 290 | + f"ChainSimulatorSetState value must be hex-encoded, got: '{v}'" |
| 291 | + ) from exc |
| 292 | + |
| 293 | + def _execute(self): |
| 294 | + """ |
| 295 | + Post hex-encoded key-value pairs to the chain simulator's |
| 296 | + address-specific set-state endpoint. |
| 297 | + """ |
| 298 | + logger = ScenarioData.get_scenario_logger(LogGroupEnum.EXEC) |
| 299 | + scenario_data = ScenarioData.get() |
| 300 | + if scenario_data.network not in self.ALLOWED_NETWORKS: |
| 301 | + raise errors.WrongNetworkForStep( |
| 302 | + scenario_data.network, self.ALLOWED_NETWORKS |
| 303 | + ) |
| 304 | + proxy = MyProxyNetworkProvider() |
| 305 | + bech32 = self.address.get_evaluated_value().to_bech32() |
| 306 | + keys = self.keys.get_evaluated_value() |
| 307 | + self._validate_hex_keys(keys) |
| 308 | + logger.info( |
| 309 | + f"Setting {len(keys)} storage key(s) for {bech32} on chain simulator" |
| 310 | + ) |
| 311 | + response = proxy.set_address_state(bech32, keys) |
| 312 | + logger.debug(f"set-state response: {response.to_dictionary()}") |
| 313 | + |
| 314 | + |
261 | 315 | @dataclass |
262 | 316 | class AccountCloneStep(Step): |
263 | 317 | """ |
|
0 commit comments