前沿测试技术MockServer

霍格沃兹测试开发学社 测试开发实战宝典
1
  • mockserver 框架
  • mockserver 实战
2

Mock多场景示例

import time
import requests
from csrmock.core.mitm.mitm_proxy import MitmProxy
from csrmock.core.mock_client import MockClient
from mtf.core.utils import Utils


class TestMock:

    def setup_class(self):
        """
        测试准备阶段，启动mock服务
        :return:
        """
        # 使用另一个线程启动mockserver
        self.process_proxy = Utils.process(target=MitmProxy.main)
        while True:
            r = set()
            r.add(Utils.connect('127.0.0.1', 8006))
            r.add(Utils.connect('127.0.0.1', 8080))
            if r == {0}:
                break
            else:
                time.sleep(1)
        # 指定代理
        self.proxies = {'http': 'http://127.0.0.1:8080', 'https': 'http://127.0.0.1:8080', }
        self.mc = MockClient('http://localhost:8006')

    def test_mock(self):
        # 测试简单mock场景,请求的所有响应数据均响应为设定的响应信息
        req = {'method': 'GET', 'path': "/get"}
        res = {'test': "ceshiren"}
        # 添加mock规则
        self.mc.mock(req, res)
        # 测试用例
        r = requests.get("https://httpbin.testing-studio.com/get", proxies=self.proxies, verify=False)
        print(r.json())

    def test_limit(self):
        # 测试limit 消耗mock场景,请求的前两次响应数据为设定的响应信息，之后为正常的响应信息
        config = {"limit": 2}
        req = {'method': 'GET', 'path': "/get"}
        res = {'status': "fail"}
        self.mc.mock(req, res, config=config)
        for i in range(1, 4):
            r = requests.get("https://httpbin.testing-studio.com/get", proxies=self.proxies, verify=False)
            # 前两次打印{'test': "ceshiren"}， 之后打印正常的响应信息
            print(r.json())

    def test_actions(self):
        req = {'method': 'GET', 'path': "/get"}
        res = {'test': "ceshiren"}
        # 响应mock数据之前,执行等待5s的操作
        config = {"action": [{"time.sleep": 5}]}
        self.mc.mock(req, res, config=config)
        start_time = time.time()
        r = requests.get("https://httpbin.testing-studio.com/get", proxies=self.proxies, verify=False)
        end_time = time.time()
        #打印结束时间-开始时间，结果应约等于设定的 5s 
        print(end_time-start_time)
3

TCP 协议过程代码示例

import time
from csrmock.core.callback_server import CallBackServer
from csrmock.core.mock_client import MockClient
from mtf.core.utils import Utils


class TestTCP:

    def setup_class(self):
        # 启动http协议mock服务
        MockClient.start_mock_server(protocol='json', port_listen=8001, port_admin=8002)
        # 启动tcp协议mock服务
        MockClient.start_mock_server(protocol='tcp', port_listen=9102, port_admin=8004)
        server = CallBackServer()
        process_callback = Utils.process(target=server.start_server, args=(8088,))
        self.mc = MockClient("http://127.0.0.1:8004")
        time.sleep(10)
#
    def test_tcp(self):
        # 添加tcp 协议的mock规则
        self.mc.mock(req= "12345678", res= "success", protocol="plain")
        # 发送socket请求，请求数据为12345678
        Utils.nc("127.0.0.1", 9102, 12345678)
        print(self.mc.history())
4

shell mock yaml 驱动测试用例示例

test_shell:
  - exec: from mtf.core.utils import Utils
  # 启动mock server
  - csrmock.core.mock_client.MockClient.start_mock_server:
      protocol: json
      port_admin: 8009
      port_listen: 8005
      forward_host: 127.0.0.1
      forward_port: 8010
  - save: p1
  # 添加mock规则
  - csrmock.core.mock_client.MockClient: http://127.0.0.1:8009
  - return_to: mock_client
  - mock_client.mock:
      req: devices
      res: 'python:res.content="{\"result\": \"hello\"}".encode()'
      protocol: json
  - mock_client.mock:
      req: forward
      #      res:
      #        python:
      #          - v=res.content.encode()+="\nshell mock demo\n"
      #          - res.content="{\"result\": v}".encode()
      res: 'python: import json; j=json.loads(res.content.decode()); j["result"]+="\nshell mock demo"; res.content=json.dumps(j).encode();'
      protocol: json
  # 启动shell mock服务，解析 shell 命令
  - process:
      target: $(csrmock.core.shell.shell_server.ShellServer.start_server)
      kwargs:
        port: 8010
  - save: p2
  - Utils.wait_port_available: [ 8010 ]
  - platform.system: [ ]
  # 判断执行的操作系统
  - if: $(self._return() == 'Windows')
    true:
      - var:
          adb: $([os.sep.join([p, 'adb']) for p in os.getenv('PATH').split(os.pathsep) if os.path.exists(os.sep.join([p, 'adb.exe']))][0])
    false:
      - var:
          adb: $([os.sep.join([p, 'adb']) for p in os.getenv('PATH').split(os.pathsep) if os.path.exists(os.sep.join([p, 'adb']))][0])
  # 使用fake 文件发起命令
  - csrmock.core.shell.__main__.FakeFile.main:
      - -p
      - ${adb}
      - -c
      - devices
  - save: r
  - assert: [ "${r}", hello ]