Python operation database, interface association and related management optimization in practical interface automation testing

1. Foreword

This article will mainly explain how Python operates the database in interface automation testing, why it is necessary to operate the database, what are the pros and cons, as well as database assertions and related interface-related tests.

2. Automated database theory and operation

2.1 Why interface automation needs to operate the database

The operation of the database in interface automation is mainly determined based on the business level. In some cases, such as querying mobile phone numbers or personal information, the database needs to be operated. Sometimes it may be necessary to delete certain content, but generally speaking, this is not done.

2.2 Advantages and Disadvantages of Interface Automation to Operate Database

"""
profit:
1. Able to solve certain business problems according to the business conditions we need

Disadvantages:
1. The operation of the database itself will make automation slower, requiring the establishment of a connection -- query, etc.
2. The database itself is a kind of dependence. We can do as little as possible without operating the database.
"""

2.3 Python operation database

To operate the database, you need to install the pymysql library first and fill in the corresponding data according to the corresponding syntax:

import pymysql

"""
1. Connect to the database: enter user name, password, address, port
2. Cursor: Use cursor to read data and modify data (execute sql statement)
3. Get the operation results
4. Close the cursor
5. Close the database connection
"""

conn = pymysql.connect(user="future",
                password="XXXXXX",
                host="XX.XX.XX.XX",
                port=3306
)

cursor = conn.cursor()

#Execute sql
sql = "SELECT mobile_phone,id,reg_name FROM XXXXXXX.member WHERE mobile_phone = 137XXXXXXXX"
cursor.execute(sql)

# Get a result and each record is represented by a tuple
res = cursor.fetchone()
print(res)

# closure
cursor.close()
conn.close()

The output data is a tuple, and the tuple data cannot be modified. We need to perform type conversion. If we want to output a dictionary, we need to add DictCursor:

import pymysql
from pymysql.cursors import DictCursor

"""
1. Connect to the database: enter user name, password, address, port
2. Cursor: Use cursor to read data and modify data (execute sql statement)
3. Get the operation results
4. Close the cursor
5. Close the database connection
"""

conn = pymysql.connect(user="future",
                password="XXXXXX",
                host="XX.XX.XX.XX",
                port=3306
)

cursor = conn.cursor(DictCursor)

#Execute sql
sql = "SELECT mobile_phone,id,reg_name FROM XXXXXXX.member WHERE mobile_phone = 137XXXXXXXX"
cursor.execute(sql)

# Get a result and each record is represented by a tuple
res = cursor.fetchone()
print(res)

# closure
cursor.close()
conn.close()

2.4 Operation database encapsulation

We need to encapsulate functions. The database content is also public content and can also be placed under the common package. After function encapsulation:

class DBHandler:

    def __init__(self, host=setting.db_host, port=setting.db_port,
                 user=setting.db_user, password=setting.db_pwd):
        self.conn = pymysql.connect(user=user,
                                    password=password,
                                    host=host,
                                    port=port,
                                    autocommit=True
                                    )

    def query_one(self, sql, cursor_type=DictCursor):
        cursor = self.conn.cursor(cursor_type)
        cursor.execute(sql)
        data = cursor.fetchone()
        cursor.close()
        return data

    def query_all(self, sql, cursor_type=DictCursor):
        cursor = self.conn.cursor(cursor_type)
        cursor.execute(sql)
        data = cursor.fetchall()
        cursor.close()
        return data

    def query(self, sql, one=True, cursor_type=DictCursor):
        if one:
            return self.query_one(sql, cursor_type)
        return self.query_all(sql, cursor_type)

    def close(self):
        self.conn.close()

    # Implement your own context manager
    # def __enter__(self):
    # return self
    #
    # def __exit__(self, exc_type, exc_val, exc_tb):
    # return self.close()


db_module = DBHandler()

if __name__ == '__main__':
    db = DBHandler()
    sql = 'select mobile_phone from futureloan.member limit 5'
    res = db.query(sql, one=False)
    db.close()
    print(res)
    #

    # with DBHandler_2() as db:
    # sql = 'select mobile_phone from futureloan.member limit 5'
    # res = db.query(sql, one=False)
    # print(res)
# Database configuration items
db_host = "XX.XX.XX.XX"
db_port=XXXX
db_user = "future"
db_pwd = "XXXXXX"

3. Verify database, interface association and project optimization

After the automated test cases are executed, the database will be verified only if the test cases are successful:

import unittest
import requests
import json

from common.db import DBHandler, db_module
from common.logger import log
from common.excel import read_excel
from common import helper
from config import path_config
from unittestreport import ddt, list_data
from config import setting

# retrieve data
data = read_excel(path_config.case_path , 'XXXXXX')

