Source code for steplib.modules.data.actions

"""Pure action functions for the data module (variables + environment)."""

from __future__ import annotations

import json
import os
import re
import time
from pathlib import Path
from typing import Any

from steplib.core.exceptions import MissingDependencyError
from steplib.modules.data.context import DataContext


def _normalize_value(value: Any) -> str:
    """Normalize a value to its string representation for comparison.

    Python's ``str(True)`` returns ``"True"``, but users naturally write
    ``"true"`` / ``"false"`` / ``"null"`` in step definitions.  This helper
    ensures booleans and ``None`` use their JSON-style lowercase
    representation.
    """
    if value is True:
        return "true"
    if value is False:
        return "false"
    if value is None:
        return "null"
    return str(value)


# --- Variable actions ---




def data_set_variable(data_ctx: DataContext, name: str, value: str) -> None:
    """Set a generic variable in the data context.

    Args:
        data_ctx: The data context to operate on.
        name: The variable name.
        value: The variable value (stored as string).

    """
    data_ctx.variables[name] = value





def data_assert_variable_equals(
    data_ctx: DataContext,
    name: str,
    expected: str,
) -> None:
    """Assert that a variable equals an expected value.

    Args:
        data_ctx: The data context to check.
        name: The variable name.
        expected: The expected value.

    Raises:
        AssertionError: If the variable does not exist or the value differs.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    actual = data_ctx.variables[name]
    if _normalize_value(actual) != _normalize_value(expected):
        raise AssertionError(
            f"Variable '{name}': expected '{expected}', got '{actual}'."
        )





def data_assert_variable_exists(data_ctx: DataContext, name: str) -> None:
    """Assert that a variable exists in the data context.

    Args:
        data_ctx: The data context to check.
        name: The variable name.

    Raises:
        AssertionError: If the variable does not exist.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")





def data_assert_variable_not_exists(data_ctx: DataContext, name: str) -> None:
    """Assert that a variable does not exist in the data context.

    Args:
        data_ctx: The data context to check.
        name: The variable name.

    Raises:
        AssertionError: If the variable exists.

    """
    if name in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' should not exist.")





def data_delete_variable(data_ctx: DataContext, name: str) -> None:
    """Delete a variable from the data context.

    Args:
        data_ctx: The data context to operate on.
        name: The variable name.

    Raises:
        KeyError: If the variable does not exist.

    """
    if name not in data_ctx.variables:
        raise KeyError(f"Variable '{name}' does not exist.")
    del data_ctx.variables[name]





def data_assert_variable_not_equals(
    data_ctx: DataContext,
    name: str,
    expected: str,
) -> None:
    """Assert that a variable does not equal a value.

    Args:
        data_ctx: The data context to check.
        name: The variable name.
        expected: The value that the variable should NOT have.

    Raises:
        AssertionError: If the variable does not exist or equals the value.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    actual = data_ctx.variables[name]
    if _normalize_value(actual) == _normalize_value(expected):
        raise AssertionError(
            f"Variable '{name}' should not equal '{expected}'."
        )





def data_assert_variable_contains(
    data_ctx: DataContext,
    name: str,
    substring: str,
) -> None:
    """Assert that a variable's string value contains a substring.

    Args:
        data_ctx: The data context to check.
        name: The variable name.
        substring: The substring to look for.

    Raises:
        AssertionError: If the variable does not exist or doesn't contain the substring.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    actual = _normalize_value(data_ctx.variables[name])
    if substring not in actual:
        raise AssertionError(
            f"Variable '{name}': expected to contain '{substring}', got '{actual}'."
        )





def data_assert_variable_is_empty(data_ctx: DataContext, name: str) -> None:
    """Assert that a variable is empty (empty string, empty list, empty dict, or None).

    Args:
        data_ctx: The data context to check.
        name: The variable name.

    Raises:
        AssertionError: If the variable does not exist or is not empty.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    value = data_ctx.variables[name]
    if value is not None and value not in ("", [], {}):
        raise AssertionError(f"Variable '{name}' is not empty: {value!r}.")





def data_assert_variable_is_not_empty(data_ctx: DataContext, name: str) -> None:
    """Assert that a variable is not empty.

    Args:
        data_ctx: The data context to check.
        name: The variable name.

    Raises:
        AssertionError: If the variable does not exist or is empty.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    value = data_ctx.variables[name]
    if value is None or value in ("", [], {}):
        raise AssertionError(f"Variable '{name}' is empty.")





