基于Python访问Hive的pytest测试代码实现

简介: 根据《用Java、Python来开发Hive应用》一文,建立了使用Python、来开发Hive应用的方法,产生的代码如下

根据《用Java、Python来开发Hive应用》一文,建立了使用Python、来开发Hive应用的方法,产生的代码如下(做了修改):

代码语言:javascript

复制

import pandas as pd
from pyhive import hive
from sqlalchemy import create_engine
from pyhive import hive
               
class Hive:
    def __init__(self):
        self.database= 'demo'
        self.host = '192.168.31.184'
        self.port = '10000'
        
    def getConnect(self):
        conn = hive.Connection(host=self.host, port=self.port,database=self.database)
        return conn;
                
    def getEngine(self):    
        # 创建 Hive 数据库连接
        hive_uri = f"hive://"+self.host+":"+self.port+"/"+self.database
        return create_engine(hive_uri)
           
    def disconnect(self,engine,conn):
        engine.dispose()
        conn.close()
           
    #执行查询
    def query(self,sql,engine,condition=None):
        try:
            if condition is None:
            # 执行 SQL 查询
                rs = pd.read_sql(sql, engine)
                return rs
            else:
                values = []
                where = " where "
                for key in condition:
                    where = where+key+" = %s and "
                    values.append(condition[key])
                where = where+"1=1"    
                sql = sql + where
                params = tuple(values)
                rs = pd.read_sql(sql, engine, params=params)
                return rs
        except Exception as e:
            print("Error occurred:", e)
           
    #添加数据
    def addDataToHiveTable(self,conn,tableName,data):
        like_array = f"array({', '.join(map(lambda x: f'\'{x}\'', data['like']))})"  # 使用单引号包裹字符串
        address_map = f"map('{list(data['address'].keys())[0]}', '{list(data['address'].values())[0]}')"  # 创建 MAP 格式
        # 创建游标
        cursor = conn.cursor()
        insertSql = "INSERT INTO person SELECT %s,%s,%s,"+like_array+","+address_map
        # 执行插入操作
        try:
            cursor.execute(insertSql, (data['id'], data['name'], data['age']))
        except Exception as e:
            print(f"Error inserting data: {e}")    
        conn.commit()
        cursor.close()
           
    #将文件中的数据加载到表中
    def loadDataForLocal(self,conn,tableName,path):
        cursor = conn.cursor()
        query = "LOAD DATA LOCAL INPATH '"+path+"' INTO TABLE "+tableName
        cursor.execute(query)
        conn.commit()
        cursor.close()
    
    #清空数据表
    def truncateTable(self,conn,tableName):
        cursor = conn.cursor()
        query = "truncate table "+tableName;
        #con.setAutoCommit(true) #确保自动提交
        cursor.execute(query)
        conn.commit()
        cursor.close()

现在,使用pytest来进行测试。

1)建立全局变量

代码语言:javascript

复制

hive = Hive()
    tableName = "person"
    sql = "SELECT * FROM "+tableName
    conn = None
    engine = None

2)建立setup_class(self)和teardown_class(self)函数

代码语言:javascript

复制

def setup_class(self):
        #导入数据路径
        path = "/home/jerry/hive/person"
        #建立连接,conn用于查询相关的SQL
        self.conn = self.hive.getConnect()
        #建立引擎,engine用于非查询相关的SQL
        self.engine = self.hive.getEngine()
        #导入测试初始化数据
        self.hive.loadDataForLocal(self.conn,self.tableName,path)
           
    def teardown_class(self):
        #清空测试数据
        self.hive.truncateTable(self.conn,self.tableName)
        #断开链接
        self.hive.disconnect(self.engine,self.conn)

3)测试查询

代码语言:javascript

复制