@ddt
class TestRegister(unittest.TestCase):

    @classmethod
    def setUpClass(cls) -> None:
        """Test class prefix"""
        cls.db = DBHandler()

    @classmethod
    def tearDownClass(cls) -> None:
        """Test class prefix"""
        cls.db.close()

    @list_data(data)
    def test_register(self, case_data):
        print(case_data)
        json_data = case_data['json']
        if '#new_phone#' in json_data:
            new_phone = helper.generate_new_phone()
            json_data = json_data.replace('#new_phone#', new_phone)

        #Convert json format string into dictionary
        json_data = json.loads(json_data)
        headers = json.loads(case_data['headers'])

        print("After replacement", json_data)

        resp = requests.request(
            method=case_data['method'],
            url= setting.host + case_data['url'],
            json=json_data,
            headers=headers
        )

        actual = resp.json()
        print(actual)
        try:
            self.assertEqual(case_data['expected'], actual['code'])
        except AssertionError as e:
            raise e

"""
1. Determine whether it is a successfully registered test case
2. Query whether the database contains records of mobile phone numbers
3. Determine whether the number of database records is 1
"""
        if actual['msg'] == "OK":
            #Create a database connection through setUp
            sql = f"select id from XXXX.XXXX where XXXXXX = {json_data['XXXXXX']}"
            result = self.db.query_all(sql)
            self.assertEqual(1, len(result))

Often the interfaces in the project will have interface associations, and we need to process the interfaces separately:

import unittest
import requests
import json
from jsonpath import jsonpath
from decimal import Decimal

from common.db import DBHandler
from common.logger import log
from common.excel import read_excel
from common import helper
from config import path_config
from unittestreport import ddt, list_data
from config import setting

# retrieve data
from middle import api

data = read_excel(path_config.case_path , 'XXXXXX')

@ddt
class TestRecharge(unittest.TestCase):

    @classmethod
    def setUpClass(cls) -> None:
        """
        1. Access the login interface and get the return value
        2. Extract data from the return value: resp["data"]["id"]
        3. Extract data from the return value: resp["data"]["token_info"]["token"]
        4. Set it as a class attribute to facilitate calling in the test case function
        :return:
        """
        resp_data = api.login()
        # Extraction method 1:
        cls.member_id = resp_data["data"]["id"]
        cls.token = resp_data["data"]["token_info"]["token"]

        # Extraction method 2: jsonpath
        cls.member_id = jsonpath(resp_data, '$..id')[0]
        cls.token = jsonpath(resp_data, '$..token')[0]
        cls.before = jsonpath(resp_data, '$..leave_amount')[0]

        #Initialize database
        cls.db = DBHandler()

    @classmethod
    def tearDownClass(cls) -> None:
        pass

    @list_data(data)
    def test_recharge(self, case_data):
        """
        1. Get the data in case_data, headers and json should be focused on.
        2. Data preprocessing: data replacement, data conversion into dictionary format: headers, json, expected
        3. Send request
        4. Assertion
        :param case_data:
        :return:
        """
        # Get the data in case_data, headers and json should be focused on.
        headers_string = case_data['headers']
        json_string = case_data['json']

        # Preprocessing of data: data replacement, data conversion into dictionary format: headers, json, expected
        if "#token#" in headers_string:
            headers_string = headers_string.replace("#token#", self.token)
        if "#member_id#" in json_string:
            json_string = json_string.replace("#member_id#", str(self.member_id))

        headers = json.loads(headers_string)
        json_data = json.loads(json_string)
        expected = json.loads(case_data['expected'])


        # Get the user's balance in the database before recharging
        sql = f"select XXXXXXX from XXXX.XXXXX where id={self.XXXXX_id}"
        before = self.db.query_one(sql)
        # {"leave_amount": 200}


        # 3, send request
        resp = requests.request(
            method=case_data['method'],
            url=setting.host + case_data['url'],
            json=json_data,
            headers=headers
        )

        # 4, assertion
        json_response = resp.json()

        try:
            for key, value in expected.items():
                self.assertEqual(value, json_response[key])
        except AssertionError as e:
            log.error(f"Test case failed: {e}")
            raise e


        if json_response['msg'] == 'OK':
            print(json_response)
            # Assert via database
            sql = f"select leave_amount from futureloan.member where id={self.member_id}"
            after = self.db.query_one(sql)
            money = str(json_data['amount'])
            after = str(after['leave_amount'])
            before = str(before['leave_amount'])
            self.assertEqual(Decimal(money), Decimal(after) - Decimal(before))

4. Summary

Regarding how Python operates the database in interface automation testing, why it is necessary to operate the database, what are the pros and cons, as well as database assertions and related interface-related tests, the author has said this. If you like it, you can like it, comment, and follow it. , follow me to give you different surprises every day.

Finally: The following is the supporting learning materials. For friends who do [software testing], it should be the most comprehensive and complete preparation warehouse. This warehouse has also accompanied me through the most difficult journey. I hope it can also help you. !

Software testing interview applet

A software test question bank that has been used by millions of people! ! ! Who is who knows! ! ! The most comprehensive interview test mini program on the Internet, you can use your mobile phone to answer questions, take the subway, bus, and roll it up!

Covering these interview question sections:

1. Basic theory of software testing, 2. web, app, interface function testing, 3. network, 4. database, 5. linux

6. Web, app, interface automation, 7. Performance testing, 8. Programming basics, 9. HR interview questions, 10. Open test questions, 11. Security testing, 12. Computer basics

How to obtain the full set of information: Click on the small card below to get it yourself