016、【前程貸—簡(jiǎn)化版，實(shí)戰(zhàn) 09】 用 faker庫(kù) 生成的隨機(jī)手機(jī)號(hào)碼，去注冊(cè)，看是否注冊(cè)過(guò) 。
2021-09-05 09:04:31

# 定義日志相關(guān)的配置
# name表示自定義日志搜集器的名字；
# level表示日志級(jí)別；
# file_name 表示生成的日志名字
# show_stream_handler表示是否在控制臺(tái)輸出日志。False表示不在控制臺(tái)顯示。
# when，表示按H(小時(shí))、D(天)、M(分鐘)、S(秒)生成日志 ；注意字母是大寫(xiě) ；
[log]
name=future_logging_collector
level=INFO
file_name=future_loan_test.log
show_stream_handler=False
when=H

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com
# Time:    2021/9/4 13:20
# Project: Future_Loan_day15
# Module:  mysql_ini.py


# 此配置文件用來(lái)配置鏈接mysql數(shù)據(jù)庫(kù)
# key已經(jīng)和代碼關(guān)聯(lián)了，不能更改key名；
MYSQL_INI = {
    "host": "api.lemonban.com",
    "port": 3306,
    "user": "future",
    "password": "123456",
    "database": "futureloan",
    "charset": "utf8"
}

import os
import json

import jsonpath
import pytest

from tools.handle_phone import HandlePhone
from tools.handler_request import HandlerRequests
from tools.handler_assert_list import HandlerAssertList
from tools.handler_excel import HandlerExcel
from tools.handler_logging import handler_logger

# 一、準(zhǔn)備 數(shù)據(jù) cases
# A、用os.path找到 xlsx 文件
file_dir = os.path.dirname(os.path.realpath('__file__'))
base_dir = os.path.dirname(file_dir)
print(f'根目錄：{base_dir}')

file_name = os.path.join(base_dir, 'test_datas', 'test_register_cases.xlsx')
print(f'excel表格的路徑：{file_name}')


# B、調(diào)用common——>excel_handler.py 中已封裝好的ExcelHandler類來(lái)操作excel表格
# # 1、打開(kāi)xlsx表格；2、根據(jù)表單名字獲取表格數(shù)據(jù)；3、讀取數(shù)據(jù)；
excel_handler = HandlerExcel(file_name)
excel_handler.select_sheet_by_name('register_case')
all_cases = excel_handler.read_all_rows_data()

# print(all_cases)
# handler_logger.info(f'=====all_cases=======：{all_cases}')


mrq = HandlerRequests()
add_assert = HandlerAssertList()
hp = HandlePhone()


class TestRegister:
    # 三、參數(shù)化測(cè)試用例
    # pytest.mark.parametrize("item", cases) 中的 "item" 必須 和 def test_my_requests(item): 中的 item 名字一樣
    @pytest.mark.parametrize("item", all_cases)
    def test_my_register(self, item):
        # handler_logger.info('=====注冊(cè)接口測(cè)試=======')
        handler_logger.info(f'"#phone#"未替換的前的數(shù)據(jù)：{item["req_data"]}=======')

        # 1、替換占位符
        # item 的值為：一行的數(shù)據(jù)，如下：
        """
            {
                'case_id': 'test_register_002',
                'case_name': '注冊(cè)成功，普通用戶',
                'method': 'POST',
                'url': 'http://api.lemonban.com/futureloan/member/register',
                'req_data': '{"mobile_phone": "#phone#","pwd": "12345678","reg_name":"Sky","type":1}',
                'assert_response_value_list': '[{"expr":"$.code","expected":0,"type":"eq"},{"expr":"$.msg","expected":"OK","type":"eq"}]',
                'actual_results': None
            }
        """

        # 1、用隨機(jī)生成的，未被注冊(cè)過(guò)的手機(jī)號(hào)碼，把 標(biāo)記 "#phone#"的字符串，替換；
　　　　　# a、查詢數(shù)據(jù)庫(kù)，并得到一個(gè)未被注冊(cè)的手機(jī)號(hào)碼
        new_phone = hp.get_phone()

        # handler_logger.info(f'=====new_phone：{new_phone}=======')
        # handler_logger.info(f'=====對(duì)比結(jié)果：{item["req_data"].find("#phone#") != -1}=======')
　　　　　　

