Utilities

def uuid4():
    """
    Generate a UUID4 string.
    
    Returns:
    str: Random UUID4 string
    """

def now():
    """
    Get current timestamp in milliseconds.
    
    Returns:
    int: Current timestamp in milliseconds since epoch
    """

def md5(*args):
    """
    Calculate MD5 hash of arguments.
    
    Parameters:
    - *args: Variable arguments to hash (strings, bytes, or other objects)
    
    Returns:
    str: Hexadecimal MD5 hash string
    """

from allure_commons.utils import uuid4, now, md5

# Generate unique test identifier
test_id = uuid4()  # "a1b2c3d4-e5f6-7890-abcd-ef1234567890"

# Get current timestamp
start_time = now()  # 1609459200000

# Generate hash for test data
test_data = "user:testuser,action:login"
data_hash = md5(test_data)  # "a1b2c3d4e5f6789..."

# Hash multiple arguments
multi_hash = md5("test", "data", 123)  # Combined hash

def platform_label():
    """
    Get platform label for test environment.
    
    Returns:
    str: Platform identifier (e.g., "python3", "cpython3")
    """

def host_tag():
    """
    Get hostname for test environment identification.
    
    Returns:
    str: Current hostname
    """

def thread_tag():
    """
    Get thread identifier for concurrent test execution.
    
    Returns:
    str: Thread identifier in format "pid-threadname"
    """

from allure_commons.utils import platform_label, host_tag, thread_tag

# Add environment labels to test
platform = platform_label()    # "python3"
hostname = host_tag()          # "test-server-01"
thread_id = thread_tag()       # "12345-MainThread"

# Use in test labeling
allure.dynamic.label("platform", platform)
allure.dynamic.label("host", hostname) 
allure.dynamic.label("thread", thread_id)

def func_parameters(func, *args, **kwargs):
    """
    Extract function parameters with proper formatting.
    
    Handles various function signatures including:
    - Regular arguments and keyword arguments
    - Default parameter values
    - Variable arguments (*args)
    - Keyword arguments (**kwargs)
    - Class methods (self/cls parameter handling)
    
    Parameters:
    - func (callable): Function to extract parameters from
    - *args: Positional arguments passed to function
    - **kwargs: Keyword arguments passed to function
    
    Returns:
    OrderedDict: Ordered dictionary of parameter names to string representations
    """

from allure_commons.utils import func_parameters

def example_function(a, b, c=3, *args, **kwargs):
    pass

# Extract parameters from function call
params = func_parameters(example_function, 1, 2, 4, 5, extra="value")
# Returns: OrderedDict([('a', '1'), ('b', '2'), ('c', '4'), ('args', '(5,)'), ('extra', "'value'")])

# Use in step decoration
@allure.step("Execute function with parameters")
def logged_function(x, y, option=None):
    # Get parameters for logging
    params = func_parameters(logged_function, x, y, option)
    allure.dynamic.parameter("function_params", str(dict(params)))
    
    # Function implementation
    return x + y if option != "subtract" else x - y

def represent(item):
    """
    Get consistent string representation of any item.
    
    Handles special cases for different data types:
    - Strings: Wrapped in single quotes
    - Bytes/bytearray: Shows type information
    - Other objects: Uses repr()
    
    Parameters:
    - item: Any object to represent
    
    Returns:
    str: String representation of the item
    """

class SafeFormatter(string.Formatter):
    """
    String formatter that safely handles missing keys.
    
    Unlike standard string formatting, missing keys are left as-is
    instead of raising KeyError.
    """
    
    def get_field(self, field_name, args, kwargs):
        """
        Get field value, returning placeholder for missing keys.
        
        Parameters:
        - field_name (str): Name of field to retrieve
        - args (tuple): Positional arguments
        - kwargs (dict): Keyword arguments
        
        Returns:
        tuple: (field_value, field_name) or (placeholder, field_name)
        """
    
    def get_value(self, key, args, kwargs):
        """
        Get value by key, raising SafeKeyOrIndexError for missing keys.
        
        Parameters:
        - key: Key to retrieve
        - args (tuple): Positional arguments
        - kwargs (dict): Keyword arguments
        
        Returns:
        Value for the key
        
        Raises:
        SafeKeyOrIndexError: For missing keys (handled internally)
        """

from allure_commons.utils import represent, SafeFormatter

# Consistent string representation
print(represent("hello"))      # "'hello'"
print(represent(123))          # "123"
print(represent(None))         # "None"
print(represent([1, 2, 3]))    # "[1, 2, 3]"
print(represent(b"bytes"))     # "<class 'bytes'>"

# Safe string formatting
formatter = SafeFormatter()

# Missing keys are preserved
result = formatter.format("Hello {name}, your score is {score}", name="Alice")
print(result)  # "Hello Alice, your score is {score}"

# Use in step titles with parameters
@allure.step("Process user {username} with action {action}")
def process_user_action(username, action, extra_data=None):
    # Step title will be formatted safely even if extra_data is referenced
    pass

def format_traceback(exc_traceback):
    """
    Format exception traceback for reporting.
    
    Parameters:
    - exc_traceback: Exception traceback object
    
    Returns:
    str: Formatted traceback string or None if no traceback
    """

def format_exception(etype, value):
    """
    Format exception type and value for reporting.
    
    Parameters:
    - etype: Exception type
    - value: Exception value/instance
    
    Returns:
    str: Formatted exception string or None if no exception
    """

from allure_commons.utils import format_traceback, format_exception
import sys

def test_with_exception_handling():
    try:
        # Test code that might fail
        result = risky_operation()
    except Exception:
        # Capture exception information
        exc_type, exc_value, exc_traceback = sys.exc_info()
        
        # Format for Allure reporting
        formatted_exception = format_exception(exc_type, exc_value)
        formatted_traceback = format_traceback(exc_traceback)
        
        # Attach to test
        allure.attach(formatted_exception, "Exception Details", 
                     allure.attachment_type.TEXT)
        allure.attach(formatted_traceback, "Stack Trace",
                     allure.attachment_type.TEXT)
        raise

def get_testplan():
    """
    Get test plan from environment configuration.
    
    Reads test plan from file specified by ALLURE_TESTPLAN_PATH
    environment variable.
    
    Returns:
    list: List of planned test identifiers, empty if no plan file
    """

from allure_commons.utils import get_testplan
import os

def should_run_test(test_id):
    """Check if test should run based on test plan."""
    # Set test plan file
    os.environ["ALLURE_TESTPLAN_PATH"] = "/tmp/testplan.json"
    
    # Get planned tests
    planned_tests = get_testplan()
    
    # Check if test is in plan
    if planned_tests:
        return test_id in [test.get('id') for test in planned_tests]
    
    # Run all tests if no plan
    return True

# Example test plan file format:
# {
#   "tests": [
#     {"id": "test_login", "selector": "tests.auth.test_login"},
#     {"id": "test_logout", "selector": "tests.auth.test_logout"}
#   ]
# }

import threading
from allure_commons.utils import uuid4, now, thread_tag

def threaded_test_execution():
    """Example of thread-safe utility usage."""
    test_id = uuid4()           # Unique per thread
    start_time = now()          # Current timestamp
    thread_id = thread_tag()    # Current thread identifier
    
    print(f"Thread {thread_id}: Test {test_id} started at {start_time}")
    
    # Simulate test execution
    time.sleep(1)
    
    end_time = now()
    print(f"Thread {thread_id}: Test completed in {end_time - start_time}ms")

# Run multiple threads
threads = []
for i in range(5):
    thread = threading.Thread(target=threaded_test_execution)
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()