Framework Architecture

The testing infrastructure follows a hybrid design pattern combining keyword-driven and data-driven methodologies. It isolates HTTP operations, configuration management, business logic, and test orchestration into distinct layers to maximize reusability and maintainability.

Directory Structure

• utils/ – HTTP client wrappers and response parsing utilities
• config/ – Environment constants and endpoint definitions
• data/ – YAML-based test scenarios and input payloads
• data_utils/ – File I/O helpers for data serialization
• tests/ – Pytest test modules and case definitions
• business/ – Domain-specific API workflow implementations
• conftest.py – Shared fixtures and collection hooks
• execute_suite.py – Runner configuration and report generation

Core Implementation

1. HTTP Utility Layer

Standard HTTP methods are encapsulated within a dedicated client class. This layer handles request dispatch, automatic status validation, and JSON field extraction via JSONPath expressions. Integration with allure.step ensures granular reporting.

import allure
import json
import requests
from jsonpath import jsonpath

class HttpClient:
    @allure.step("Executing GET request")
    def fetch(self, endpoint, query_params=None, **options):
        response = requests.get(endpoint, params=query_params, **options)
        response.raise_for_status()
        return response

    @allure.step("Executing POST request")
    def submit(self, endpoint, payload=None, **options):
        response = requests.post(endpoint, json=payload, **options)
        response.raise_for_status()
        return response

    @allure.step("Extracting value from JSON response")
    def parse_json_field(self, response_text, field_path):
        parsed_data = json.loads(response_text)
        matched_values = jsonpath(parsed_data, f'$..{field_path}')
        return matched_values[0] if matched_values else None

2. Configuration and Test Data

Global endpoints are centralized to prevent duplication. Test scenarios are defined externally using YAML, allowing dynamic parameterization without modifying the core logic.

BASE_URI = "http://api.staging-env.internal"
SERVICE_PORT = "8080"

- scenario_id: valid_login
  title: Successful authentication with valid credentials
  credentials:
    username: "admin_user"
    password: "secure_pass"
  expected_status: "Login successful"
- scenario_id: invalid_creds
  title: Authentication failure with incorrect credentials
  credentials:
    username: "wrong_user"
    password: "secure_pass"
  expected_status: "Invalid credentials"

3. Data Serialization Helper

YAML files are parsed safely using context-managed file operations to prevent resource leaks.

import yaml
from pathlib import Path

def read_test_data(file_path: str) -> list:
    target = Path(file_path)
    with target.open(encoding="utf-8") as stream:
        return yaml.safe_load(stream)

4. Session Fixtures and Collection Hooks

A session-scoped fixture authenticates once, caches the access token, and shares the HTTP client across all tests. Unicode encoding hooks ensure console readability for multilingual identifiers.

import pytest
import allure
from utils.http_client import HttpClient
from config.constants import BASE_URI, SERVICE_PORT

@pytest.hookimpl(tryfirst=True)
def pytest_collection_modifyitems(config, items):
    for item in items:
        item.name = item.name.encode("utf-8").decode("unicode_escape")
        item.nodeid = item.nodeid.encode("utf-8").decode("unicode_escape")

@pytest.fixture(scope="session")
def auth_session_context():
    client = HttpClient()
    login_url = f"{BASE_URI}:{SERVICE_PORT}/v1/authenticate"
    payload = {"username": "admin", "password": "secure_pass"}
    
    with allure.step("Initiating session-wide authentication"):
        response = client.submit(login_url, payload)
        token = client.parse_json_field(response.text, "accessToken")
        
    return {"client": client, "token": token, "initial_response": response}

5. Business Logic Abstraction

Complex API chains are modeled as workflow classes. Each method consumes the session context, executes HTTP calls, validates assertions, and propagates necessary identifiers to subsequent steps. Dependencies are resolved internally to keep test cases declarative.

import allure
from config.constants import BASE_URI, SERVICE_PORT

