Files
H100PowerManger/test/integration/check_db.py
T

28 lines
893 B
Python

#!/usr/bin/env python3
#=======================================================================
# check_db.py: 校验 pPowerManger 写入的 power_data.db 是否有 CCU 反馈记录
#(验证 UDP 反馈 → SystemData → SQLite 全链路)。
#=======================================================================
import sys, os, sqlite3
def main():
db = os.environ.get('PM_DB', '/src/bin/power_data.db')
if not os.path.exists(db):
print(f"DB not found: {db}")
return 1
conn = sqlite3.connect(db)
cur = conn.cursor()
try:
cur.execute("SELECT COUNT(*) FROM ccuState")
n = cur.fetchone()[0]
except Exception as e: # noqa: BLE001
print(f"query failed: {e}")
conn.close()
return 1
conn.close()
print(f"ccuState rows = {n}")
return 0 if n > 0 else 1
if __name__ == '__main__':
sys.exit(main())