TesterHome

游戏测试 python 编写游戏测试机器人客户端 (一)

A_Jian · 2020年12月08日 · 最后由 goodpassion 回复于 2021年01月28日 · 9596 次阅读
本帖已被设为精华帖！

游戏测试机器人搭建 - Player Actor

前言：

第一次写博客，文笔有限，加上半路出身的游测，大佬们别往死里喷，错的地方请指正，在正式介绍之前先简单说下框架：Python 的 pykka(Actor 模型)，测试框架 pytest 和测试报告 allure。粗略框架介绍

阅读前的基础知识

Python 的 socket 编程，多进程，多线程，队列

Actors 做什么

当一个 actor 接收到消息后，它能做如下三件事中的一件：

  • Create more actors; 创建其他 actors
  • Send messages to other actors; 向其他 actors 发送消息
  • Designates what to do with the next message. 指定下一条消息到来的行为

Actor 的组成

Actor 是由状态（state）、行为（behavior）、邮箱（mailbox）三者组成的。

  • 状态（state）：状态是指 actor 对象的变量信息，状态由 actor 自身管理，避免并发环境下的锁和内存原子性等问题。
  • 行为（behavior）：行为指定的是 actor 中计算逻辑，通过 actor 接收到的消息来改变 actor 的状态。
  • 邮箱（mailbox）：邮箱是 actor 之间的通信桥梁，邮箱内部通过 FIFO 消息队列来存储发送发消息，而接收方则从邮箱中获取消息。

Actors 一大重要特征在于 actors 之间相互隔离，它们并不互相共享内存。这点区别于上述的对象。也就是说，一个 actor 能维持一个私有的状态，并且这个状态不可能被另一个 actor 所改变。
具体 Actor 介绍可参考：传送门 >> JunChow520 的 Actor 模型介绍

pykka 框架的使用

传送门 >> pykka 使用说明
引用说明书的例子：

# !/usr/bin/env python3
import pykka

GetMessages = object()

class PlainActor(pykka.ThreadingActor):
    def __init__(self):
        super().__init__()
        self.stored_messages = []

    def on_receive(self, message):
        if message is GetMessages:
            return self.stored_messages
        else:
            self.stored_messages.append(message)


if __name__ == '__main__':
    actor = PlainActor.start()
    actor.tell({'no': 'Norway', 'se': 'Sweden'})
    actor.tell({'a': 3, 'b': 4, 'c': 5})
    print(actor.ask(GetMessages))
    actor.stop()

环境

  • Python:3.7
  • Windows/Linux

正文

机器人架构图

robot

测试机器人要用的 3 个核心 Actor(Player,Send,Recv)

Player Actor

Player 对象初始化

  • 传递参数：用户名、服务器 IP 和 Port、性别、职业，根据项目类型增减；
  • 其他 Actor 的调用：
    • Recv Actor：必要的，socket 接收数据，后面章节介绍；
    • Send Actor：必要的，socket 发送数据，后面章节介绍；
    • 可自由增加其他的 Actor；
  • 初始化可设置角色的相关属性，后面根据服务端返回数据进行相关赋值，如：角色的 ID，name，资产货币，攻击防御等 ；
