cleveragents-core/features/steps/security_pyyaml_dependency_steps.py

"""Step definitions for PyYAML security dependency verification (#9055).

Note: the shared steps for reading pyproject.toml and verifying importability
are defined in ``tdd_a2a_sdk_dependency_steps.py`` to avoid AmbiguousStep
conflicts.  This file contains only the PyYAML-specific step definitions.
"""

from __future__ import annotations

import re
from typing import Any

from behave import then, when
from behave.runner import Context
from packaging.version import parse as parse_version


@then('the dependency list should include a package matching "{pattern}"')
def step_dependency_matches_pattern(context: Context, pattern: str) -> None:
    """Assert that a dependency in the project dependencies matches the given regex pattern."""
    deps: list[str] = context.project_dependencies
    found = any(re.search(pattern, dep) for dep in deps)
    assert found, (
        f"No dependency matching '{pattern}' found in "
        f"[project.dependencies]. Current deps: {deps}"
    )


@when("I read the PyYAML dependency specification from pyproject.toml")
def step_read_pyyaml_dependency(context: Context) -> None:
    """Read and parse the PyYAML dependency from pyproject.toml."""
    content = context.pyproject_path.read_bytes()
    data: dict[str, Any] = _load_toml(content)
    deps: list[str] = data.get("project", {}).get("dependencies", [])

    # Find the pyyaml dependency entry
    pyyaml_dep = None
    for dep in deps:
        if dep.strip().startswith("pyyaml"):
            pyyaml_dep = dep.strip()
            break

    assert pyyaml_dep is not None, (
        f"PyYAML dependency not found in [project.dependencies]. Current deps: {deps}"
    )

    # Parse the minimum version from the constraint
    # Examples: "pyyaml>=6.0.3", "pyyaml>=6.0", "pyyaml==6.0.2"
    match = re.search(r"(?:>=|==|~>)([\d]+(?:\.[\d]+)*(?:\.\w+)*)", pyyaml_dep)
    assert match is not None, (
        f"Could not parse version from PyYAML dependency: {pyyaml_dep}"
    )

    context.pyyaml_version_spec = pyyaml_dep
    context.pyyaml_min_version = match.group(1)


@then('the minimum version should be at least "{expected}"')
def step_pyyaml_minimum_version(context: Context, expected: str) -> None:
    """Assert that the PyYAML minimum version is >= expected version."""
    min_version = context.pyyaml_min_version
    assert parse_version(min_version) >= parse_version(expected), (
        f"PyYAML minimum version {min_version} is less than required {expected}. "
        f"Dependency spec: {context.pyyaml_version_spec}"
    )


def _flatten_toml_dict(d: dict[Any, Any]) -> dict[str, Any]:
    """Recursively convert a tomlkit dict to a plain Python dict.

    tomlkit wraps every value in a proxy object that behaves like the
    underlying Python type but is not identical to it.  This helper
    strips those wrappers so the result is a plain ``dict[str, Any]``
    compatible with ``tomllib`` output.
    """
    result: dict[str, Any] = {}
    for key, value in d.items():
        if isinstance(value, dict):
            result[key] = _flatten_toml_dict(value)
        else:
            result[key] = value
    return result


def _toml_to_python(data: object) -> dict[str, Any]:
    """Convert a tomlkit document to a plain Python dict.

    Delegates recursive flattening to the module-level
    ``_flatten_toml_dict`` helper so the logic stays testable in
    isolation and ruff does not flag nested function definitions.
    """
    if isinstance(data, dict):
        return _flatten_toml_dict(data)
    raise TypeError(f"Expected dict-like TOML table, got {type(data).__name__}")


def _load_toml(content: bytes) -> dict[str, Any]:
    """Load a TOML file from bytes. Uses tomllib if available, else fallback."""
    try:
        import tomllib

        return tomllib.loads(content.decode("utf-8"))
    except ImportError:  # pragma: no cover — Python <3.11
        # Fallback for older Python versions without tomllib
        import tomlkit

        parsed_doc = tomlkit.parse(content.decode("utf-8"))
        return _toml_to_python(parsed_doc)