接口测试 解析和增量导出 Swagger 的 JSON 文件

Keith Mo · 2017年07月21日 · 最后由 王_test 回复于 2021年05月25日 · 6118 次阅读

用了 Swagger 的程序在运行时会扫描类和方法里的注解,生成 JSON 文档,默认地址为接口访问地址后面加 /v2/api-docs

本地启动的服务默认为 http://localhost:8080/v2/api-docs

Swagger UI 生成的网页的数据源就是这 JSON 文件

缺点:

  • 只有服务启动了才能查看,没有离线版
  • 只支持全量导出,有些由于历史遗留问题没细分的服务可能有几千个接口,不方便查看
  • 靠 Swagger 配置类里设的分组难以满足 “只看这版本新增接口” 这样的需求

思路

写脚本解析 Swagger 生成的 json 文件,可以干很多事:

  • 对比新旧 2 份文件,分别导出新增、删除的接口
  • 按关键字导出特定接口
  • 修改页面发请求时访问的服务器地址
  • ……

json 文件格式如下,只列出一定会出现的属性:

{
    "swagger": "2.0",
    "host": "xxx",
    "basePath": "/",
    "tags":[
        {"name":"xxx-controller","description":"xxx"},
        ...
    ],
    "paths": {
        "<接口地址1>": { ... },
        "<接口地址2>": { ... },
        ...
    },
    "definitions": {
        "<实体类1>": { ... },
        "<实体类2>": { ... },
        ...
    }
}
  • host:服务器域名或 IP
  • basePath:必须以 / 开头,访问某服务的接口时网关里设的那个前缀。如果 host 指定端口直连,这里为/
  • tags:默认为 Controller 类名(@Api注解了的类)。建议保持默认,不需要设@Api(tags={"xx", "yy"})@Api(value="xxx")
  • paths:接口(@ApiOperation注解了的方法)
  • definitions:实体类(@ApiModel注解了的类)

其中 host 和 basePath 最重要,这里设置得不对,页面上就没法访问到服务


以下脚本原本打算用于 Confluence 的 Open API 插件,把 json 导出来之后粘贴到插件里就有跟 swagger ui 一样的效果


但那插件非常不方便,后来有了这帖里的集中展示之后就不用那插件了,export_apis 和 export_models 这 2 个方法也显得多余了(当时想着万一还想加 1 个就粘贴文件里的内容插进去)


实现

新建 Python 文件,比如叫做 swagger_json_dump.py

再另外用 Java 搭个 Spring Boot + Swagger 的 demo,看看不同设置下的输出,边调边写,都是体力活

# coding=utf-8

import argparse
import errno
import json
import os
import re
import sys
from datetime import datetime
from os import path

import requests

HOST = 'localhost:8080'
BASE_PATH = '/'


def check_python_version():
    if sys.version_info[0] < 3 or (sys.version_info[0] == 3 and sys.version_info[1] < 2):
        print('Must use Python 3.2+. Current version:', sys.version)
        sys.exit(1)


def parse_cmdline_args():
    parser = argparse.ArgumentParser()

    parser.add_argument('--url', '-u',
                        help='Swagger API docs URL. Format: "http://<host>[:<port>][/<basePath>]/v2/api-docs" (default: http://localhost:8080/v2/api-docs)')
    parser.add_argument('--host', '-H',
                        help='Domain name or IP. (default: "localhost:8080")')
    parser.add_argument('--basePath', '-b',
                        help='URL prefix for all API paths, relative to the host root. It must start with a leading slash "/". (default: "/" )')
    parser.add_argument('--outputDir', '-o', help='Output directory path. (default: current working directory)')
    parser.add_argument('--compare', '-c',
                        help='Compare to another (old) Swagger JSON file and export newly added APIs and models.')
    parser.add_argument('--api', '-a',
                        help='Export specified API(s). Comma-separated (partial) API path(s), case sensitive. eg. "register,login"')
    parser.add_argument('--model', '-m',
                        help='Export specified model(s). Comma-separated (partial) model name(s), case sensitive. eg. "User,Files"')

    return parser.parse_args()


def get_swagger_json(url):
    try:
        r = requests.get(url)
    except:
        raise Exception('[ERROR] Invalid URL or service is not running. URL: {}'.format(url))

    if r.status_code == 200:
        try:
            response = r.json()
            if response.get('swagger'):
                return response
            else:
                raise Exception('[ERROR] Not a Swagger JSON file:\n' + str(response))
        except:
            raise
    else:
        raise Exception('[ERROR] Cannot get JSON. Status code: {}, content: {}'.format(r.status_code, r.text))


def make_dir(dir_path):
    if not path.exists(dir_path):
        try:
            os.mkdir(dir_path)
        except OSError as e:
            if e.errno != errno.EEXIST:
                raise


