Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions qase-robotframework/changelog.md
Original file line number Diff line number Diff line change
@@ -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
Expand Down
2 changes: 1 addition & 1 deletion qase-robotframework/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"}]
Expand Down
285 changes: 280 additions & 5 deletions qase-robotframework/src/qase/robotframework/listener.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import logging
import pathlib
import re
import uuid

from filelock import FileLock
Expand Down Expand Up @@ -193,15 +194,243 @@ 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 ${<Response [200]>.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: ${<Object [info]>.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)):
if hasattr(result.body[i], "type") and result.body[i].type == "MESSAGE":
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
Expand Down Expand Up @@ -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()),
Expand All @@ -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
Expand Down