简介面向光伏监控系统测试人员、运维人员及需要对接台电DREAMS规范的云平台厂商这份测试工具包用于验证云端太阳能监控系统是否符合台电DREAMS要求。核心围绕DNP 3.0协议实现发电数据回传与控制指令交互可测试变流器实功、虚功、功率因数上限及VPset设定等调控能力重点校验数据上报链路与控制命令下发的正确性。工具适配Linux环境推荐Ubuntu 18.04需安装Docker与Docker Compose便于快速拉起测试环境压缩包共147个文件、大小仅351KB以JavaScript和JSON配置为主辅以C/HPP核心逻辑、Shell脚本、Dockerfile及SSL证书配置整体轻量且目录模块清晰。当前已有141人学习参考。内容包含主程序、SOE事件处理、消息发送器等关键模块源码并提供SSL/TLS加密通信配置、Nginx站点配置与示例证书方便测试人员在实际部署前完成协议兼容性验证、异常排查及参数调整尤其适合同步验证多电站场景下与台电DREAMS的对接行为。1. 验证云光伏监控系统对接 Tai Power DREAMS这个测试存储库解决的是验收的什么问题做光伏监控平台的对点测试久了你会发现一个规律大部分项目在功能演示时一切正常真正翻车的是并网验收那一关。电网侧的 DREAMS 系统不是拿你 PPT 上的功能清单来验收它只看通信规约、数据模型、控制响应和异常行为够不够格。你手里有没有一套能让测试人员照着跑、跑完就能出报告的验证资产决定了并网窗口是两周还是两个月。dreams-cloud-joint-tester 就是干这个的用存储库把 DREAMS 的验收要求翻译成可重复执行的测试用例、仿真网关和追溯报告让云光伏监控系统的验证从「人肉对点、现场排查」变成「用例驱动、一跑出报告」。适合正在做云光伏监控平台交付、对接 Tai Power DREAMS 并网要求却又苦于没有电网侧真实环境可用的测试人员、集成商交付工程师和对点工程师。2. DREAMS 需求拆解与验证矩阵先定测什么再写测试脚本很多测试存储库最后沦为「代码垃圾场」核心原因是一上来就写脚本没先回答一个问题DREAMS 到底在验收什么。如果你没有把需求拆开脚本写得再漂亮验收条目对不上依然是白干。这一章先把 DREAMS 侧的要求拆成四个需求域再谈验证矩阵怎么设计最后说存储库的资产构成。2.1 需求域数据模型、通信规约与控制行为Tai Power DREAMS 对云光伏监控系统的要求本质上可以归纳为「电网调度侧对分布式电源的可见性与可控性」。可见性是说调度侧必须实时知道每个光伏电站发了多少电、并网开关什么状态、有没有告警可控性是说调度侧下发的指令云侧必须接得住、执行得到位、还能把执行结果明确回给调度侧。围绕这两点需求通常落在四个域。第一个域是通信规约。DREAMS 侧对外提供标准的远动通道常见做法是用 IEC 60870-5-104 承载遥信遥测遥控用 IEC 61850 描述设备模型底层采集器对逆变器则常用 Modbus TCP。规约层验证的要点不是你用了什么协议而是报文格式、端口号、链路建立方式、心跳超时参数是否和调度侧约定一致。第二个域是数据模型包括点表映射、数据类型、缩放因子、死区阈值。光伏监控系统内部可能把有功功率存成 floatDREAMS 侧期望的是带缩放因子的整型不提前对齐数据一上送就错位。第三个域是控制行为包括遥控指令的确认机制、限功率调度的执行时限、指令超时重发规则。第四个域是异常行为包括通信断链后的重连、重复帧处理、时间戳校验、防孤岛动作逻辑。需求域典型验收点测试手段通信规约端口、心跳、报文类型、链路建立仿真网关抓报文比对格式数据模型点表映射、缩放因子、死区注入固定值校验上送结果控制行为指令确认、执行时限、重发机制下发指令断言响应时序异常行为断链重连、重复帧、时间戳故障注入观察恢复行为这四个域在实际验收中不会分开考通常是一条链路走到底建链、上送遥测、调度下发指令、云侧执行并回报、然后模拟链路中断看恢复。所以测试存储库不能只按协议栈组织还要按「链路场景」组织。2.2 验证矩阵把验收条目翻译成可执行用例我在几个交付项目里踩过同一个坑验收条目写在文档里测试用例写在代码里两边各说各话。到了现场业主问「第 4.2 条限功率功能验证了吗」你翻半天找不到对应脚本。后来我养成一个习惯先做一张验证矩阵表每条验收条目对应一个或多个用例 ID再组织代码。验证矩阵的列建议固定为条目编号、需求域、验收描述、用例 ID、前置条件、预期结果。比如调度侧要求「收到有功调节指令后 10 秒内开始执行60 秒内到达目标值」对应的用例就是test_derating_step_60pct前置条件是当前出力在 100%预期结果是 60 秒内遥测值进入目标值正负 2% 区间。这张表本身就是交付物测试报告里附上它验收人员一眼就能看出覆盖度。矩阵设计时有一个必须注意的点不要只按「功能正常」来设计用例。DREAMS 验收往往会包含边界和异常比如指令超时没收到执行回报、遥信抖动、功率调节回差。每一条正向验收条目至少要配一条异常路径用例。限功率功能正向用例验证 60% 指令能执行到位异常用例就得验证「指令下发后 5 秒内没收到确认云侧是否按约定重发」以及「重发超过上限是否报错」。验证矩阵不覆盖异常现场验收时基本会被打回。2.3 存储库的定位用例库、脚本库、仿真资产三位一体dreams-cloud-joint-tester 这种存储库在项目里的价值不是「一堆 pytest 文件」而是把三样东西沉淀下来。第一是可执行的用例库每个用例对应一个验收行为跑完自动断言第二是仿真资产包括 DREAMS 网关的模拟器、报文样本、点表配置让测试人员在没有真实调度侧环境时也能练手第三是追溯报告能把用例执行结果映射回验收条目。仿真资产这块最容易被低估。真实对接 DREAMS 的窗口期很短现场环境不是你想用就能用。测试人员如果在等待窗口期只能干瞪眼交付节奏就被卡死。所以我一般会在存储库里放一个轻量级的 DREAMS 网关模拟器云侧系统连它的端口它按约定的规约回应相当于把电网侧那个「黑匣子」搬到了本地。模拟器和真实系统的差异是客观存在的但没有模拟器连基本链路都跑不通更别提发现问题了。3. 搭建最小可跑通的验证环境仿真网关、目录结构与第一条用例不少测试人员拿到存储库第一反应是「先把所有脚本跑一遍」结果几十个用例一起失败根本没法定位问题。正确的姿势是先搭一个最小可跑通的环境从一条链路、一条用例开始确定「云侧能连上仿真网关、遥信能上送、断言能通过」再逐步叠加。这一章从零开始把环境立起来。3.1 目录结构让用例按验收条目组织而不是按脚本类型堆目录结构决定了这个存储库能不能长期维护。我见过最失败的布局是把所有脚本平铺在一个文件夹里文件名从test1.py到test30.py三个月后没人知道每个脚本在验什么。建议按「仿真资产、测试用例、配置、报告」分层用例目录内再按需求域拆分这样新增验收条目时能直接找到该改哪里。dreams-cloud-joint-tester/ ├── conftest.py ├── config/ │ ├── point_table.yaml # 点表映射云侧与DREAMS之间的遥信遥测点对应关系 │ └── simulator.yaml # 仿真网关的IP、端口、心跳、重连参数 ├── simulators/ │ └── dreams_gateway.py # 模拟DREAMS侧规约行为的网关仿真器 ├── testcases/ │ ├── test_telemetry.py # 遥信遥测上送类用例 │ ├── test_control.py # 遥控指令与确认类用例 │ ├── test_derating.py # 限功率调度类用例 │ └── test_abnormal.py # 断链重连、重复帧、异常响应类用例 ├── payloads/ │ ├── iec104_telemetry.xml # 遥测报文模板 │ └── iec104_command.xml # 遥控报文模板 ├── reports/ │ └── pytest_report.html # 测试报告与追溯矩阵输出目录 └── requirements/ └── acceptance_trace.csv # 验收条目与用例ID的关联矩阵conftest.py放共享 fixture比如仿真网关的启动和停止、测试数据集的加载。配置文件与代码分离是必须的现场联调时 IP、端口、点表几乎一定会变改配置比改脚本快得多也不容易引入代码错误。payloads目录放报文模板抓包排错时拿来直接比对省去翻文档的时间。3.2 仿真网关在没有真实电网侧环境时先跑通链路仿真网关是这套环境的地基。它的职责是模拟 DREAMS 侧的规约行为监听端口、接受云侧连接、接收上送数据、下发控制指令、按约定返回确认。最小实现不需要完整覆盖协议栈只需要做到「云侧能连上、能收发报文」即可。# simulators/dreams_gateway.py import socket import threading import time class DREAMSGwSimulator: 极简DREAMS仿真网关监听端口逐行接收报文并回确认 def __init__(self, host: str, port: int): self.host host self.port port self._server socket.socket(socket.AF_INET, socket.SOCK_STREAM) self._running False self.received_payloads [] # 记录收到的报文供断言查询 def start(self): self._server.bind((self.host, self.port)) self._server.listen(1) self._running True threading.Thread(targetself._serve, daemonTrue).start() def _serve(self): while self._running: conn, _ self._server.accept() threading.Thread(targetself._handle_conn, args(conn,), daemonTrue).start() def _handle_conn(self, conn): data conn.recv(1024) if data: self.received_payloads.append(data) # 回一个固定确认帧真实场景需要按规约内容解析后组装 conn.sendall(bACK) conn.close() def stop(self): self._running False self._server.close()这个仿真器故意做得极简目的是先跑通链路而不是模拟完整规约。received_payloads列表把收到的原始报文记下来测试用例可以直接断言「云侧上送的数据是否真的到达了网关侧」。注意_handle_conn里目前是收到什么就回一个通用确认真实项目中这里要替换成按规约解析请求、按点表组装响应后面讲到遥控指令时会看到更完整的版本。参数上只有主机和端口两个simulator.yaml里配一次即可如果你要仿真多路连接把listen(1)改成listen(5)并给每个连接维护独立线程。3.3 第一条用例从「上送遥信」开始链路打通后的第一条用例建议从最简单的遥信开始云侧把「并网断路器合闸」这个遥信点上送仿真网关侧断言收到了正确的点号和值。这个用例覆盖了建链、编码、传输、接收四条链路环节是后续一切用例的骨架。# testcases/test_telemetry.py import pytest from simulators.dreams_gateway import DREAMSGwSimulator pytest.fixture def gateway(): gw DREAMSGwSimulator(127.0.0.1, 2404) gw.start() yield gw gw.stop() pytest.fixture def cloud_client(gateway): # 模拟云侧监控平台主动连接仿真网关生产中此处是对接被测系统接口 from dreams_tester import PVGatewayClient client PVGatewayClient(127.0.0.1, 2404) client.connect() yield client client.close() def test_report_breaker_closed(gateway, cloud_client): # 步骤云侧上送遥信“并网断路器合闸”点号5001值1 seq cloud_client.send_telemetry(point5001, value1, quality0) # 断言网关侧在5秒内收到该点号且质量位正常 payload gateway.wait_for_point(5001, timeout5) assert payload is not None, 网关未收到点号5001的遥信 assert payload[value] 1 assert payload[quality] 0fixture保证了每个用例的网关环境是全新启动的避免上一个用例的连接残留影响下一个。send_telemetry在真实项目里会被替换成调用被测云光伏监控系统的接口或直接构造协议报文注入在最小环境里它只是往仿真网关写一条带点号、值、质量位的报文。wait_for_point轮询received_payloads里解析出的点表状态超时 5 秒就返回None断言失败时能明确看出是「数据没到」还是「数据错了」。跑用例的命令不用复杂pytest -v --tbshort --htmlreports/pytest_report.html-v把每条用例的结果打出来--tbshort只显示断言摘要--html生成给验收人员看的报告文件。第一次跑这条用例时重点不是看它通过而是看仿真网关的日志和抓包数据确认云侧是真的按约定链路连上来了而不是断言碰巧通过。4. 把核心验收场景写成用例遥控确认、限功率调度与异常响应遥信能上送只是证明链路通了。DREAMS 验收真正严苛的是控制行为指令发下去云侧不仅要执行还要在规定时间内回报确认。这一章选三个最核心的验收场景展开遥控指令链路、限功率调度、异常响应。每个场景都会给到可抄的用例骨架和参数设计思路。4.1 遥控链路指令下发、确认与执行结果核对遥控的核心不是「下发指令」本身而是「指令被确认」和「执行结果被回报」这两个动作是否完整。实际验收时电网侧下发分闸指令云侧要先回一个「收到指令」的确认然后在规定时间内执行再上送执行后的遥信状态。缺任何一环验收都不算过。# testcases/test_control.py import time def test_remote_control_open_breaker(gateway, cloud_client): # 步骤1仿真网关下发“遥控分闸”指令点号5002 gateway.send_command(point5002, commandopen, timeout_ms3000) # 步骤2断言云侧在3秒内回报“已收到指令” ack cloud_client.wait_ack(point5002, timeout3) assert ack is not None, 云侧未返回指令确认 # 步骤3断言云侧在10秒内执行到位遥信点5001变成0 state cloud_client.wait_telemetry(point5001, value0, timeout10) assert state is True, 指令已确认但未执行到位这里两条超时参数是关键。timeout_ms3000是仿真网关侧等待确认的窗口对应真实调度侧对云平台的确认时限要求timeout10是执行结果回报的窗口依据并网协议里「遥控执行应在 10 秒内完成」这类约定来设。这两个参数不要拍脑袋要来自需求文档否则测出来的结论在验收现场站不住脚。这个用例容易出错的地方是断言行「云侧未返回确认」时你分不清是云侧没收到指令还是收到了但没回确认。排错时先看仿真网关侧日志里指令是否发送成功再在云侧抓包看应用层有没有响应最后才是断言层的问题。还有一点wait_ack和wait_telemetry是两次独立轮询中间不建议加固定sleep而是用轮询加超时的方式既能覆盖响应慢的情况又不会让用例整体耗时失控。4.2 限功率调度有功调节的阶梯与回差测试限功率调度是光伏并网验收的高频场景。电网侧下发有功调节指令要求云光伏系统在限定时间内把出力降到目标值。测试时一般按阶梯走100% → 60% → 30% → 恢复每个阶梯等出力稳定后再断言。# testcases/test_derating.py def test_derating_step_60_percent(gateway, cloud_client): # 前置当前出力在100%通过遥测确认基线 baseline cloud_client.read_telemetry(point4001) # 4001为有功功率遥测点 assert baseline[value] 100.0 # 步骤1仿真网关下发限功率60%指令 gateway.send_command(point5006, commandderate, value60) # 步骤260秒内遥测值进入目标区间60% ± 2% target_min, target_max 58.0, 62.0 ok cloud_client.wait_telemetry_in_range(point4001, lotarget_min, hitarget_max, timeout60) assert ok, f限功率未在规定时间内到位目标区间 {target_min}-{target_max}% # 步骤3多等10秒确保出力稳定不再跌出区间 time.sleep(10) final cloud_client.read_telemetry(point4001) assert target_min final[value] target_max回差 2% 不是随手写的它来自协议里对功率调节精度的约定。不同项目可能要求正负 2% 或正负 5%测试前必须确认。这个用例的另一个价值是覆盖「稳定保持」很多系统刚调节到位时满足要求但 10 秒后出力缓慢漂移出区间这在现场验收会被判不合格。阶梯测试建议写一个参数化用例把目标值、目标区间、等待时间作为参数列表一次性覆盖 60%、30%、恢复三个阶段。参数化用例还能顺便验证调节方向比如从低功率恢复到 100% 的路径经常出现爬坡速率过慢的问题。4.3 异常响应断链重连、重复帧与时间戳校验异常响应是存储库最该提前写、却最容易被拖到最后的部分。现场验收时测试人员会人为制造通信中断、报文异常观察云侧能否按约定恢复。断链重连是最基础的异常场景云侧与 DREAMS 网关之间链路中断云侧要能感知并自动重连重连后数据能继续上送。# testcases/test_abnormal.py def test_link_break_and_reconnect(gateway, cloud_client): # 步骤1建立链路先让一条遥测正常通过 cloud_client.send_telemetry(point4001, value50.0, quality0) assert gateway.wait_for_point(4001, timeout5) is not None # 步骤2模拟链路中断直接断开云侧连接 cloud_client.simulate_link_break() # 步骤310秒内云侧应自动重连 reconnected cloud_client.wait_reconnect(timeout10) assert reconnected is True, 链路中断后云侧未在10秒内重连 # 步骤4重连后遥测继续上送点号值不丢不乱 cloud_client.send_telemetry(point4001, value51.0, quality0) payload gateway.wait_for_point(4001, timeout5) assert payload[value] 51.0断链用例的隐藏重点是重连后的数据完整性。很多系统重连能成功但重连后上送的数据丢了时间戳或者质量位异常这是验收时最容易被揪出来的细节。时间戳校验要单独写用例云侧上送的数据带上时间仿真网关侧断言时间戳和网关本地时间的偏差在约定范围内比如 5 秒。时钟偏差问题在第 5 章会专门展开。这三种异常场景建议按优先级落地断链重连最高重复帧次之时间戳校验最后。重复帧测试需要仿真网关支持把同一报文发送两次云侧应能识别重复并丢弃或只处理一次不产生重复执行动作。5. 对接 DREAMS 的 5 个常见翻车点与排查路径存储库里的用例跑不过不一定是云侧功能问题很多时候是测试环境和测试方法本身埋了雷。这一章写五个我在对接项目里反复遇到、每次都要排查半天的经典问题。5.1 时钟不同步导致时间戳断言失败现象所有遥信遥测用例单独跑都通过但加上时间戳校验后集体失败报错信息是「时间戳偏差超出阈值」。 原因云光伏监控系统所在服务器和运行仿真网关的机器没有做时间同步两边时钟差了几十秒。时间戳断言用的是本机时间差天然把所有报文都判死。 解决测试环境搭建清单里把 NTP 同步列为前置条件所有涉及时间戳的断言不要直接对比绝对值而是对比「上报时间」和「网关接收时间」的差值同时给差值留出合理的抖动余量比如正负 5 秒。我一般在conftest.py里做一次环境自检NTP 不同步直接 fail避免后面所有用例白跑。5.2 浮点缩放因子不一致导致功率读数偏移现象限功率用例里的遥测值断言永远差一截目标 60% 读出来是 59.2%而且不是固定偏差。 原因云侧内部用浮点存储功率上送时报文里按 DREAMS 约定的缩放因子换算。但点表配置里有一处把 10 倍缩放写成了 100 倍导致所有功率相关点位的读数整体偏移一个量级换算回来就是百分比差一截。 解决先把point_table.yaml里每个模拟量的缩放因子和主站侧确认一遍再看具体点位的原始值和转换后值是否成固定倍数关系。排查技巧是给 4001 注入一个整数原始值比如 500看云侧上送出来是多少如果读出来是 5000缩放因子差 10 倍改配置即可。这条经验总结就是点表配置的「玄学」问题90% 是缩放因子和字节序。5.3 只测正流程没测指令超时重发现象功能联调时一切正常现场验收时调度侧模拟 5 秒不回确认云侧没有任何重发动作直接被判不合格。 原因开发侧只实现了「收到指令 → 执行 → 回报」这条正向链路没有实现「确认超时 → 重发指令」的机制测试侧用例也只覆盖了正向流程缺陷被一起带到了现场。 解决控制类用例必须配超时重发测试——仿真网关下发指令后故意不回应确认断言云侧在约定的重发间隔后会再次发送相同指令并且重发次数不超过协议上限。这类用例不需要真实等待全部超时周期把重发间隔从配置里临时调短到 500ms跑完再改回来。5.4 测试数据集互相污染现象用例单独跑全过pytest 一起跑就开始随机失败且失败的用例每次还不一样。 原因多个用例共用了同一套测试数据集或同一个仿真网关状态。比如限功率用例把出力调到了 30%下一个遥信用例又把 30% 当作基线来断言数据被前一个用例污染了。 解决每个用例独立准备数据集fixture 里每条用例都用新启动的仿真网关实例和干净的配置文件。如果一定要共享状态用 pytest 的scopemodule并显式声明依赖顺序但我的建议是宁可多花几秒启动时间也不要共享可变状态。这条做好了存储库的「血泪经验」能少一半。5.5 模拟器「太乖」真实链路抖动一来就崩现象模拟器环境全部通过真实对接 DREAMS 时链路经常中断云侧重连逻辑被反复触发之前没暴露的问题全出来了。 原因本地仿真网关是稳定环境没有模拟真实的链路抖动、网络延时、报文乱序。云侧代码里不少假设在稳定环境下成立到了真实网络里就暴露。 解决在仿真网关里加一个「抖动模式」随机注入报文延时、乱序、重复帧。建议新建test_abnormal_jitter.py专门跑这类用例每次 CI 都执行一轮而不是只在验收前临时测。故障注入常态化之后云侧代码对网络问题的容忍度会被真实提升而不是「在模拟器里看着没问题」。6. 把存储库做成长期回归资产追溯报告与故障注入存储库能跑通是第一步能长期用下去才是它真正的价值。我最后想讲的一个习惯是把每个用例 ID 和验收条目编号做硬关联跑完测试直接生成追溯报告。做法是在requirements/acceptance_trace.csv里维护「条目编号、用例 ID、验证结果」三列pytest 跑完后用一个脚本读取结果回填报告。过程很笨但每次验收答辩时把这份追溯报告往桌上一放比口头解释一百句都管用。故障注入部分我建议做三种常态化链路拔插、报文乱序、应答延时。每种都写成一个可配置的测试开关用环境变量控制仿真网关要不要开启抖动模式。默认 CI 不开因为稳定用例是日常回归的基线每周跑一次「抖动全开」的完整套件专门找云侧代码的网络脆弱点。这个节奏让存储库既能守住基线又能持续暴露边界问题。我现在的习惯是每次改完解析配置或点表先跑一遍test_abnormal.py再跑全量确认没有把历史行为改坏。这个顺序帮我拦下过好几次回归问题。说实话测试存储库这东西前期投入不小但它能把并网验收从「看运气」变成「看报告」这笔账怎么算都划算。希望帮到你。本文还有配套的精品资源点击获取