diff --git a/.showrunner/flowrunner-cli.json b/.showrunner/flowrunner-cli.json index 77a15ae..6a4a6b5 100644 --- a/.showrunner/flowrunner-cli.json +++ b/.showrunner/flowrunner-cli.json @@ -190,7 +190,7 @@ ] }, "image": { - "repo_tag": "razor29/flowrunner-cli:v1.1.4", + "repo_tag": "razor29/flowrunner-cli:v1.2.0", "description": "" } -} \ No newline at end of file +} diff --git a/README.md b/README.md index 573e26b..8a16f17 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # FlowRunner (Automated API Flow Execution Engine) -**Version:** 1.1.3 +**Version:** 1.2.0 **Status:** Stable ## 1. Overview @@ -26,16 +26,19 @@ FlowRunner is a powerful, UI-less engine designed for the automated execution of FlowRunner is designed to execute flows exported from a companion graphical flow authoring application, ensuring consistency between flow design and automated execution. -## 2. Key Features (Version 1.1.3) +## 2. Key Features (Version 1.2.0) * **Flow Execution:** * Runs multi-step API flows defined in a JSON format. - * Supports **Request Steps** (GET, POST, PUT, PATCH, DELETE, etc.), **Condition Steps** (if/then/else), and **Loop Steps** (for-each). + * Supports **Request Steps** (GET, POST, PUT, PATCH, DELETE, etc.), **Condition Steps** (if/then/else), **Loop Steps** (for-each), and **Transform Steps** (ordered data operations). * **Variable Management:** * **Static Variables:** Define global key-value pairs for a flow run. * **Dynamic Extraction:** Extract status codes, headers, and body values (including implicit `body.` paths) into variables. * **Substitution:** Use `{{variableName}}` syntax in URLs, headers, and request bodies. `##VAR:string:name##` and `##VAR:unquoted:name##` allow precise JSON value injection. - * **Special Variables:** The `{{RANDOM_IP}}` variable generates a random public IPv4 address once per flow run. The IP remains consistent throughout all steps in a single flow execution cycle, making it ideal for headers like `X-Forwarded-For` or `X-Real-IP`. The IP is regenerated for each new flow cycle. + * **Special Variables:** Generated once per flow iteration and cached for the run. + * `{{RANDOM_IP}}`: Random public IPv4 address (stable during a single flow iteration). + * `{{RANDOM_INT}}` / `{{RANDOM_INT(min,max)}}`: Random integer (default range 0–1000000). + * `{{RANDOM_STRING}}` / `{{RANDOM_STRING(length)}}`: Random alphanumeric string (default length 12). * **URL Handling:** * **Global Target URL:** Configure a primary base URL for all flow operations. * **DNS Override:** Optionally override DNS resolution for the global target URL. @@ -327,7 +330,30 @@ Defines iteration over a list. * Inside the loop `steps`, `{{loopVariable}}` (e.g., `{{item}}`) and `{{loopVariable_index}}` will be available in the context. -### 5.3. URL Construction Logic (Version 1.1.3 Behavior) +#### 5.2.4. `type: "transform"` + +Defines ordered data transformations that write results into the context. + +```json +{ + "id": "string", + "name": "string (optional)", + "type": "transform", + "ops": [ + { + "op": "string (e.g., math_add, json_set, base64_encode)", + "set": "string (context variable name to store result)", + "args": [ /* optional positional arguments */ ], + "options": { /* optional op-specific options */ } + } + ] +} +``` + +* **Supported ops:** `base64_encode`, `base64_decode`, `jwt_encode`, `jwt_decode`, `json_set`, `math_add`, `math_sub`, `math_mul`, `math_div`, `to_number`, `to_string`, `to_boolean`, `boolean_not`. +* **References:** Use `{ "ref": "path.to.value" }` or `"{{path.to.value}}"` inside `args`/`options` to pull from context. + +### 5.3. URL Construction Logic (Version 1.2.0 Behavior) The final URL for each Request step depends on `override_step_url_host`: @@ -341,7 +367,7 @@ The final URL for each Request step depends on `override_step_url_host`: * If it is relative, it is appended to `flow_target_url`. 4. **DNS Override:** When `flow_target_dns_override` is set, requests are directed to that IP while the `Host` header reflects the original hostname. -**URL Override Update (v1.1.3):** `config.override_step_url_host` now controls how final request URLs are built. When `true` (default) the scheme/host/port come exclusively from `flow_target_url` and the step only provides the path/query. Set to `false` to allow absolute step URLs as in v1.0.0. +**URL Override Update (v1.2.0):** `config.override_step_url_host` controls how final request URLs are built. When `true` (default) the scheme/host/port come exclusively from `flow_target_url` and the step only provides the path/query. Set to `false` to allow absolute step URLs as in v1.0.0. ## 6. Deployment & Usage (Docker) @@ -349,11 +375,15 @@ Refer to the provided `Dockerfile` and `requirements.txt`. 1. **Build the Docker Image:** ```bash - docker build -t flowrunner-engine:1.1.4 . + docker build -t flowrunner-engine:1.2.0 . ``` 2. **Run the Container:** ```bash - docker run -d -p 8080:8080 --name my-flowrunner flowrunner-engine:1.1.4 + docker run -d -p 8080:8080 --name my-flowrunner flowrunner-engine:1.2.0 + ``` +3. **Prebuilt Image (Recommended):** + ```bash + docker run -d -p 8080:8080 --name my-flowrunner razor29/flowrunner-cli:v1.2.0 ``` * The API will be available on `http://localhost:8080`. * Consider volume mounting for persistent configurations or logs if needed. @@ -477,6 +507,7 @@ Stop with `Ctrl+C` when finished. } ``` In this example, the same random IP is used across all steps in a single flow execution cycle. When the flow repeats (new cycle), a different random IP is generated. + * **Randomized Values:** Use `{{RANDOM_INT(min,max)}}` and `{{RANDOM_STRING(length)}}` in URLs, headers, or JSON bodies to generate stable-per-iteration randomized values. * Verify conditional logic: Is the `conditionData` correct and evaluating as intended? * Inspect loop sources: Is the `source` variable resolving to a valid list? * Examine extraction rules: Are paths correct? Are there extraction failure warnings in logs or metrics? diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index f147b16..748e24f 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,10 +1,10 @@ -# Release Notes: FlowRunner CLI v1.1.3 +# Release Notes: FlowRunner CLI v1.2.0 ## Highlights -- **Container Control Core v2.0 integration** for unified lifecycle management and API endpoints. -- **Run once and step delay overrides** via updated `run_local_flow.sh` script and direct invoker. -- **Enhanced release pipeline** with GitHub workflow updates. -- **Documentation updates** and improved local execution options. +- **Transform step support** for ordered operations (base64/JWT, JSON set, math, and conversions). +- **New special variables** `RANDOM_INT` and `RANDOM_STRING` (cached per flow iteration). +- **Updated flow compatibility** with FlowRunner UI v1.2.0 exports. +- **Documentation updates** and refreshed Docker image tag. -This release aligns the CLI with the latest Container Control capabilities while giving operators more control when executing flows locally. See the README for detailed usage instructions. +Use the published image: `razor29/flowrunner-cli:v1.2.0`. See the README for detailed usage instructions. diff --git a/flow_runner.py b/flow_runner.py index 38d3e2c..c7d5a28 100644 --- a/flow_runner.py +++ b/flow_runner.py @@ -2,6 +2,9 @@ import asyncio import aiohttp +import base64 +import hashlib +import hmac import json import random import time @@ -135,17 +138,30 @@ class LoopStep(BaseStep): loopVariable: str = Field(..., description="Name for the variable representing each item in the loop (e.g., 'item')") steps: List["FlowStep"] = Field(default_factory=list, description="Steps to execute for each item in the loop") +class TransformOp(BaseModel): + op: str = Field(..., description="Transform operation name") + set: str = Field(..., description="Context variable name to store the result") + args: List[Any] = Field(default_factory=list, description="Arguments for the transform operation") + options: Dict[str, Any] = Field(default_factory=dict, description="Options for the transform operation") + + model_config = ConfigDict(extra="ignore") + +class TransformStep(BaseStep): + type: Literal['transform'] = Field(..., description="Specifies the step type as 'transform'") + ops: List[TransformOp] = Field(default_factory=list, description="List of transform operations to execute") + # ------------------------------------------------------------------ # Make FlowStep a discriminated union using Annotated # ------------------------------------------------------------------ FlowStep = Annotated[ - Union[RequestStep, ConditionStep, LoopStep], + Union[RequestStep, ConditionStep, LoopStep, TransformStep], Field(discriminator='type') ] # Update nested references in ConditionStep and LoopStep ConditionStep.model_rebuild() LoopStep.model_rebuild() +TransformStep.model_rebuild() class FlowMap(BaseModel): id: Optional[str | int] = Field( @@ -608,6 +624,546 @@ def set_value_in_context(context: Dict[str, Any], key: str, value: Any): logger.error(f"Unexpected error setting context key '{key}' at path '{processed_path}': {e}", exc_info=False) +# --------------------------- +# Special Variables & Transform Ops +# --------------------------- + +_RANDOM_INT_DEFAULT_MIN = 0 +_RANDOM_INT_DEFAULT_MAX = 1000000 +_RANDOM_STRING_DEFAULT_LENGTH = 12 +_RANDOM_STRING_MAX_LENGTH = 256 +_RANDOM_STRING_CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + +_TRANSFORM_OP_DEFS: Dict[str, Dict[str, Any]] = { + "base64_decode": {"args": 1, "options": {"base64": "url", "as": "text"}}, + "base64_encode": {"args": 1, "options": {"base64": "url", "padding": "strip"}}, + "jwt_decode": {"args": 1, "options": {"base64": "url", "stripBearer": "true"}}, + "jwt_encode": {"args": 3, "options": {"base64": "url", "signatureMode": "reuse", "algorithm": "HS256"}}, + "json_set": {"args": 3, "options": {}}, + "math_add": {"args": 2, "options": {}}, + "math_sub": {"args": 2, "options": {}}, + "math_mul": {"args": 2, "options": {}}, + "math_div": {"args": 2, "options": {}}, + "to_number": {"args": 1, "options": {}}, + "to_string": {"args": 1, "options": {}}, + "to_boolean": {"args": 1, "options": {}}, + "boolean_not": {"args": 1, "options": {}}, +} + + +def _parse_function_args(ref: str, name: str) -> Optional[List[str]]: + pattern = re.compile(rf"^{re.escape(name)}\s*(?:\(([^)]*)\))?$") + match = pattern.match(ref) + if not match: + return None + args = match.group(1) + if not args: + return [] + return [part.strip() for part in args.split(",") if part.strip()] + + +def _parse_int(value: Any) -> Optional[int]: + if value is None: + return None + text = str(value).strip() + if not text: + return None + try: + return int(text, 10) + except ValueError: + try: + return int(float(text)) + except (ValueError, TypeError): + return None + + +def _generate_random_int(min_value: int, max_value: int) -> int: + if max_value < min_value: + min_value, max_value = max_value, min_value + return random.randint(min_value, max_value) + + +def _generate_random_string(length: int) -> str: + safe_length = max(1, min(length, _RANDOM_STRING_MAX_LENGTH)) + return "".join(random.choice(_RANDOM_STRING_CHARS) for _ in range(safe_length)) + + +def _normalize_ref_path(ref_path: str) -> str: + if not ref_path or not isinstance(ref_path, str): + return "" + trimmed = ref_path.strip() + if trimmed.startswith("{{") and trimmed.endswith("}}"): + return trimmed[2:-2].strip() + return trimmed + + +def _normalize_boolean(value: Any, fallback: bool) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + lower = value.strip().lower() + if lower == "true": + return True + if lower == "false": + return False + return fallback + + +def _normalize_transform_op(op: Any) -> Dict[str, Any]: + if isinstance(op, TransformOp): + safe_op: Dict[str, Any] = op.model_dump() + elif isinstance(op, dict): + safe_op = op + else: + safe_op = {} + + op_name = safe_op.get("op") + if op_name not in _TRANSFORM_OP_DEFS: + op_name = "base64_decode" + + defn = _TRANSFORM_OP_DEFS[op_name] + raw_args = safe_op.get("args") + args = list(raw_args) if isinstance(raw_args, list) else [] + args = args[: defn["args"]] + while len(args) < defn["args"]: + args.append("") + + options: Dict[str, Any] = {} + provided_options = safe_op.get("options") + if isinstance(provided_options, dict): + for key, default in defn["options"].items(): + if key in provided_options: + options[key] = provided_options[key] + elif default is not None: + options[key] = default + else: + options = {key: value for key, value in defn["options"].items() if value is not None} + + set_name = safe_op.get("set") if isinstance(safe_op.get("set"), str) else "" + + return { + "op": op_name, + "set": set_name, + "args": args, + "options": options, + } + + +def _resolve_transform_value(value: Any, context: Dict[str, Any], evaluate_path: Optional[Any]) -> Any: + if isinstance(value, dict) and not isinstance(value, TransformOp): + keys = list(value.keys()) + if len(keys) == 1 and keys[0] == "ref": + path = _normalize_ref_path(value.get("ref", "")) + if not path: + raise ValueError("Transform reference is empty.") + if not callable(evaluate_path): + raise ValueError("Transform reference requires a path evaluator.") + resolved = evaluate_path(context, path) + if resolved is _MISSING: + raise ValueError(f'Transform reference "{{{{{path}}}}}" is undefined.') + return resolved + resolved_obj: Dict[str, Any] = {} + for key, item in value.items(): + resolved_obj[key] = _resolve_transform_value(item, context, evaluate_path) + return resolved_obj + + if isinstance(value, list): + return [_resolve_transform_value(item, context, evaluate_path) for item in value] + + if isinstance(value, str): + trimmed = value.strip() + if trimmed.startswith("{{") and trimmed.endswith("}}"): + path = _normalize_ref_path(trimmed) + if not path: + raise ValueError("Transform reference is empty.") + if not callable(evaluate_path): + raise ValueError("Transform reference requires a path evaluator.") + resolved = evaluate_path(context, path) + if resolved is _MISSING: + raise ValueError(f'Transform reference "{{{{{path}}}}}" is undefined.') + return resolved + return value + + +def _normalize_base64_variant(value: Any, fallback: str) -> str: + return value if value in ("standard", "url") else fallback + + +def _normalize_base64_padding(value: Any, fallback: str) -> str: + return value if value in ("keep", "strip", "add") else fallback + + +def _normalize_decode_output(value: Any) -> str: + return value if value in ("text", "json") else "text" + + +def _normalize_signature_mode(value: Any) -> str: + return value if value in ("reuse", "none", "sign") else "reuse" + + +def _normalize_signature_algorithm(value: Any) -> str: + return value if value in ("HS256", "HS384", "HS512") else "HS256" + + +def _normalize_text_input(input_value: Any) -> str: + if input_value is None: + return "" + if isinstance(input_value, str): + return input_value + try: + return json.dumps(input_value) + except Exception: + return str(input_value) + + +def _add_base64_padding(value: str) -> str: + mod = len(value) % 4 + if mod == 0: + return value + return value + "=" * (4 - mod) + + +def _apply_base64_variant(base64_value: str, variant: str, padding: str) -> str: + value = base64_value + if variant == "url": + value = value.replace("+", "-").replace("/", "_") + if padding == "strip": + value = value.rstrip("=") + elif padding == "add": + value = _add_base64_padding(value) + return value + + +def _normalize_base64_for_decode(value: str, variant: str) -> str: + normalized = re.sub(r"\s+", "", value) + if variant == "url": + normalized = normalized.replace("-", "+").replace("_", "/") + return _add_base64_padding(normalized) + + +def _encode_text_to_base64(text: str) -> str: + encoded = base64.b64encode(text.encode("utf-8")) + return encoded.decode("ascii") + + +def _decode_base64_to_text(value: str) -> str: + decoded = base64.b64decode(value) + return decoded.decode("utf-8", errors="replace") + + +def _normalize_json_input(value: Any, label: str) -> str: + if value is None: + raise ValueError(f"{label} is required.") + if isinstance(value, str): + trimmed = value.strip() + if not trimmed: + raise ValueError(f"{label} is empty.") + try: + json.loads(trimmed) + return trimmed + except json.JSONDecodeError as error: + raise ValueError(f"{label} must be valid JSON: {error.msg}") from error + try: + return json.dumps(value) + except Exception as error: + raise ValueError(f"{label} could not be stringified: {error}") from error + + +def _to_number(value: Any) -> float: + if value is None: + return 0 + if isinstance(value, bool): + return 1.0 if value else 0.0 + if isinstance(value, (int, float)) and not isinstance(value, bool): + if isinstance(value, float) and math.isnan(value): + raise ValueError(f'Value "{value}" is not a number.') + return float(value) + if isinstance(value, str): + text = value.strip() + if text == "": + return 0.0 + try: + if re.match(r"^[+-]?0[xX][0-9a-fA-F]+$", text) or re.match(r"^[+-]?0[bB][01]+$", text) or re.match(r"^[+-]?0[oO][0-7]+$", text): + return float(int(text, 0)) + return float(text) + except ValueError as error: + raise ValueError(f'Value "{value}" is not a number.') from error + if isinstance(value, list): + if not value: + return 0.0 + if len(value) == 1: + return _to_number(value[0]) + raise ValueError(f'Value "{value}" is not a number.') + raise ValueError(f'Value "{value}" is not a number.') + + +def _to_string_value(value: Any) -> str: + return "" if value is None else str(value) + + +def _to_boolean(value: Any) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + lower = value.strip().lower() + if lower == "true": + return True + if lower == "false": + return False + return bool(value) + + +def _tokenize_path(path: str) -> List[str]: + if not path: + return [] + tokens: List[str] = [] + current: List[str] = [] + in_bracket = False + quote: Optional[str] = None + for ch in path: + if in_bracket: + if quote: + if ch == quote: + quote = None + else: + current.append(ch) + continue + if ch in ("\"", "'"): + quote = ch + continue + if ch == "]": + if current: + tokens.append("".join(current)) + current = [] + in_bracket = False + continue + if ch != "[": + current.append(ch) + continue + if ch == ".": + if current: + tokens.append("".join(current)) + current = [] + continue + if ch == "[": + if current: + tokens.append("".join(current)) + current = [] + in_bracket = True + continue + current.append(ch) + if current: + tokens.append("".join(current)) + return [token for token in tokens if token] + + +def _clone_value(value: Any) -> Any: + if isinstance(value, list): + return [_clone_value(item) for item in value] + if isinstance(value, dict): + return {key: _clone_value(val) for key, val in value.items()} + return value + + +def _set_path_value(root: Any, tokens: List[str], value: Any) -> Any: + if not tokens: + return _clone_value(value) + + if isinstance(root, list): + copy_root: Any = [_clone_value(item) for item in root] + elif isinstance(root, dict): + copy_root = {key: _clone_value(val) for key, val in root.items()} + else: + copy_root = {} + + current = copy_root + for index, token in enumerate(tokens): + is_last = index == len(tokens) - 1 + next_token = tokens[index + 1] if not is_last else None + next_is_index = bool(next_token and next_token.isdigit()) + + def assign(container: Any, key: str, val: Any) -> None: + if isinstance(container, list) and key.isdigit(): + idx = int(key) + if idx >= len(container): + container.extend([None] * (idx - len(container) + 1)) + container[idx] = val + elif isinstance(container, dict): + container[key] = val + + def fetch(container: Any, key: str) -> Any: + if isinstance(container, list) and key.isdigit(): + idx = int(key) + if 0 <= idx < len(container): + return container[idx] + return None + if isinstance(container, dict): + return container.get(key) + return None + + if is_last: + assign(current, token, _clone_value(value)) + break + + existing = fetch(current, token) + if isinstance(existing, (list, dict)): + next_value = _clone_value(existing) + else: + next_value = [] if next_is_index else {} + + assign(current, token, next_value) + current = next_value + + return copy_root + + +def _json_set(target: Any, path: str, value: Any) -> Any: + tokens = _tokenize_path(str(path or "").strip()) + if not tokens: + raise ValueError("JSON set requires a path.") + return _set_path_value(target, tokens, value) + + +def _execute_transform_op(op: Dict[str, Any], context: Dict[str, Any], evaluate_path: Optional[Any]) -> Any: + resolved_args = [_resolve_transform_value(arg, context, evaluate_path) for arg in op.get("args", [])] + resolved_options = { + key: _resolve_transform_value(val, context, evaluate_path) + for key, val in (op.get("options") or {}).items() + } + op_name = op.get("op") + if op_name == "base64_decode": + variant = _normalize_base64_variant(resolved_options.get("base64"), "url") + output_as = _normalize_decode_output(resolved_options.get("as")) + normalized = _normalize_base64_for_decode(str(resolved_args[0] if resolved_args else ""), variant) + text = _decode_base64_to_text(normalized) + if output_as == "json": + try: + return json.loads(text) + except json.JSONDecodeError as error: + raise ValueError(f"Base64 decode produced invalid JSON: {error.msg}") from error + return text + if op_name == "base64_encode": + variant = _normalize_base64_variant(resolved_options.get("base64"), "url") + padding = _normalize_base64_padding(resolved_options.get("padding"), "strip" if variant == "url" else "keep") + text = _normalize_text_input(resolved_args[0] if resolved_args else "") + base64_value = _encode_text_to_base64(text) + return _apply_base64_variant(base64_value, variant, padding) + if op_name == "jwt_decode": + token = str(resolved_args[0] if resolved_args else "").strip() + strip_bearer = _normalize_boolean(resolved_options.get("stripBearer"), True) + if strip_bearer and re.match(r"^bearer\s+", token, re.IGNORECASE): + token = re.sub(r"^bearer\s+", "", token, flags=re.IGNORECASE).strip() + parts = token.split(".") + if len(parts) < 2: + raise ValueError("JWT must contain at least header and payload segments.") + variant = _normalize_base64_variant(resolved_options.get("base64"), "url") + header_json = _execute_transform_op( + {"op": "base64_decode", "args": [parts[0]], "options": {"base64": variant, "as": "text"}}, + context, + evaluate_path, + ) + payload_json = _execute_transform_op( + {"op": "base64_decode", "args": [parts[1]], "options": {"base64": variant, "as": "text"}}, + context, + evaluate_path, + ) + try: + header = json.loads(header_json) + except json.JSONDecodeError as error: + raise ValueError(f"JWT header is not valid JSON: {error.msg}") from error + try: + payload = json.loads(payload_json) + except json.JSONDecodeError as error: + raise ValueError(f"JWT payload is not valid JSON: {error.msg}") from error + signature = parts[2] if len(parts) > 2 else "" + return { + "header": header, + "payload": payload, + "signature": signature, + "parts": { + "header": parts[0], + "payload": parts[1], + "signature": signature, + }, + } + if op_name == "jwt_encode": + variant = _normalize_base64_variant(resolved_options.get("base64"), "url") + header_json = _normalize_json_input(resolved_args[0] if len(resolved_args) > 0 else None, "JWT header") + payload_json = _normalize_json_input(resolved_args[1] if len(resolved_args) > 1 else None, "JWT payload") + header_part = _execute_transform_op( + {"op": "base64_encode", "args": [header_json], "options": {"base64": variant, "padding": "strip" if variant == "url" else "keep"}}, + context, + evaluate_path, + ) + payload_part = _execute_transform_op( + {"op": "base64_encode", "args": [payload_json], "options": {"base64": variant, "padding": "strip" if variant == "url" else "keep"}}, + context, + evaluate_path, + ) + signature_mode = _normalize_signature_mode(resolved_options.get("signatureMode")) + signature_part = "" + if signature_mode == "reuse": + signature_part = "" if len(resolved_args) < 3 or resolved_args[2] is None else str(resolved_args[2]) + elif signature_mode == "sign": + algorithm = _normalize_signature_algorithm(resolved_options.get("algorithm")) + secret = resolved_options.get("secret") + if secret is None or secret == "": + raise ValueError("JWT signing requires a secret.") + hash_map = { + "HS256": hashlib.sha256, + "HS384": hashlib.sha384, + "HS512": hashlib.sha512, + } + to_sign = f"{header_part}.{payload_part}" + digestmod = hash_map[algorithm] + raw = hmac.new(str(secret).encode("utf-8"), to_sign.encode("utf-8"), digestmod).digest() + base64_value = base64.b64encode(raw).decode("ascii") + signature_part = _apply_base64_variant( + base64_value, + variant, + "strip" if variant == "url" else "keep", + ) + return f"{header_part}.{payload_part}.{signature_part}" + if op_name == "json_set": + if len(resolved_args) < 3: + raise ValueError("JSON set requires target, path, and value.") + return _json_set(resolved_args[0], resolved_args[1], resolved_args[2]) + if op_name == "math_add": + return _to_number(resolved_args[0]) + _to_number(resolved_args[1]) + if op_name == "math_sub": + return _to_number(resolved_args[0]) - _to_number(resolved_args[1]) + if op_name == "math_mul": + return _to_number(resolved_args[0]) * _to_number(resolved_args[1]) + if op_name == "math_div": + divisor = _to_number(resolved_args[1]) + if divisor == 0: + raise ValueError("Division by zero.") + return _to_number(resolved_args[0]) / divisor + if op_name == "to_number": + return _to_number(resolved_args[0]) + if op_name == "to_string": + return _to_string_value(resolved_args[0]) + if op_name == "to_boolean": + return _to_boolean(resolved_args[0]) + if op_name == "boolean_not": + return not _to_boolean(resolved_args[0]) + raise ValueError(f'Unsupported transform op "{op_name}".') + + +def execute_transform_ops(ops: Any, context: Dict[str, Any], evaluate_path: Optional[Any] = None) -> Dict[str, Any]: + output = {"updatedVars": [], "warnings": []} + ops_list = ops if isinstance(ops, list) else [] + for index, op in enumerate(ops_list): + normalized = _normalize_transform_op(op) + set_name = normalized.get("set") + if not isinstance(set_name, str) or not set_name: + raise ValueError(f"Transform op {index + 1} is missing a valid output variable.") + value = _execute_transform_op(normalized, context, evaluate_path) + context[set_name] = value + output["updatedVars"].append(set_name) + return output + + def get_header_value_case_insensitive(headers: Dict[str, Any], header_name: str) -> Optional[Any]: """ Return the header value for header_name from headers using case-insensitive lookup. @@ -1047,6 +1603,63 @@ def get_active_user_count(self) -> int: # but currently only accessed internally or via control thread which should be safe enough. return self._active_users_count + def _get_random_cache(self, context: Dict[str, Any]) -> Dict[str, Any]: + cache = context.get("_RANDOM_CACHE") + if not isinstance(cache, dict): + cache = {} + context["_RANDOM_CACHE"] = cache + return cache + + def _resolve_special_variable( + self, + var_path: str, + context: Dict[str, Any], + *, + as_raw: bool = False, + ) -> Optional[Any]: + if not isinstance(context, dict): + return None + + trimmed = var_path.strip() + if trimmed == "RANDOM_IP": + if "_RANDOM_IP" not in context: + random_ip = self.generate_random_ip() + context["_RANDOM_IP"] = random_ip + logger.debug(f"Generated random IP for flow run: {random_ip}") + return context["_RANDOM_IP"] + + int_args = _parse_function_args(trimmed, "RANDOM_INT") + if int_args is not None: + min_val = _RANDOM_INT_DEFAULT_MIN + max_val = _RANDOM_INT_DEFAULT_MAX + if len(int_args) == 1: + parsed_max = _parse_int(int_args[0]) + max_val = parsed_max if parsed_max is not None else _RANDOM_INT_DEFAULT_MAX + elif len(int_args) >= 2: + parsed_min = _parse_int(int_args[0]) + parsed_max = _parse_int(int_args[1]) + min_val = parsed_min if parsed_min is not None else _RANDOM_INT_DEFAULT_MIN + max_val = parsed_max if parsed_max is not None else _RANDOM_INT_DEFAULT_MAX + cache = self._get_random_cache(context) + if trimmed not in cache: + cache[trimmed] = _generate_random_int(min_val, max_val) + logger.debug(f"Generated random int for '{trimmed}': {cache[trimmed]}") + return cache[trimmed] if as_raw else str(cache[trimmed]) + + string_args = _parse_function_args(trimmed, "RANDOM_STRING") + if string_args is not None: + length = _RANDOM_STRING_DEFAULT_LENGTH + if len(string_args) >= 1: + parsed_length = _parse_int(string_args[0]) + length = parsed_length if parsed_length is not None else _RANDOM_STRING_DEFAULT_LENGTH + cache = self._get_random_cache(context) + if trimmed not in cache: + cache[trimmed] = _generate_random_string(length) + logger.debug(f"Generated random string for '{trimmed}': {cache[trimmed]}") + return cache[trimmed] + + return None + def _substitute_variables( self, data: Union[str, Dict, List], @@ -1071,6 +1684,16 @@ def _substitute_variables( if len(parts) != 2: raise ValueError("Invalid ##VAR format, expected type:path") var_type, var_path = parts + var_path = var_path.strip() + + special_value = self._resolve_special_variable(var_path, context, as_raw=(var_type == "unquoted")) + if special_value is not None: + if var_type == "string": + return str(special_value) + if var_type == "unquoted": + return special_value + logger.warning(f"Unsupported ##VAR type: '{var_type}' in token '{data}'. Treating as string.") + return str(special_value) # Get value using the robust getter value = get_value_from_context(context, var_path) @@ -1101,7 +1724,7 @@ def _substitute_variables( # Regular {{variable.or[0].path}} Substitution - For URLs, headers, string parts of body # This always results in a string substitution. - pattern = r"\{\{([\w\.\[\]]+?)\}\}" # Non-greedy match inside braces + pattern = r"\{\{([^}]+)\}\}" # Non-greedy match inside braces new_string = data try: # Use finditer for non-overlapping matches and correct replacement @@ -1118,29 +1741,21 @@ def _substitute_variables( # Append the literal text before the match result_parts.append(data[last_end:start]) - # Handle special reserved variable: RANDOM_IP - if var_path == 'RANDOM_IP': - # Generate random IP once per flow run and cache it in context - if '_RANDOM_IP' not in context: - random_ip = self.generate_random_ip() - context['_RANDOM_IP'] = random_ip - logger.debug(f"Generated random IP for flow run: {random_ip}") - value_str = context['_RANDOM_IP'] - result_parts.append(value_str) - last_end = end - continue - - # Get the value from context - value = get_value_from_context(context, var_path) - - # Determine the string representation for substitution - if value is _MISSING: - logger.warning(f"Variable '{{{{{var_path}}}}}' not found in context. Substituting with empty string.") - value_str = "" - elif value is None: - value_str = "" # Substitute None as empty string in {{}} context + special_value = self._resolve_special_variable(var_path, context, as_raw=False) + if special_value is not None: + value_str = str(special_value) else: - value_str = str(value) # Convert other types to string + # Get the value from context + value = get_value_from_context(context, var_path) + + # Determine the string representation for substitution + if value is _MISSING: + logger.warning(f"Variable '{{{{{var_path}}}}}' not found in context. Substituting with empty string.") + value_str = "" + elif value is None: + value_str = "" # Substitute None as empty string in {{}} context + else: + value_str = str(value) # Convert other types to string if for_url: # Avoid encoding full URLs or plain hostnames with optional ports @@ -2148,6 +2763,30 @@ async def _execute_loop_step( break + async def _execute_transform_step( + self, + step: TransformStep, + context: Dict[str, Any], + depth: int, + user_id_log: str, + ) -> None: + """Executes a TransformStep and updates context with outputs.""" + indent = " " * depth + step_identifier = f"'{step.name}' ({step.id})" if step.name else f"({step.id})" + try: + ops = step.ops or [] + logger.debug(f"{indent}User {user_id_log}: Transform {step_identifier}: Executing {len(ops)} ops.") + output = execute_transform_ops(ops, context, evaluate_path=get_value_from_context) + logger.debug(f"{indent}User {user_id_log}: Transform {step_identifier}: Updated vars {output.get('updatedVars', [])}.") + except Exception as exc: + logger.error( + f"{indent}User {user_id_log}: Transform {step_identifier} failed: {exc}", + exc_info=self.config.debug, + ) + set_value_in_context(context, 'flow_error', f"Transform {step_identifier} failed: {exc}") + return + + async def _execute_steps( self, steps: List[Union[FlowStep, Dict]], # Input list might contain dicts or models @@ -2185,7 +2824,7 @@ async def _execute_steps( # --- FIX: Dynamic Validation of Steps --- step_instance = None # Holds the validated Pydantic model instance - if isinstance(step_data, (RequestStep, ConditionStep, LoopStep)): + if isinstance(step_data, (RequestStep, ConditionStep, LoopStep, TransformStep)): # Already a validated model (likely from top-level parsing) step_instance = step_data elif isinstance(step_data, dict): @@ -2207,6 +2846,8 @@ async def _execute_steps( step_instance = ConditionStep.model_validate(step_data) elif step_type == 'loop': step_instance = LoopStep.model_validate(step_data) + elif step_type == 'transform': + step_instance = TransformStep.model_validate(step_data) else: raise ValueError(f"Unknown step type: {step_type}") @@ -2320,6 +2961,15 @@ async def _execute_steps( user_id_log, ) + # --- Transform Step --- + elif isinstance(step_instance, TransformStep): + await self._execute_transform_step( + step_instance, + context, + depth, + user_id_log, + ) + else: # Should be unreachable with validated models logger.error(f"{indent}User {user_id_log}: Encountered unknown step instance type '{type(step_instance).__name__}' for {step_identifier}. Halting sequence.") set_value_in_context(context, 'flow_error', f"Unknown step type {type(step_instance).__name__}") @@ -2560,6 +3210,7 @@ def generate_random_ip(self) -> str: FlowMap.model_rebuild() ConditionStep.model_rebuild() LoopStep.model_rebuild() +TransformStep.model_rebuild() StartRequest.model_rebuild() # Also rebuild RequestStep explicitly in case forward refs were missed? Not strictly needed here. RequestStep.model_rebuild() # Add just in case, though not strictly necessary here diff --git a/roadmap.md b/roadmap.md index 7fefa7e..4697a37 100644 --- a/roadmap.md +++ b/roadmap.md @@ -254,9 +254,11 @@ By meticulously addressing these points, the AI agent can successfully upgrade t ## Project Roadmap -### v1.1.3 (Current) +### v1.2.0 (Current) - [x] **Continuous Flow Runner (Simplified):** FlowRunner now runs indefinitely when started via the API and resets context between iterations. - [x] **Validation & Error Handling Improvements:** API responses and internal logging provide clearer details on invalid configurations and flow errors. - [x] **UI/UX Tweaks:** Logging levels can be adjusted via configuration; metrics endpoints expose detailed runtime information. - [x] **Automated Testing Framework:** Unit and end-to-end test suites exercise the FlowRunner core and API. +- [x] **Transform Steps:** Transform operations execute in order and update context variables. +- [x] **Randomization Variables:** `RANDOM_INT` and `RANDOM_STRING` are supported and cached per flow iteration. diff --git a/tests/e2e/container_control_stub.py b/tests/e2e/container_control_stub.py new file mode 100644 index 0000000..ebb9530 --- /dev/null +++ b/tests/e2e/container_control_stub.py @@ -0,0 +1,216 @@ +# Test-only container control stub used by E2E tests. + +import asyncio +import threading +import time +from typing import Any, Dict, Optional, Tuple + +import psutil +from fastapi import FastAPI, HTTPException +from fastapi.responses import PlainTextResponse +from pydantic import ValidationError + +from flow_runner import FlowRunner, Metrics, StartRequest, logger as flow_logger + +app = FastAPI() + +flow_runner: Optional[FlowRunner] = None +background_thread: Optional[threading.Thread] = None +event_loop: Optional[asyncio.AbstractEventLoop] = None +metrics: Optional[Metrics] = None + +current_settings: Dict[str, Any] = {"app_status": "initializing"} +_state_lock = threading.Lock() + + +def _app_status_to_gauge(status: str) -> int: + mapping = { + "initializing": 0, + "running": 1, + "stopped": 2, + "error": 3, + } + return mapping.get(status, 3) + + +def _run_flow_runner_in_thread(start_request: StartRequest) -> None: + global flow_runner, event_loop, metrics + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + event_loop = loop + metrics = Metrics() + + flowmaps = start_request.flowmaps or ([start_request.flowmap] if start_request.flowmap else []) + flow_runner = FlowRunner( + start_request.config, + start_request.flowmap, + metrics, + flowmaps=flowmaps, + ) + + try: + current_settings["app_status"] = "running" + loop.run_until_complete(flow_runner.start_generating()) + except Exception as exc: + flow_logger.error(f"FlowRunner thread error: {exc}") + current_settings["app_status"] = "error" + finally: + try: + loop.run_until_complete(loop.shutdown_asyncgens()) + except Exception: + pass + loop.close() + event_loop = None + flow_runner = None + if current_settings.get("app_status") == "running": + current_settings["app_status"] = "stopped" + + +def _force_stop_flow_runner() -> None: + global flow_runner, background_thread, event_loop + if flow_runner and event_loop and event_loop.is_running(): + try: + future = asyncio.run_coroutine_threadsafe(flow_runner.stop_generating(), event_loop) + future.result(timeout=5) + except Exception as exc: + flow_logger.warning(f"Failed to stop FlowRunner: {exc}") + + if background_thread and background_thread.is_alive(): + background_thread.join(timeout=5) + + flow_runner = None + event_loop = None + background_thread = None + current_settings["app_status"] = "stopped" + + +def _get_metrics_snapshot() -> Tuple[float, float, int]: + rps_value = 0.0 + avg_flow_ms = 0.0 + active_users = 0 + if flow_runner: + try: + active_users = flow_runner.get_active_user_count() + except Exception: + active_users = 0 + if metrics: + if event_loop and event_loop.is_running(): + try: + rps_future = asyncio.run_coroutine_threadsafe(metrics.get_rps(), event_loop) + rps_value = float(rps_future.result(timeout=1)) + except Exception: + rps_value = float(getattr(metrics, "last_rps_value", 0.0) or 0.0) + try: + avg_future = asyncio.run_coroutine_threadsafe(metrics.get_average_flow_duration_ms(), event_loop) + avg_flow_ms = float(avg_future.result(timeout=1)) + except Exception: + avg_flow_ms = 0.0 + else: + rps_value = float(getattr(metrics, "last_rps_value", 0.0) or 0.0) + return rps_value, avg_flow_ms, active_users + + +def _sanitize_validation_errors(errors: list[Dict[str, Any]]) -> list[Dict[str, Any]]: + sanitized: list[Dict[str, Any]] = [] + for err in errors: + entry = dict(err) + ctx = entry.get("ctx") + if isinstance(ctx, dict): + entry["ctx"] = {key: str(value) for key, value in ctx.items()} + sanitized.append(entry) + return sanitized + + +@app.post("/api/start") +async def start_flow_runner(payload: Dict[str, Any]) -> Dict[str, str]: + global background_thread + try: + start_request = StartRequest.model_validate(payload) + except ValidationError as exc: + raise HTTPException(status_code=400, detail=_sanitize_validation_errors(exc.errors())) from exc + except Exception as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + with _state_lock: + if background_thread and background_thread.is_alive(): + _force_stop_flow_runner() + + background_thread = threading.Thread( + target=_run_flow_runner_in_thread, + args=(start_request,), + daemon=True, + ) + background_thread.start() + + return {"message": "Flow runner started with the provided flowmap"} + + +@app.post("/api/stop") +async def stop_flow_runner() -> Dict[str, str]: + if not background_thread or not background_thread.is_alive(): + if current_settings.get("app_status") != "stopped": + current_settings["app_status"] = "stopped" + return {"message": "Flow runner is already stopped."} + + _force_stop_flow_runner() + return {"message": "Flow runner forcibly stopped."} + + +@app.get("/api/health") +async def health() -> Dict[str, str]: + return {"status": "healthy", "app_status": current_settings.get("app_status", "error")} + + +@app.get("/api/metrics") +async def metrics_endpoint() -> Dict[str, Any]: + cpu_percent = float(psutil.cpu_percent(interval=None)) + mem = psutil.virtual_memory() + net = psutil.net_io_counters() + rps_value, avg_flow_ms, active_users = _get_metrics_snapshot() + + return { + "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "app_status": current_settings.get("app_status", "error"), + "container_status": "running", + "network": { + "bytes_sent": int(getattr(net, "bytes_sent", 0)), + "bytes_recv": int(getattr(net, "bytes_recv", 0)), + "packets_sent": int(getattr(net, "packets_sent", 0)), + "packets_recv": int(getattr(net, "packets_recv", 0)), + }, + "system": { + "cpu_percent": cpu_percent, + "memory_percent": float(getattr(mem, "percent", 0.0)), + "memory_available_mb": float(getattr(mem, "available", 0.0)) / (1024 * 1024), + "memory_used_mb": float(getattr(mem, "used", 0.0)) / (1024 * 1024), + }, + "metrics": { + "rps": rps_value, + "active_simulated_users": active_users, + "average_flow_duration_ms": avg_flow_ms, + }, + } + + +@app.get("/metrics") +async def prometheus_metrics() -> PlainTextResponse: + rps_value, avg_flow_ms, active_users = _get_metrics_snapshot() + status_value = _app_status_to_gauge(current_settings.get("app_status", "error")) + lines = [ + "# HELP app_status Application status (initializing=0, running=1, stopped=2, error=3).", + "# TYPE app_status gauge", + f"app_status {status_value}", + "# HELP flow_runner_rps Current requests-per-second generated by flows.", + "# TYPE flow_runner_rps gauge", + f"flow_runner_rps {rps_value}", + "# HELP flow_runner_active_users Number of active simulated users.", + "# TYPE flow_runner_active_users gauge", + f"flow_runner_active_users {active_users}", + "# HELP flow_runner_avg_flow_duration_ms Average flow duration in milliseconds.", + "# TYPE flow_runner_avg_flow_duration_ms gauge", + f"flow_runner_avg_flow_duration_ms {avg_flow_ms}", + ] + return PlainTextResponse("\n".join(lines)) + + +__all__ = ["app", "background_thread", "current_settings", "_force_stop_flow_runner"] diff --git a/tests/e2e/test_container_control_api.py b/tests/e2e/test_container_control_api.py index 7a35542..be61ec7 100644 --- a/tests/e2e/test_container_control_api.py +++ b/tests/e2e/test_container_control_api.py @@ -44,7 +44,7 @@ def decorator(fn): import pytest_asyncio import json -import container_control +from tests.e2e import container_control_stub as container_control from tests.e2e.mock_server import create_mock_server, shutdown_mock_server diff --git a/tests/unit/test_flow_runner.py b/tests/unit/test_flow_runner.py index f8542ed..ad11135 100644 --- a/tests/unit/test_flow_runner.py +++ b/tests/unit/test_flow_runner.py @@ -53,6 +53,7 @@ def decorator(fn): RequestStep, LoopStep, ConditionStep, + TransformStep, ConditionData, Metrics, StartRequest, @@ -160,6 +161,50 @@ async def test_substitute_variables_string_and_markers(base_config, empty_flow): assert runner._substitute_variables("Missing {{none}}", context) == "Missing " +def test_substitute_random_variables_cached(base_config, empty_flow): + runner = make_runner(base_config, empty_flow) + context = {} + first_int = runner._substitute_variables("{{RANDOM_INT}}", context) + second_int = runner._substitute_variables("{{RANDOM_INT}}", context) + assert first_int == second_int + assert int(first_int) >= 0 + + first_range = runner._substitute_variables("{{RANDOM_INT(5, 5)}}", context) + second_range = runner._substitute_variables("{{RANDOM_INT(5, 5)}}", context) + assert first_range == "5" + assert first_range == second_range + + first_str = runner._substitute_variables("{{RANDOM_STRING(8)}}", context) + second_str = runner._substitute_variables("{{RANDOM_STRING(8)}}", context) + assert first_str == second_str + assert len(first_str) == 8 + + +def test_substitute_random_unquoted_marker(base_config, empty_flow): + runner = make_runner(base_config, empty_flow) + context = {} + value = runner._substitute_variables("##VAR:unquoted:RANDOM_INT(1, 1)##", context) + assert value == 1 + + +@pytest.mark.asyncio +async def test_transform_step_updates_context(base_config, empty_flow): + runner = make_runner(base_config, empty_flow) + context = {"payload": {"exp": 100}} + step = TransformStep( + id="t1", + name="Transform", + type="transform", + ops=[ + {"op": "math_add", "set": "sum", "args": [1, 2]}, + {"op": "json_set", "set": "payload", "args": [{"ref": "payload"}, "exp", 110]}, + ], + ) + await runner._execute_transform_step(step, context, depth=0, user_id_log="test") + assert context["sum"] == 3 + assert context["payload"]["exp"] == 110 + + def test_extract_data_status_headers_and_body(base_config, empty_flow): runner = make_runner(base_config, empty_flow) ctx: Dict[str, Any] = {} @@ -592,7 +637,7 @@ async def test_simulate_user_lifecycle_run_once(monkeypatch, empty_flow): contexts = [] - async def fake_execute_steps(steps, session_obj, base_headers=None, flow_headers=None, context=None, depth=0): + async def fake_execute_steps(steps, session=None, base_headers=None, flow_headers=None, context=None, depth=0, **kwargs): contexts.append(context.copy()) monkeypatch.setattr(runner, "_execute_steps", fake_execute_steps)