Python脚本连接Oracle数据库并验证成功

# yaml文件存储数据->root\Data\oracle_admin_f_shozaiko.yaml

# Test Data for Oracle DB: ADMIN->F_SHOZAIKO
-
  name: connect to Oracle
  request:
    uname: admin
    upwd: P823!ApoL
    host: rf-oms.cbfvvrud0bld.ap-northeast-1.rds.amazonaws.com:1521/rfoms
    query: SELECT * FROM F_SHOZAIKO WHERE COLORCD='01' AND RYOZAIKOSU='10'
  validate:
    HINBAN: 02494

------------------------------------------------------------------------------------------------------------------------------------------------------------------

# yaml_utils.py方法读取yaml文件的数据->root\utilities\yaml_utils.py

import yaml
import cx_Oracle as oracle

class YamlUtils:

    def read_yaml(self, yaml_file):
        """read yaml file data and return content"""
        try:
            with open(yaml_file, "r", encoding='utf-8') as f:
                data = yaml.load(f, Loader=yaml.FullLoader)
                # print(data, type(data)) # debug
                return data
        except:
            return None

if __name__ == '__main__':
    yaml_utils = YamlUtils()
    yaml_utils.read_yaml(yaml_file)

------------------------------------------------------------------------------------------------------------------------------------------------------------------

# test_xx.py->root\TestCases\test_oracle_admin_f_shozaiko.py

import cx_Oracle as oracle
import pytest
from utilities.yaml_utils import YamlUtils

class TestOracleAdminF_SHOZAIKO():

    @pytest.mark.parametrize('args', YamlUtils().read_yaml('../Data/oracle_admin_f_shozaiko.yaml'))
    def test_admin_f_shozaiko(self, args):
        """This function is aim to connect Oracle DB and execute SQL query then verify the data"""

        """1.parameters group"""
        uname = args['request']['uname']
        upwd = args['request']['upwd']
        host = args['request']['host']
        query = args['request']['query']
        HINBAN = args['validate']['HINBAN']

        """2.connect to Oracle DB"""
        conn = oracle.connect('{uname}'.format(uname=uname), '{upwd}'.format(upwd=upwd), '{host}'.format(host=host))
        cursor = conn.cursor()

        """3.execute SQL query to collect data and assert"""
        sql = cursor.execute("{}".format(query))
        whole_data = sql.fetchall()
        list_data = []
        for i in whole_data:
            for j in i:
                list_data.append(j)
        assert list_data[0] == HINBAN

if __name__ == '__main__':
    to = TestOracleAdminF_SHOZAIKO()
    to.test_admin_f_shozaiko()