数据库验证流程十一脚本编写

air_case/b6_spd3_core_business_flow_database_check/a_b4流程十一请领业务流_二级库向直送_数据库验证.air/a_b4流程十一请领业务流_二级库向直送_数据库验证.py

@@ -1109,6 +1109,7 @@ try:
         for i in list_data:
             check_mcms_purchase(int(i))
         order_dp(1)  # 1低值
+
         order_dp(2)  # 2高值
         order_dp(3)  # 3试剂
         check_accept_order(1)  # 低值

common/db/sql/sql_tools_out.py

+import pymysql
+
+from common.fileUtls import FileUtils
+
+
+def get_db():
+    # 打开数据库连接
+    db = pymysql.connect(host="10.17.65.108", user="root", password="Cmic.2023", database="spd_hdiorder_test2",
+                                     charset="utf8")
+    return db
+
+
+def get_sql_conn():
+    """
+    获取数据库连接
+    """
+    db = get_db()
+    cursor = db.cursor()
+    return db, cursor
+
+
+def get_index_dict(cursor):
+    """
+    获取数据库对应表中的字段名
+    """
+    index_dict = dict()
+    index = 0
+    for desc in cursor.description:
+        index_dict[desc[0]] = index
+        index = index + 1
+    return index_dict
+
+
+def get_dict_data_sql(cursor, sql):
+    """
+    运行sql语句，获取结果，并根据表中字段名，转化成dict格式（默认是tuple格式）
+    """
+    cursor.execute(sql)
+    data = cursor.fetchall()
+    index_dict = get_index_dict(cursor)
+    res = []
+    for datai in data:
+        resi = dict()
+        for indexi in index_dict:
+            resi[indexi] = datai[index_dict[indexi]]
+        res.append(resi)
+    return res
+
+
+
+def check_mcms_dept_buy_car_sql(sql):
+    db, cursor = get_sql_conn()
+    res_dict = get_dict_data_sql(cursor, sql)
+    print('res_dict', res_dict)
+    cursor.close()
+    return res_dict
+def check_mcms_dept_buy_sql(sql):
+    db, cursor = get_sql_conn()
+    res_dict = get_dict_data_sql(cursor, sql)
+    print('res_dict', res_dict)
+    cursor.close()
+    return res_dict
+def check_mcms_pur_plan_sql(sql):
+    db, cursor = get_sql_conn()
+    res_dict = get_dict_data_sql(cursor, sql)
+    print('res_dict', res_dict)
+    cursor.close()
+    return res_dict
+def get_create_time(sql):
+    db, cursor = get_sql_conn()
+    cursor.execute(sql)
+    results = cursor.fetchone()
+    str1 = str(results)
+    new_str1 = str1.replace(',)', '')
+    new_str1 = new_str1.replace('(date', 'date')
+    # new_str1 = new_str1.replace(',', '')
+    # print(new_str1)
+    cursor.close()
+    return new_str1
+def get_expdt_date(sql):
+    db, cursor = get_sql_conn()
+    cursor.execute(sql)
+    results = cursor.fetchone()
+    str1 = str(results)
+    new_str1 = str1.replace(',)', '')
+    new_str1 = new_str1.replace('(date', 'date')
+    # new_str1 = new_str1.replace(',', '')
+    # print(new_str1)
+    cursor.close()
+    return new_str1
+def get_id(sql):
+    db, cursor = get_sql_conn()
+    cursor.execute(sql)
+    results = cursor.fetchone()
+    str1 = str(results)
+    new_str1 = str1.replace(',)', '')
+    new_str1 = new_str1.replace('(', '')
+    new_str1 = new_str1.replace(',', '')
+    new_str1 = new_str1.replace("'", '')
+    # print(new_str1)
+    cursor.close()
+    return new_str1
+def get_dept_buy_id(sql):
+    db, cursor = get_sql_conn()
+    cursor.execute(sql)
+    results = cursor.fetchall()
+    cursor.close()
+    return results
+def process_tuple(tuple1):
+    str1 = str(tuple1)
+    new_str1 = str1.replace(',)', '')
+    new_str1 = new_str1.replace('(', '')
+    new_str1 = new_str1.replace(',', '')
+    new_str1 = new_str1.replace("'", '')
+    return new_str1
+
+def get_mdm_goods_code(sql):
+    db, cursor = get_sql_conn()
+    cursor.execute(sql)
+    results = cursor.fetchall()
+    cursor.close()
+    return results
+# hos_goods_id = 'h034700004220'
+# main2(hos_goods_id)
\ No newline at end of file