ReqFlow: Python API Testing Made Easy

ReqFlow is a robust Python library for efficient and intuitive API testing. Designed with a fluent and flexible interface, it simplifies the process of crafting and validating HTTP requests. Lowering the entry barrier, perfect for both beginners and advanced users, ReqFlow's RestAssured-like approach makes Python API testing straightforward and powerful.

Features

• Fluent API: Build and send HTTP requests effortlessly.
• Response handling and validations
• PyDantic Integration: Customizable response validation.
• Asynchronous Support: Perform concurrent API requests.
• Utility Methods: Common assertions and response manipulations.
• Reporting and PyTest Integration

The tool is still in development, braking changes are possible. Any feedback and contributions are highly appreciated.

Documentation

Detailed documentation can be found at reqflow.org

Installation

Install ReqFlow using pip:

pip install reqflow 

Quick Start

Contents

• Send a Request
• Headers
• Query parameters
• Response
• Cookies
• Authentication
• Assertions
• PyDantic Response Validation
• Asynchronous requests with ReqFlow
• Upload files
• Download files
• Logging and PyTest Integration

Let's make a simple request to HTTPBin API by create a new client and making a GET request to the /get endpoint and asserting the response status code is 200:

client = Client("https://httpbin.org")
given(client).when("GET", "/get").then().status_code(200)

Alternatively, the request can be sent without explicitly defined client object:

given(url="https://httpbin.org").when("GET", "/get").then().status_code(200)

For other HTTP methods, you can use the GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS parameters:

PAYLOAD = {"foo": "bar"}

given(client).body(PAYLOAD).when("POST", "/post").then()...

given(client).body(PAYLOAD).when("PUT", "/put").then()...

given(client).body(PAYLOAD).when("PATCH", "/patch").then()...

given(client).when("DELETE", "/delete").then()...

...

Headers

To set a header for your request, one can use the header method traling the given method:

given(client).header("Content-Type", "application/json")\
    .when("POST", "/post")\
    .then()...

In case you want to set multiple headers, you can use the headers method:

HEADERS = {"Content-Type": "application/json", "Accept": "application/json"}

given(client).headers(HEADERS).when("POST", "/post").then()...

To retrieve one or multiple response headers:

hdr = given(client).when("GET", "/get").then().get_header("Content-Type")
hdr
>>> "application/json"

hdrs = given(client).when("GET", "/get").then().get_headers()
hdrs
>>> {"Content-Type": "application/json", "Content-Length": "1234", ...}

Query parameters

To set up query parameters in the URL, use the query_param method:

PARAMS = {"foo": "bar"}
given(client).query_param(PARAMS).when("GET", "/get").then()...

Response operations

If you want to retrieve the response object, you can use the then.().get_response() method:

r = given(client).when("GET").then().get_response()
r
>>> <UnifiedResponse object at 0x108f81es0>

The API reference for the UnifiedResponse object can be found here.

To retrieve the response content, you can use the then.().get_content() method:

data = given(client).when("GET").then().get_content()
data
>>> {...}

Cookies

To set a cookie for your request, one can use the cookie method trailing the given method:

cks = {"cookie1": "value1", "cookie2": "value2"}
given(client).cookies(cks).when("GET", "https://httpbin.org/cookies")\
    .then()...

To retrieve one or multiple response cookies:

cks = given(client).when("GET", "https://httpbin.org/cookies")\
    .then().get_cookies()
ck
>>> {"cookie1": "value1", "cookie2": "value2"}

Authentication

Reqflow supports the following authentication methods:

• Basic Authentication
• OAuth2.0 Authentication
• API Keys

Basic Authentication

To set up basic authentication, use the with_auth method trailing the when method:

given(client)\
        .when("GET", "/basic-auth/user/passwd").with_auth("user", "passwd")\
        .then()...

OAuth2 Authentication (Bearer Token)

The Bearer token can be set either explicitly in header or via the with_oauth2 method:

given(client).when("GET", "/bearer").with_oauth2(token)\
        .then()...

API Keys

API Key authorization method represents a wrapper for setting a header with a known name and value in the form of an API key.

given(client).when("GET", "/bearer").with_api_key(HEADER_NAME, API_KEY)\
        .then()...

Assertions

ReqFlow provides a set of assertions to validate the response parameters as well as the embedded assertion functions to validate the response content.

Assertion Functions

The following embedded assertion functions are available:

• contains_string()
• equal_to()
• not_equal_to()
• greater_than()
• less_than()
• list_contains()
• is_none()
• is_not_none()
• matches_regex()
• and_(*assertion_functions)
• or_(*assertion_functions)

The list of assertion functions and with the descriptions can be found here.

Status Code

given(client).when("GET", "/get").then().status_code(200)

Response Time

given(client).when("GET", "/get?foo=bar").then()\
    .assert_response_time(max_time=0.5)

Cookies