class WorkflowManager:
    def __init__(self, session_ctx):
        self.ctx = session_ctx
        self.http = self.ctx["client"]

    @allure.step("Authenticating user with data-driven parameters")
    def execute_login(self, test_case):
        allure.dynamic.title(test_case["title"])
        endpoint = f"{BASE_URI}:{SERVICE_PORT}/v1/authenticate"
        resp = self.http.submit(endpoint, test_case["credentials"])
        with allure.step("Validating login response"):
            status_msg = self.http.parse_json_field(resp.text, "message")
            assert status_msg == test_case["expected_status"]
        return resp

    def fetch_user_profile(self):
        with allure.step("Retrieving user profile details"):
            endpoint = f"{BASE_URI}:{SERVICE_PORT}/v1/profile"
            headers = {"Authorization": f"Bearer {self.ctx['token']}"}
            resp = self.http.fetch(endpoint, headers=headers)
            profile_data = resp.json()
            assert profile_data.get("username") == "wind_clear_swordsman"
            return profile_data

    def add_product_to_cart(self, product_id=9001):
        with allure.step("Adding item to shopping cart"):
            profile = self.fetch_user_profile()
            endpoint = f"{BASE_URI}:{SERVICE_PORT}/v1/cart/add"
            headers = {"Authorization": f"Bearer {self.ctx['token']}"}
            payload = {
                "userId": profile["userId"],
                "openId": profile["openId"],
                "productId": product_id
            }
            resp = self.http.submit(endpoint, payload=payload, headers=headers)
            assert resp.json()["status"] == "success"
            return resp.json()

    def finalize_order(self, product_id=9001):
        with allure.step("Processing checkout and order creation"):
            cart_info = self.add_product_to_cart(product_id)
            endpoint = f"{BASE_URI}:{SERVICE_PORT}/v1/orders/create"
            headers = {"Authorization": f"Bearer {self.ctx['token']}"}
            payload = {
                "userId": cart_info["userId"],
                "openId": cart_info["openId"],
                "productId": product_id,
                "cartId": cart_info["cartId"]
            }
            resp = self.http.submit(endpoint, payload=payload, headers=headers)
            assert resp.json()["status"] == "success"

6. Test Case Composition

Test modules remain lightweight, focusing solely on scenario orchestration. Pytest's parametrization engine drives login validation, while fixtures inject the authenticated session into stateful workflows.

import pytest
import allure
from data_utils.yaml_loader import read_test_data
from business.workflows import WorkflowManager

@allure.epic("E-Commerce API Test Suite")
class TestEcommerceOperations:
    @allure.feature("Authentication Flow")
    @allure.story("Credential Validation")
    @pytest.mark.parametrize("case_data", read_test_data("./data/scenarios.yaml"))
    def test_01_login_scenarios(self, case_data, auth_session_context):
        manager = WorkflowManager(auth_session_context)
        manager.execute_login(case_data)

    @allure.feature("Profile Management")
    @allure.story("Data Retrieval")
    def test_02_get_profile(self, auth_session_context):
        manager = WorkflowManager(auth_session_context)
        manager.fetch_user_profile()

    @allure.feature("Shopping Cart")
    @allure.story("Item Addition")
    def test_03_add_to_cart(self, auth_session_context):
        manager = WorkflowManager(auth_session_context)
        manager.add_product_to_cart()

    @allure.feature("Order Processing")
    @allure.story("Checkout Execution")
    def test_04_create_order(self, auth_session_context):
        manager = WorkflowManager(auth_session_context)
        manager.finalize_order()

7. Execution Pipeline

The runner script configures Pytest arguments, directs results to the Allure output directory, and triggers the interactive report server up on completion.

import os
import pytest

def execute():
    report_dir = "./allure-results"
    pytest.main([
        "-v",
        "--tb=short",
        "--alluredir", report_dir,
        "--clean-alluredir",
        "./tests/test_ecommerce.py"
    ])
    os.system(f"allure open {report_dir}")

if __name__ == "__main__":
    execute()