def data_assert_variable_has_length(
    data_ctx: DataContext,
    name: str,
    expected: int,
) -> None:
    """Assert that a variable has a specific length.

    Works with strings, lists, dicts, and any object with ``__len__``.

    Args:
        data_ctx: The data context to check.
        name: The variable name.
        expected: The expected length.

    Raises:
        AssertionError: If the variable does not exist or has a different length.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    value = data_ctx.variables[name]
    try:
        actual = len(value)
    except TypeError as exc:
        raise AssertionError(
            f"Variable '{name}' of type {type(value).__name__} has no length."
        ) from exc
    if actual != expected:
        raise AssertionError(
            f"Variable '{name}': expected length {expected}, got {actual}."
        )





def data_copy_variable(
    data_ctx: DataContext,
    source: str,
    target: str,
) -> None:
    """Copy a variable to a new name.

    Args:
        data_ctx: The data context to operate on.
        source: The source variable name.
        target: The target variable name.

    Raises:
        KeyError: If the source variable does not exist.

    """
    if source not in data_ctx.variables:
        raise KeyError(f"Source variable '{source}' does not exist.")
    data_ctx.variables[target] = data_ctx.variables[source]





def data_clear_variables(data_ctx: DataContext) -> None:
    """Clear all variables from the data context."""
    data_ctx.variables = {}





def data_set_variable_json(
    data_ctx: DataContext,
    name: str,
    json_str: str,
) -> None:
    """Set a variable to a parsed JSON value.

    Args:
        data_ctx: The data context to operate on.
        name: The variable name.
        json_str: A JSON string to parse and store.

    Raises:
        json.JSONDecodeError: If the string is not valid JSON.

    """
    data_ctx.variables[name] = json.loads(json_str)





def data_set_env_from_variable(
    data_ctx: DataContext,
    variable: str,
    key: str,
) -> None:
    """Set an environment variable from a data variable's value.

    Args:
        data_ctx: The data context (used for backup tracking).
        variable: The source data variable name.
        key: The environment variable name to set.

    Raises:
        KeyError: If the source variable does not exist.

    """
    if variable not in data_ctx.variables:
        raise KeyError(f"Variable '{variable}' does not exist.")
    data_set_env_var(data_ctx, key, _normalize_value(data_ctx.variables[variable]))





def data_load_yaml_file(data_ctx: DataContext, path: str, name: str) -> None:
    """Load a YAML file into a variable as a dict.

    Args:
        data_ctx: The data context to operate on.
        path: Path to the YAML file.
        name: The variable name to store the parsed content.

    Raises:
        MissingDependencyError: If PyYAML is not installed.
        FileNotFoundError: If the file does not exist.

    """
    try:
        import yaml
    except ImportError as exc:
        raise MissingDependencyError("data", "pyyaml") from exc

    file_path = Path(path)
    if not file_path.exists():
        raise FileNotFoundError(f"YAML file not found: {path}")

    with file_path.open(encoding="utf-8") as f:
        data_ctx.variables[name] = yaml.safe_load(f)





def data_load_json_file(data_ctx: DataContext, path: str, name: str) -> None:
    """Load a JSON file into a variable as a dict.

    Args:
        data_ctx: The data context to operate on.
        path: Path to the JSON file.
        name: The variable name to store the parsed content.

    Raises:
        FileNotFoundError: If the file does not exist.

    """
    file_path = Path(path)
    if not file_path.exists():
        raise FileNotFoundError(f"JSON file not found: {path}")

    with file_path.open(encoding="utf-8") as f:
        data_ctx.variables[name] = json.load(f)





def data_extract_key_path(
    data_ctx: DataContext,
    source: str,
    key_path: str,
    target: str,
) -> None:
    """Extract a value from a variable using dot-path navigation.

    Navigates nested dicts/lists using dot-separated keys. List indices
    are supported via integer keys (e.g. ``items.0.name``).

    Args:
        data_ctx: The data context to operate on.
        source: The source variable name.
        key_path: Dot-separated path (e.g. ``user.address.city``).
        target: The target variable name to store the extracted value.

    Raises:
        KeyError: If the source variable does not exist.
        KeyError: If any key in the path is not found.

    """
    if source not in data_ctx.variables:
        raise KeyError(f"Source variable '{source}' does not exist.")

    current: Any = data_ctx.variables[source]
    for key in key_path.split("."):
        if isinstance(current, list):
            try:
                idx = int(key)
            except ValueError as exc:
                raise KeyError(
                    f"Cannot index list with non-integer key '{key}' in path '{key_path}'."
                ) from exc
            if idx < 0 or idx >= len(current):
                raise KeyError(
                    f"Index {idx} out of range in path '{key_path}'."
                )
            current = current[idx]
        elif isinstance(current, dict):
            if key not in current:
                raise KeyError(
                    f"Key '{key}' not found in path '{key_path}'."
                )
            current = current[key]
        else:
            raise KeyError(
                f"Cannot navigate into non-dict/list value at '{key}' in path '{key_path}'."
            )

    data_ctx.variables[target] = current



# --- Environment variable actions ---




def data_set_env_var(data_ctx: DataContext, key: str, value: str) -> None:
    """Set an environment variable, backing up the original for restoration.

    Args:
        data_ctx: The data context (used for backup tracking).
        key: The environment variable name.
        value: The value to set.

    """
    if key not in data_ctx._env_backup:
        data_ctx._env_backup[key] = os.environ.get(key)
    os.environ[key] = value





def data_delete_env_var(data_ctx: DataContext, key: str) -> None:
    """Delete an environment variable, backing up the original for restoration.

    Args:
        data_ctx: The data context (used for backup tracking).
        key: The environment variable name.

    """
    if key not in data_ctx._env_backup:
        data_ctx._env_backup[key] = os.environ.get(key)
    os.environ.pop(key, None)





def data_assert_env_equals(key: str, expected: str) -> None:
    """Assert that an environment variable equals an expected value.

    Args:
        key: The environment variable name.
        expected: The expected value.

    Raises:
        AssertionError: If the env var does not exist or differs.

    """
    actual = os.environ.get(key)
    if actual is None:
        raise AssertionError(f"Environment variable '{key}' is not set.")
    if actual != expected:
        raise AssertionError(
            f"Environment variable '{key}': expected '{expected}', got '{actual}'."
        )





def data_assert_env_exists(key: str) -> None:
    """Assert that an environment variable exists.

    Args:
        key: The environment variable name.

    Raises:
        AssertionError: If the env var does not exist.

    """
    if key not in os.environ:
        raise AssertionError(f"Environment variable '{key}' is not set.")





def data_assert_env_not_equals(key: str, expected: str) -> None:
    """Assert that an environment variable does not equal a value.

    Args:
        key: The environment variable name.
        expected: The value that the env var should NOT have.

    Raises:
        AssertionError: If the env var does not exist or equals the value.

    """
    actual = os.environ.get(key)
    if actual is None:
        raise AssertionError(f"Environment variable '{key}' is not set.")
    if actual == expected:
        raise AssertionError(
            f"Environment variable '{key}' should not equal '{expected}'."
        )





def data_assert_env_not_exists(key: str) -> None:
    """Assert that an environment variable does not exist.

    Args:
        key: The environment variable name.

    Raises:
        AssertionError: If the env var exists.

    """
    if key in os.environ:
        raise AssertionError(f"Environment variable '{key}' should not be set.")





def data_store_env_var(
    data_ctx: DataContext,
    key: str,
    variable: str,
) -> None:
    """Store an environment variable's value into a data variable.

    Args:
        data_ctx: The data context to operate on.
        key: The environment variable name.
        variable: The target variable name.

    Raises:
        AssertionError: If the env var does not exist.

    """
    if key not in os.environ:
        raise AssertionError(f"Environment variable '{key}' is not set.")
    data_ctx.variables[variable] = os.environ[key]





def data_load_env_file(data_ctx: DataContext, path: str) -> None:
    """Load environment variables from a .env-style file.

    Parses simple ``KEY=VALUE`` lines. Lines starting with ``#`` are
    ignored. Quoted values (single or double) are unquoted.

    Args:
        data_ctx: The data context (used for backup tracking).
        path: Path to the .env file.

    Raises:
        FileNotFoundError: If the file does not exist.

    """
    file_path = Path(path)
    if not file_path.exists():
        raise FileNotFoundError(f"Env file not found: {path}")

    with file_path.open(encoding="utf-8") as f:
        for raw_line in f:
            line = raw_line.strip()
            if not line or line.startswith("#"):
                continue
            if "=" not in line:
                continue
            key, _, value = line.partition("=")
            key = key.strip()
            value = value.strip()
            if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
                value = value[1:-1]
            data_set_env_var(data_ctx, key, value)



# --- Extended variable assertions ---




def data_assert_variable_matches(
    data_ctx: DataContext, name: str, pattern: str
) -> None:
    """Assert that a variable's string value matches a regex pattern.

    Args:
        data_ctx: The data context.
        name: Variable name.
        pattern: Regex pattern to match.

    Raises:
        AssertionError: If the variable does not match or does not exist.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    value = _normalize_value(data_ctx.variables[name])
    try:
        if not re.search(pattern, value):
            raise AssertionError(
                f"Variable '{name}' value '{value}' does not match pattern '{pattern}'."
            )
    except re.error as exc:
        raise AssertionError(f"Invalid regex pattern '{pattern}': {exc}") from exc





