likes
comments
collection
share

零散逻辑验证不再烦恼：基于Python和Redis的实践

作者站长头像
站长
· 阅读数 59

在开发过程中，经常需要验证某个逻辑，或者某种设计方案，但是我们Android的编译运行会随着项目的迭代变慢，此时验证问题较为麻烦，很多工程师会选择新建一个新的项目去验证，但是新建项目会面临很多基础组件的调用问题，如果在项目之外创建的项目为其配备对应的基件，由会面临开头的问题。

这时，拥有一个能验证零散逻辑，且不会造成上述问题的工程就尤为重要了。

经过筛选，我们可以创建一个python 项目，在平时的开发中，我们可以快速验证某些逻辑，当然，这个验证还是要看你有没有这个需求。

为你的项目创建Python 项目

如果你的团队测试有python经验，那可以向他们取经，如果没有，就要酌情新建了，要尊重测试岗位的同事。

我创建的需求是，我需要一个测试平台，验证IM相关的一些逻辑是否可行，刚好公司没有测试接口的项目，所以直接搭建了一个python 项目

一、环境

  • Mac pro 12.6 版本
  • Pycharm (PyCharm 2022.3.1 (Community Edition))
  • Python 3.x
  • redis 缓存
  • 传输数据格式： protocol buffers（PB）

二、基建搭建

创建项目就不用说了，使用pycharm 创建即可

2.1 pb 编译

我们可以同Android开发做对比，pb 的使用流程应该是一样的。

protocol buffers 提供了Java、kotlin及python的编译，这直接决定了使用pb传输的公司能否使用python进行接口自动化测试。在python 中使用pb传输，步骤如下：

  1. 安装必要的库：你需要安装Google的protobuf库，它提供了用于编写和读取protocol buffers的Python API。可以使用以下命令来安装：
pip install protobuf
  1. 编写protocol buffer定义文件：使用protobuf语言编写定义文件，定义接口请求和响应的消息格式。(这一步基本上不需要，我们项目都是定义好的)
  2. 生成Python代码：使用protobuf编译器生成Python代码，这些代码包括消息类、编码和解码函数等。可以使用以下命令来生成Python代码：
protoc --python_out=. your_proto_file.proto
  1. 编写测试脚本：使用Python编写测试脚本，根据你的测试需求构造请求消息，发送请求到服务器，并验证响应消息是否符合预期。(这是自动化测试流程，当然我需要的是逻辑验证，这里就按自己的需求即可)

例子：

用于向服务器发送一个消息，并验证服务器的响应

import your_proto_file_pb2
import requests

# 构造请求消息
request_msg = your_proto_file_pb2.RequestMessage()
request_msg.user_id = 123
request_msg.request_type = "get_data"

# 发送请求
response = requests.post('http://your-server.com/api', data=request_msg.SerializeToString())

# 解析响应消息
response_msg = your_proto_file_pb2.ResponseMessage()
response_msg.ParseFromString(response.content)

# 验证响应消息是否符合预期
assert response_msg.result == "success"

是不是非常简单。

2.1.1 存在问题
  1. 实际需求中，pb 的数量巨多
  2. python 入门真的非常简单，上述代码谁都能整明白，但是为什么做不好一个项目呢？
2.1.2 解决问题
  1. 实际需求中，pb 的数量巨多

利用脚本，统一编译， 使用shell 非常简单（哈哈 在python项目中使用shell）,我就直接写出来了。

