cleveragents-core/features/steps/auth_refresh.py

"""Step definitions for JWT refresh endpoint BDD scenarios.

This module implements steps that mimic the behaviour of the ASGI
auth/refresh endpoint using the ``cleveragents`` token helpers directly
instead of firing real HTTP requests.  It lets tests assert on the
Python API surface (status codes, JSON bodies, token introspection)
without needing a live server process.
"""

from __future__ import annotations

import json
from typing import TYPE_CHECKING

from behave import given, then, when

if TYPE_CHECKING:
    from behave.runner import Context

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


# ---------------------------------------------------------------------------
# Given
# ---------------------------------------------------------------------------


@given("the application is running")
def step_app_running(context: Context) -> None:
    """Application is assumed to be running for testing."""
    pass


@given("I have a valid refresh token")
def step_valid_refresh_token(context: Context) -> None:
    """Generate and store a valid refresh token."""
    context.refresh_token = create_refresh_token("test-user")


# ---------------------------------------------------------------------------
# When
# ---------------------------------------------------------------------------


@when('I send a POST request to "/auth/refresh" with JSON body:')
def step_send_post_auth_refresh_json(context: Context) -> None:
    """Send a POST request to the auth refresh endpoint with a JSON body.

    The multi-line string in the Gherkin step becomes ``context.text``.
    """
    try:
        body = json.loads(context.text)
    except (json.JSONDecodeError, TypeError):
        body = None

    context.request_body = body
    context.request_content_type = "application/json"

    if body is None:
        context.response_status = 400
        context.response_data = {"error": "Invalid JSON body"}
    else:
        _process_refresh_request(context, body)


@when(
    'I send a POST request to "/auth/refresh" with content type "{content_type}" and body:'
)
def step_send_post_auth_refresh_content_type(
    context: Context, content_type: str
) -> None:
    """Send a POST request with a specific content type."""
    context.request_content_type = content_type
    context.request_body = context.text

    if "application/json" not in content_type:
        context.response_status = 400
        context.response_data = {"error": "Content-Type must be application/json"}
    else:
        try:
            body = json.loads(context.text)
            context.request_body = body
            _process_refresh_request(context, body)
        except json.JSONDecodeError:
            context.response_status = 400
            context.response_data = {"error": "Invalid JSON body"}


@when(
    'I send a POST request to "/auth/refresh" with JSON body containing a non-refresh token'
)
def step_send_non_refresh_token(context: Context) -> None:
    """Send an access token instead of a refresh token."""
    context.request_body = {"refresh_token": create_access_token("test-user")}
    _process_refresh_request(context, context.request_body)


@when('I send a POST request to "/auth/refresh" with the stored refresh token')
def step_send_stored_refresh_token(context: Context) -> None:
    """Send a POST request using the refresh token stored in context."""
    body = {"refresh_token": context.refresh_token}
    context.request_body = body
    context.request_content_type = "application/json"
    _process_refresh_request(context, body)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _process_refresh_request(context: Context, body: object) -> None:
    """Execute the refresh-token flow and capture the result on *context*.

    Args:
        context: The Behave context object.
        body: The parsed JSON body (must be a ``dict``).
    """
    from cleveragents.core.jwt_service import refresh_tokens

    refresh_token_val: object = (
        body.get("refresh_token") if isinstance(body, dict) else body
    )

    if not refresh_token_val or not isinstance(refresh_token_val, str):
        context.response_status = 400
        context.response_data = {"error": "refresh_token is required"}
        return

    try:
        result = refresh_tokens(refresh_token_val)
    except Exception:
        context.response_status = 401
        context.response_data = {"error": "Invalid or expired refresh token"}
        return

    context.response_status = 200
    context.response_data = result
    context.new_access_token = result["access_token"]
    context.new_refresh_token = result["refresh_token"]


# ---------------------------------------------------------------------------
# Then
# ---------------------------------------------------------------------------


@then("the response status code should be {status_code:d}")
def step_status_code(context: Context, status_code: int) -> None:
    """Verify the response status code."""
    assert context.response_status == status_code, (
        f"Expected {status_code}, got {context.response_status}"
    )


@then('the response should contain "{text}"')
def step_response_contains(context: Context, text: str) -> None:
    """Verify the response contains expected text."""
    if isinstance(context.response_data, dict):
        response_text = json.dumps(context.response_data)
    else:
        response_text = str(context.response_data)
    assert text in response_text, f"Expected '{text}' in response: {response_text}"


@then("the new access token should be a valid JWT")
def step_verify_access_token_json(context: Context) -> None:
    """Verify the new access token is a valid JWT with correct claims."""
    token = context.new_access_token
    assert token is not None, "No access token in response"

    payload = decode_token(token)
    assert payload.get("type") == "access", (
        f"Token type is '{payload.get('type')}', expected 'access'"
    )
    assert "sub" in payload, "Token missing 'sub' claim"
    assert "iat" in payload, "Token missing 'iat' claim"
    assert "exp" in payload, "Token missing 'exp' claim"


@then("the new refresh token should be a valid refresh token")
def step_verify_refresh_token(context: Context) -> None:
    """Verify the new refresh token is a valid refresh token."""
    token = context.new_refresh_token
    assert token is not None, "No refresh token in response"

    payload = decode_token(token)
    assert payload.get("type") == "refresh", (
        f"Token type is '{payload.get('type')}', expected 'refresh'"
    )
    assert "sub" in payload, "Token missing 'sub' claim"
    assert "iat" in payload, "Token missing 'iat' claim"
    assert "exp" in payload, "Token missing 'exp' claim"