如何用pytest配合BAC0库测试BACnet UDP楼宇自动化应用?
关于pytest与BAC0配合测试BACnet服务器的问题
背景
我从未使用过pytest,想确认能否将其与Python BACnet库BAC0配合使用,是否有样板脚本,以及我的测试思路是否可行。
我开发了一个BACnet服务器应用,包含1个可发现的BACnet(UDP协议)点位用于需求响应信号级别,代码如下:
def make_bacnet_app(): def check_event_status(): adr_sig_object = bacnet.this_application.get_object_name('ADR-Event-Level') adr_sig_object.presentValue = Real(ven_client.bacnet_payload_value) # create discoverable BACnet object _new_objects = analog_value( name='ADR-Event-Level', description='SIMPLE SIGNAL demand response level', presentValue=0,is_commandable=False ) # create BACnet app bacnet = BAC0.lite() _new_objects.add_objects_to_application(bacnet) print('BACnet APP Created Success!') bacnet_sig_handle = RecurringTask(check_event_status,delay=5) bacnet_sig_handle.start() return bacnet if __name__ == "__main__": print("Starting main loop") t2 = threading.Thread( target=lambda: make_bacnet_app()) t2.setDaemon(True) t2.start()
通过发送GET请求到http://192.168.0.100:8080/trigger/1可触发该BACnet点位值变化,且已通过BACnet扫描工具验证功能正常。另外,服务器与客户端代码不能在同一设备运行,否则会因BACnet端口占用报错。
测试思路
希望用pytest完成以下测试步骤:
- 在远程机器启动
server.py(可自行通过SSH操作) - 在远程机器启动客户端BAC0脚本(可自行通过SSH操作)
- 向服务器发送GET请求
- 用BAC0读取BACnet点位,验证值为0
- 等待100秒
- 再次读取点位,验证值为1
- 等待100秒
现有客户端测试代码
import BAC0 bacnet = BAC0.lite() reading = bacnet.read("192.168.0.100 analogValue 0 presentValue") print("The signal is: ",reading)
测试建议
1. pytest与BAC0的兼容性
完全可以配合使用。pytest负责用例组织、断言和执行流程管控,BAC0负责BACnet协议的读写操作,二者功能互补,没有冲突。
2. 测试思路的可行性
整体思路可行,但有两处细节需要调整:
- 步骤2中的“客户端BAC0脚本”应改为测试端的BACnet读取程序(而非服务器端关联的客户端),因为服务器和测试客户端不能同设备运行,测试脚本需在独立机器上执行,通过SSH启动远程服务器后,在测试机上用BAC0读取远程点位。
- 步骤7的“等待100秒”若无后续验证动作,可考虑移除,或明确等待后的测试目标。
3. 样板测试脚本
以下是整合SSH操作、HTTP请求、BAC0读取的pytest样板脚本,需提前安装依赖:pip install pytest paramiko requests BAC0
import pytest import paramiko import requests import time import BAC0 from contextlib import contextmanager # 远程服务器配置 REMOTE_HOST = "192.168.0.100" REMOTE_USER = "your_ssh_user" REMOTE_PASSWORD = "your_ssh_password" SERVER_PATH = "/path/to/server.py" @contextmanager def remote_ssh_session(): """创建SSH会话上下文,自动关闭连接""" ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(REMOTE_HOST, username=REMOTE_USER, password=REMOTE_PASSWORD) try: yield ssh finally: ssh.close() def start_remote_server(ssh): """启动远程BACnet服务器,后台运行避免SSH断开后进程终止""" command = f"nohup python3 {SERVER_PATH} > server.log 2>&1 &" ssh.exec_command(command) # 等待服务器初始化完成 time.sleep(5) print("远程服务器启动完成") def stop_remote_server(ssh): """停止远程BACnet服务器,测试后清理进程""" command = "pkill -f server.py" ssh.exec_command(command) print("远程服务器已停止") @pytest.fixture(scope="module") def setup_environment(): """测试环境初始化:启动远程服务器,创建BAC0客户端""" with remote_ssh_session() as ssh: start_remote_server(ssh) # 测试机上初始化BAC0客户端,避免端口冲突 bacnet_client = BAC0.lite() yield bacnet_client, ssh # 测试结束后清理资源 stop_remote_server(ssh) bacnet_client.disconnect() def test_bacnet_point_change(setup_environment): bacnet_client, ssh = setup_environment server_ip = REMOTE_HOST # 发送GET请求触发点位变化 requests.get(f"http://{server_ip}:8080/trigger/1") # 读取点位,验证初始值为0 reading = bacnet_client.read(f"{server_ip} analogValue 0 presentValue") assert reading == 0, f"初始值验证失败,读取值为{reading}" # 等待100秒 time.sleep(100) # 再次读取,验证值为1 reading = bacnet_client.read(f"{server_ip} analogValue 0 presentValue") assert reading == 1, f"触发后值验证失败,读取值为{reading}" # 等待100秒(若有后续验证需求可补充逻辑) time.sleep(100)
4. 注意事项
- 远程进程管理:用
nohup后台启动服务器,避免SSH会话断开导致进程终止;测试结束后用pkill清理进程,避免残留。 - 端口冲突规避:若测试机上有其他BACnet服务占用默认端口(47808),可初始化BAC0时指定其他端口:
BAC0.lite(port=47809)。 - 等待时间优化:100秒等待可根据实际业务逻辑调整,也可改用轮询方式(每隔几秒读取一次,直到值变为1或超时),提升测试效率。
- SSH安全优化:建议使用SSH密钥认证替代密码,避免密码硬编码,修改
paramiko连接方式为密钥登录即可。 - 日志排查:远程服务器启动异常时,可通过
cat /path/to/server.log查看日志定位问题。
内容的提问来源于stack exchange,提问作者bbartling
相关产品推荐
相关产品推荐