def export_whole_json(swagger_json, output_path):
    make_dir(output_path)

    with open(path.join(output_path, 'all.json'), 'w', encoding='utf-8') as f:
        json.dump(swagger_json, f, ensure_ascii=False)


def export_apis(swagger_json, output_path):
    make_dir(output_path)

    api_dir = path.join(output_path, 'api')
    make_dir(api_dir)

    tags = swagger_json['tags']
    for tag in tags:
        tag_name = tag['name']
        tag_dir = path.join(api_dir, tag_name)
        make_dir(tag_dir)

        apis = swagger_json['paths']
        for api in apis:
            if tag_name in json.dumps(apis[api], ensure_ascii=False):
                api_file = path.join(tag_dir, api.replace('/', '--') + '.json')
                with open(path.join(tag_dir, api_file), 'w', encoding='utf-8') as f:
                    json.dump(apis[api], f, ensure_ascii=False)


def export_models(swagger_json, output_path):
    make_dir(output_path)

    model_dir = path.join(output_path, 'model')
    make_dir(model_dir)

    models = swagger_json['definitions']
    for model in models:
        with open(path.join(model_dir, model + '.json'), 'w', encoding='utf-8') as f:
            json.dump(models[model], f, ensure_ascii=False)


def search_refs(string):
    return re.findall('"\$ref": "#/definitions/(.+?)"', string)


def write_selected_elements_to_file(selections, elements, file_obj):
    refs = []
    elements_to_write = []

    for selection in set(selections):
        for element in elements:
            if selection in element:
                content = json.dumps(elements[element], ensure_ascii=False)
                elements_to_write.append('\t\t"{}": {},\n'.format(element, content))

                refs.extend(search_refs(content))

    for element in set(elements_to_write):
        file_obj.write(element)

    return list(set(refs))


def write_all_ref_models_to_file(ref_list, model_list, file_obj):
    ref_list = list(set(ref_list))
    models_to_write = []

    while len(ref_list):
        for ref in set(ref_list):
            if ref in model_list:
                content = json.dumps(model_list[ref], ensure_ascii=False)
                models_to_write.append('\t\t"{}": {},\n'.format(ref, content))

                ref_list.extend(search_refs(content))
                if ref in ref_list:
                    ref_list.remove(ref)

    for model in set(models_to_write):
        file_obj.write(model)


def export_selected_docs(swagger_json, output_file, api_list=None, model_list=None):
    if not api_list:
        api_list = []
    if not model_list:
        model_list = []

    apis = swagger_json['paths']
    models = swagger_json['definitions']
    refs = []

    with open(output_file, 'w', encoding='utf-8') as f:
        f.write('{\n\t"swagger": "2.0",\n')
        f.write('\t"info": {},\n'.format(json.dumps(swagger_json['info'], ensure_ascii=False)))
        f.write('\t"host": "{}",\n'.format(HOST))
        f.write('\t"basePath": "{}",\n'.format(BASE_PATH))
        f.write('\t"tags": {},\n'.format(json.dumps(swagger_json['tags'], ensure_ascii=False)))

        f.write('\t"paths": {\n')
        refs.extend(write_selected_elements_to_file(api_list, apis, f))
        f.write('\t\t"":{}\n')  # Use empty element to avoid trailing comma issue
        f.write('\t},\n')

        f.write('\t"definitions": {\n')
        refs.extend(write_selected_elements_to_file(model_list, models, f))
        write_all_ref_models_to_file(refs, models, f)
        f.write('\t\t"":{}\n')  # Use empty element to avoid trailing comma issue
        f.write('\t}\n')
        f.write('}')


def export_diff_docs(old_file_path, output_path):
    if path.exists(old_file_path) and path.isfile(old_file_path):
        with open(old_file_path, encoding='utf-8') as f:
            old_swagger_json = json.load(f)

        new_apis = set(j['paths'].keys())
        old_apis = set(old_swagger_json['paths'].keys())

        added = new_apis - old_apis
        missing = old_apis - new_apis
        print('\nAdded APIs:\n{}'.format(added))
        print('Missing APIs:\n{}'.format(missing))

        if len(added):
            output_added = path.join(output_path, 'diff-added.json')
            export_selected_docs(j, output_added, added)
            print('\nAdded APIs are saved to:\n{}'.format(output_added))
        else:
            print('[INFO] No new APIs found.')

        if len(missing):
            output_missing = path.join(output_path, 'diff-missing.json')
            export_selected_docs(old_swagger_json, output_missing, missing)
            print('\nMissing APIs are saved to:\n{}'.format(output_missing))
    else:
        print('[ERROR] No such file: {}'.format(old_file_path))