　　　　 # b、判斷字符串中是否含有 #phone#,如果有把它替換成 剛隨機(jī)生成的手機(jī)號(hào)碼
        if item["req_data"] and item["req_data"].find("#phone#") != -1:
            item['req_data'] = item['req_data'].replace("#phone#", new_phone)

        if item["assert_response_value_list"] and item["assert_response_value_list"].find("#phone#") != -1:
            item["assert_response_value_list"] = item["assert_response_value_list"].replace("#phone#", new_phone)

        # 2、把json格式的字符串轉(zhuǎn)換成 字典
        temp_req_data_dict = json.loads(item['req_data'])
        handler_logger.info(f'用隨機(jī)手機(jī)號(hào)碼替換"#phone#"得到的數(shù)據(jù)：{temp_req_data_dict}=======')

        # 3、發(fā)起請(qǐng)求
        resp = mrq.send_reuqests(item['method'], item['url'], temp_req_data_dict)
        print(resp.json())
        # 4、斷言
        add_assert.assert_response_value(item["assert_response_value_list"], resp.json())

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com

from faker import Faker
from tools.handler_mysql import handler_mysql
from tools.handler_logging import handler_logger


class HandlePhone:
    def __init__(self):
        self.fk = Faker(locale='zh-cn')

    # 如果系統(tǒng)對(duì)手機(jī)號(hào)前綴有要求的，要做前綴校驗(yàn)
    def __faker_phone(self):
        phone = self.fk.phone_number()
        handler_logger.info(f'隨機(jī)生成的手機(jī)號(hào)為：{phone}')
        return phone

    # 獲取數(shù)據(jù)庫(kù)查詢結(jié)果
    def __select_phone(self, phone):
        sql = "select * from member where mobile_phone = '{}'".format(phone)
        select_phone_result = handler_mysql.get_db_all_data(sql=sql)
        return select_phone_result

    # 獲取未注冊(cè)的手機(jī)號(hào)
    # 1、生成一個(gè)隨機(jī)手機(jī)號(hào)碼
    # 2、用生成的手機(jī)號(hào)碼去數(shù)據(jù)庫(kù)查詢，如果查詢結(jié)果大于1行說(shuō)明已經(jīng)存在，被注冊(cè)過(guò)；
    def get_phone(self):
        while True:
            phone = self.__faker_phone()  # 隨機(jī)生成手機(jī)號(hào)
            select_phone_result = self.__select_phone(phone)  # 拿到數(shù)據(jù)庫(kù)執(zhí)行結(jié)果
            if len(select_phone_result) > 0:
                continue
            else:
                return phone


if __name__ == '__main__':
    cl = HandlePhone()
    result = cl.get_phone()
    print(result)

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com
# Time:    2021/9/4 16:41
# Project: Future_Loan_day15
# Module:  handler_assert_list.py

import jsonpath
import ast


# 封裝 assert_list列 添加斷言
class HandlerAssertList:

    def assert_response_value(self, assert_response_value_list, response_dict):
        """
        根據(jù)響應(yīng)的值斷言
        :param assert_response_value_list: 從excel表格中獲取到的斷言字符串,比如：'[{"expr":"$.code","expected":0,"type":"eq"},{"expr":"$.msg","expected":"賬號(hào)已存在","type":"eq"}]'
        :param response_dict: 發(fā)起請(qǐng)求后得到的響應(yīng)結(jié)果
        :return:
        """
        # 把字符串轉(zhuǎn)換成python列表
        # 比eval安全一點(diǎn)，轉(zhuǎn)換成列表。eval去掉最外層引號(hào)后還會(huì)自動(dòng)計(jì)算，literal_eval僅去掉最外層引號(hào)；
        check_list = ast.literal_eval(assert_response_value_list)
        # print(check_list)

        # 每一個(gè)單元格所有斷言的比對(duì)結(jié)果存放在check_res
        check_res = []
        for check in check_list:
            # 通過(guò)jsonpath表達(dá)式，從響應(yīng)結(jié)果中拿到實(shí)際結(jié)果
            actual = jsonpath.jsonpath(response_dict, check["expr"])
            if isinstance(actual, list):
                actual = actual[0]
            # 與實(shí)際結(jié)果比對(duì)
            if check["type"] == "eq":
                # print(f'
實(shí)際比對(duì)結(jié)果：{actual == check["expected"]}')
                check_res.append(actual == check["expected"])

        # print(f'所有斷言結(jié)果：{check_res}')
        # 如果 斷言列表中有False，拋出 斷言異常
        if False in check_res:
            raise AssertionError