def data_assert_variable_starts_with(
    data_ctx: DataContext, name: str, text: str
) -> None:
    """Assert that a variable's string value starts with the given text.

    Args:
        data_ctx: The data context.
        name: Variable name.
        text: Expected prefix.

    Raises:
        AssertionError: If the variable does not start with the text or does not exist.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    value = _normalize_value(data_ctx.variables[name])
    if not value.startswith(text):
        raise AssertionError(
            f"Variable '{name}' value '{value}' does not start with '{text}'."
        )





def data_assert_variable_ends_with(
    data_ctx: DataContext, name: str, text: str
) -> None:
    """Assert that a variable's string value ends with the given text.

    Args:
        data_ctx: The data context.
        name: Variable name.
        text: Expected suffix.

    Raises:
        AssertionError: If the variable does not end with the text or does not exist.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    value = _normalize_value(data_ctx.variables[name])
    if not value.endswith(text):
        raise AssertionError(
            f"Variable '{name}' value '{value}' does not end with '{text}'."
        )





def data_increment_variable(
    data_ctx: DataContext, name: str, amount: int = 1
) -> None:
    """Increment a numeric variable by a given amount.

    Args:
        data_ctx: The data context.
        name: Variable name.
        amount: Amount to increment (default 1).

    Raises:
        KeyError: If the variable does not exist.
        ValueError: If the variable is not numeric.

    """
    if name not in data_ctx.variables:
        raise KeyError(f"Variable '{name}' does not exist.")
    current = data_ctx.variables[name]
    if isinstance(current, bool):
        raise ValueError(
            f"Variable '{name}' value '{current}' is a boolean, not numeric."
        )
    try:
        if isinstance(current, float):
            numeric: int | float = current + amount
        else:
            try:
                numeric = int(current) + amount
            except (TypeError, ValueError):
                numeric = float(current) + amount
    except (TypeError, ValueError) as exc:
        raise ValueError(
            f"Variable '{name}' value '{current}' is not numeric."
        ) from exc
    data_ctx.variables[name] = numeric





