You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.06 10:05:38