fix: pybot 审计修复——插件异常类型、I2C 引用校验、卡尔曼除零守卫、CLI 端到端测试

This commit is contained in:
qing committed 2026-08-10 16:16:40 +00:00
1 parent 09dc36d591
commit a8ffa73a6b
6 files changed
+70 -6

No files matched your search

+1
View File
@@ -25,3 +25,4 @@ from .parser import ( # noqa: F401
to_int,
)
from .robot import RobotStruct # noqa: F401
from .plugin import PluginNotFound, PluginProtocolError, PluginVar # noqa: F401
+6 -1
View File
@@ -114,7 +114,12 @@ class MPU6050Var(HardwareVar):
addr = get_mt(matrix, i, matrix.index_of_col(0, "addr"))
log.info("Create a MPU6050_BIND HVar: %s, use i2c interf: %s, "
"i2c addr: %s", name, i2c_name, addr)
fd = M[i2c_name].play({})["fd"] # 引用 I2C 变量取 fd
i2c_var = M[i2c_name]
if not isinstance(i2c_var, I2CVar):
raise TypeError(
f"@MPU6050_BIND '{name}' references '{i2c_name}', "
f"but it is a {type(i2c_var).__name__}, not an I2CVar")
fd = i2c_var.play({})["fd"]
return cls(name, MPU6050(fd, to_int(addr)))
+1 -1
View File
@@ -21,7 +21,7 @@ class Kalman:
def filter(self, dat):
x_mid = self.a * self.x_last # (1) x(k|k-1) = A*X(k-1|k-1)
p_mid = self.a * self.p_last + self.q # (2) p(k|k-1) = A*p(k-1|k-1)*A'+Q
kg = p_mid / (p_mid + self.r) # (4) kg(k) = p(k|k-1)/(p(k|k-1)+R)
kg = p_mid / max(p_mid + self.r, 1e-10) # (4) kg(k) = p(k|k-1)/(p(k|k-1)+R)
x_now = x_mid + kg * (dat - x_mid) # (3) x(k|k) = X(k|k-1)+kg*(Z(k)-X(k|k-1))
p_now = (1.0 - kg) * p_mid # (5) p(k|k) = (1-kg)*P(k|k-1)
self.p_last = p_now
+12 -3
View File
@@ -37,6 +37,17 @@ class PluginNotFound(Exception):
self.searched = searched
class PluginProtocolError(Exception):
"""插件缺少 build / call / destroy 之一"""
def __init__(self, name, path, missing):
super().__init__(
f"plugin '{name}' at {path} missing function: {missing}(ctx)")
self.name = name
self.path = path
self.missing = missing
def search_dirs():
"""插件搜索目录(按优先级):$PYBOT_MODULES_PATH → 包内置 → cwd"""
dirs = []
@@ -68,9 +79,7 @@ def load_module(name):
spec.loader.exec_module(module)
for fn in ("build", "call", "destroy"):
if not callable(getattr(module, fn, None)):
raise PluginNotFound(
name, [path]) from ValueError(
f"plugin '{name}' missing function {fn}(ctx)")
raise PluginProtocolError(name, path, fn)
log.debug("loaded plugin %s from %s", name, path)
return module
+49
View File
@@ -0,0 +1,49 @@
"""CLI 端到端测试:--dry-run / --test roundtrip / stdin 输入"""
import io
import sys
from tests.conftest import SCRIPTS_DIR
from pybot.main import main
def test_stdin_input(monkeypatch):
"""stdin 输入(模拟 cat script | pybot)"""
src = (SCRIPTS_DIR / "std.mt").read_text(encoding="utf-8")
monkeypatch.setattr(sys, "stdin", io.StringIO(src))
monkeypatch.setattr(sys, "argv", ["pybot", "--dry-run"])
# 验证不抛异常且返回 0
assert main(["--dry-run"]) == 0
def test_cli_help():
"""--help 返回 0"""
try:
main(["--help"])
except SystemExit as e:
assert e.code == 0
def test_test_roundtrip(tmp_path):
"""--test roundtrip in.mt out.mt 往返"""
src = (SCRIPTS_DIR / "std.mt").read_text(encoding="utf-8")
infile = tmp_path / "in.mt"
infile.write_text(src, encoding="utf-8")
outfile = tmp_path / "out.mt"
assert main(["--test", "roundtrip", str(infile), str(outfile)]) == 0
# 输出可被再次解析
from pybot.parser import MatrixParser
parser = MatrixParser(outfile.read_text(encoding="utf-8"))
blocks = []
while True:
obj = parser.parse()
if obj is None:
break
blocks.append(obj)
assert len(blocks) == 7 # std.mt 含 7 个 @块
def test_test_unknown_mode():
"""未知 --test 模式返回 2"""
assert main(["--test", "unknown", "/dev/null", "/dev/null"]) == 2
+1 -1
View File
@@ -36,7 +36,7 @@ def test_missing_protocol_function(tmp_path, monkeypatch):
(tmp_path / "Broken.py").write_text(
"def build(ctx): return {}\n", encoding="utf-8")
monkeypatch.setenv("PYBOT_MODULES_PATH", str(tmp_path))
with pytest.raises(plugin.PluginNotFound):
with pytest.raises(plugin.PluginProtocolError):
plugin.load_module("Broken")