#!/bin/bash
PB_DIR=/proto
KOTLIN_OUT_DIR=../
for file in ${PB_DIR}/*.proto; do
  # 获取文件名（不包含扩展名）
  echo ${file}
  filename=$(basename -- "${file}")
  filename="${filename%.*}"


  protoc --proto_path=${PB_DIR} --python_out=${KOTLIN_OUT_DIR} ${file}

done

执行脚本之后，会统一生成pb文件。

  1. python 入门真的非常简单，上述代码谁都能整明白，但是为什么做不好一个项目呢？

这就是后续要讨论的，将代码工程化。

2.2 搭建网络请求框架

在Android 开发中，网络请求我们一般都是统一初始化，后来的便利性架构，可以添加拦截器等很爽的处理方式，python 的网络请求非常简单。

import requests
import hashlib
import time
from collections import OrderedDict

class NetworkUtils:
    
    def __init__(self):
        self.url = "baseurl"
        # 有序字典
        self.headers = OrderedDict()
        self.payload = None
        # SSL 证书验证
        self.cert = ('client.crt',
                     'client.key')
        self.verify = 'ca.crt'
        self.response = None

    def set_headers(self, head_map):
        for key, value in head_map.items():
            self.headers[key] = value

    def set_payload(self, payload):
        self.payload = payload

    def set_cert(self, cert):
        self.cert = cert

    def set_verify(self, verify):
        self.verify = verify

    def request(self, path):
        # 可以做统一的拦截工作
        self.response = requests.request("POST", self.url,
                                         headers=self.headers,
                                         data=self.payload,
                                         cert=self.cert,
                                         verify=self.verify)

    @staticmethod
    def get_sign(map, key, body):
        #hashlib 加密数据等
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

2.3 使用

def test_guest():
    try:
        # 构造请求消息 
        request = your_proto_file_pb2.RequestMessage() request_msg.user_id = 123
        request.request_type = "get_data"
        # 转成二进制 pb 序列化
        payload = request.SerializeToString()
        network_utils = NetworkUtils()
        network_utils.set_payload(payload)
        network_utils.set_headers(OrderedDict())
        network_utils.request("path")
        # 解析响应消息 
        response_msg = your_proto_file_pb2.ResponseMessage()
        response_msg.ParseFromString(response.content)
        print(res_response)
    except Exception as ex:
        print("Found Error in auth phase:%s" % str(ex))


if __name__ == '__main__':
    test_guest()

三、缓存

3.1 常见的缓存

在Python中，可以使用多种方式缓存数据，下面是其中的几种：

  1. 使用Python内置的functools.lru_cache装饰器，它可以自动地为函数添加缓存机制，从而避免函数重复计算。使用functools.lru_cache时需要注意函数的参数和返回值必须是可哈希的。
import functools

@functools.lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)
  1. 使用Python内置的cache模块，它提供了一个简单易用的内存缓存实现，可以方便地缓存任意可哈希的对象。
import cache

my_cache = cache.Cache()
my_cache.set("key", "value")
value = my_cache.get("key")
  1. 使用第三方的缓存库，例如redis、memcached等，这些库可以将缓存数据存储在内存或磁盘中，支持多种数据结构，可以实现分布式缓存等高级功能。
import redis

redis_client = redis.Redis(host='localhost', port=6379)
redis_client.set('key', 'value')
value = redis_client.get('key')

3.2 使用建议

  1. 使用本地内存缓存

如果测试数据量较小，可以使用Python内置的dict或者第三方库（如cachetools）来实现本地内存缓存。这种方法的优点是速度快，不需要网络请求和数据序列化等操作，但缺点是缓存的生命周期与应用程序相同，程序重启后缓存会被清空。

  1. 使用本地文件缓存

如果测试数据量较大，或者需要持久化缓存数据，可以将缓存数据保存在本地文件中，例如使用Python内置的shelve模块，或者第三方库（如diskcache）。这种方法的优点是可以实现持久化缓存，缺点是速度较慢，需要进行数据序列化和文件读写操作。

  1. 使用远程缓存

如果需要在不同的机器之间共享缓存数据，可以使用远程缓存，例如使用Redis等内存数据库。这种方法的优点是可以实现分布式缓存，多个应用程序之间可以共享缓存数据，缺点是需要网络请求，速度相对较慢。

针对服务接口自动测试的缓存问题，需要根据实际情况和需求选择合适的缓存方式。如果测试数据量较小，可以考虑使用本地内存缓存，如果需要持久化缓存数据，可以使用本地文件缓存，如果需要分布式缓存，可以使用远程缓存。同时，需要注意缓存的生命周期、缓存数据的一致性以及缓存的清理等问题。

3.3 redis

Redis是一个高性能、非关系型的内存数据库，常用于缓存、队列、计数器、排行榜等场景。在Python中，可以使用第三方库redis-py来连接和操作Redis数据库。下面介绍一下Redis在Python中的使用方式，并提供一个基于redis-py的工具类实现。

3.3.1 安装redis-py库
pip install redis
3.3.2 连接Redis数据库

在使用redis-py库操作Redis数据库之前，需要先建立与Redis数据库的连接。redis-py提供了Redis类来连接Redis数据库

import redis

# 建立与Redis数据库的连接
r = redis.Redis(host='localhost', port=6379, db=0)

以上代码建立了一个默认的Redis连接，连接到本地Redis服务器，端口号为6379，使用默认的0号数据库。如果需要连接其他Redis服务器或其他数据库，可以修改host、port和db参数。

3.3.3 存储数据

连接到Redis数据库后，可以使用redis-py提供的方法来存储数据

# 存储字符串类型的数据
r.set('name', 'Tom')

# 存储哈希类型的数据
r.hset('user', 'name', 'Tom')
r.hset('user', 'age', 18)

# 存储列表类型的数据
r.lpush('mylist', 'a', 'b', 'c')
3.3.4 获取数据
# 获取字符串类型的数据
name = r.get('name')
print(name)

# 获取哈希类型的数据
user = r.hgetall('user')
print(user)

# 获取列表类型的数据
mylist = r.lrange('mylist', 0, -1)
print(mylist)
3.3.5 删除数据
# 删除字符串类型的数据
r.delete('name')

# 删除哈希类型的数据
r.hdel('user', 'age')

# 删除列表类型的数据
r.ltrim('mylist', 1, -1)

3.3 redis 工具类封装

基于redis-py的Redis工具类，可以方便地对Redis进行连接、存储、获取和删除等操作：

import redis

class RedisClient:
    def __init__(self, host='localhost', port=6379, db=0):
        self.host = host
        self.port = port
        self.db = db
        self.client = redis.Redis(host=self.host, port=self.port, db=self.db)

    def set(self, key, value, expire=None):
        self.client.set(key, value, ex=expire)

    def get(self, key):
        return self.client.get(key)

    def hset(self, name, key, value):
        self.client.hset(name, key, value)

    def hgetall(self, name):
        return self.client.hgetall(name)

    def lpush(self, name, *values):
        self.client.lpush(name
    def lrange(self, name, start=0, end=-1):
        return self.client.lrange(name, start, end)

    def delete(self, *keys):
        self.client.delete(*keys)

    def hdel(self, name, *keys):
        self.client.hdel(name, *keys)

    def ltrim(self, name, start, end):
        self.client.ltrim(name, start, end)

实现了常见的Redis操作，包括set/get、hset/hgetall、lpush/lrange、delete、hdel和ltrim等方法。使用时，只需要创建RedisClient对象并调用相应的方法即可

# 创建RedisClient对象
redis_client = RedisClient()

# 存储数据
redis_client.set('name', 'Tom')
redis_client.hset('user', 'name', 'Tom')
redis_client.lpush('mylist', 'a', 'b', 'c')

# 获取数据
name = redis_client.get('name')
user = redis_client.hgetall('user')
mylist = redis_client.lrange('mylist', 0, -1)

# 删除数据
redis_client.delete('name')
redis_client.hdel('user', 'name')
redis_client.ltrim('mylist', 1, -1)

总结

在开发过程中，为了快速验证某些逻辑，可以考虑创建一个Python项目。并要多使用面向对象的方式编码，可以提高代码的可读性和可维护性。

另外，在进行接口自动测试时，可以使用Python中的缓存工具，例如Redis，来提高接口的性能和效率。通过使用Redis的缓存，可以减少请求的响应时间，提高系统的性能和可用性。

这不，有了这个基础，尽情玩吧！！！

转载自:https://juejin.cn/post/7224330515535904825
评论
请登录