Behave - Basic step setup for APIs

api_steps.py

"""Step implementations for API tests."""
import logging
from argparse import ArgumentError

import jsonpath_ng
import requests
from behave import given, then, when, step
from behave.runner import Context

logger = logging.getLogger(__name__)


@given("the service is running")
def step_service_is_running(context: Context):
    """Verify the service is running."""
    assert hasattr(context, 'base_url'), "Service base URL not found in context"
    assert context.base_url, "Service base URL is empty"


@when('I send a {method} request to "{endpoint}"')
def step_send_get_request(context: Context, method: str, endpoint: str):
    """Send a request to a specified endpoint."""
    try:
        url = context.build_url(endpoint)
        headers = context.next_request.headers
        body = context.next_request.body

        logger.info(f"Sending {method} request to: {url}\nBody: {body}")
        context.response = requests.request(method, url, headers=headers, data=body)

        context.reset_next_request()

        # log the entire response, including payload
        logger.info(f"Response: {context.response.status_code} {context.response.reason}\n{context.response.content}")

    except Exception as e:
        logger.error(f"Failed to send {method} request to {endpoint}: {e}")
        raise


@then("the response status code should be {status_code:d}")
def step_check_status_code(context: Context, status_code: int):
    """Check if the response status code is the expected one."""
    assert context.response.status_code == status_code, (
        f"Expected status code {status_code}, but got {context.response.status_code}"
    )


@given("I have the request body:")
def step_next_request_body(context: Context):
    context.next_request.body = context.render(context.text)


@given("I have request headers:")
def step_next_request_headers(context: Context):
    try:
        headers = get_kv_table(context)
        context.next_request.headers = headers
    except Exception as e:
        logger.error(f"Failed to set request headers: {e}")
        raise


def get_kv_table(context: Context, render_values=True) -> dict[str, str]:
    """ convert <key, value> table to dict[str, str] """
    try:
        table = context.table

        if len(table.headings) > 2:
            raise ArgumentError("cannot convert table with > 2 columns to key-value pairs")

        # use the headings as the first row
        out = {table.headings[0]: table.headings[1]} | dict(table)

        if render_values:
            try:
                rendered = {k: context.render(v) for k, v in out.items()}
                return rendered
            except Exception as e:
                logger.error(f"Failed to render template values: {e}")
                logger.error(f"Available variables: {getattr(context, 'variables', {})}")
                raise AssertionError(f"Failed to render values: {e}")

        return out
    except Exception as e:
        logger.error(f"Failed to process table: {e}")
        raise


@step("the response JSON should match:")
def step_response_json_should_match(context: Context):
    try:
        response_json = context.response.json()
    except Exception as e:
        logger.error(f"Failed to parse response as JSON: {e}")
        logger.error(f"Response content: {context.response.content}")
        raise AssertionError(f"Response is not valid JSON: {e}")

    json_path_matchers = get_kv_table(context)

    for jsonpath, raw_expected_value in json_path_matchers.items():
        expected_value = context.render(raw_expected_value)

        try:
            # Use jsonpath_ng.parse() to create a parser, then find() to get matches
            jsonpath_expr = jsonpath_ng.parse(jsonpath)
            matches = jsonpath_expr.find(response_json)

            if not matches:
                raise AssertionError(
                    f"JSON path '{jsonpath}' not found in response: {response_json}"
                )

            actual_value = matches[0].value

            if str(actual_value) != str(expected_value):
                raise AssertionError(
                    f"Expected JSON path '{jsonpath}' to have value '{expected_value}', "
                    f"but got '{actual_value}'"
                )
        except Exception as e:
            raise


@given("I have the following variables:")
def step_variables(context: Context):
    context.variables |= get_kv_table(context)

environment.py

"""Environment setup for behave tests."""

import asyncio
import logging
import os
import threading
import time
from typing import Optional, Dict

import jinja2
import requests
import uvicorn
from dotenv import load_dotenv
from pydantic import BaseModel

from lifecapades.api.app import app
from lifecapades.config import get_settings

load_dotenv(".env.test")
settings = get_settings()

logger = logging.getLogger(__name__)

logging.basicConfig(
    level=logging.getLevelName(os.getenv("LOG_LEVEL", settings.LOG_LEVEL)),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler()]
)