@allure.feature('Python访问Hive数据库')
    @allure.story('根据query进行查询')
    @allure.severity('Critical')
    #测试根据Query查询
    def test_query(self):
        #建立查询
        rs = self.hive.query(self.sql,self.engine)
        #获得记录个数
        row_count = len(rs)
        #验证记录个数
        assert row_count == 6
        #遍历记录
        for index, row in rs.iterrows():
            #验证编号行是不是数字
            assert isinstance(row.iloc[0], int)
            #验证姓名行是不是包含"elite"
            assert "elite" in row.iloc[1]
            #验证年龄行是不是数字
            assert isinstance(row.iloc[2], int)
            #验证爱好行是不是包含"basketball"
            assert "basketball" in row.iloc[3]
            #验证地址行是不是包含"address"    
            assert "address" in row.iloc[4]
           
    @allure.feature('Python访问Hive数据库')
    @allure.story('带一个的条件查询')
    @allure.severity('Normal')
    def test_query_with_one_condition(self):
        #查询条件
        condition={"name":"elite1"}
        #建立查询
        rs = self.hive.query(self.sql,self.engine,condition)
        #获得记录个数
        row_count = len(rs)
        #验证记录个数
        assert row_count == 1
        #遍历记录
        for index, row in rs.iterrows():
            #验证是不是符合查询条件
            #验证姓名行是不是包含"elite"
            assert "elite1" == row.iloc[1]
            
           
    @allure.feature('Python访问Hive数据库')    
    @allure.story('带两个的条件查询')
    @allure.severity('Normal')
    def test_query_with_Two_condition(self):
        #查询条件
        condition={"name":"elite1","age":"20"}
        #建立查询
        rs = self.hive.query(self.sql,self.engine,condition)
        #获得记录个数
        row_count = len(rs)
        #验证记录个数
        assert row_count == 1
        #遍历记录
        for index, row in rs.iterrows():
            #验证是不是符合查询条件
            #验证姓名行是不是"elite"
            assert "elite1" == row.iloc[1]
            #验证年龄行是不是50
            assert "50" == row.iloc[2]
            #验证编号行是不是数字
            assert isinstance(row.iloc[0], int)
            #验证姓名行是不是包含"elite"
            assert "elite" in row.iloc[1]    
            #验证年龄行是不是数字
            assert isinstance(row.iloc[2], int)
            #验证爱好行是不是包含"basketball"
            assert "basketball" in row.iloc[3]
            #验证地址行是不是包含"address"
            assert "address" in row.iloc[4]
           
    @allure.feature('Python访问Hive数据库')
    @allure.story('带两个的条件查询')
    @allure.severity('Normal')
    def test_query_with_Two_condition(self):
        #查询条件
        condition={"name":"elite1","age":"20"}
        #建立查询
        rs = self.hive.query(self.sql,self.engine,condition)
        #获得记录个数
        row_count = len(rs)
        #验证记录个数
        assert row_count == 1
        #遍历记录
        for index, row in rs.iterrows():
            #验证是不是符合查询条件    
            #验证姓名行是不是"elite"
            assert "elite1" == row.iloc[1]
            #验证年龄行是不是20
            assert 20 == row.iloc[2]
                     
    @allure.feature('Python访问Hive数据库')
    @allure.story('带三个的条件查询')
    @allure.severity('Normal')
    def test_query_with_three_condition(self):
        #查询条件
        condition={"id":"1","name":"elite0","age":"10"}
        #建立查询
        rs = self.hive.query(self.sql,self.engine,condition)
        #获得记录个数
        row_count = len(rs)
        #验证记录个数
        assert row_count == 1
        #遍历记录
        for index, row in rs.iterrows():
            #验证是不是符合查询条件
            #验证编号行是不是数字    
            assert 1 == row.iloc[0]
            #验证姓名行是不是包含"elite"
            assert "elite0" in row.iloc[1]
            #验证年龄行是不是数字
            assert 10 == row.iloc[2]

4)测试添加数据

代码语言:javascript

复制

@allure.feature('Python访问Hive数据库')
    @allure.story('插入数据')
    @allure.severity('Normal')
    def test_addDataToHiveTable(self):
        #构造插入数据
        data = {
        'id': "50",
        'name': "Jerry",
        'age': "50",
        'like': ["basketball", "music", "dance"],
        'address': {"address": "xx"}
        }
        #插入数据
        self.hive.addDataToHiveTable(self.conn,self.tableName,data)
        #查询插入数据是否可以查询出来
        condition = {"name":"Jerry","age":"50"}    
        rs = self.hive.query(self.sql,self.engine,condition)
        row_count = len(rs)
        assert row_count == 1
        #验证插入数据
        for index, row in rs.iterrows():
            assert "Jerry" in row.iloc[1]
            assert "50" in str(row.iloc[2])

主函数改为

代码语言:javascript

复制

if __name__ == '__main__':
pytest.main(['-sv', '-q', '--alluredir', './report/xml'])

建立项目文件

代码语言:javascript

复制

environment.properties
Project Name=Hive
Author = Jerry Gu
System Version= Win11
java version "17.0.10"
Allure Version= 2.20.1

