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

A_Jian · 2020年12月08日 · 最后由 A_Jian 回复于 2020年12月11日 · 1729 次阅读
本帖已被设为精华帖!

游戏测试机器人搭建 - 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 ActorRecv 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
或扫描二维码回复机器人脚本即可
微信二维码

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
共收到 3 条回复 时间 点赞
陈子昂 将本帖设为了精华贴 12月08日 13:01

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

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

skottZy 回复

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

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