bianxinhua / partner_api · 提交

转到一个项目

作者 dengyuting
提交 bdaa0f47d475e387913515405a0578a744183ba1 1 个父辈 80321ce3

增加数据清理

内嵌 并排对比
正在显示 7 个修改的文件 包含 373 行增加 和 320 行删除
name: /api/time/get_timestamp
request:
headers:
User-Agent: okhttp/3.12.3
method: GET
url: http://tongji.fjmaimaimai.com/api/time/get_timestamp
validate:
- eq:
- content.msg
- success
- eq:
- content.code
- 0
\ No newline at end of file
... ...
name: 删除意向订单
base_url: ${ENV(base_manager_url)}
variables:
id: 1027
id: ${get_order_id_from_database(214, 1)}
request:
headers:
Content-Type: application/json;charset=utf-8
... ...
... ... @@ -4,8 +4,8 @@ import time,datetime
from urllib import parse
from helpfunc.login_func import *
# from helpfunc.pg_db_func import *
from helpfunc.db_func import *
from helpfunc.pg_db_func import *
# from helpfunc.db_func import *
from helpfunc.header_func import *
from random import randint
# from helpfunc.pg_app_db_func import *
... ...
# @Time :2019/11/26 11:25
# @Author :dengyuting
# @File :db_func.py
import pymysql
db_config ={"host": "101.37.68.23",
"port": 3306,
"user": "dengyuting",
"password": "dyt_123456",
"db": "opportunity_test",
"charset": 'utf8'}
def get_id1_from_database(tbl, company_id, column, column_value):
db = pymysql.connect(**db_config)
cursor = db.cursor()
company_id = str(company_id)
column_value = str(column_value)
sql= "select id from " + tbl + " where company_id=" + company_id + " and " + column + "=" + column_value + \
" and delete_at=0"
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def get_chanceid_from_database(company_id):
db = pymysql.connect(**db_config)
cursor = db.cursor()
company_id = str(company_id)
sql= "select id from chance where company_id=" + company_id + " order by create_at desc"
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def setup_hook_clean_db(companyid):
"""
初始化时清理数据库中对应公司的历史数据
:return:
"""
db = pymysql.connect(**db_config)
cursor = db.cursor()
companyid = str(companyid)
# get_adminid = 'select admin_id from company where id='+companyid
# get_user_company_id = 'select id from user_company where company_id='+companyid
try:
# cursor.execute(get_adminid)
# result = cursor.fetchone()
# admin_id = result[0]
# admin_id = str(admin_id)
# cursor.execute(get_user_company_id)
# result = cursor.fetchone()
# user_company_id = result[0]
# user_company_id = str(user_company_id)
# cursor.execute('delete from department where company_id=' + companyid + ' and is_top!=1')
# cursor.execute('delete from user_company where company_id=' + companyid + ' and user_id!='+admin_id)
# cursor.execute('delete from user_department where company_id=' + companyid + ' and user_company_id!='+ user_company_id)
# cursor.execute('delete from position where company_id=' + companyid)
# cursor.execute('delete from user_position where company_id=' + companyid)
# cursor.execute('delete from role where company_id=' + companyid + ' and is_default!=1')
# cursor.execute('delete from user_role where company_id=' + companyid)
# cursor.execute('delete from chance_type where company_id=' + companyid)
# cursor.execute('delete from audit_template where company_id=' + companyid)
# cursor.execute('delete from audit_form where company_id=' + companyid)
cursor.execute('delete a from chance_data a left JOIN chance b on a.chance_id=b.id where b.company_id=' + companyid)
db.commit()
cursor.execute('delete a from audit_flow_process a left JOIN chance b on a.chance_id=b.id where b.company_id=' + companyid)
db.commit()
cursor.execute('delete a from chance_department a left JOIN chance b on a.chance_id=b.id where b.company_id=' + companyid)
db.commit()
cursor.execute('delete from user_msg where receive_user_id in (select id from user_company where company_id=' + companyid + ') ' \
'or sender_user_id in (select id from user_company where company_id=' + companyid +')')
db.commit()
cursor.execute('delete from comment where user_id in (select id from user_company where company_id=' + companyid + ')')
db.commit()
cursor.execute('delete from chance_favorite where user_id in (select id from user_company where company_id=' + companyid + ')')
db.commit()
#这句要放在最后执行
cursor.execute('delete from chance where company_id=' + companyid)
db.commit()
print("delete OK")
except Exception as err:
# 发生错误时回滚
print("this is:", err)
db.rollback()
db.close()
\ No newline at end of file
# # @Time :2019/11/26 11:25
# # @Author :dengyuting
# # @File :db_func.py
# import pymysql
#
# db_config ={"host": "101.37.68.23",
# "port": 3306,
# "user": "dengyuting",
# "password": "dyt_123456",
# "db": "opportunity_test",
# "charset": 'utf8'}
#
# def get_id1_from_database(tbl, company_id, column, column_value):
# db = pymysql.connect(**db_config)
# cursor = db.cursor()
# company_id = str(company_id)
# column_value = str(column_value)
# sql= "select id from " + tbl + " where company_id=" + company_id + " and " + column + "=" + column_value + \
# " and delete_at=0"
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
#
# def get_chanceid_from_database(company_id):
# db = pymysql.connect(**db_config)
# cursor = db.cursor()
# company_id = str(company_id)
# sql= "select id from chance where company_id=" + company_id + " order by create_at desc"
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
#
# def setup_hook_clean_db(companyid):
# """
# 初始化时清理数据库中对应公司的历史数据
# :return:
# """
# db = pymysql.connect(**db_config)
# cursor = db.cursor()
# companyid = str(companyid)
# # get_adminid = 'select admin_id from company where id='+companyid
# # get_user_company_id = 'select id from user_company where company_id='+companyid
# try:
# # cursor.execute(get_adminid)
# # result = cursor.fetchone()
# # admin_id = result[0]
# # admin_id = str(admin_id)
# # cursor.execute(get_user_company_id)
# # result = cursor.fetchone()
# # user_company_id = result[0]
# # user_company_id = str(user_company_id)
# # cursor.execute('delete from department where company_id=' + companyid + ' and is_top!=1')
# # cursor.execute('delete from user_company where company_id=' + companyid + ' and user_id!='+admin_id)
# # cursor.execute('delete from user_department where company_id=' + companyid + ' and user_company_id!='+ user_company_id)
# # cursor.execute('delete from position where company_id=' + companyid)
# # cursor.execute('delete from user_position where company_id=' + companyid)
# # cursor.execute('delete from role where company_id=' + companyid + ' and is_default!=1')
# # cursor.execute('delete from user_role where company_id=' + companyid)
# # cursor.execute('delete from chance_type where company_id=' + companyid)
# # cursor.execute('delete from audit_template where company_id=' + companyid)
# # cursor.execute('delete from audit_form where company_id=' + companyid)
# cursor.execute('delete a from chance_data a left JOIN chance b on a.chance_id=b.id where b.company_id=' + companyid)
# db.commit()
# cursor.execute('delete a from audit_flow_process a left JOIN chance b on a.chance_id=b.id where b.company_id=' + companyid)
# db.commit()
# cursor.execute('delete a from chance_department a left JOIN chance b on a.chance_id=b.id where b.company_id=' + companyid)
# db.commit()
# cursor.execute('delete from user_msg where receive_user_id in (select id from user_company where company_id=' + companyid + ') ' \
# 'or sender_user_id in (select id from user_company where company_id=' + companyid +')')
# db.commit()
# cursor.execute('delete from comment where user_id in (select id from user_company where company_id=' + companyid + ')')
# db.commit()
# cursor.execute('delete from chance_favorite where user_id in (select id from user_company where company_id=' + companyid + ')')
# db.commit()
# #这句要放在最后执行
# cursor.execute('delete from chance where company_id=' + companyid)
# db.commit()
# print("delete OK")
# except Exception as err:
# # 发生错误时回滚
# print("this is:", err)
# db.rollback()
# db.close()
\ No newline at end of file
... ...
... ... @@ -5,242 +5,268 @@
# @file: pg_db_func.py
import psycopg2
# 获得连接
def GetConnect():
conn = False
try:
conn = psycopg2.connect(database="partner_test",
user="bianxinhua",
password="bianxinhua_123",
host="114.55.200.59",
port="31543")
user="bianxinhua",
password="bianxinhua_123",
host="114.55.200.59",
port="31543")
except Exception as err:
print("连接数据库失败，%s" % err)
return conn
#获取超级管理员账号
def get_admin_account_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
# sql= "select id from " + '''"''' + tbl + '''"'''+ ' ' + 'order by id desc '
sql = '''select account from \"{}\" where is_default ='t' '''.format(tbl)
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#获取管理员的id
def get_admin_id_from_database():
db = GetConnect()
cursor = db.cursor()
sql = '''select id from admin_user where is_usable = 't' order by id desc '''
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#获取管理员的account
def get_account_from_database():
db = GetConnect()
cursor = db.cursor()
sql= '''select account from admin_user where is_usable = 't' order by id desc '''
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#实际订单的发货单号
def get_actual_delivery_code_from_database():
db = GetConnect()
cursor = db.cursor()
sql = '''select delivery_code from order_base where order_type = 1 order by id DESC'''
#print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#实际订单的id
def get_actual_id_from_database():
db = GetConnect()
cursor = db.cursor()
sql = '''select id from order_base where order_type = 1 order by id DESC'''
#print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#意向订单id
def get_purpose_id_from_database():
db = GetConnect()
cursor = db.cursor()
# sql= "select id from " + '''"''' + tbl + '''"'''+ ' ' + 'order by id desc '
sql = '''select id from order_base where order_type = 2 order by id DESC'''
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#意向订单的产品ID
def get_ID_from_database(id):
db = GetConnect()
cursor = db.cursor()
sql= '''select id from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchall()
return result[id][0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#分红管理的产品ID
def get_dividends_ID_from_database(id):
db = GetConnect()
cursor = db.cursor()
sql= '''select id from order_good where order_id = {} order by id ASC '''.format(get_actual_id_from_database())
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchall()
return result[id][0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#意向订单的产品名称
def get_productName_from_database(id):
db = GetConnect()
cursor = db.cursor()
sql= '''select good_name from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchall()
return result[id][0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#意向订单的产品数量
def get_orderNum_from_database(id):
# 清除数据
def setup_hook_clean_db(companyid):
"""
初始化时清理数据库中对应公司的历史数据
:return:
"""
db = GetConnect()
cursor = db.cursor()
sql= '''select plan_good_number from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchall()
return result[id][0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#意向订单的产品单价
def get_univalence_from_database(id):
db = GetConnect()
cursor = db.cursor()
sql= '''select price from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchall()
return result[id][0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#意向订单的合伙人分红比例
def get_partnerRatio_from_database(id):
db = GetConnect()
cursor = db.cursor()
sql= '''select partner_bonus_percent from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchall()
return result[id][0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#获取合伙人所属区域
def get_area_from_database():
db = GetConnect()
cursor = db.cursor()
sql = '''select region_info from partner_info where account = '13459147023' '''
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]['regionName']
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#获取合伙人的业务员姓名
def get_salesman_from_database():
db = GetConnect()
cursor = db.cursor()
sql = '''select salesman from partner_info where account = '13459147023' '''
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0][0]['name']
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#获取合伙人的状态
def get_status_from_database():
db = GetConnect()
cursor = db.cursor()
sql = '''select status from partner_info where account = '13459147023' '''
# print(sql)
companyid = str(companyid)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
cursor.execute('delete from order_base where company_id=' + companyid)
db.commit()
cursor.execute('delete from order_good where company_id=' + companyid)
db.commit()
print("delete OK")
except Exception as err:
# 发生错误时回滚
print(err)
print("this is:", err)
db.rollback()
db.close()
# #获取超级管理员账号
# def get_admin_account_from_database(tbl):
# db = GetConnect()
# cursor = db.cursor()
# # sql= "select id from " + '''"''' + tbl + '''"'''+ ' ' + 'order by id desc '
# sql = '''select account from \"{}\" where is_default ='t' '''.format(tbl)
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #获取管理员的id
# def get_admin_id_from_database():
# db = GetConnect()
# cursor = db.cursor()
# sql = '''select id from admin_user where is_usable = 't' order by id desc '''
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #获取管理员的account
# def get_account_from_database():
# db = GetConnect()
# cursor = db.cursor()
# sql= '''select account from admin_user where is_usable = 't' order by id desc '''
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
#
# #实际订单的发货单号
# def get_actual_delivery_code_from_database():
# db = GetConnect()
# cursor = db.cursor()
# sql = '''select delivery_code from order_base where order_type = 1 order by id DESC'''
# #print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
#
# #实际订单的id
# def get_actual_id_from_database():
# db = GetConnect()
# cursor = db.cursor()
# sql = '''select id from order_base where order_type = 1 order by id DESC'''
# #print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #意向订单id
# def get_purpose_id_from_database():
# db = GetConnect()
# cursor = db.cursor()
# # sql= "select id from " + '''"''' + tbl + '''"'''+ ' ' + 'order by id desc '
# sql = '''select id from order_base where order_type = 2 order by id DESC'''
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #意向订单的产品ID
# def get_ID_from_database(id):
# db = GetConnect()
# cursor = db.cursor()
# sql= '''select id from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchall()
# return result[id][0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #分红管理的产品ID
# def get_dividends_ID_from_database(id):
# db = GetConnect()
# cursor = db.cursor()
# sql= '''select id from order_good where order_id = {} order by id ASC '''.format(get_actual_id_from_database())
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchall()
# return result[id][0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #意向订单的产品名称
# def get_productName_from_database(id):
# db = GetConnect()
# cursor = db.cursor()
# sql= '''select good_name from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchall()
# return result[id][0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #意向订单的产品数量
# def get_orderNum_from_database(id):
# db = GetConnect()
# cursor = db.cursor()
# sql= '''select plan_good_number from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchall()
# return result[id][0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #意向订单的产品单价
# def get_univalence_from_database(id):
# db = GetConnect()
# cursor = db.cursor()
# sql= '''select price from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchall()
# return result[id][0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #意向订单的合伙人分红比例
# def get_partnerRatio_from_database(id):
# db = GetConnect()
# cursor = db.cursor()
# sql= '''select partner_bonus_percent from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchall()
# return result[id][0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #获取合伙人所属区域
# def get_area_from_database():
# db = GetConnect()
# cursor = db.cursor()
# sql = '''select region_info from partner_info where account = '13459147023' '''
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]['regionName']
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #获取合伙人的业务员姓名
# def get_salesman_from_database():
# db = GetConnect()
# cursor = db.cursor()
# sql = '''select salesman from partner_info where account = '13459147023' '''
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0][0]['name']
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
# #获取合伙人的状态
# def get_status_from_database():
# db = GetConnect()
# cursor = db.cursor()
# sql = '''select status from partner_info where account = '13459147023' '''
# # print(sql)
# try:
# cursor.execute(sql)
# result = cursor.fetchone()
# return result[0]
# except Exception as err:
# # 发生错误时回滚
# print(err)
# db.rollback()
# db.close()
... ...
config:
name: start_testcases
#清理数据
setup_hooks:
- ${setup_hook_clean_db(214)}
teardown_hooks:
- ${sleep(2)}
teststeps:
- name: start_testcases
api: api/app_api/login/gettimestamp_api.yml
\ No newline at end of file
... ...
... ... @@ -2,6 +2,9 @@ config:
name: 合伙人测试集
testcases:
- name: start....初始化数据库
testcase: testcases/common/start_testcases.yml
############################分红管理模块开始#################################
- name: DIV001-事业分红列表-数据源来自于实发订单
testcase: testcases/dividends/DIV001_dividendslist_testcases.yml
... ...