代码语言:javascript

复制

pytest --alluredir=.\report\xml
copy environment.properties .\report\xml
allure serve .\report\xml\

目录
相关文章
|
1天前
|
测试技术 数据安全/隐私保护 开发者
自动化测试的奥秘:如何用Selenium和Python提升软件质量
【9月更文挑战第35天】在软件开发的海洋中,自动化测试是那艘能引领我们穿越波涛的帆船。本文将揭开自动化测试的神秘面纱,以Selenium和Python为工具,展示如何构建一个简单而强大的自动化测试框架。我们将从基础出发,逐步深入到高级应用,让读者能够理解并实现自动化测试脚本,从而提升软件的质量与可靠性。
|
12天前
|
设计模式 SQL 安全
PHP中的设计模式:单例模式的深入探索与实践在PHP的编程实践中,设计模式是解决常见软件设计问题的最佳实践。单例模式作为设计模式中的一种,确保一个类只有一个实例,并提供全局访问点,广泛应用于配置管理、日志记录和测试框架等场景。本文将深入探讨单例模式的原理、实现方式及其在PHP中的应用,帮助开发者更好地理解和运用这一设计模式。
在PHP开发中,单例模式通过确保类仅有一个实例并提供一个全局访问点,有效管理和访问共享资源。本文详细介绍了单例模式的概念、PHP实现方式及应用场景,并通过具体代码示例展示如何在PHP中实现单例模式以及如何在实际项目中正确使用它来优化代码结构和性能。
|
15天前
|
Web App开发 测试技术 持续交付
自动化测试的利器:Selenium与Python的完美结合
【9月更文挑战第21天】在软件开发的世界里,测试是确保产品质量的关键步骤。随着敏捷开发和持续集成的流行,自动化测试工具变得尤为重要。本文将介绍如何使用Selenium和Python进行高效的自动化测试,不仅提供代码示例,还深入探讨如何设计测试用例、选择正确的测试框架、以及如何整合到CI/CD流程中。无论你是初学者还是有经验的开发者,这篇文章都将为你提供宝贵的见解和实用的技巧。
26 3
|
24天前
|
SQL JavaScript 前端开发
基于Java访问Hive的JUnit5测试代码实现
根据《用Java、Python来开发Hive应用》一文,建立了使用Java、来开发Hive应用的方法,产生的代码如下
49 6
WK
|
10天前
|
存储 安全 索引
如何在Python中访问字典中的值
在Python中,访问字典(Dictionary)中的值非常简单。字典是一种无序的集合,它存储了键值对(key-value pairs),其中每个键都是唯一的,并映射到一个值上。要访问字典中的值,你需要使用键作为索引。
WK
10 0
|
23天前
|
敏捷开发 测试技术 持续交付
自动化测试之美:如何用Selenium和Python打造高效测试脚本
【9月更文挑战第13天】在软件开发的海洋中,自动化测试是那抹不可或缺的亮色。它不仅提升了测试效率,还保障了产品质量。本文将带你领略使用Selenium和Python构建自动化测试脚本的魅力所在,从环境的搭建到脚本的编写,再到问题的排查,每一步都是对软件质量把控的深刻理解和实践。让我们开始这段探索之旅,解锁自动化测试的秘密吧!
16 0
|
23天前
|
jenkins 测试技术 持续交付
自动化测试的高效之路:如何利用Python进行Web应用测试
【9月更文挑战第13天】在软件开发的快节奏中,自动化测试是确保质量和效率的关键。本文将引导你了解如何使用Python语言及其强大的测试框架来提升Web应用的测试效率。我们将一起探索编写简洁而强大的测试脚本的技巧,以及如何通过持续集成(CI)实现自动化测试流程。准备好让你的测试工作飞一般的感觉!
|
2天前
|
机器学习/深度学习 Linux Python
Python编程教学
Python教学
24 13
|
1天前
|
安全 Python
Python 高级编程:高效读取 txt 文件的技巧与实践
在 Python 中,读取 txt 文件是常见操作。本文介绍了使用 `with` 语句自动管理文件资源、逐行读取文件、读取特定字节范围内容、处理编码问题以及使用缓冲读取提高性能等高级方法,确保代码高效且安全。通过这些技巧,你可以更灵活地处理文件内容,并避免资源泄漏等问题。原文链接:https://www.wodianping.com/app/2024-10/44183.html
32 18
下一篇
无影云桌面