Commit 0197ef9c authored by “xbozhong's avatar “xbozhong

测试1

parent 577b71fa
...@@ -6,16 +6,13 @@ import random ...@@ -6,16 +6,13 @@ import random
from Common.handle_db import HandleDB from Common.handle_db import HandleDB
db = HandleDB()
# 获取一个没有注册的手机号码 # 获取一个没有注册的手机号码
def get_new_phone(): def get_new_phone(db):
# 1、获取手机号码 # 1、获取手机号码
mobile_phone = __generator_phone() mobile_phone = __generator_phone()
# 2、对手机号码进行校验 # 2、对手机号码进行校验
while True: while True:
data = check_phone_in_db(mobile_phone) data = check_phone_in_db(mobile_phone, db)
if data == 0: if data == 0:
return mobile_phone return mobile_phone
else: else:
...@@ -35,7 +32,7 @@ def __generator_phone(): ...@@ -35,7 +32,7 @@ def __generator_phone():
# 通过随机生产的号码去数据库进行查询 # 通过随机生产的号码去数据库进行查询
def check_phone_in_db(mobile_phone): def check_phone_in_db(mobile_phone, db):
sql = 'SELECT * FROM FlowMicros.micros_customer WHERE phone = "{}";'.format(mobile_phone) sql = 'SELECT * FROM FlowMicros.micros_customer WHERE phone = "{}";'.format(mobile_phone)
data = db.select_count(sql) data = db.select_count(sql)
return data return data
......
...@@ -9,9 +9,9 @@ from Common.my_logger import logger ...@@ -9,9 +9,9 @@ from Common.my_logger import logger
from Common.handle_config import conf from Common.handle_config import conf
def send_request(method, url, data=None, token=None): def send_request(method, url, data=None, token=None, Authorization=None):
# 请求头信息 # 请求头信息
headers = __handle_headers(token) headers = __handle_headers(token, Authorization)
# url处理,拼接获取完整的url # url处理,拼接获取完整的url
url = __pre_url(url) url = __pre_url(url)
# data处理,把字符转化为字典 # data处理,把字符转化为字典
...@@ -33,10 +33,12 @@ def send_request(method, url, data=None, token=None): ...@@ -33,10 +33,12 @@ def send_request(method, url, data=None, token=None):
return resp return resp
def __handle_headers(token=None): def __handle_headers(token=None, Authorization=None):
headers = {'App': 'galaxy-admin', 'Content-Type': 'application/json'} headers = {'App': 'galaxy-admin', 'Content-Type': 'application/json'}
if token: if token:
headers['token'] = '{}'.format(token) headers["token"] = "{}".format(token)
if Authorization:
headers["Authorization"] = "{}".format(Authorization)
return headers return headers
......
[server] [server]
base_url = http://middle-platform.galaxy-immi.com:8183 ;base_url = http://middle-platform.galaxy-immi.com:8183
base_url = http://test.crm.galaxy-immi.com
[log] [log]
name = HXX name = HXX
......
import os.path import os.path
import pytest import pytest
import time
from Common.handle_excel import HandleExcel from Common.handle_excel import HandleExcel
from Common.handle_path import datas_dir from Common.handle_path import datas_dir
...@@ -9,24 +10,27 @@ from Common.handle_data import replace_case_by_regular ...@@ -9,24 +10,27 @@ from Common.handle_data import replace_case_by_regular
from Common.handle_envdata import EnvData from Common.handle_envdata import EnvData
from Common.handle_requests import send_request from Common.handle_requests import send_request
from Common.handle_extract_data_from_response import extract_data_from_response from Common.handle_extract_data_from_response import extract_data_from_response
from Common.handle_db import HandleDB
@pytest.fixture(scope="class") @pytest.fixture(scope="class")
def get_newPhone(): def get_newPhone():
logger.info("************ 优才流程 开始执行 ***************") logger.info("************ 优才流程 开始执行 ***************")
db = HandleDB()
# 获取手机号码 # 获取手机号码
mobile = get_new_phone() mobile = get_new_phone(db)
# 设置手机号码为全局变量 # 设置手机号码为全局变量
setattr(EnvData, "mobile", mobile) setattr(EnvData, "mobile", mobile)
# setattr(EvnData, "username", "赵六") setattr(EnvData, "username", "hxx_alan测试")
# setattr(EvnData, "copywriter_id", 1391) # setattr(EvnData, "copywriter_id", 1391)
yield yield db
db.close()
logger.info("************ 优才流程 执行结束 ***************") logger.info("************ 优才流程 执行结束 ***************")
@pytest.mark.usefixtures("get_newPhone") @pytest.mark.usefixtures("get_newPhone")
class TestTalentedPerson: class TestTalentedPerson:
print(get_newPhone)
# 第一步,读取excel中的数据 # 第一步,读取excel中的数据
exc = HandleExcel(os.path.join(datas_dir, "接口业务流.xlsx"), "优才") exc = HandleExcel(os.path.join(datas_dir, "接口业务流.xlsx"), "优才")
cases = exc.read_all_datas() cases = exc.read_all_datas()
...@@ -34,12 +38,13 @@ class TestTalentedPerson: ...@@ -34,12 +38,13 @@ class TestTalentedPerson:
exc.close_file() exc.close_file()
print(cases) print(cases)
# # 如果存在distribution变量,说明该订单被分配了,需要跳过分配客户接口 # # 如果存在distribution变量,说明该订单被分配了,需要跳过分配客户接口
if hasattr(EnvData, "distribution"): # if hasattr(EnvData, "distribution"):
del cases[12:15] # del cases[12:15]
@pytest.mark.parametrize("case", cases) @pytest.mark.parametrize("case", cases)
def test_talented_person(self, case): def test_talented_person(self, case, get_newPhone):
logger.info("********** 执行用例{}:{} **********".format(case["id"], case["title"])) logger.info("********** 执行用例{}:{} **********".format(case["id"], case["title"]))
# 第二步,替换request_data中的未知数 # 第二步,替换request_data中的未知数
case = replace_case_by_regular(case) case = replace_case_by_regular(case)
...@@ -47,7 +52,18 @@ class TestTalentedPerson: ...@@ -47,7 +52,18 @@ class TestTalentedPerson:
# 第三步,执行请求 # 第三步,执行请求
if hasattr(EnvData, "token"): if hasattr(EnvData, "token"):
response = send_request(case["method"], case["url"], case["request_data"], token=getattr(EnvData, "token")) if case["title"] == "提交修改后的文档" or case["title"] == "提交":
time.sleep(90)
id_list = getattr(EnvData, "id_list")
dic = {}
for i in range(len(id_list)):
dic["id"] = id_list[i]
response = send_request(case["method"], case["url"], dic, token=getattr(EnvData, "token"),
Authorization=getattr(EnvData, "client_token"))
time.sleep(90)
else:
response = send_request(case["method"], case["url"], case["request_data"],
token=getattr(EnvData, "token"))
else: else:
response = send_request(case["method"], case["url"], case["request_data"]) response = send_request(case["method"], case["url"], case["request_data"])
...@@ -55,9 +71,17 @@ class TestTalentedPerson: ...@@ -55,9 +71,17 @@ class TestTalentedPerson:
if case["extract_data"]: if case["extract_data"]:
extract_data_from_response(case["extract_data"], response.json()) extract_data_from_response(case["extract_data"], response.json())
# 断言 # 第五步,执行sql
# 任务流转接口
if case["check_sql"] and case["check_keyword"] == "order_task_id":
order_task_id = get_newPhone.select_one_data(case["check_sql"])["id"]
setattr(EnvData, "order_task_id", order_task_id)
elif case["check_sql"] and case["check_keyword"] == "id_list":
ids = get_newPhone.select_all_data(case["check_sql"])
id_list = []
for i in ids:
id_list.append(i["id"])
setattr(EnvData, "id_list", id_list)
# 第五步,断言
assert response.json()["code"] == 200
import pytest import pytest
if __name__ == '__main__': if __name__ == '__main__':
pytest.main(["-s", "-v", "--alluredir=Outputs/reports", "--clean-alluredir"]) pytest.main(["-s", "-v",
\ No newline at end of file "--reruns", "2", "--reruns-delay", "5",
"--alluredir=Outputs/reports",
"--clean-alluredir"])
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment