接口自動化基礎（五）資料驅動

封裝思想

關鍵資料檔案

配置資料：{project_name}.conf.yaml 存放配置資訊，如環境、郵箱、資料庫資訊等

業務資料：{testcase_name}.api.yaml 存放接口資訊，如：method、url、header、body等

用例資料：{testcase_name}.data.yaml 存放接口入參（測試用例）

過程資料：{testcase_name}.step.yaml 存放過程資料（借鑒httprunner），一般不用這個，可以忽略

配置：根據配置⽂件擷取初始配置和依賴

接口封裝：接口調⽤進⾏抽象封裝，類似PageObject效果

業務流程：業務⽤例設計，含有多個api形成的流程定義；不要再包含任何接口實作細節；斷⾔

用例建構

使⽤package管理業務子產品，使⽤class管理業務功能，使⽤method完成業務具體⾏為，使⽤配置⽂件讀取初始配置，使⽤繼承規劃⽤例執⾏順序，使⽤testcase完成測試⽤例的落地，使⽤assertion完成業務正确性校驗，使⽤資料⽂件管理⽤例的資料驅動，使⽤jenkins完成持續內建。

具體實作

目錄一覽：

用例驅動

最簡單，使用@pytest.mark.parametrize裝飾器實作

步驟：

1. 将用例寫在yaml檔案中
2. 使用 yaml.safe_load讀取檔案
3. 拿到讀取的内容，使用@pytest.mark.parametrize傳參

以管理企業标簽為例：

# base_api.py   封裝資料驅動架構，所有業務代碼的父類
class BaseApi:

    @classmethod
    def yaml_load(cls, path):
    # 封裝yaml檔案的加載
        with open(path) as f:
            return yaml.safe_load(f)
            
    def jsonpath(self, path, r=None):
        if r is None:
            r = self.r.json()  # 拿到上一次使用format方法中的r
        return jsonpath(r, path)
           

# tag.py 業務代碼
class  Tag(BaseApi):
  
  def create_tags(self, token, name):
        url = 'https://qyapi.weixin.qq.com/cgi-bin/tags/create'
      	method = 'post'
        params = {'access_token': token}
        headers = {'Content-Type': 'application/json;charset=utf8'}
        json = {'tag':{'name': name}}
        res = request.request(method = method, url = url, params = params, headers = headers, json=json)
        return res

    def get_tags(self, token):
        url = 'https://qyapi.weixin.qq.com/cgi-bin/tags/get'
        method = 'get'
        params = {'access_token': token}
        res = request.request(method = method, url = url, params = params)
        return self.api_send(self.data['get_tags'])

    def delete_tags(self, token, id):
        url = 'https://qyapi.weixin.qq.com/cgi-bin/tags/delete'
        method = 'post'
        params = {'access_token': token}
        headers = {'Content-Type': 'application/json;charset=utf8'}
        json = {'tag':{'id': id}}
        res = request.request(method = method, url = url, params = params, headers = headers, json=json)
        return res
           

# test_tag.py  case檔案
import pytest
from api.tag import Tag
from api.token import Token

class TestTag:
    data = Tag.yaml_load("../data/test_tag.data.yaml")
    token = Token().get_token('xxx')
    t = Tag()

    # @pytest.mark.parametrize('name', ['黃銅', '白銀', '黃金'])
    @pytest.mark.parametrize('name', data['test_delete'])
    def test_delete(self, name):
      	# 删除标簽用例
        # 如果該标簽已存在，就删除
        self.t.get_tags(token)
        x = self.tag.jsonpath(f'$..tag[[email protected]=="{name}"]')  # jsonpath，json的元素定位方式，類似于xml中的xpth，這裡我們用的是自己封裝後的jsonpath方法（在BaseApi中），此時不用傳r，因為上一句中我們調用的get方法中使用了format方法，r便自動傳入了我們封裝的jsonpath方法中;是以隻要我們調用了format方法，并保證這個r是适合的，就不用傳r
        if isinstance(x, list) and len(x) > 0:
            self.t.delete_tags(token, tag_id=[x[0]['id']])

        # 環境幹淨之後開始測試
        self.t.get_tags(token)
        path = '$..tag[?(@.name!="")]'
        size = len(self.tag.jsonpath(path))

        # 添加新标簽并斷言
        self.t.create_tags(token, name)
        self.t.get_tags(token)
        assert len(self.tag.jsonpath(path)) == size + 1

        # 删除标簽
        tag_id = self.tag.jsonpath(f'$..tag[?(@.name == "{name}")]')[0]['id']
        self.t.delete_tags(token, tag_id)

        # 擷取标簽并斷言
        self.t.get_tags(token)
        assert len(self.tag.jsonpath(path)) == size
           