# !/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   player.py    
@Contact :   512759438@qq.com
@Author  :   Jian
'''


import pykka
import websocket
import traceback as tb
from proto import ProtoHandler
from remote import RemoteHandler
from util import MsgSwitch, RecvActor, SendActor


TEST_CASE_CALL = None
class Player(pykka.ThreadingActor):
    def __init__(self, user_name='', server='SERVER_1',sex=1, job=1):
        super(Player, self).__init__()
        self.host = SERVER_LIST[server]['GAME_HOST']
        self.port = SERVER_LIST[server]['GAME_PORT']
        self.web_host = "x.x.x.x"
        self.recv_actor = None
        self.send_actor = None
        self.socket = None
        self.proto_handler = ProtoHandler(self)
        self.remote_handler = RemoteHandler(self)
        '''测试用例执行时需要调用player'''
        global TEST_CASE_CALL
        TEST_CASE_CALL = self

        self.player_id = None
        self.state_user_id = 0
        self.state_user_name = user_name
        self.sys_count = 0

封装消息发送，用来给自己发送消息或者其它 Actor 调用

def send_msg(self, msg_type=None, data=None):
    '''
    :param msg_type:消息类型 
    :param data: 数据
    '''
    self.actor_ref.tell({
        'msg': msg_type,
        'data': data
    })

Player Actor 实例化之后第一个执行的地方

  • 不指定用户的话则随机账号登录
  • MSG_GUEST_LOGIN: 消息类型
  • 接下来就把消息告诉自己，on_receive 会接收到这条消息
def on_start(self):
    if self.state_user_name is '':
        self.send_msg(MSG_GUEST_LOGIN)
    else:
        self.send_msg(MSG_LOGIN_INFO)

接收消息及消息处理

  • 接收的消息会放进 MsgSwitch 进行参数检查，MsgSwitch 参考 C 语言 switch
  • 接下来是用户注册、创角、登录流程
  • MSG_LOGIN_INFO：获取玩家的信息
    • 获取用户信息后通知 MSG_LOGIN
  • MSG_LOGIN：玩家开始连接 socket，有角色则开始登录，没有角色则通知 MSG_CREATE_PLAYER
    • 连接 socket 后开启 Send Actor 和 Recv Actor
    • 给 Send Actor 发送消息：角色登录需要的参数，Send Actor会进行打包和发给服务端
  • MSG_CREATE_PLAYER：
    • 把需要的创角参数传给Send Actor，打包后通知服务端要创角
  • MSG_PROTO：Recv Actor 从 socket.recv 接收的数据进行反序列化后会发送过来到这处理
    • 如果需要对服务端返回的协议数据进行自动处理，可以在 proto 模块写上对应处理方法，MSG_PROTO 消息类型接收的每一条协议数据都会去 proto 模块查找有没有对应的处理方法 (hasattr 和 getattr)
  • MSG_REMOTE_CMD：后续写到 remote 再进行结合一起写，不影响运行
def on_receive(self, msg):
    for case in MsgSwitch(msg):
        # 获取用户信息
        if case(MSG_LOGIN_INFO):
            account_info = Account(self.state_user_name).login_info()
            if account_info['code_str'] == 'OK':
                user_into = account_info['user']
                self.create_player_params  = {
                    'rd3_token': user_into['token'],
                    'rd3_userId': user_into['userId'],
                    'server_list_type': 0,
                    'sid': 1,
                    'token': user_into['token'],
                }
                self.create_player_params.update(Account(self.state_user_name).data)
                self.create_player_params.pop('password')
                self.create_player_params['cmd'] = 'game_login'
                self.send_msg(MSG_LOGIN)
            else:print(f'获取角色信息ERROR, 原因: {account_info["code_str"]},{account_info["code"]}')
            break

        # 用户登录
        if case(MSG_LOGIN):
            self.socket = websocket.create_connection(f'ws://{self.host}:{self.port}/')
            self.recv_actor = RecvActor.start(self, self.socket)
            self.send_actor = SendActor.start(self, self.socket)
            self.send_actor.tell({MSG_PROTO: self.create_player_params})
            break
        # 用户创角
        if case(MSG_CREATE_PLAYER):
            create_data = {
                'nickname': self.state_user_name,
                'rd3_token': self.create_player_params['rd3_token'],
                'rd3_userId': self.create_player_params['rd3_userId'],
                'sid': self.create_player_params['sid'],
                'token': self.create_player_params['token'],
            }
            self.send_actor.tell({MSG_PROTO: create_data})
            break

        # 服务端返回协议处理
        if case(MSG_PROTO):  
            method, data = msg['data']
            if hasattr(self.proto_handler, method):
                getattr(self.proto_handler, method)(data)
            else:
                print(f"没有为协议: {method} 定义处理方法, 请前往 proto.py 文件中定义!")
            break
        # 控制台调用命令
        if case(MSG_REMOTE_CMD):
            method = msg['method']
            method = (type(method) is int and "r" + str(method)) or (type(method) is str and method)
            if hasattr(self.remote_handler, method):
                getattr(self.remote_handler, method)(msg['data'])
            else:
                print(f"没有为远程命令: {method} 定义处理方法, 请前往 remote.py 文件中定义!")
            break

封装远程命令

  • 角色登录后可通过 pykka.ActorRegistry.get_by_class_name('Player') 获取实例对象用远程命令遥控角色
def remote_msg(self, method:str=None, data=None):
    '''
    调用remote里的方法
    :param method: 方法名
    :param data: 传入的参数 元组
    '''
    self.actor_ref.tell({
        'msg': MSG_REMOTE_CMD,
        'method': method,
        'data': data
    })

停止 Player Actor

  • 先停止其他的 Actor 再关闭 socket，最后关掉自己
def on_stop(self):
    self.recv_actor.stop()
    self.send_actor.stop()
    self.socket.close()
    self.socket.shutdown()
    self.stop()

log 收集

# 打印报错消息
@GetLog(level='fatal')
def on_failure(self, exception_type, exception_value, traceback):
    logging.fatal(f'Player: {self.state_user_name} is down.')
    logging.fatal(f"ErrorType  => {exception_type}")
    logging.fatal(f"ErrorValue => {exception_value}")
    logging.fatal(f"TraceBack  => {tb.print_tb(traceback)}")
    self.on_stop()

后续文章传送门

python 编写游戏测试机器人客户端 (一)
python 编写游戏测试机器人客户端 (二)
python 编写游戏测试机器人客户端 (三)
python 编写游戏测试机器人客户端 (四)

到这里 Player Actor 已经写完了，目前是无法单独运行的，需要结合后面的 Send Actor 和 Recv Actor 才能运行，写的不清晰的欢迎在评论留言

最后的最后，各位的关注、点赞、收藏、碎银子打赏是对我最大的支持，谢谢大家！
需要源码的小伙伴关注微信公众号 ID：gameTesterGz
或扫描二维码回复机器人脚本即可
微信二维码

「原创声明：保留所有权利，禁止转载」
15 个赞
如果觉得我的文章对您有用，请随意打赏。您的支持将鼓励我继续创作！
打赏支持
共收到 4 条回复 时间 点赞
陈子昂 将本帖设为了精华贴 12月08日 21:01
陈子昂 #1 · 2020年12月08日 2 个赞

加精理由是 pykka 的应用和 actor 模式，这块在测试上属于相对跨学科，建议都可以学习下。并且是个系列作品，期待后续。

skottZy #3 · 2020年12月10日

case 这一块用反射去做好一点吧,getattr(object, case)(args)

A_Jian #4 · 2020年12月11日 Author
对 skottZy 回复

我这里 MsgSwitch 是个迭代器，返回一个 match 方法对接收的消息类型进行检查

需要 登录 后方可回复, 如果你还没有账号请点击这里 注册。