cleveragents-core/features/steps/auth_refresh_steps.py

"""Step definitions for JWT refresh endpoint ASGI-level BDD scenarios.

These steps complement ``asgi_app_steps.py`` with auth-specific request
helpers and response assertions for the ``/auth/refresh`` endpoint.
The ``@given("the ASGI app module is loaded")`` step is defined in
``asgi_app_steps.py`` and shared across all ASGI feature tests.
"""

from __future__ import annotations

import asyncio
import json
from typing import Any

from behave import then, when
from behave.runner import Context

from cleveragents.a2a.asgi import app
from cleveragents.core.jwt_service import (
    create_access_token,
    create_refresh_token,
    decode_token,
)

SendMessage = dict[str, Any]


def _make_asgi_post_request(
    path: str,
    body: bytes = b"",
    content_type: str = "application/json",
) -> list[SendMessage]:
    """Simulate an ASGI HTTP POST request.

    Args:
        path: Request path.
        body: Request body bytes.
        content_type: Content-Type header value.

    Returns:
        List of messages sent via the ASGI send callable.
    """
    sent_messages: list[SendMessage] = []

    async def receive() -> dict[str, Any]:
        if body:
            return {"type": "http.request", "body": body, "more_body": False}
        return {"type": "http.request", "body": b"", "more_body": False}

    async def send(message: SendMessage) -> None:
        sent_messages.append(message)

    headers: list[tuple[bytes, bytes]] = [
        (b"content-type", content_type.encode("utf-8")),
    ]
    scope: dict[str, Any] = {
        "type": "http",
        "method": "POST",
        "path": path,
        "headers": headers,
    }
    asyncio.run(app(scope, receive, send))
    return sent_messages


def _extract_json_body(sent_messages: list[SendMessage]) -> dict[str, Any]:
    """Extract and parse the JSON body from ASGI sent messages.

    Args:
        sent_messages: ASGI messages captured during request handling.

    Returns:
        Parsed JSON dictionary.

    Raises:
        AssertionError: If no response body was sent.
    """
    body_msg = next(
        (msg for msg in sent_messages if msg.get("type") == "http.response.body"),
        None,
    )
    if body_msg is None:
        raise AssertionError("No response body sent")
    return json.loads(body_msg["body"].decode("utf-8"))


@when('I send an HTTP POST request to "{path}" with empty JSON body')
def step_post_empty_json(context: Context, path: str) -> None:
    """Send a POST with an empty object as JSON body."""
    context.asgi_sent_messages = _make_asgi_post_request(path, body=b"{}")


@when('I send an HTTP POST request to "{path}" with invalid JSON')
def step_post_invalid_json(context: Context, path: str) -> None:
    """Send a POST with an invalid JSON string."""
    context.asgi_sent_messages = _make_asgi_post_request(path, body=b"{invalid}")


@when('I send an HTTP POST request to "{path}" with malformed token')
def step_post_malformed_token(context: Context, path: str) -> None:
    """Send a POST with a string that is not a valid JWT."""
    payload = json.dumps({"refresh_token": "not-a-valid-jwt-token"}).encode("utf-8")
    context.asgi_sent_messages = _make_asgi_post_request(path, body=payload)


@when('I send an HTTP POST request to "{path}" with an access token')
def step_post_access_token(context: Context, path: str) -> None:
    """Send a POST with a valid access token where a refresh token is expected."""
    access_tok = create_access_token("test-user")
    payload = json.dumps({"refresh_token": access_tok}).encode("utf-8")
    context.asgi_sent_messages = _make_asgi_post_request(path, body=payload)


@when('I send an HTTP POST request to "{path}" with a valid refresh token')
def step_post_valid_refresh_token(context: Context, path: str) -> None:
    """Send a POST with a valid refresh token."""
    refresh_tok = create_refresh_token("test-user")
    payload = json.dumps({"refresh_token": refresh_tok}).encode("utf-8")
    context.asgi_sent_messages = _make_asgi_post_request(path, body=payload)
    context._refresh_token_value = refresh_tok


@then("the response contains valid new tokens")
def step_response_has_new_tokens(context: Context) -> None:
    """Verify response contains decodable access and refresh tokens."""
    data = _extract_json_body(context.asgi_sent_messages)
    assert "access_token" in data, "Response missing access_token"
    assert "refresh_token" in data, "Response missing refresh_token"
    decode_token(data["access_token"])
    decode_token(data["refresh_token"])


@then("the token_type is bearer")
def step_token_type_is_bearer(context: Context) -> None:
    """Verify the response token_type is 'bearer'."""
    data = _extract_json_body(context.asgi_sent_messages)
    assert data.get("token_type") == "bearer", (
        f"Expected token_type 'bearer', got {data.get('token_type')!r}"
    )


@then("the returned access token can be decoded")
def step_access_token_decodable(context: Context) -> None:
    """Verify the returned access token decodes correctly."""
    data = _extract_json_body(context.asgi_sent_messages)
    payload = decode_token(data["access_token"])
    assert payload.get("type") == "access", (
        f"Expected type 'access', got {payload.get('type')!r}"
    )
    assert payload.get("sub") == "test-user", (
        f"Expected sub 'test-user', got {payload.get('sub')!r}"
    )


@then("the decoded access token has correct subject")
def step_access_token_subject(context: Context) -> None:
    """Verify the access token subject matches the original."""
    data = _extract_json_body(context.asgi_sent_messages)
    payload = decode_token(data["access_token"])
    assert payload.get("sub") == "test-user"


@then("the returned refresh token can be decoded")
def step_refresh_token_decodable(context: Context) -> None:
    """Verify the returned refresh token decodes correctly."""
    data = _extract_json_body(context.asgi_sent_messages)
    payload = decode_token(data["refresh_token"])
    assert payload.get("type") == "refresh", (
        f"Expected type 'refresh', got {payload.get('type')!r}"
    )


@then("the decode type field is refresh")
def step_decode_type_is_refresh(context: Context) -> None:
    """Verify the decode type field indicates a refresh token."""
    data = _extract_json_body(context.asgi_sent_messages)
    payload = decode_token(data["refresh_token"])
    assert payload.get("type") == "refresh"