# test_tag.data.yaml   yaml檔案
"test_delete": ["demo1", "demo2", "demo3", "黃金"," ", "????" ]
           

業務驅動

将業務過程資料寫在yaml方法中，包括請求方法，請求位址，請求參數，請求body，請求頭等，在BaseApi中封裝好通用的請求方法，業務類中直接調用，如下：

# BaseApi.py
class BaseApi:
    yaml_params = {}  # 定義好yaml_params，在業務方法中傳入，以便yaml檔案的内容替換
    
    def api_send(self, req: dict):  # req 傳入加載好的yaml檔案
        req['params']['access_token'] = self.get_token(self.secret)  # 給params的token指派
        
        # 模闆内容替換  yaml檔案中的變量指派
        raw = yaml.dump(req)
        for key, value in self.yaml_params.items():
            raw = raw.replace(f"${{{key}}}", repr(value))
        req = yaml.safe_load(raw)

        r = requests.request(
            req['method'],  # 傳入yaml檔案的method
            url=req['url'],  # 傳入yaml檔案的url
            params=req['params'],  # 傳入yaml檔案的params
            json=req['json']  # 傳入yaml檔案的json
        )
        self.format(r)
        return r.json()
           

# test_tag.api.yaml  業務過程的yaml檔案
add:
  method: post
  url: https://qyapi.weixin.qq.com/cgi-bin/externalcontact/add_corp_tag
  params:
    access_token: ${token}
  json:
    group_id: ${group_id}
    tag:
      - name: ${name}
           

# tag.py 業務檔案
class Tag(Token):

    def __init__(self):
        self.data = self.api_load('../data/test_tag.step.yaml')  # 把yaml檔案中所有的api加載到目前類中 
        
    def get(self, group_id, name):
        self.yaml_params['group_id'] = group_id  # 調用時給BaseApi中的yaml_params傳值，
        self.yaml_params['name'] = name          # 以便yaml檔案的内容替換
        res = self.api_send(self.data['add'])   # 發請求
        return res
           

過程驅動

# base_api.py
def steps_run(self, steps: list):

    for step in steps:
        raw = yaml.dump(step)
        for key, value in self.params.items():
            raw = raw.replace(f"${{{key}}}", repr(value))
        step = yaml.safe_load(raw)

        if isinstance(step, dict):
            if "method" in step.keys():
                method = step['method'].split('.')[-1]
                getattr(self, method)(**step)
            if "extract" in step.keys():
                self.data[step["extract"]] = getattr(self, 'jsonpath')(**step)
            if "assertion" in step.keys():
                assertion = step["assertion"]
                if isinstance(assertion, str):
                    assert eval(assertion)
                if assertion[1] == "eq":
                    assert assertion[0] == assertion[2]
           

steps:
- {method: tag.get}
- {path: "$..tag[?(@.name==${name})]", extract: before}
- {method: tag.add, name: "${name}" }
- {method: get}
- {path: "$..tag[?(@.name==${name})]", extract: after}
- {assertion: [1, eq, 1]}
- {assertion: "len([1,2]) < len([1])" }