How We Built a Scalable API Automation Platform for Multi‑Environment DevOps Integration

1. Motivation

In fast‑iteration R&D scenarios, traditional testing faces two bottlenecks: manual testing is inefficient due to repetitive execution and environment dependence, and script automation has a high programming barrier, making it costly for teams.

2. Platform Capabilities

Interface Automation – supports HTTP/HTTPS calls, multiple script languages (BeanShell, Python2/3, JavaScript, Groovy), database operations, test scenario orchestration, test plan management, CI/CD pipeline triggers, scheduled tasks, and Feishu group notifications.

Multi‑Network Environment Testing – supports testing in test, simulation, and pre‑release environments.

3. Construction Process

The platform was built in four stages: basic capability construction, test‑plan and CI/CD integration, limited pilot rollout with expanded network support, and finally migration from MeterSphere with pain‑point resolution.

4. Architecture

Layered Architecture

Capability layer – defines interfaces, test cases, scenarios, plans, environments, logs, and results.

Data layer – stores interface cases, scenario data, and execution results.

Infrastructure layer – provides the underlying environment and external services.

Deployment Architecture – a single case set can be executed across multiple environments via centralized deployment.

5. Core Functions

Interface Cases – basic configuration of HTTP requests, pre‑/post‑scripts, pre‑/post‑SQL, parameter extraction, and assertions (regex, JSONPath, XPath, script, schema).

Scenario Orchestration – drag‑and‑drop composition of cases, support for custom requests, scripts, conditional, loop, wait, and transaction controllers.

Variable Tracking – snapshots of environment variables before and after each step enable quick change tracking.

6. Technical Challenges & Solutions

Efficient Case Modeling – extended OpenAPI3 to describe HTTP cases, added pre/post operations and assertion steps.

API Management Integration – auto‑populate request parameters from the API catalog.

Mock Data Generation – use Mockjs variables for request parameters.

JMeter Function Support – generate parameter values via JMeter functions.

cURL Import – directly convert exported cURL commands into test cases.

Variable Change Tracking – store variable snapshots for diff comparison.

Cross‑Network Security Restrictions – implement JVM‑wide ProxySelector with ThreadLocal IP/domain lists to enforce proxy rules and prevent environment leakage.

Script Engine Network Limits – intercept socket operations in Python2/Beanshell/JavaScript to apply the same proxy restrictions; Python3 uses JEP and is exempt.

import socket
import re
import ipaddress
original_socket = socket.socket
original_create_connection = socket.create_connection

def proxy_select(host, port):
    # custom proxy logic
    pass

def is_ip_in_subnet(ip, subnet):
    # subnet check
    pass

def connect_through_proxy(sock, target_host, target_port):
    proxy_host, proxy_port, selected_proxy = proxy_select(target_host, target_port)
    sock.connect((proxy_host, proxy_port))
    if target_port == 443:
        connect_request = f"CONNECT {target_host}:{target_port} HTTP/1.1
Host: {target_host}

"
        sock.sendall(connect_request.encode())
        response = sock.recv(4096)
        if not response.startswith(b"HTTP/1.1 200"):
            raise Exception(f"Proxy connection failed: {response}")

def create_connection_internal(address, *args, **kwargs):
    host, port = address
    allowed, use_proxy, selected_proxy = proxy_select(host, port)
    if not allowed:
        raise Exception(f"Connection to {host}:{port} is not allowed")
    sock = original_socket(socket.AF_INET, socket.SOCK_STREAM)
    if use_proxy:
        connect_through_proxy(sock, host, port)
    else:
        sock.connect(address)
    return sock

class InterceptedSocket(original_socket):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._intercept_connect = True
    def connect(self, address):
        if not self._intercept_connect:
            return super().connect(address)
        host, port = address
        allowed, use_proxy, selected_proxy = proxy_select(host, port)
        if not allowed:
            raise Exception(f"Connection to {host}:{port} is not allowed")
        if use_proxy:
            self._intercept_connect = False
            try:
                connect_through_proxy(self, host, port)
            finally:
                self._intercept_connect = True
        else:
            super().connect(address)

def intercepted_create_connection(address, *args, **kwargs):
    return create_connection_internal(address, *args, **kwargs)

def apply_interception():
    socket.socket = InterceptedSocket
    socket.create_connection = intercepted_create_connection

def remove_interception():
    socket.socket = original_socket
    socket.create_connection = original_create_connection

7. Business Support & Promotion

The platform reduces test case authoring effort, integrates with DevOps pipelines and Feishu for instant failure notifications, and automates end‑of‑day batch verification, dramatically improving testing efficiency.

Key promotion considerations include usability (complex scenario configuration cost) and compatibility (migration from MeterSphere requires data conversion and user adaptation).

8. Future Plans

Implement a scheduling mechanism with private and shared resource clusters to improve stability.

Integrate the platform into the quality middle‑platform, introducing iterative and full‑coverage case libraries.

Refactor architecture to separate management console and execution agents, deploying agents within target network environments to resolve service call and resource access issues.