if __name__ == '__main__':
    # 測(cè)試代碼
    assert_response_value_list = '[{"expr":"$.code","expected":0,"type":"eq"},' 
                                 '{"expr":"$.msg","expected":"OK","type":"eq"}]'
    response_dict = {
        "code": 0,
        "msg": "OK",
        "data": {
            "id": 123660458,
            "reg_name": "Sky",
            "mobile_phone": "13321886699"
        },
        "copyright": "Copyright 檸檬班 ? 2017-2020 湖南省零檬信息技術(shù)有限公司 All Rights Reserved"
    }

    add_assert = HandlerAssertList()
    check_res = add_assert.assert_response_value(assert_response_value_list, response_dict)

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com
# Time:    2021/9/4 17:58
# Project: Future_Loan_day15
# Module:  handler_conf.py

from configparser import ConfigParser


class HandlerConf(ConfigParser):

    def __init__(self, file_name):
        super().__init__()
        self.read(file_name, encoding='utf-8')

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com
# Time:    2021/9/4 16:34
# Project: Future_Loan_day15
# Module:  handler_excel.py

import openpyxl

from tools.handler_logging import handler_logger


# 封裝一個(gè)xlsx表格操作類
class HandlerExcel:
    # 操作一個(gè)excel表格：
    # 第一步：打開(kāi)工作簿
    # 第二步：選取表單
    # 第三步：讀取數(shù)據(jù)
    # 第四步：關(guān)閉打開(kāi)的工作簿

    def __init__(self, xlsx_file_path: str):
        """
        傳入一個(gè)xlsx文件路徑，用load_workbook()方法加載，如果文件加載不成功，拋出異常。如果成功，打開(kāi)一個(gè)工作簿。
        :param xlsx_file_path: xlsx文件路徑
        """
        try:
            self.wb = openpyxl.load_workbook(xlsx_file_path)
        except FileNotFoundError as ffe:
            # print('打開(kāi)文件失敗')
            handler_logger.error(ffe)
            raise
        # 不確定打開(kāi)的是哪個(gè)表單
        self.sh = None

    def close_workbook(self):
        """
        關(guān)閉當(dāng)前打開(kāi)的工作簿
        :return:
        """
        self.wb.close()

    def select_sheet_by_name(self, sheet_name: str):
        """
        根據(jù)傳入的工作表的名字，打開(kāi)工作表。
        :param sheet_name: 作表的名字
        """
        self.sh = self.wb[f'{sheet_name}']

    def read_all_rows_data(self):
        """
        從選定的表單當(dāng)中，第一行作為key.
        將后面的每一行數(shù)據(jù)，與第一行拼接成一個(gè)字典數(shù)據(jù)，作為一條測(cè)試用例數(shù)據(jù)。
        將所有測(cè)試用例數(shù)據(jù)，添加到一個(gè)列表當(dāng)中。
       :return: 測(cè)試用例數(shù)據(jù)列表
        """
        # 獲取表單的所有行，即獲取表單的所有數(shù)據(jù)
        sheet_all_rows = list(self.sh.values)
        # 把第一行作為數(shù)據(jù)的keys
        keys = sheet_all_rows[0]
        # print(keys)

        # 定義 cases_list 存放測(cè)試用例
        cases_list = []

        # 以下代碼功能：excel表單第2行開(kāi)始的每一行測(cè)試數(shù)據(jù)，與第一行的keys拼接成一個(gè)字典。
        for single_row in sheet_all_rows[1:]:
            case_dict = dict(zip(keys, single_row))
            cases_list.append(case_dict)
        return cases_list


if __name__ == '__main__':
    eh = HandlerExcel(r'D:SkyWorkSpaceWorkSpaceAPI_testlmFuture_Loan'
                      r'Future_Loan_day16	est_datas	est_register_cases.xlsx')
    eh.select_sheet_by_name('register_case')
    print(eh.read_all_rows_data())
    eh.close_workbook()

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com


import logging
# 自定義一個(gè)日志模塊
import os
# 導(dǎo)入 ConfigParser 類 ，用來(lái)操作 config.ini 文件 ；
import time
from configparser import ConfigParser
from logging import Logger
from logging import handlers


