背景

作为一个中间件的测试工程师，如何对于中间件提供的命令进行自动化的回归，这一直是一个难题，市面上好像缺乏了对于命令进行自动化回归的合理解决方案。
常见方式有下面两种：

• 直接写字符串的命令，然后使用各种编程语言的SSH库进行连接，然后执行命令字符串，获取执行结果，如果需要对传入的参数进行校验，还需要各种提取字符串，或者设置各种命令变量，比较繁琐
• 编写命令对象，通过操作命令对象set/get方式，生成命令字符串给SSH库进行命令执行，如果需要对传入的参数进行校验，只需要通过命令对象get的方式进行获取，当然缺点就是，每次都需要进行大量的工作去编写命令对象，重复性劳动

通过对比之后，会发现，使用第二种方式更加直观、简洁，提取响应数据也更加方便，但是第二种方式对于大量重复性的去编写命令对象的方式，也让人十分恶心。如果有一个工具能够帮助大家去生成命令对象，然后通过调用命令对象执行请求响应的方式就能让大家从这种困境中走出来。

所以引入了cmdlinker这个开源python库，cmdlinker能够帮助我们做什么？

• 通过编写命令对象yaml的方式，生成命令对象，无需手动进行命令对象代码的编写
• 命令对象自带runner，可以直接获取命令执行的请求响应
• 支持本地shell执行及远程ssh执行的方式

安装

pip3 install cmdlinker

文档地址

https://github.com/chineseluo/cmdlinker

案例演示：

使用docker命令举例，查询容器，获取容器配置信息，断言容器的ID是否包含请求的ID
编写yaml文件：

entry: "docker"
mapping_entry: "Docker"
module_name: "docker"
class_name: "Docker"
out_path: "./"
mode: "SSH"
parameters:- mapping_name: "ps"original_cmd: "ps"value: False # 是否需要值mutex: False # 是否互斥default: None- mapping_name: "inspect"original_cmd: "inspect"value: True # 是否需要值mutex: False # 是否互斥default: None- mapping_name: "f"original_cmd: "-f"value: True # 是否需要值mutex: False # 是否互斥default: None

执行生成命令:

CmdLinker init -f ./docker.yaml

生成python命令对象