def data_assert_variable_greater_than(
    data_ctx: DataContext, name: str, value: str
) -> None:
    """Assert that a variable's numeric value is greater than a threshold.

    Args:
        data_ctx: The data context.
        name: Variable name.
        value: Threshold value (compared as float).

    Raises:
        AssertionError: If the variable is not greater than the value or does not exist.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    raw = data_ctx.variables[name]
    if isinstance(raw, bool):
        raise AssertionError(
            f"Variable '{name}' is a boolean, not a numeric value."
        )
    try:
        current = float(raw)
        threshold = float(value)
    except (TypeError, ValueError) as exc:
        raise AssertionError(
            f"Variable '{name}' or threshold '{value}' is not numeric."
        ) from exc
    if not current > threshold:
        raise AssertionError(
            f"Variable '{name}' value {current} is not greater than {threshold}."
        )





def data_assert_variable_less_than(
    data_ctx: DataContext, name: str, value: str
) -> None:
    """Assert that a variable's numeric value is less than a threshold.

    Args:
        data_ctx: The data context.
        name: Variable name.
        value: Threshold value (compared as float).

    Raises:
        AssertionError: If the variable is not less than the value or does not exist.

    """
    if name not in data_ctx.variables:
        raise AssertionError(f"Variable '{name}' does not exist.")
    raw = data_ctx.variables[name]
    if isinstance(raw, bool):
        raise AssertionError(
            f"Variable '{name}' is a boolean, not a numeric value."
        )
    try:
        current = float(raw)
        threshold = float(value)
    except (TypeError, ValueError) as exc:
        raise AssertionError(
            f"Variable '{name}' or threshold '{value}' is not numeric."
        ) from exc
    if not current < threshold:
        raise AssertionError(
            f"Variable '{name}' value {current} is not less than {threshold}."
        )



# --- Utility actions ---




def data_wait(seconds: float) -> None:
    """Sleep for a given number of seconds.

    Args:
        seconds: Number of seconds to sleep.

    """
    time.sleep(seconds)