class HandlerLogger(Logger):

    def __init__(self):

        # 一、用 ConfigParser類 來(lái)操作 config.ini 配置文件 ；
        # 實(shí)例化一個(gè) ConfigParser ；
        conf = ConfigParser()

        # 1、獲取config.ini文件
        common_dir = os.path.dirname(os.path.realpath(__file__))
        base_dir = os.path.dirname(common_dir)
        config_ini_file = os.path.join(base_dir, 'conf', 'config.ini')

        # 2、從配置文件獲取值
        conf.read(config_ini_file, encoding='utf-8')
        logger_name = conf.get('log', 'name')
        level = conf.get('log', 'level')
        file_name = conf.get('log', 'file_name')
        show_stream_handler = conf.get('log', 'show_stream_handler')
        when = conf.get('log', 'when')

        # 二、設(shè)置自定義日志搜集器名字、設(shè)置日志級(jí)別；
        super().__init__(logger_name, level)

        # 三、定義日志輸出格式, 使用Formatter類實(shí)例化一個(gè)日志格式類；
        fmt = '%(asctime)s, %(levelname)s, %(message)s, %(name)s, %(pathname)s,line=%(lineno)d'
        # fmt = '%(asctime)s, %(levelname)s, %(message)s, %(name)s,line=%(lineno)d'
        formatter = logging.Formatter(fmt)

        # 四A、日志默認(rèn)輸出到控制臺(tái)，如果設(shè)置為False，日志將不輸出到控制臺(tái)；
        if show_stream_handler == 'True':
            stream_handler = logging.StreamHandler()
            # 設(shè)置渠道當(dāng)中的日志格式
            stream_handler.setFormatter(formatter)
            # 將渠道與實(shí)例日志搜集器綁定
            self.addHandler(stream_handler)

        # 四B、把日志輸出到文件file
        # 首先拼接存放log的file文件
        logs_file = os.path.join(base_dir, 'logs', file_name)
        print(logs_file)
        if logs_file:
            file_handle = handlers.TimedRotatingFileHandler(filename=logs_file,
                                                            when=when,
                                                            encoding='utf-8',
                                                            interval=1,
                                                            backupCount=5)

            # 設(shè)置渠道當(dāng)中的日志格式
            file_handle.setFormatter(formatter)
            # 將渠道與實(shí)例日志搜集器綁定
            self.addHandler(file_handle)


# 生成一個(gè) handler_logger 實(shí)例，在其他所有模塊中導(dǎo)入該模塊時(shí)，共用這一個(gè)日志搜集實(shí)例。handler_logger 類似于 全局變量
# 日志搜集是典型的單列設(shè)計(jì)模式 (單實(shí)例模式) 。
handler_logger = HandlerLogger()

if __name__ == '__main__':
    handler_logger = HandlerLogger()
    for i in range(10):
        time.sleep(1)
        handler_logger.debug('=====debug=====')
        handler_logger.info('=====info=====')
        handler_logger.warning('=====warning=====')
        handler_logger.error('=====error=====')

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com
# Time:    2021/9/4 13:18
# Project: Future_Loan_day15
# Module:  handler_mysql.py

import pymysql

from conf.mysql_ini import MYSQL_INI


# 封裝操作Mysql數(shù)據(jù)庫(kù)類
class HandleMysql:
    def __init__(self):
        """
        1、初始化建立連接，創(chuàng)建數(shù)據(jù)庫(kù)連接
        charset='utf8', 注意，不是utf-8 哦
        返回?cái)?shù)據(jù)格式控制: cursorclass=pymysql.cursors.DictCursor  加上這個(gè)表示返回字典格式的數(shù)據(jù)；不加的話，以元組的形式返回；
        """
        self.connection = pymysql.connect(
            host=MYSQL_INI['host'],
            port=MYSQL_INI['port'],
            user=MYSQL_INI['user'],
            password=MYSQL_INI['password'],
            database=MYSQL_INI['database'],
            charset=MYSQL_INI['charset'],
            cursorclass=pymysql.cursors.DictCursor)
        # 2、創(chuàng)建游標(biāo)
        self.cur = self.connection.cursor()

    # 獲取 查詢得到的行數(shù)(數(shù)量)
    def get_count(self, sql):
        count = self.cur.execute(sql)
        return count

    # 獲取一條數(shù)據(jù)，一般都是最前面的那條數(shù)據(jù)
    def get_db_one_data(self, sql):
        self.cur.execute(sql)
        return self.cur.fetchone()

    # 獲取全部數(shù)據(jù)
    def get_db_all_data(self, sql):
        self.cur.execute(sql)
        return self.cur.fetchall()

    # 關(guān)閉數(shù)據(jù)庫(kù)連接
    def close(self):
        self.cur.close()
        self.connection.close()


# 使用單例模式，后續(xù)導(dǎo)入數(shù)據(jù)庫(kù)就只導(dǎo)入 handler_mysql
handler_mysql = HandleMysql()