if __name__ == '__main__':
    check_python_version()

    args = parse_cmdline_args()

    if args.url:
        match = re.search('https?://(.+?)(?:/v2/api-docs.*?)', args.url)
        if match:
            HOST = match.groups()[0]
            if '/' in HOST:
                match2 = re.search('(.+?)(/.*)', HOST)
                if match2:
                    HOST = match2.groups()[0]
                    BASE_PATH = match2.groups()[1]

    # overwrite "host" and "basePath" in Swagger JSON
    if args.host:
        HOST = args.host
    if args.basePath:
        BASE_PATH = args.basePath

    swagger_url = args.url or 'http://{}{}/v2/api-docs'.format(HOST, BASE_PATH)

    output_dir = path.abspath(path.expanduser(args.outputDir)) if args.outputDir else path.join(
        os.getcwd(), 'swagger_json_docs-{}'.format(datetime.now().strftime('%Y%m%d-%H%M%S')))

    j = get_swagger_json(swagger_url)

    export_whole_json(j, output_dir)
    export_apis(j, output_dir)
    export_models(j, output_dir)
    print('Saved to:\n{}'.format(output_dir))

    if args.compare:
        old_file = args.compare
        export_diff_docs(old_file, output_dir)

    if args.api or args.model:
        selected_apis = args.api.strip().split(',') if args.api else []
        selected_models = args.model.strip().split(',') if args.model else []
        output_file = path.join(output_dir, 'selected.json')

        export_selected_docs(j, output_file, selected_apis, selected_models)
        print('\nSelected APIs and models are saved to:\n{}'.format(output_file))

注意插了"":{}作为pathsdefinitions数组的最后 1 个元素

因为 JSON 不允许数组和对象最后 1 项后面带逗号,又不知有多少项,插个空元素避免问题,基本不影响页面展示


用法

需要 Python 3.2+

查看帮助

python3 swagger_json_dump.py --help

连本地服务

python3 swagger_json_dump.py

# 默认从 http://localhost:8080/v2/api-docs 获取json文件

默认在当前工作目录生成名为swagger_json_docs-<年月日>-<时分秒>的文件夹

  • 整个 json 的内容原样保存到 all.json
  • 实体类相关的内容($.definitions.<类名> 的值)拆分保存在 model/<类名>.json 下
  • 接口相关的内容($.paths.<接口地址> 的值)拆分保存在 api/<--path--to–api>.json 下(文件名里不能有/,用 -- 代替)

指定 swagger url

python3 swagger_json_dump.py --url "http://localhost:8080/v2/api-docs?group=user"
python3 swagger_json_dump.py --url "http://192.168.3.231/faq/v2/api-docs"

如果 swagger 返回 json 的地址不是默认的/v2/api-docs(例如在配置类里设了分组,要在后面加 ?group=XXX

又或者服务不在本地等等,都要指定 url

覆盖输出文档里的 hostbasePath 属性

python3 swagger_json_dump.py --url "http://localhost:8080/v2/api-docs" --host "192.168.3.231" --basePath "/faq"

这 2 个属性决定 swagger 生成的网页做接口测试时连什么地址

脚本默认从 url 里提取,如果 url 和目标服务器不一样,需要手动指定

(例如上面 url 指向本地服务,但想接口写好之后让人在开发环境调试)

python3 swagger_json_dump.py --host "192.168.3.231" --basePath "/faq"

# 从 http://192.168.3.231/faq/v2/api-docs 获取json文件

如果使用默认地址、无分组,只是服务器不一样,可以只指定这 2 个来代替输完整 url

指定输出目录

python3 swagger_json_dump.py --output "~/apidoc"

找出改动过的接口

python3 swagger_json_dump.py --compare "<之前的json文件路径>"

从线上拿最新的 json 跟保存的文件比较,如果有改动,新增的接口保存到 diff-added.json,删除的接口保存到 diff-missing.json

接口用到的数据模型都会加进文件里

导出部分接口

python3 swagger_json_dump.py --api "/collect"
python3 swagger_json_dump.py --api "findByLabelName,findByUserId"

指定完整或部分地址,区分大小写,多个关键字用逗号分隔

保存到 selected.json,引用到的模型也包含在里面

导出部分模型

python3 swagger_json_dump.py --model "User,Files"

同样保存到 selected.json,不指定导出的接口时,文件里只有模型没有接口


我不懂 Python,以上全是我编的,我实在编不下去了……😆

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
共收到 2 条回复 时间 点赞

效果不太好啊,Post ,delete 类型的接口基本不会识别出来。。

因为同一个叫 v1/disease 的接口里面有 post get 两种 所以循环取 api 的时候取不全,那怎么改呢

需要 登录 后方可回复, 如果你还没有账号请点击这里 注册