xiaoyezihanghui 2019-10-23
import base64 import json import requests class TestEnCode: origin_url = 'http://0.0.0.0:8000/json_1.txt' url = "http://0.0.0.0:8000/base64_1.txt" def test_get(self): r = requests.get(url=self.url) print(r.content) data = self.decode(r.content) j = json.loads(data) print(j) assert len(j['topics']) == 2 def test_encode(self): r = requests.get(url=self.origin_url) print(r.json()) assert len(r.json()['topics']) == 2 def decode(self, raw): return base64.b64decode(raw) def test_base64_api(self): """返回结果有加密""" res = ApiRequest() res_data = { "schema": "http", "encoding": "base64", "method": "get", "url": "http://docker.testing-studio.com/base64_1.txt", "headers": None } j = res.send(res_data) assert len(j['topics']) == 2 def test_api(self): """返回结果没有加密""" res = ApiRequest() res_data = { "schema": "http",# 请求方式 "encoding": "",# 是否加密 "method": "get",# 请求方法 "url": "http://0.0.0.0:8000/json_1.txt", "headers": None } j = res.send(res_data) assert len(j['topics']) == 2 class ApiRequest: """处理接口返回数据加密""" def send(self, data: dict): if data["schema"] == "http": # 把host修改为iP,并附加host到headers """支持多环境测试""" env = { "docker.testing-studio.com": { "test": "192.168.1.37", "dev": "192.168.2.14" }, "default": "test" } # 井host修改为IP data["url"] = str(data['url']).replace("docker.testing-studio.com", env['docker.testing-studio.com'][env['default']]) # 添加host到headers中 data["headers"]["host"] = "docker.testing-studio.com" res = requests.request(data["method"], data['url'], headers=data["headers"]) if data['encoding'] == 'base64': return json.loads(base64.b64decode(res.content)) else: return json.loads(res.content) elif "dubbol" == data['schema']: pass elif data['urllib'] == "urllib": pass elif data['schema'] == "websocket": pass else: pass