if __name__ == '__main__':

    # 1、建立鏈接
    handler_mysql = HandleMysql()

    # 2、執(zhí)行 sql 語(yǔ)句
    # 先在 sql客戶端 (Navicat Premium 12免安裝) 執(zhí)行以下sql語(yǔ)句，查詢結(jié)果
    phone = 18837906872
    sql_str_2 = f"select * from member where mobile_phone='{phone}'"    # 方式二
    # 執(zhí)行sql語(yǔ)句
    results = handler_mysql.get_db_one_data(sql_str_2)
    print(results)

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com
import requests
from tools.handler_logging import handler_logger


class HandlerRequests:

    def __init__(self):
        """
        初始化方法,初始化請(qǐng)求頭；
        """
        self.headers = {"X-Lemonban-Media-Type": "lemonban.v2"}
        handler_logger.info(f'請(qǐng)求頭為：{self.headers}')

    # 方法 post/put...  json=xxx, get方法用 params=xxx
    def send_reuqests(self, method, url, req_data, token=None):
        """
        調(diào)用requests庫(kù)里面的方法去發(fā)起請(qǐng)求，并得到響應(yīng)結(jié)果；
        :param url: 接口url
        :param method: 請(qǐng)求方法，get,psot
        :param req_data: 請(qǐng)求數(shù)據(jù)
        :param token: 如果有token，添加token
        """
        handler_logger.info(f'請(qǐng)求方法為：{method}')
        handler_logger.info(f'請(qǐng)求url為：{url}')
        handler_logger.info(f'請(qǐng)求數(shù)據(jù)為：{req_data}')

        # 如果有token，添加token
        self.__del_header(token)

        # 注意 request() 方法是不帶s的，requests庫(kù)是帶s的；
        if method.upper() == "GET":
            resp = requests.request(method, url, params=req_data, headers=self.headers)
            return resp
        if method.upper() == "POST":
            # 為了便于學(xué)習(xí)，簡(jiǎn)單的認(rèn)為就是用 json 格式傳；
            resp = requests.request(method, url, json=req_data, headers=self.headers)
            return resp

    def __del_header(self, token=None):
        """
        如果有token，添加token處理
        :param token: 如果有token，添加token處理
        """
        if token:
            self.headers["Authorization"] = f"Bearer {token}"


if __name__ == '__main__':
    handler_requests = HandlerRequests()

# -*- coding:utf-8 -*-
# Author:  Sky
# Email:   2780619724@qq.com
# Time:    2021/9/4 10:02
# Project: Future_Loan_day15
# Module:  操作mysql.py


import pymysql

# 1、建立連接，創(chuàng)建數(shù)據(jù)庫(kù)連接
# charset='utf8', 注意，不是utf-8 哦
# 返回?cái)?shù)據(jù)格式控制: cursorclass=pymysql.cursors.DictCursor  加上這個(gè)表示返回字典格式的數(shù)據(jù)；不加的話，以元組的形式返回；
connection = pymysql.connect(host='api.lemonban.com',
                             port=3306,
                             user='future',
                             password='123456',
                             database='futureloan',
                             charset='utf8',
                             cursorclass=pymysql.cursors.DictCursor)

# 2、創(chuàng)建游標(biāo)
cur = connection.cursor()

# 3、執(zhí)行sql語(yǔ)句 , 返回?cái)?shù)據(jù)
sql_str = "select * from member where reg_name='娜娜'"   # 方式一

# phone = 18837906872
# sql_str_2 = f"select * from member where mobile_phone='{phone}'"    # 方式二

affected_rows = cur.execute(sql_str)    # 返回影響的行數(shù)
print(f'返回影響的行數(shù)：{affected_rows}')

# 4、獲取查詢的結(jié)果
first_data = cur.fetchone()  # 獲取一條數(shù)據(jù)，第一條數(shù)據(jù)；字典形式返回 ；
all_data = cur.fetchall()  # 獲取全部數(shù)據(jù)；列表形式返回 ；
two_data = cur.fetchmany(size=2)   # 獲取前2行數(shù)據(jù)；列表形式返回 ；
print(f'獲取第一條數(shù)據(jù)：{first_data}')
print(f'獲取全部數(shù)據(jù)：{all_data}')
print(f'獲取前2行數(shù)據(jù)：{two_data}')

# 5、關(guān)閉數(shù)據(jù)庫(kù)連接
cur.close()     # 首先關(guān)閉游標(biāo)
connection.close()      # 關(guān)閉數(shù)據(jù)庫(kù)