moler是帮助构建自动化测试的库
moler的Python项目详细描述
目录
- moler信息
- moler用法示例
- API设计推理
- 设计的API
< > >
摩尔
moler（moler name origin）是python库 它为构建自动化测试提供了"砖块"。 所有这些"砖块"都有明确的职责，有相似的api， 遵循相同的构造模式（这样就很容易创建新的结构模式）。
它们在这里：
- 命令作为自立对象
- 允许命令触发和解析封装在单个对象中（降低维护成本）
- 事件观察者和回调（报警是事件示例）
- 允许在线反应（不是离线后处理）
- 在后台运行观察者/命令
- 允许将测试逻辑分解为并行运行的多个命令
- 允许处理意外的系统行为（重新启动、警报）
- 状态机->；断开连接后自动连接
- 增加框架自动恢复功能，并帮助解决"哪里出了问题"
- 测试所用设备的所有连接的自动日志记录
- 通过将日志集中在被测系统的不同部分来减少调查时间
moler用法示例
让我们看看莫勒的行动。下面是一个假设的用例："查找所有python进程的pids"：
αααα1- 要获得命令，我们要求设备"给我这样的命令"。
- 要运行命令，我们只需将其作为函数调用（command object是可调用的）
- 命令返回的通常是dict或dict列表-易于处理
上面的代码显示：
PID: 1817 CMD: /usr/bin/python /usr/share/system-config-printer/applet.py PID: 21825 CMD: /usr/bin/python /home/gl/moler/examples/command/unix_ps.py
它怎么知道"我的机器"是什么意思？代码从
my_devices.yml配置文件加载定义：DEVICES:MyMachine:DEVICE_CLASS:moler.device.unixremote.UnixLocalRebexTestMachine:DEVICE_CLASS:moler.device.unixremote.UnixRemoteCONNECTION_HOPS:UNIX_LOCAL:# from stateUNIX_REMOTE:# to stateexecute_command:ssh# via commandcommand_params:# with paramsexpected_prompt:demo@host:test.rebex.netlogin:demopassword:passwordset_timeout:False# remote doesn't support: export TMOUT
我们的配置中有远程机器。让我们检查一下是否有"readme.txt"文件 在那台机器上（还有一些关于文件的信息）：
remote_unix=DeviceFactory.get_device(name='RebexTestMachine')# it starts in local shellremote_unix.goto_state(state="UNIX_REMOTE")# make it go to remote shellls_cmd=remote_unix.get_cmd(cmd_name="ls",cmd_params={"options":"-l"})remote_files=ls_cmd()if'readme.txt'inremote_files['files']:print("readme.txt file:")readme_file_info=remote_files['files']['readme.txt']forattrinreadme_file_info:print(" {:<18}: {}".format(attr,readme_file_info[attr]))
正如您可能注意到的，设备是状态机。状态转换在内部定义 配置文件位于
connection\u hops下。请注意，只有配置文件 知道"我需要使用ssh进行远程访问"-客户机代码只是说"转到远程"。 因此，您可以在不更改主代码的情况下交换"How to reach remote"。上面的代码显示：
readme.txt file: permissions : -rw------- hard_links_count : 1 owner : demo group : users size_raw : 403 size_bytes : 403 date : Apr 082014 name : readme.txt并行做多件事怎么样？让我们ping google 在询问test.rebex.net关于readme.txt文件的信息时：
my_unix=DeviceFactory.get_device(name='MyMachine')host='www.google.com'ping_cmd=my_unix.get_cmd(cmd_name="ping",cmd_params={"destination":host,"options":"-w 6"})remote_unix=DeviceFactory.get_device(name='RebexTestMachine')remote_unix.goto_state(state="UNIX_REMOTE")ls_cmd=remote_unix.get_cmd(cmd_name="ls",cmd_params={"options":"-l"})print("Start pinging {} ...".format(host))ping_cmd.start()# run command in backgroundprint("Let's check readme.txt at {} while pinging {} ...".format(remote_unix.name,host))remote_files=ls_cmd()# foreground "run in the meantime"file_info=remote_files['files']['readme.txt']print("readme.txt file: owner={fi[owner]}, size={fi[size_bytes]}".format(fi=file_info))ping_stats=ping_cmd.await_done(timeout=6)# await background commandprint("ping {}: {}={}, {}={} [{}]".format(host,'packet_loss',ping_stats['packet_loss'],'time_avg',ping_stats['time_avg'],ping_stats['time_unit']))
Start pinging www.google.com ... Let's check readme.txt at RebexTestMachine while pinging www.google.com ... readme.txt file: owner=demo, size=403 ping www.google.com: packet_loss=0, time_avg=35.251 [ms]
除了可调用的命令对象作为"未来"工作（结果承诺）。 您可以在后台启动它，然后等待它完成以获取结果。
如果我们使用日志相关信息增强配置：
LOGGER:PATH:./logsDATE_FORMAT:"%H:%M:%S"
然后上述代码将自动创建molers的主日志（
moler.log） 显示所有设备上的活动：22:30:19.723 INFO moler |More logs in: ./logs 22:30:19.747 INFO MyMachine |Connection to: 'MyMachine' has been opened. 22:30:19.748 INFO MyMachine |Changed state from 'NOT_CONNECTED' into 'UNIX_LOCAL' 22:30:19.866 INFO MyMachine |Event 'moler.events.unix.wait4prompt.Wait4prompt':'[re.compile('^moler_bash#')]' started. 22:30:19.901 INFO RebexTestMachine |Connection to: 'RebexTestMachine' has been opened. 22:30:19.901 INFO RebexTestMachine |Changed state from 'NOT_CONNECTED' into 'UNIX_LOCAL' 22:30:19.919 INFO RebexTestMachine |Event 'moler.events.unix.wait4prompt.Wait4prompt':'[re.compile('demo@')]' started. 22:30:19.920 INFO RebexTestMachine |Event 'moler.events.unix.wait4prompt.Wait4prompt':'[re.compile('^moler_bash#')]' started. 22:30:19.921 INFO RebexTestMachine |Command 'moler.cmd.unix.ssh.Ssh':'TERM=xterm-mono ssh -l demo test.rebex.net' started. 22:30:19.921 INFO RebexTestMachine |TERM=xterm-mono ssh -l demo test.rebex.net 22:30:20.763 INFO RebexTestMachine |********* 22:30:20.909 INFO RebexTestMachine |Changed state from 'UNIX_LOCAL' into 'UNIX_REMOTE' 22:30:20.917 INFO RebexTestMachine |Command 'moler.cmd.unix.ssh.Ssh' finished. 22:30:20.919 INFO MyMachine |Command 'moler.cmd.unix.ping.Ping':'ping www.google.com -w 6' started. 22:30:20.920 INFO MyMachine |ping www.google.com -w 6 22:30:20.920 INFO RebexTestMachine |Command 'moler.cmd.unix.ls.Ls':'ls -l' started. 22:30:20.922 INFO RebexTestMachine |ls -l 22:30:20.985 INFO RebexTestMachine |Command 'moler.cmd.unix.ls.Ls' finished. 22:30:26.968 INFO MyMachine |Command 'moler.cmd.unix.ping.Ping' finished. 22:30:26.992 INFO RebexTestMachine |Event 'moler.events.unix.wait4prompt.Wait4prompt': '[re.compile('^moler_bash#')]' finished. 22:30:27.011 INFO RebexTestMachine |Event 'moler.events.unix.wait4prompt.Wait4prompt': '[re.compile('demo@')]' finished. 22:30:27.032 INFO MyMachine |Event 'moler.events.unix.wait4prompt.Wait4prompt': '[re.compile('^moler_bash#')]' finished.正如您可能注意到的，主日志从高级视图-数据显示代码进度 连接时不可见，只显示设备上运行的命令的活动。
如果您想详细了解每台设备上发生了什么，请将其保存在设备日志中。 moler为每个设备创建日志
moler.rebextestmachine.log：frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
0以及moler.mymachine.log
frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
1前面的例子要求设备创建命令。我们也可以自己创建命令 连接到操作：
frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
2请注意，连接是上下文管理器执行打开/关闭操作。
frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
3重复使用自由
库可以让您自由地重用哪个部分。我们是"只拿你需要的东西"的粉丝。
您可以使用配置文件或通过python调用进行配置。
frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
4您可以手动使用设备或创建命令
您可以自己建立连接：
frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
5您甚至可以安装自己的实现来代替每个连接类型的默认实现
API设计推理
命令的主要目标是它的使用简单：只需运行它并返回结果即可。
命令对调用方隐藏：
- 如何实现"运行"
- 如何获取要分析的输出数据
- 它如何解析数据
命令显示给调用方：
- 启动/停止或等待其完成的API
- 查询其结果或结果就绪性的api
命令的作用是期货和承诺
在开始之后，我们等待它的结果，它被解析出通常作为dict提供的命令输出。 运行该命令并分析其输出可能需要一些时间，因此直到该点结果计算尚未完成。
作为未来的命令
- 它通过连接在设备/shell上启动一些命令 （当未来函数开始执行时）
- 它解析通过这种连接传入的数据 （未来的功能正在处理）
- 它存储解析的结果 （未来函数在计算结果中得出结论）
- 它提供了返回结果的方法 （正如future函数通过"return"或"yield"语句所做的那样）
- 结果还没准备好"刚刚"调用命令 （与功能相比，它是未来的）
所以命令应该有未来的api。
引自卢克·斯内林格的《专业蟒蛇》：
< Buff行情>未来是一个独立的对象。它独立于正在运行的实际函数。 它只存储状态和结果信息。
命令的不同之处在于两者都是：
- 类似于执行计算的对象的函数
- 存储计算结果的类似未来的对象。
命令与连接观察员
命令只是连接观察员的"活动版本"。
连接观察者是被动的，因为它只是观察一些数据的连接； 可能只是异步出现的数据（警报、重新启动或任何您想要的）。 这里的意图是责任划分：一个观察者在寻找警报， 另一个用于重新启动。
命令是活动的，因为它在连接时会主动触发一些输出 通过该连接发送命令字符串。所以，它激活了一些动作 在连接后面的设备上。这个动作在设备术语中是"命令"。 就像bash控制台/设备上的ping。它产生"命令"输出。 这个输出是moler作为连接观察者的命令所寻找的。
最著名的python的未来
<表><广告>API concurrent.futures.future 异步。未来 存储结果 ：白色复选标记： 设置结果（）：白色复选标记： 设置结果（）结果检索 ：白色复选标记： result（）：白色复选标记： result（）存储失败原因 ：白色复选标记： 设置异常（）：白色复选标记： 设置异常（）故障原因检索 ：白色复选标记： exception（）：白色复选标记： exception（）停止 ：白色复选标记： cancel（）：白色复选标记： cancel（）检查是否停止 ：白色复选标记： 已取消（）：白色复选标记： 已取消（）检查是否正在运行 ：白色复选标记： running（）：没有输入符号： （但是abstractEventLoop.running（））检查是否完成 ：白色复选标记： done（）：白色复选标记： done（）订阅完成 ：白色复选标记： add\u done\u callback（）：白色复选标记： add\u done\u callback（）取消订阅完成 ：无输入符号： ：白色复选标记： 删除已完成的回调函数启动可调用以"作为未来"运行是由未来对象外部的实体完成的
<表><广告><> API < /广告><正文>concurrent.futures
通过执行器对象（线程/进程）启动异步
通过模块LVL功能或EV回路启动开始可调用 提交（fn，*args，**kwargs）
可调用的计划执行为
fn（*args**kwargs）->；未来确保未来（coro_或_future）->；任务
future=运行协同程序线程安全（coro，循环）在数据迭代器上启动相同的可调用 映射（fn，*iterables，超时）->；迭代器 join_future=asyncio.gather（*map（f，iterable））
循环。运行直至完成（join_future）等待未来的完成是由未来对象外部的实体完成的
<表><广告><> API < /广告><正文>并发.未来
等待模块级功能异步
等待模块LVL功能或EV循环等待完成 完成，不完成=等待（未来，超时）->；未来 完成，未完成=等待等待等待（未来）
结果=等待收集（未来）
结果=等待未来
结果=从未来获得收益
结果=等待协程
结果=从协程获得收益
结果=从等待获得收益（未来，超时）
循环。运行直至完成（未来）->；阻塞运行完成时进行处理 对于"完成"状态（期货，超时）->；期货 对于"完成"状态（期货，超时）->；期货 命令的基本差异
与concurrent.futures和asyncio相反，我们不希望命令由某个外部实体运行。 为了使用简单，我们希望它是自执行的。 我们想接受命令，然后对它说：
- "运行"或"后台运行"
- 而不是"嗨，外部运行程序，你能为我运行/运行后台命令吗"
设计API
- 创建命令对象 < > >
- 同步运行/blocking并一次性获得结果的行为类似于函数调用，因为命令是可调用的。 < > >
- 异步/非阻塞运行 < > >
- 从背景切换到前景 < > >
- 命令"hi command，这就是我想要你做的"（上面的api说"hi you there，这就是我想要你用命令做的事情"）是"内部的"
- 它直接（python的禅）显示了我们在等待什么
- timeout是必需的参数（而不是concurrent.futures中的参数），因为我们不希望命令无休止地执行（用户必须知道等待命令完成的最坏情况超时是什么）
frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
6run as callable具有很大的优势，因为它非常适合python生态系统。
frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
7功能示例：
frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
8frommoler.configimportload_configfrommoler.device.deviceimportDeviceFactoryload_config(config='my_devices.yml')# description of available devicesmy_unix=DeviceFactory.get_device(name='MyMachine')# take specific device out of available onesps_cmd=my_unix.get_cmd(cmd_name="ps",# take command of that devicecmd_params={"options":"-ef"})processes_info=ps_cmd()# run the command, it returns resultforproc_infoinprocesses_info:if'python'inproc_info['CMD']:print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
9异步：变量看起来像：
PID: 1817 CMD: /usr/bin/python /usr/share/system-config-printer/applet.py PID: 21825 CMD: /usr/bin/python /home/gl/moler/examples/command/unix_ps.py
0并且并发。期货变量看起来像：
PID: 1817 CMD: /usr/bin/python /usr/share/system-config-printer/applet.py PID: 21825 CMD: /usr/bin/python /home/gl/moler/examples/command/unix_ps.py
1moler的api映射到上述众所周知的api
PID: 1817 CMD: /usr/bin/python /usr/share/system-config-printer/applet.py PID: 21825 CMD: /usr/bin/python /home/gl/moler/examples/command/unix_ps.py
2欢迎加入QQ群-->： 979659372
推荐PyPI第三方库
- 命令作为自立对象
