基于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\

目录
相关文章
|
6天前
|
数据采集 供应链 API
实战指南:通过1688开放平台API获取商品详情数据(附Python代码及避坑指南)
1688作为国内最大的B2B供应链平台,其API为企业提供合法合规的JSON数据源,直接获取批发价、SKU库存等核心数据。相比爬虫方案,官方API避免了反爬严格、数据缺失和法律风险等问题。企业接入1688商品API需完成资质认证、创建应用、签名机制解析及调用接口四步。应用场景包括智能采购系统、供应商评估模型和跨境选品分析。提供高频问题解决方案及安全合规实践,确保数据安全与合法使用。立即访问1688开放平台,解锁B2B数据宝藏!
|
7天前
|
API 开发工具 Python
【Azure Developer】编写Python SDK代码实现从China Azure中VM Disk中创建磁盘快照Snapshot
本文介绍如何使用Python SDK为中国区微软云(China Azure)中的虚拟机磁盘创建快照。通过Azure Python SDK的Snapshot Class,指定`location`和`creation_data`参数,使用`Copy`选项从现有磁盘创建快照。代码示例展示了如何配置Default Azure Credential,并设置特定于中国区Azure的`base_url`和`credential_scopes`。参考资料包括官方文档和相关API说明。
|
2月前
|
存储 缓存 Java
Python高性能编程:五种核心优化技术的原理与Python代码
Python在高性能应用场景中常因执行速度不及C、C++等编译型语言而受质疑,但通过合理利用标准库的优化特性,如`__slots__`机制、列表推导式、`@lru_cache`装饰器和生成器等,可以显著提升代码效率。本文详细介绍了这些实用的性能优化技术,帮助开发者在不牺牲代码质量的前提下提高程序性能。实验数据表明,这些优化方法能在内存使用和计算效率方面带来显著改进,适用于大规模数据处理、递归计算等场景。
76 5
Python高性能编程:五种核心优化技术的原理与Python代码
|
3月前
|
Python
课程设计项目之基于Python实现围棋游戏代码
游戏进去默认为九路玩法,当然也可以选择十三路或是十九路玩法 使用pycharam打开项目,pip安装模块并引用,然后运行即可, 代码每行都有详细的注释,可以做课程设计或者毕业设计项目参考
85 33
|
3月前
|
IDE 测试技术 开发工具
10个必备Python调试技巧:从pdb到单元测试的开发效率提升指南
在Python开发中,调试是提升效率的关键技能。本文总结了10个实用的调试方法,涵盖内置调试器pdb、breakpoint()函数、断言机制、logging模块、列表推导式优化、IPython调试、警告机制、IDE调试工具、inspect模块和单元测试框架的应用。通过这些技巧,开发者可以更高效地定位和解决问题,提高代码质量。
396 8
10个必备Python调试技巧:从pdb到单元测试的开发效率提升指南
|
3月前
|
JavaScript API C#
【Azure Developer】Python代码调用Graph API将外部用户添加到组,结果无效,也无错误信息
根据Graph API文档,在单个请求中将多个成员添加到组时,Python代码示例中的`members@odata.bind`被错误写为`members@odata_bind`,导致用户未成功添加。
59 10
|
3月前
|
存储 应用服务中间件 开发工具
对象存储OSS-Python设置代理访问请求
通过 Python SDK 配置 nginx 代理地址请求阿里云 OSS 存储桶服务。示例代码展示了如何使用 RAM 账号进行身份验证,并通过代理下载指定对象到本地文件。
120 15
|
3月前
|
数据可视化 Python
以下是一些常用的图表类型及其Python代码示例,使用Matplotlib和Seaborn库。
通过这些思维导图和分析说明表,您可以更直观地理解和选择适合的数据可视化图表类型,帮助更有效地展示和分析数据。
116 8
|
3月前
|
Python
探索Python中的装饰器:简化代码,增强功能
在Python的世界里,装饰器就像是给函数穿上了一件神奇的外套,让它们拥有了超能力。本文将通过浅显易懂的语言和生动的比喻,带你了解装饰器的基本概念、使用方法以及它们如何让你的代码变得更加简洁高效。让我们一起揭开装饰器的神秘面纱,看看它是如何在不改变函数核心逻辑的情况下,为函数增添新功能的吧!
|
3月前
|
程序员 测试技术 数据安全/隐私保护
深入理解Python装饰器:提升代码重用与可读性
本文旨在为中高级Python开发者提供一份关于装饰器的深度解析。通过探讨装饰器的基本原理、类型以及在实际项目中的应用案例,帮助读者更好地理解并运用这一强大的语言特性。不同于常规摘要,本文将以一个实际的软件开发场景引入,逐步揭示装饰器如何优化代码结构,提高开发效率和代码质量。
80 6

热门文章

最新文章