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

在开发过程中，经常需要验证某个逻辑，或者某种设计方案，但是我们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. 
   
2. 安装必要的库：你需要安装Google的protobuf库，它提供了用于编写和读取protocol buffers的Python API。可以使用以下命令来安装：
   

pip install protobuf

2. 
   
3. 编写protocol buffer定义文件：使用protobuf语言编写定义文件，定义接口请求和响应的消息格式。(这一步基本上不需要，我们项目都是定义好的)
   
4. 生成Python代码：使用protobuf编译器生成Python代码，这些代码包括消息类、编码和解码函数等。可以使用以下命令来生成Python代码：
   

protoc --python_out=. your_proto_file.proto

4. 
   
5. 编写测试脚本：使用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. 
   
2. 实际需求中，pb 的数量巨多
   
3. python 入门真的非常简单，上述代码谁都能整明白，但是为什么做不好一个项目呢？
   

2.1.2 解决问题

1. 
   
2. 实际需求中，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文件。

2. 
   
3. 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. 
   
2. 使用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)

2. 
   
3. 使用Python内置的cache模块，它提供了一个简单易用的内存缓存实现，可以方便地缓存任意可哈希的对象。
   

import cache


my_cache = cache.Cache()

my_cache.set("key", "value")

value = my_cache.get("key")

3. 
   
4. 使用第三方的缓存库，例如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. 
   
2. 使用本地内存缓存
   

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

2. 
   
3. 使用本地文件缓存
   

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

3. 
   
4. 使用远程缓存
   

如果需要在不同的机器之间共享缓存数据，可以使用远程缓存，例如使用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的缓存，可以减少请求的响应时间，提高系统的性能和可用性。

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