from loguru import logger
from typing import Union, Text
from cmdlinker.builtin.exception import CmdLinkerMutexException, CmdLinkerMulMutexException
from cmdlinker.builtin.ssh_utils import SSHClient
from cmdlinker.builtin.shell_utils import ShellClientclass Cmds:def __init__(self):self.index = 0self.CMD_LIST = []class Runner:def __init__(self):self._mutexs = []self._gcs = []self._not_mutexs = []self.pre: Runner = objectself.root: Runner = objectself.next: Union[Runner] = objectself.main_cmd: Text = Noneself.cmds: Cmds = Noneself.ssh_client: SSHClient = Noneself.sudo: bool = Trueself.timeout: int = 20def _exclude(self, cmd_obj):if not cmd_obj.mark:logger.debug(f"命令对象：{cmd_obj.__class__.__name__}未被使用")return Falseelse:return Truedef _get_execute_cmd(self, cmd_obj):if cmd_obj.need_value:return f"{cmd_obj.meta_data['original_cmd']} {cmd_obj.value}"else:return f"{cmd_obj.meta_data['original_cmd']}"def _get_log_desc(self):if self.pre == self.root:log_desc = "子"elif self.pre is None and self.root == self:log_desc = "根"else:log_desc = "父"return log_descdef cmd_checker(self):log_desc = self._get_log_desc()logger.info("==" * 20 + f"开启{log_desc}命令{self.__class__.__name__}合法性检查" + "==" * 20)for cmd_obj_str in vars(self).keys():cmd_obj = getattr(self, cmd_obj_str)if isinstance(cmd_obj, BaseCmd):if cmd_obj_str == "next":continueif not self._exclude(cmd_obj):continueif cmd_obj.mutex:self._mutexs.append(cmd_obj)logger.debug(f"{self.main_cmd}添加互斥对象：{cmd_obj},最新互斥对象列表：")else:self._not_mutexs.append(cmd_obj)logger.debug(f"{self.main_cmd}添加非互斥对象：{cmd_obj},最新非互斥对象列表：{self._not_mutexs}")if len(self._mutexs) > 1:raise CmdLinkerMulMutexException(self.__class__.__name__, self._mutexs)if len(self._mutexs) != 0 and len(self._not_mutexs) != 0:raise CmdLinkerMutexException(self.__class__.__name__, self._mutexs, self._not_mutexs)if self.pre == self.root:logger.debug(f"启用命令对象{self.__class__.__name__} 根命令: {self.root.__class__.__name__} 检查")self.pre.cmd_checker()elif self.pre is None and self.root == self:passelse:logger.debug(f"启用命令对象{self.__class__.__name__}父命令: {self.pre.__class__.__name__} 检查")self.pre.cmd_checker()self._not_mutexs = []self._mutexs = []logger.info("==" * 20 + f"{log_desc}命令{self.__class__.__name__}合法性检查通过" + "==" * 20)def runner(self):self.cmd_checker()cmd = self.exec_cmd()self.cmds.CMD_LIST = []return self.ssh_client.run_cmd(cmd, timeout=self.timeout)def collector(self):return self.cmds.CMD_LIST.sort(key=lambda cmd_obj: cmd_obj.index)def exec_cmd(self):self.cmds.CMD_LIST.sort(key=lambda cmd_obj: cmd_obj.index)cmd_list = [self.main_cmd] + [self._get_execute_cmd(cmd) for cmd in self.cmds.CMD_LIST]if self.sudo:cmd_list.insert(0, "sudo")logger.info(f"执行命令列表：{cmd_list}")cmd = " ".join(cmd_list)return cmdclass BaseCmd(Runner):def __init__(self):super().__init__()self.cmds: Cmds = Noneself.meta_data = Noneself.mutex = Noneself.need_value = Noneself.has_child_cmd = Noneself.gc = Noneself.child_cmd = Noneself.default_value = Noneself.value = Noneself.mark = Noneself.index = Noneself.level = 0self._mutexs = []self._gcs = []self._not_mutexs = []self.main_cmd = "docker"self.pre: object = objectself.root: object = objectself.next: Union[object] = objectself.ssh_client: SSHClient = Noneclass Ps(BaseCmd):def __init__(self, root_obj, pre_obj):super().__init__()self.pre: Docker = pre_objself.root: Docker = root_objself.next: Union[Ps] = selfself.meta_data = {'mapping_name': 'ps', 'original_cmd': 'ps', 'value': False, 'mutex': False, 'default': 'None','has_child_cmd': False, 'child_cmds': [], 'parent_cmd': 'Docker', 'root_cmd': 'Docker'}self.level = 2self.mutex = Falseself.need_value = Falseself.has_child_cmd = Falseself.child_cmds = []self.gc = Falseself.default_value = "None"self.value = self.default_valueself.mark = Falseclass Inspect(BaseCmd):def __init__(self, root_obj, pre_obj):super().__init__()self.pre: Docker = pre_objself.root: Docker = root_objself.next: Union[Inspect] = selfself.meta_data = {'mapping_name': 'inspect', 'original_cmd': 'inspect', 'value': True, 'mutex': False,'default': 'None', 'has_child_cmd': False, 'child_cmds': [], 'parent_cmd': 'Docker','root_cmd': 'Docker'}self.level = 2self.mutex = Falseself.need_value = Trueself.has_child_cmd = Falseself.child_cmds = []self.gc = Falseself.default_value = "None"self.value = self.default_valueself.mark = Falseclass Format(BaseCmd):def __init__(self, root_obj, pre_obj):super().__init__()self.pre: Docker = pre_objself.root: Docker = root_objself.next: Union[Format] = selfself.meta_data = {'mapping_name': 'format', 'original_cmd': '--format', 'value': True, 'mutex': False,'default': 'None', 'has_child_cmd': False, 'child_cmds': [], 'parent_cmd': 'Docker','root_cmd': 'Docker'}self.level = 2self.mutex = Falseself.need_value = Trueself.has_child_cmd = Falseself.child_cmds = []self.gc = Falseself.default_value = "None"self.value = self.default_valueself.mark = Falseclass Docker(Runner):def __init__(self, host=None, name=None, pwd=None, port="22", timeout="60", sudo=False):super().__init__()self.cmds: Cmds = Cmds()self.pre: object = Noneself.root: Docker = selfself.next: Union[Ps, Inspect, Format,] = Noneself.main_cmd = "docker"self._mutexs = []self._gcs = []self._not_mutexs = []self.mode = "SSH"self.ssh_client = SSHClient(host, name, pwd, port)self._ps: Ps = Ps(self, self)self._inspect: Inspect = Inspect(self, self)self._format: Format = Format(self, self)def _set_ps(self):self._ps.mark = Trueself._ps.index = self.cmds.indexself.cmds.index += 1self.next = self._psself.cmds.CMD_LIST.append(self._ps)self._ps.cmds = self.cmdsself._ps.ssh_client = self.ssh_clientdef tset_ps(self):"""传递TRANSMIT模式，可以获取子命令对象，可通过，root/pre/next，控制命令对象的根/父/子级"""self._set_ps()return self._psdef hset_ps(self, ):"""保持HOLD模式，该方法返回该对象本身，不返回子对象"""self._set_ps()return selfdef _set_inspect(self, value=None):self._inspect.mark = Trueself._inspect.index = self.cmds.indexself.cmds.index += 1self.next = self._inspectself.cmds.CMD_LIST.append(self._inspect)self._inspect.cmds = self.cmdsself._inspect.ssh_client = self.ssh_clientif self._inspect.need_value:self._inspect.value = valuedef tset_inspect(self, value=None):"""传递TRANSMIT模式，可以获取子命令对象，可通过，root/pre/next，控制命令对象的根/父/子级"""self._set_inspect(value=value)return self._inspectdef hset_inspect(self, value=None):"""保持HOLD模式，该方法返回该对象本身，不返回子对象"""self._set_inspect(value=value)return selfdef _set_format(self, value=None):self._format.mark = Trueself._format.index = self.cmds.indexself.cmds.index += 1self.next = self._formatself.cmds.CMD_LIST.append(self._format)self._format.cmds = self.cmdsself._format.ssh_client = self.ssh_clientif self._format.need_value:self._format.value = valuedef tset_format(self, value=None):"""传递TRANSMIT模式，可以获取子命令对象，可通过，root/pre/next，控制命令对象的根/父/子级"""self._set_format(value=value)return self._formatdef hset_format(self, value=None):"""保持HOLD模式，该方法返回该对象本身，不返回子对象"""self._set_format(value=value)return selfdef ps(self) -> Ps:return self._psdef inspect(self) -> Inspect:return self._inspectdef format(self) -> Format:return self._format

