How to Build a Scalable API Automation Framework for Search Services

Background and Design Thinking

The article starts by defining what an API is in the context of a search service architecture, illustrating how multiple modules (dispatcher, searcher, merger, etc.) communicate via JSON, text, or HTML responses, and why API automation is valuable for high‑frequency, isolated interfaces.

Why Conduct API Testing

API testing offers a high ROI because it is easier to automate than lower‑level components, especially for services with many independent endpoints such as search APIs.

Automation Tool Core Features

Supports multiple request methods (GET, POST, HTTPS, encrypted, authenticated, etc.).

Separates test data from code for easy maintenance.

Provides online monitoring and alerting (IP/port separation, dynamic URL composition, scheduled runs, email notifications).

Flexible execution: independent or batch calls via parameter passing.

Clear, structured test reports with log aggregation and post‑processing.

Tool Architecture

The layered diagram (Figure 4) shows a configuration layer, a common‑function layer, and a test‑case layer.

Execution Process

Three stages are described:

Preparation – load test data, assemble full request URLs, and define expected results.

Crawling – invoke each API, capture response content, and compare with expectations.

Result Output – analyze logs, format reports, and send emails.

Implementation Details

Data Management

All mutable input parameters, scenario values, and expected results are stored in CSV‑style text files (comma‑separated columns). Examples of ordinary and special data construction for translation, map, and video interfaces are shown (Figures 6‑9).

Configuration Layer

Two parts: a config.ini file per API containing thread count, log path, ports, etc., and a Python configuration module for JSON nodes, report titles, and data‑center IP mappings (Figure 10‑11).

Common Function Layer

Key utilities include a data‑reader class and generic API‑crawling functions.

# encoding: utf-8
import csv
class DataReader:
  def __init__(self, data_file_path):
    self.data_file_path = data_file_path
    self.data_file = open(self.data_file_path)
    # filter lines starting with '#'
    self.data_reader = csv.DictReader(filter(lambda row: row[0] != '#', self.data_file))
    self.datas = []
    for row in self.data_reader:
      self.datas.append(row)
  def getDatas(self):
    return self.datas

def getPostDate(base_url, data):
    base64data = base64.b64decode(data)
    req_header = {'Host':'tip.f.360.cn'}
    return_dict = {}
    return_dict["url"] = base_url
    try:
        req = urllib2.Request(base_url, data=base64data, headers=req_header)
        res = urllib2.urlopen(req, timeout=3)
        doc = res.read()
        status = res.getcode()
    except Exception as e:
        return return_dict
    else:
        return_dict["data"] = doc
        return_dict["status"] = status
    return return_dict

def getDataforNlp(base_url, **args):
    urlargs = ''
    for key in args:
        urlargs += "&" + key + "=" + args[key]
    url = base_url + urlargs
    return_dict = {"url": url}
    try:
        req = requests.get(url, timeout=1, headers=req_header)
        response_time = req.elapsed.microseconds
        status = req.status_code
        return_dict["status"] = status
        return_dict["response_time"] = "%sms" % (response_time/1000)
    except Exception as e:
        print(e)
    else:
        doc_ = json.loads(req.text)
        return_dict["data"] = doc_
    return return_dict

Test Case Management

Test cases are organized per API in separate *.py files using unittest. Each file contains multiple test_* functions that validate response data.

from common import commonMethod   // import utility functions
from common import CommonTestCase // setup/teardown base class
import unittest
from conf import data            // import JSON node config
class Mip(CommonTestCase.CommonTestCase):
    def test_result(self):
        def fuc(row):
            query = row["query"].strip()
            test_result = "Fail"
            except_type = "mip" if query.startswith("mip:") else "amp"
            message = ''
            log = []
            url = self.__class__.url
            return_dataall = commonMethod.getDataForImg(url, query, "url=")
            return_data = return_dataall.get("data")
            allurl = return_dataall.get("url")
            status = return_dataall.get("status")
            response_time = return_dataall.get("response_time")
            content = commonMethod.dict_get(return_data, data.DataResult.content)
            errorno = commonMethod.dict_get(return_data, data.DataResult.todaynewserrorno)
            if return_data:
                if content:
                    htm_type = "mip" if "<html mip" in second_line or "mip>" in second_line or "mip=" in second_line else "amp"
                    if except_type == htm_type:
                        test_result = "Pass"
                    else:
                        message = "格式不正确query显示为%s但content中目前为%s" % (except_type, htm_type)
                else:
                    message = "接口访问失败未返回content数据statuscode=%s|response_time=%sms" % (status, response_time)
            else:
                message = "接口访问失败未返回data数据statuscode=%s|response_time=%sms" % (status, response_time)
            log = [test_result, query, message, pageurl, content]
            if test_result == 'Pass':
                self.pass_num += 1
                log.insert(0, self.comm_log_str)
                self.logger.info(u'{0[0]},{0[1]},{0[2]},{0[3]}'.format(log))
            elif test_result == 'Fail':
                self.fail_num += 1
                log.insert(0, self.comm_log_str)
                self.logger.error(u'{0[0]},{0[1]},{0[2]},{0[3]},{0[4]},{0[5]}'.format(log))
        commonMethod.pool_map(fuc, self.datas, self.thread_num)

Logging and Reporting

Logs are structured and stored in a database; a formatted report can be generated and sent via email, as shown in the screenshot (Figure 12).

CI Integration and Online Monitoring

The tool is triggered by shell scripts that accept parameters such as port, hostname, recipient, and working directory. It can be invoked from Jenkins pipelines for continuous integration and from cron jobs for periodic monitoring, with alerts sent via email, SMS, or corporate messaging platforms.

Practical Cases in Search Service

Continuous Integration: The Merger service pipeline automatically builds and deploys without QA intervention.

Automated email notifications are sent after each successful build.

Online Monitoring: Cron jobs iterate over all data centers, sending alerts via email, SMS, or corporate chat when errors are detected. A whitelist mechanism prevents duplicate alerts for known issues.

Quantified Benefits

Automated testing of over 70 APIs across more than ten business lines, accumulating >8,000 test cases.

Integrated into multiple CI pipelines, achieving fully automated server‑side testing without QA hand‑over.

Provides developers with local debugging capabilities for individual APIs.

Continuous online monitoring has identified numerous production issues (third‑party failures, environment problems, data updates) and helped prevent repeat incidents.