given(client).query_param(params).when("GET", "/cookies/set").then()\
        .assert_cookie('chocolate', equal_to('chip'))

Headers

    given(client).when("GET", "/get?foo=bar")\
        .then().assert_header("Content-Type", equal_to("application/json"))

Response Content

To validate a specific response content value, the assert_body can be used along with the embedded assertion functions. The parameter search is implemented by using the jsonpath-ng package.

given(client).when("GET", "/get?foo=bar").then()\
    .status_code(200).\
    assert_body("args.foo", equal_to("bar"))

PyDantic Response Validation

PyDantic integration allows to define precise data structures and use them as a blueprint for the response validation. The validation is performed by the validate_data method and passing the PyDantic model as a parameter.

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    username: str
    email: str
    address: dict
    phone: str
    website: str
    company: dict
    
given(client).when("GET", "/users/1").then()\
    .status_code(200)\
    .validate_data(User)

Asynchronous Requests

ReqFlow supports asynchronous requests to enhance performance for I/O-bound operations.

import asyncio
import pytest 


@pytest.mark.asyncio
async def test_get_request_async():
    client = Client(base_url="https://httpbin.org")
    result = await given(client).when("GET", "/get?foo=bar").then_async()
    result.status_code(200).assert_body("args.foo", equal_to("bar"))


@pytest.mark.asyncio
@pytest.mark.parametrize("test_data", [
    (["/get", "/ip", "/user-agent"], [{"test": f"value{i}"} for i in range(50)])
])
async def test_async_performance(test_data):
    endpoints, param_list = test_data
    async with Client(base_url="https://httpbin.org") as client:

        tasks = [
            given(client).query_param(params).when("GET", endpoint).then_async()
            for params in param_list
            for endpoint in endpoints
        ]
        
        results = await asyncio.gather(*tasks)
        for result in results:
            result.status_code(200)

For more information on async support, please refer to the documentation.

Upload files

To upload a file to a particular endpoint, use the file_upload method specifying the field_name and the path to the file:

given(client).file_upload(field_name="userfile", file_path="data/test.png")\
    .when("POST", "/doc/file_upload.html")\
    .then().status_code(200)

field_name must be the same as the name of the form field in the request.

Download files

To download a file or save the response content to a file with a desired format, use the save_response_to_file method specifying the file_path parameter:

given(client).when("GET").then()\
    .status_code(200)\
    .save_response_to_file(file_path="file.pdf")

Logging

ReqFlow supports logging to aggregate the test results and provide a detailed overview of the execution across all client objects. To enable logging, set the logging argument to True when creating a new client object:

client = Client("https://httpbin.org", logging=True)

With the logging enabled, all requests/responses made by the client object will be stored in the GlobalLogger object

from reqflow.utils.logger import GlobalLogger
from reqflow import Client, given

client = Client(base_url="https://httpbin.org", logging=True)
given(client).when("GET", "/get?foo=bar").then().status_code(200)

logs = GlobalLogger.get_logs()
print(logs)

>>> [
        {'function': 'test_function_name',
        'request': {...request details...},
        'response': {...response details...}
    ]

The logger saves the following information:

• function - the name of the test function (or the function from where the then method was called)
• request - the request details (method, url, headers, body, query parameters, redirect and timeout options, cookies)
• response - the response details (status code, headers, content, cookies, response time)

Generating Reports

HTML Report

To generate an HTML report, use the generate_html_report method from the GlobalLogger object:

from reqflow.utils.logger import GlobalLogger
from reqflow import Client, given

client = Client(base_url="https://httpbin.org", logging=True)
given(client).when("GET", "/get?foo=bar").then().status_code(200)

GlobalLogger.generate_html_report(file_path="/path/to/report.html", report_title="Smoke Test")

JSON Report

To generate a JSON report, use the generate_json_report method from the GlobalLogger object:

from reqflow.utils.logger import GlobalLogger
from reqflow import Client, given

client = Client(base_url="https://httpbin.org", logging=True)
given(client).when("GET", "/get?foo=bar").then().status_code(200)
# OR
given(url="https://httpbin.org", logging=True).when("GET", "/get?foo=bar").then().status_code(200)


GlobalLogger.generate_json_report(file_path="/path/to/report.json")

PyTest Integration

To integrate ReqFlow reporting/logging with PyTest, one can use PyTest's fixtures and hooks in the conftest.py file:

import pytest
from reqflow.utils.logger import GlobalLogger

@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_protocol(item, nextitem):
    yield

@pytest.hookimpl
def pytest_sessionfinish(session, exitstatus):
    logs = GlobalLogger.get_logs()
    if logs:
        GlobalLogger.generate_html_report(file_path="test_report.html", report_title="Aggregated Requests")
        GlobalLogger.generate_json_report(file_path="test_report.json")
    GlobalLogger.clear_logs()

With the example above, the report will be generated after the test session is finished. The results will be aggregated across all test functions and clients within the session.