执行测试

进行请求响应断言（PS:自行替换下面的host/name/pwd/port等参数验证）

if __name__ == '__main__':docker = Docker(host="192.168.1.2", name="root", pwd="xxxx")ps_res = docker.hset_ps().hset_format("{{.ID}}").runner()inspect_res = docker.hset_inspect(ps_res["stdout"]).runner()assert ps_res["stdout"].strip() in inspect_res["stdout"][0]["Id"]

检查日志

C:\Users\Dell\AppData\Local\Programs\Python\Python39\python.exe E:\开源项目\CmdLinker\tests\docker.py 
2025-01-03 16:38:03.116 | INFO     | __main__:cmd_checker:53 - ========================================开启根命令Docker合法性检查========================================
2025-01-03 16:38:03.116 | DEBUG    | __main__:cmd_checker:66 - docker添加非互斥对象：<__main__.Ps object at 0x000001ADA171E9A0>,最新非互斥对象列表：[<__main__.Ps object at 0x000001ADA171E9A0>]
2025-01-03 16:38:03.116 | DEBUG    | __main__:_exclude:31 - 命令对象：Inspect未被使用
2025-01-03 16:38:03.116 | DEBUG    | __main__:cmd_checker:66 - docker添加非互斥对象：<__main__.Format object at 0x000001ADA171EAF0>,最新非互斥对象列表：[<__main__.Ps object at 0x000001ADA171E9A0>, <__main__.Format object at 0x000001ADA171EAF0>]
2025-01-03 16:38:03.116 | INFO     | __main__:cmd_checker:81 - ========================================根命令Docker合法性检查通过========================================
2025-01-03 16:38:03.116 | INFO     | __main__:exec_cmd:97 - 执行命令列表：['sudo', 'docker', 'ps', '--format {{.ID}}']
2025-01-03 16:38:03.693 | INFO     | cmdlinker.builtin.logger_operation:console_output:18 - 
================== cmd request  details ==================
execute cmd on host : 192.168.1.2
execute cmd : sudo docker ps --format {{.ID}}2025-01-03 16:38:03.694 | WARNING  | cmdlinker.builtin.ssh_utils:run_cmd:67 - 尝试对response stdout进行json转换失败，原样输出
2025-01-03 16:38:03.694 | INFO     | cmdlinker.builtin.logger_operation:console_output:18 - 
================== cmd response  details ==================
status_code : 0
execute_time : 578ms
stdout   : 7393aa6334e6stderr   : 2025-01-03 16:38:03.694 | INFO     | __main__:cmd_checker:53 - ========================================开启根命令Docker合法性检查========================================
2025-01-03 16:38:03.694 | DEBUG    | __main__:cmd_checker:66 - docker添加非互斥对象：<__main__.Ps object at 0x000001ADA171E9A0>,最新非互斥对象列表：[<__main__.Ps object at 0x000001ADA171E9A0>]
2025-01-03 16:38:03.694 | DEBUG    | __main__:cmd_checker:66 - docker添加非互斥对象：<__main__.Inspect object at 0x000001ADA171EA60>,最新非互斥对象列表：[<__main__.Ps object at 0x000001ADA171E9A0>, <__main__.Inspect object at 0x000001ADA171EA60>]
2025-01-03 16:38:03.694 | DEBUG    | __main__:cmd_checker:66 - docker添加非互斥对象：<__main__.Format object at 0x000001ADA171EAF0>,最新非互斥对象列表：[<__main__.Ps object at 0x000001ADA171E9A0>, <__main__.Inspect object at 0x000001ADA171EA60>, <__main__.Format object at 0x000001ADA171EAF0>]
2025-01-03 16:38:03.694 | INFO     | __main__:cmd_checker:81 - ========================================根命令Docker合法性检查通过========================================
2025-01-03 16:38:03.694 | INFO     | __main__:exec_cmd:97 - 执行命令列表：['sudo', 'docker', 'inspect 7393aa6334e6\n']
2025-01-03 16:38:03.922 | INFO     | cmdlinker.builtin.logger_operation:console_output:18 - 
================== cmd request  details ==================
execute cmd on host : 47.97.37.176
execute cmd : sudo docker inspect 7393aa6334e62025-01-03 16:38:03.922 | INFO     | cmdlinker.builtin.logger_operation:console_output:18 - 
================== cmd response  details ==================
status_code : 0
execute_time : 228ms
stdout   : [{"Id": "7393aa6334e6a091aae54790f5a6e3505f6dbab9515c5643978afbdc221975dd","Created": "2024-06-26T07:04:49.182101334Z","Path": "docker-entrypoint.sh","Args": ["--requirepass","yourpassword"],"State": {"Status": "running","Running": true,"Paused": false,"Restarting": false,"OOMKilled": false,"Dead": false,"Pid": 27544,"ExitCode": 0,"Error": "","StartedAt": "2024-06-26T07:04:49.54446658Z","FinishedAt": "0001-01-01T00:00:00Z"},"Image": "sha256:7614ae9453d1d87e740a2056257a6de7135c84037c367e1fffa92ae922784631","ResolvConfPath": "/var/lib/docker/containers/7393aa6334e6a091aae54790f5a6e3505f6dbab9515c5643978afbdc221975dd/resolv.conf","HostnamePath": "/var/lib/docker/containers/7393aa6334e6a091aae54790f5a6e3505f6dbab9515c5643978afbdc221975dd/hostname","HostsPath": "/var/lib/docker/containers/7393aa6334e6a091aae54790f5a6e3505f6dbab9515c5643978afbdc221975dd/hosts","LogPath": "/var/lib/docker/containers/7393aa6334e6a091aae54790f5a6e3505f6dbab9515c5643978afbdc221975dd/7393aa6334e6a091aae54790f5a6e3505f6dbab9515c5643978afbdc221975dd-json.log","Name": "/some-redis","RestartCount": 0,"Driver": "overlay2","Platform": "linux","MountLabel": "","ProcessLabel": "","AppArmorProfile": "","ExecIDs": null,"HostConfig": {"Binds": null,"ContainerIDFile": "","LogConfig": {"Type": "json-file","Config": {}},"NetworkMode": "default","PortBindings": {"6379/tcp": [{"HostIp": "","HostPort": "6379"}]},"RestartPolicy": {"Name": "no","MaximumRetryCount": 0},"AutoRemove": false,"VolumeDriver": "","VolumesFrom": null,"CapAdd": null,"CapDrop": null,"CgroupnsMode": "host","Dns": [],"DnsOptions": [],"DnsSearch": [],"ExtraHosts": null,"GroupAdd": null,"IpcMode": "private","Cgroup": "","Links": null,"OomScoreAdj": 0,"PidMode": "","Privileged": false,"PublishAllPorts": false,"ReadonlyRootfs": false,"SecurityOpt": null,"UTSMode": "","UsernsMode": "","ShmSize": 67108864,"Runtime": "runc","ConsoleSize": [0,0],"Isolation": "","CpuShares": 0,"Memory": 0,"NanoCpus": 0,"CgroupParent": "","BlkioWeight": 0,"BlkioWeightDevice": [],"BlkioDeviceReadBps": null,"BlkioDeviceWriteBps": null,"BlkioDeviceReadIOps": null,"BlkioDeviceWriteIOps": null,"CpuPeriod": 0,"CpuQuota": 0,"CpuRealtimePeriod": 0,"CpuRealtimeRuntime": 0,"CpusetCpus": "","CpusetMems": "","Devices": [],"DeviceCgroupRules": null,"DeviceRequests": null,"KernelMemory": 0,"KernelMemoryTCP": 0,"MemoryReservation": 0,"MemorySwap": 0,"MemorySwappiness": null,"OomKillDisable": false,"PidsLimit": null,"Ulimits": null,"CpuCount": 0,"CpuPercent": 0,"IOMaximumIOps": 0,"IOMaximumBandwidth": 0,"MaskedPaths": ["/proc/asound","/proc/acpi","/proc/kcore","/proc/keys","/proc/latency_stats","/proc/timer_list","/proc/timer_stats","/proc/sched_debug","/proc/scsi","/sys/firmware"],"ReadonlyPaths": ["/proc/bus","/proc/fs","/proc/irq","/proc/sys","/proc/sysrq-trigger"]},"GraphDriver": {"Data": {"LowerDir": "/var/lib/docker/overlay2/da926ce47de0e966f295ddf249be649da1f2c1ea162be28fb84a5c5d28db3d96-init/diff:/var/lib/docker/overlay2/cb066f8e71551530105d92d22aee4ac7d10beeff480389676e102be0d701e534/diff:/var/lib/docker/overlay2/b69aadb0a11079daa266a36401ba0ce147450516b500c645eff85176f8e7919b/diff:/var/lib/docker/overlay2/a57e79ac2a99e28c990210b22b831dcd8fad2c98d810de241fc8a3d99b64953a/diff:/var/lib/docker/overlay2/482ff82ce4a7763a75994358255c4e604abab9fcb77fd1f5f2d815a0a98a323b/diff:/var/lib/docker/overlay2/8e951ea76637a9804eb44f89eb8353a5a71da11d304056b22e2d1886e7da9c2d/diff:/var/lib/docker/overlay2/e88bc247f216bd19bd2329b2feeda9cc387df6241598b863b201cafe7ff0db3a/diff","MergedDir": "/var/lib/docker/overlay2/da926ce47de0e966f295ddf249be649da1f2c1ea162be28fb84a5c5d28db3d96/merged","UpperDir": "/var/lib/docker/overlay2/da926ce47de0e966f295ddf249be649da1f2c1ea162be28fb84a5c5d28db3d96/diff","WorkDir": "/var/lib/docker/overlay2/da926ce47de0e966f295ddf249be649da1f2c1ea162be28fb84a5c5d28db3d96/work"},"Name": "overlay2"},"Mounts": [{"Type": "volume","Name": "502c8ea157173ae65d24a2e0173c0e990e5ba1398013ad01a048e7a594a62608","Source": "/var/lib/docker/volumes/502c8ea157173ae65d24a2e0173c0e990e5ba1398013ad01a048e7a594a62608/_data","Destination": "/data","Driver": "local","Mode": "","RW": true,"Propagation": ""}],"Config": {"Hostname": "7393aa6334e6","Domainname": "","User": "","AttachStdin": false,"AttachStdout": false,"AttachStderr": false,"ExposedPorts": {"6379/tcp": {}},"Tty": false,"OpenStdin": false,"StdinOnce": false,"Env": ["PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin","GOSU_VERSION=1.12","REDIS_VERSION=6.2.6","REDIS_DOWNLOAD_URL=http://download.redis.io/releases/redis-6.2.6.tar.gz","REDIS_DOWNLOAD_SHA=5b2b8b7a50111ef395bf1c1d5be11e6e167ac018125055daa8b5c2317ae131ab"],"Cmd": ["--requirepass","yourpassword"],"Image": "redis","Volumes": {"/data": {}},"WorkingDir": "/data","Entrypoint": ["docker-entrypoint.sh"],"OnBuild": null,"Labels": {}},"NetworkSettings": {"Bridge": "","SandboxID": "b98966d73f26b520343cc69867d8ac56dd3ac65655375c2edb047cabc9eec9c8","HairpinMode": false,"LinkLocalIPv6Address": "","LinkLocalIPv6PrefixLen": 0,"Ports": {"6379/tcp": [{"HostIp": "0.0.0.0","HostPort": "6379"}]},"SandboxKey": "/var/run/docker/netns/b98966d73f26","SecondaryIPAddresses": null,"SecondaryIPv6Addresses": null,"EndpointID": "63a20510e954d1b81181d319b2d004e68918a2926d3db1e71e846197b79c72e1","Gateway": "172.17.0.1","GlobalIPv6Address": "","GlobalIPv6PrefixLen": 0,"IPAddress": "172.17.0.2","IPPrefixLen": 16,"IPv6Gateway": "","MacAddress": "02:42:ac:11:00:02","Networks": {"bridge": {"IPAMConfig": null,"Links": null,"Aliases": null,"NetworkID": "7fdaf78d05fd8a3d1c3c2ae0151022ae8c326582524327e18403fdab4a26f16d","EndpointID": "63a20510e954d1b81181d319b2d004e68918a2926d3db1e71e846197b79c72e1","Gateway": "172.17.0.1","IPAddress": "172.17.0.2","IPPrefixLen": 16,"IPv6Gateway": "","GlobalIPv6Address": "","GlobalIPv6PrefixLen": 0,"MacAddress": "02:42:ac:11:00:02","DriverOpts": null}}}}
]
stderr   : 
Process finished with exit code 0

我们可以很方便的使用返回的status_code/stdout/stderr进行请求响应断言

---

OpenSourceTest cmdlinker 社区
欢迎小伙伴加群，讨论cmdlinker相关问题，或提出优化建议！
QQ群（自动化测试-夜行者）：816489363