diff --git a/qase-robotframework/changelog.md b/qase-robotframework/changelog.md index c4933ea7..d0f52282 100644 --- a/qase-robotframework/changelog.md +++ b/qase-robotframework/changelog.md @@ -1,3 +1,9 @@ +# qase-robotframework 4.0.7 + +## What's new + +- Fixed an issue with handling steps which contains variables. + # qase-robotframework 4.0.6 ## What's new diff --git a/qase-robotframework/pyproject.toml b/qase-robotframework/pyproject.toml index 5f656d63..e042708d 100644 --- a/qase-robotframework/pyproject.toml +++ b/qase-robotframework/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "qase-robotframework" -version = "4.0.6" +version = "4.0.7" description = "Qase Robot Framework Plugin" readme = "README.md" authors = [{name = "Qase Team", email = "support@qase.io"}] diff --git a/qase-robotframework/src/qase/robotframework/listener.py b/qase-robotframework/src/qase/robotframework/listener.py index eceeb0a9..b35b8479 100644 --- a/qase-robotframework/src/qase/robotframework/listener.py +++ b/qase-robotframework/src/qase/robotframework/listener.py @@ -1,5 +1,6 @@ import logging import pathlib +import re import uuid from filelock import FileLock @@ -193,7 +194,235 @@ def __extract_tests_with_suites(self, suite, parent_suites=None): return test_dict - def __parse_steps(self, result) -> List[Step]: + def __extract_resolved_variables(self, body_element, accumulated_vars: dict = None) -> dict: + """ + Extract resolved variable values from log messages in the body element. + Robot Framework logs variable assignments in format: ${variable} = value + Returns a dictionary mapping variable names to their resolved values. + Accumulates variables from all nested messages recursively. + """ + if accumulated_vars is None: + accumulated_vars = {} + + if not hasattr(body_element, "body"): + return accumulated_vars + + # Pattern to match variable assignments: ${variable} = value + # This matches patterns like: ${full_url} = https://jsonplaceholder.typicode.com/users + # Also handles nested variables like ${response.status_code} + # The pattern captures everything after '=' until end of line or next variable assignment + var_pattern = re.compile(r'\$\{([^}]+)\}\s*=\s*(.+?)(?=\s+\$\{|$)') + + # Also try to extract from BuiltIn API for variables that might not be in messages + try: + builtin = BuiltIn() + except Exception: + builtin = None + + for item in body_element.body: + if hasattr(item, "type") and item.type == "MESSAGE": + # Try different possible attributes for message content + message_text = None + if hasattr(item, "message"): + message_text = str(item.message) + elif hasattr(item, "msg"): + message_text = str(item.msg) + elif hasattr(item, "text"): + message_text = str(item.text) + + if message_text: + # Try to match variable assignment pattern + match = var_pattern.search(message_text) + if match: + var_name = match.group(1) + var_value = match.group(2).strip() + # Remove trailing commas, semicolons, or other punctuation that might be captured + var_value = var_value.rstrip(',;') + # Store both with and without ${} for flexible matching + accumulated_vars[var_name] = var_value + accumulated_vars[f"${{{var_name}}}"] = var_value + # Also handle nested variable access like response.status_code + if '.' in var_name: + parts = var_name.split('.') + base_var = parts[0] + accumulated_vars[f"${{{base_var}}}"] = var_value # Store base variable too + + # Also try to extract variable names mentioned in the message + # and resolve them via BuiltIn API if available + if builtin: + # Find all variable references in the message + var_refs = re.findall(r'\$\{([^}]+)\}', message_text) + for var_ref in var_refs: + if var_ref not in accumulated_vars and f"${{{var_ref}}}" not in accumulated_vars: + try: + var_value = builtin.get_variable_value(f"${{{var_ref}}}") + if var_value is not None: + accumulated_vars[var_ref] = str(var_value) + accumulated_vars[f"${{{var_ref}}}"] = str(var_value) + except Exception: + pass + + # Recursively process nested body elements to accumulate variables + if hasattr(item, "body") and item.body: + self.__extract_resolved_variables(item, accumulated_vars) + + return accumulated_vars + + def __resolve_variables_in_data(self, data: str, resolved_vars: dict) -> str: + """ + Replace variable placeholders in data string with their resolved values. + Uses regex to match variable patterns and replace them with resolved values. + Aggressively uses BuiltIn API to resolve any remaining variables. + """ + if not data: + return data + + result = data + # Sort by length (longest first) to handle nested variables correctly + sorted_vars = sorted(resolved_vars.items(), key=lambda x: len(x[0]), reverse=True) + + for var_placeholder, var_value in sorted_vars: + # Replace all occurrences of the variable placeholder + # Use regex to match exact variable patterns (e.g., ${var} but not ${var_suffix}) + if var_placeholder.startswith('${') and var_placeholder.endswith('}'): + # Escape special regex characters in the variable name + escaped_var = re.escape(var_placeholder) + # Match the variable as a whole word/pattern + pattern = re.compile(escaped_var) + result = pattern.sub(str(var_value), result) + else: + # For non-brace variables, use simple string replacement + result = result.replace(var_placeholder, str(var_value)) + + # Aggressively try to resolve any remaining variables using BuiltIn API + # This handles cases where variables weren't captured in messages, + # including object attributes like response.status_code + try: + builtin = BuiltIn() + + # Find all ${variable} patterns (scalar variables) + scalar_vars = re.findall(r'\$\{([^}]+)\}', result) + for var_name in scalar_vars: + # Skip if already resolved + if var_name in resolved_vars or f"${{{var_name}}}" in resolved_vars: + continue + + try: + # Handle cases like ${.status_code} or ${response.status_code} + if '.' in var_name and ('<' in var_name or '>' in var_name or not var_name.startswith('<')): + # This is an object attribute access + # Pattern: ${.attribute} or ${object.attribute} + parts = var_name.split('.') + # Remove angle brackets and brackets if present from base name + base_name = parts[0].strip('<>[]').strip() + # Remove any text in brackets like [200] + base_name = re.sub(r'\[.*?\]', '', base_name).strip() + + # Try to get base variable + base_value = builtin.get_variable_value(f"${{{base_name}}}") + if base_value is not None: + # Try to get nested attribute + current = base_value + for attr in parts[1:]: + if hasattr(current, attr): + current = getattr(current, attr) + elif isinstance(current, dict) and attr in current: + current = current[attr] + elif hasattr(current, '__getitem__'): + try: + current = current[attr] + except (KeyError, TypeError, IndexError): + current = None + break + else: + current = None + break + + if current is not None: + result = result.replace(f"${{{var_name}}}", str(current)) + continue + + # Try to get variable value normally + var_value = builtin.get_variable_value(f"${{{var_name}}}") + if var_value is not None: + # Replace all occurrences + result = result.replace(f"${{{var_name}}}", str(var_value)) + except Exception: + # If variable resolution fails, try alternative approaches + if '.' in var_name: + try: + # Try to split and resolve nested attributes + parts = var_name.split('.') + # Remove angle brackets and brackets if present + base_name = parts[0].strip('<>[]').strip() + # Remove any text in brackets + base_name = re.sub(r'\[.*?\]', '', base_name).strip() + + # Try to get base variable + base_value = builtin.get_variable_value(f"${{{base_name}}}") + if base_value is not None: + # Try to get nested attribute + current = base_value + for attr in parts[1:]: + if hasattr(current, attr): + current = getattr(current, attr) + elif isinstance(current, dict) and attr in current: + current = current[attr] + elif hasattr(current, '__getitem__'): + try: + current = current[attr] + except (KeyError, TypeError, IndexError): + current = None + break + else: + current = None + break + + if current is not None: + result = result.replace(f"${{{var_name}}}", str(current)) + except Exception: + pass + + # Find all @{variable} patterns (list variables) + list_vars = re.findall(r'@\{([^}]+)\}', result) + for var_name in list_vars: + try: + var_value = builtin.get_variable_value(f"@{{{var_name}}}") + if var_value is not None: + # Convert list to string representation + if isinstance(var_value, (list, tuple)): + list_str = ', '.join(str(item) for item in var_value) + result = result.replace(f"@{{{var_name}}}", list_str) + else: + result = result.replace(f"@{{{var_name}}}", str(var_value)) + except Exception: + pass + + # Also handle &{variable} patterns (dictionary variables) + dict_vars = re.findall(r'&\{([^}]+)\}', result) + for var_name in dict_vars: + try: + var_value = builtin.get_variable_value(f"&{{{var_name}}}") + if var_value is not None: + result = result.replace(f"&{{{var_name}}}", str(var_value)) + except Exception: + pass + + except Exception: + # If BuiltIn is not available or fails, continue with what we have + pass + + return result + + def __parse_steps(self, result, accumulated_vars: dict = None) -> List[Step]: + """ + Parse test steps from Robot Framework result, resolving variable values. + Accumulates resolved variables from all previous steps to resolve variables + in subsequent steps. + """ + if accumulated_vars is None: + accumulated_vars = {} + steps = [] for i in range(len(result.body)): @@ -201,7 +430,7 @@ def __parse_steps(self, result) -> List[Step]: continue if hasattr(result.body[i], "type") and result.body[i].type == "IF/ELSE ROOT": - condition_steps = self.__parse_condition_steps(result.body[i]) + condition_steps = self.__parse_condition_steps(result.body[i], accumulated_vars.copy()) for step in condition_steps: steps.append(step) continue @@ -236,6 +465,48 @@ def __parse_steps(self, result) -> List[Step]: if hasattr(body_element, "values") and body_element.values: data = ' '.join(str(val) for val in body_element.values) + # Extract resolved variable values from log messages in this step + # This will also accumulate variables from nested messages + step_resolved_vars = self.__extract_resolved_variables(body_element, accumulated_vars.copy()) + + # Merge newly resolved variables into accumulated variables + accumulated_vars.update(step_resolved_vars) + + # Also try to resolve variables directly from BuiltIn API before resolving in data + # This helps with variables that weren't logged but are available in the context + if data: + try: + builtin = BuiltIn() + # Find all variable patterns in data + all_vars = re.findall(r'[\$@&]\{([^}]+)\}', data) + for var_name in all_vars: + if var_name not in accumulated_vars and f"${{{var_name}}}" not in accumulated_vars: + try: + # Try scalar variable + var_value = builtin.get_variable_value(f"${{{var_name}}}") + if var_value is not None: + accumulated_vars[var_name] = str(var_value) + accumulated_vars[f"${{{var_name}}}"] = str(var_value) + except Exception: + try: + # Try list variable + var_value = builtin.get_variable_value(f"@{{{var_name}}}") + if var_value is not None: + if isinstance(var_value, (list, tuple)): + var_str = ', '.join(str(item) for item in var_value) + else: + var_str = str(var_value) + accumulated_vars[var_name] = var_str + accumulated_vars[f"@{{{var_name}}}"] = var_str + except Exception: + pass + except Exception: + pass + + # Resolve variables in data using all accumulated variables + if data and accumulated_vars: + data = self.__resolve_variables_in_data(data, accumulated_vars) + step = Step( step_type=StepType.GHERKIN, id=str(uuid.uuid4()), @@ -255,19 +526,23 @@ def __parse_steps(self, result) -> List[Step]: step.execution.end_time = result.body[i].end_time.timestamp() if hasattr(result.body[i], "body"): - step.steps = self.__parse_steps(result.body[i]) + # Recursively parse nested steps, passing accumulated variables + step.steps = self.__parse_steps(result.body[i], accumulated_vars.copy()) steps.append(step) return steps - def __parse_condition_steps(self, result_step) -> List[Step]: + def __parse_condition_steps(self, result_step, accumulated_vars: dict = None) -> List[Step]: + if accumulated_vars is None: + accumulated_vars = {} + steps = [] for body_element in result_step.body: if hasattr(body_element, "type"): step = Listener._create_gherkin_step_with_type(body_element) - child_steps = self.__parse_steps(body_element) + child_steps = self.__parse_steps(body_element, accumulated_vars.copy()) Listener._set_step_status_based_on_children(step, child_steps) step.steps = child_steps