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.

• Fluent API: Build and send HTTP requests effortlessly.
• Response handling and validations
• PyDantic Integration: Customizable response validation.
• 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.

Detailed documentation can be found at reqflow.org

Install ReqFlow using pip:

pip install reqflow 

• Send a Request
• Headers
• Query parameters
• Response
• Cookies
• Authentication
• Assertions
• PyDantic Response Validation
• 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()...

...

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", ...}

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()...

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
>>> {...}

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"}

Reqflow supports the following authentication methods:

• Basic Authentication
• OAuth2.0 Authentication
• API Keys

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()...

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 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()...

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

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.

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

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

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

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

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 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)

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.

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")

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)

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")

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")

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.