class TestServer:
    """Test server wrapper for uvicorn."""

    def __init__(self, host="127.0.0.1", port=settings.PORT):
        self.host = host
        self.port = port
        self.server = None
        self.thread = None
        self.should_exit = False

    def start(self):
        """Start the server in a separate thread."""
        config = uvicorn.Config(
            app,
            host=self.host,
            port=self.port,
            log_level="error",
            access_log=False
        )
        self.server = uvicorn.Server(config)
        self.thread = threading.Thread(target=self._run_server, daemon=True)
        self.thread.start()

    def _run_server(self):
        """Run the server."""
        asyncio.run(self.server.serve())

    def stop(self):
        """Stop the server."""
        if self.server:
            self.server.should_exit = True
        if self.thread:
            self.thread.join(timeout=5)


def wait_for_server(base_url: str, timeout: int = 30) -> bool:
    """Wait for the server to be ready by polling the health endpoint."""
    start_time = time.time()
    while time.time() - start_time < timeout:
        try:
            response = requests.get(f"{base_url}/health", timeout=1)
            if response.status_code == 200:
                return True
        except requests.exceptions.RequestException:
            pass
        time.sleep(0.1)
    return False


def before_all(context):
    """Set up the test environment before all tests.

    This includes starting the FastAPI server in a separate thread.
    """

    # Get settings from loaded .env.test configuration
    settings = get_settings()

    # Start the test server using settings from .env.test
    context.test_server = TestServer(host="127.0.0.1", port=settings.PORT)
    context.test_server.start()

    # Store settings in context for use in tests
    context.settings = settings

    # Wait for the server to be ready
    context.base_url = f"http://{context.test_server.host}:{context.test_server.port}"
    if not wait_for_server(context.base_url):
        context.test_server.stop()
        raise RuntimeError("Server failed to start within timeout period")

    print(f"Test server started at {context.base_url}")
    print(f"Using configuration: ENV={settings.ENV}, PORT={settings.PORT}")

    context.variables = {"settings": get_settings()}

    # helpers
    context.render = lambda text: jinja2.Template(text).render(context.variables)
    context.build_url = lambda path: context.render(f"{context.base_url}{path}")
    context.reset_next_request = lambda: setattr(context, "next_request", NextRequest())


def after_all(context):
    """Tear down the test environment after all tests.

    This includes stopping the FastAPI server.
    """
    if hasattr(context, 'test_server'):
        context.test_server.stop()
        print("Test server stopped")


def before_scenario(context, scenario):
    """Set up a clean state before each scenario.

    This includes initializing a fresh NextRequest object.
    """
    context.reset_next_request()


def after_scenario(context, scenario):
    """Handle any unhandled exceptions after each scenario."""
    if hasattr(context, 'exception') and context.exception:
        scenario.skip(reason=f"Unhandled exception: {context.exception}")


def before_step(context, step):
    """Clear any previous exceptions before each step."""
    context.exception = None


def after_step(context, step):
    """Catch unhandled exceptions after each step."""
    if step.exception:
        context.exception = step.exception
        logger.error(f"Step failed: {step.name}", exc_info=step.exception)


class NextRequest(BaseModel):
    body: Optional[str] = None
    headers: Optional[Dict[str, str]] = None

health.feature

Feature: Health Check

  Background:
    Given the service is running
    And I have the following variables:
      | token | my-api-token |

  Scenario: Check the health endpoint
    When I send a GET request to "/health"
    Then the response status code should be 200
    And the response JSON should match:
      | $.status   | ok                    |
      | $.app_name | {{settings.APP_NAME}} |

  Scenario: Echo!
    Given I have the request body:
      """
      {
        "message": "Hello, World!"
      }
      """
    And I have request headers:
      | Authorization | Bearer {{token}} |
      | Content-Type  | application/json |
    When I send a POST request to "/echo"
    Then the response status code should be 200
    And the response JSON should match:
      | $.message | Hello, World! |

  Scenario: Interpolated Echo!
    Given I have the following variables:
      | message | Hello, World! |
    Given I have the request body:
      """
      {
        "message": "{{message}}"
      }
      """
    And I have request headers:
      | Authorization | Bearer {{token}} |
      | Content-Type  | application/json |
    When I send a POST request to "/echo"
    Then the response status code should be 200
    And the response JSON should match:
      | $.message | {{message}} |

routes.py

from typing import Annotated

from fastapi import APIRouter, Depends
from pydantic import BaseModel

from lifecapades.config import Settings, get_settings

router = APIRouter()


class EchoRequest(BaseModel):
    message: str


class EchoResponse(BaseModel):
    message: str


@router.get("/health")
async def health(settings: Annotated[Settings, Depends(get_settings)]):
    """A simple health check endpoint that returns the app name and environment."""
    return {
        "status": "ok",
        "app_name": settings.APP_NAME,
        "environment": settings.ENV,
    }


@router.post("/echo")
async def echo(request: EchoRequest) -> EchoResponse:
    return EchoResponse(message=request.message)