qiuruiandClaude Sonnet 4.6
54340472ed
- [I-1] collect_local 无匹配文件时不写入空字符串 key - [I-2] run_collection 闭包绑定添加注释，明确避免循环捕获问题 - [I-4] 变量名 l 改为 line，SSH 状态行过滤添加注释 - [m-1] run_collection cfg 参数添加 AgentsConfig 类型注解 - [m-3] collect_device 中 ws 改为 workspace，语义更清晰 - [m-4] 测试文件 DeviceConfig 导入移至顶部，删除函数内重复 import - [m-5] 测试名称 test_save_requires_env_collector_role 改为 test_env_collector_role_can_write_env_facts，内容与实现对齐 （set_fact 无角色限制，权限测试改用 set_goal） - [I-5] 测试断言改用 .get() 与 I-1 修复后行为保持一致 - [I-6] fake_run 添加 remote_cwd 参数验证 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
117 lines
4.8 KiB
Python
117 lines
4.8 KiB
Python
from __future__ import annotations
|
||
import os
|
||
import pytest
|
||
from unittest.mock import MagicMock, patch
|
||
from rockchip_agents.config import AgentsConfig, ClaudeConfig, DeviceConfig, FeishuConfig, SchedulerConfig
|
||
from rockchip_agents.tools.env_collector import EnvCollector, run_collection
|
||
from rockchip_agents.tools.project_memory import ProjectMemory
|
||
|
||
|
||
def test_collect_local_env_vars(tmp_path):
|
||
"""collect_local 应收集含 MODEL/RKNN/ONNX 关键词的环境变量。"""
|
||
env = {
|
||
"RKNN_MODEL_PATH": "/data/rockchip/models",
|
||
"HOME": "/home/pi",
|
||
"ONNX_WEIGHT_DIR": "/data/weights",
|
||
}
|
||
with patch.dict(os.environ, env, clear=True):
|
||
col = EnvCollector()
|
||
facts = col.collect_local(scan_dirs=[])
|
||
assert facts.get("local.envvar.RKNN_MODEL_PATH") == "/data/rockchip/models"
|
||
assert facts.get("local.envvar.ONNX_WEIGHT_DIR") == "/data/weights"
|
||
assert "local.envvar.HOME" not in facts
|
||
|
||
|
||
def test_collect_local_file_scan(tmp_path):
|
||
"""collect_local 文件扫描应找到 *.rknn / *.onnx / *.pt 文件。"""
|
||
(tmp_path / "model.rknn").touch()
|
||
(tmp_path / "sub").mkdir()
|
||
(tmp_path / "sub" / "det.onnx").touch()
|
||
col = EnvCollector()
|
||
facts = col.collect_local(scan_dirs=[str(tmp_path)])
|
||
assert str(tmp_path / "model.rknn") in facts["local.models.rknn"]
|
||
assert str(tmp_path / "sub" / "det.onnx") in facts["local.models.onnx"]
|
||
# [I-5] I-1 修复后无 .pt 文件时 key 不写入，.get() 返回 "" 与实现一致
|
||
assert facts.get("local.models.pt", "") == ""
|
||
|
||
|
||
def test_collect_local_skips_venv(tmp_path):
|
||
"""collect_local 应跳过 venv / __pycache__ 目录。"""
|
||
venv = tmp_path / "venv" / "lib"
|
||
venv.mkdir(parents=True)
|
||
(venv / "ignore.rknn").touch()
|
||
col = EnvCollector()
|
||
facts = col.collect_local(scan_dirs=[str(tmp_path)])
|
||
assert facts.get("local.models.rknn", "") == ""
|
||
|
||
|
||
def test_collect_device_success():
|
||
"""collect_device 成功时应返回 models/disk/packages facts。"""
|
||
dev = DeviceConfig(host="1.2.3.4", user="pi", password="pi",
|
||
workspace="/home/pi", type="linux", connect="ssh")
|
||
|
||
# [I-6] 添加参数验证，确保 remote_cwd 传递正确
|
||
def fake_run(cmd: str, remote_cwd: str = "", timeout: int = 60) -> str:
|
||
assert remote_cwd == "/home/pi", f"unexpected remote_cwd: {remote_cwd}"
|
||
if "find" in cmd:
|
||
return "[rc=0]\n/home/pi/model.rknn\n"
|
||
if "df" in cmd:
|
||
return "[rc=0]\n/dev/root 29G 8.2G 19G\n"
|
||
if "pip" in cmd:
|
||
return "[rc=0]\nrknn-toolkit-lite2 2.3.2\nopencv-python 4.13.0\n"
|
||
return "[rc=0]\n"
|
||
|
||
col = EnvCollector()
|
||
col._run_on_device = fake_run
|
||
facts = col.collect_device("rk3588", dev)
|
||
assert "/home/pi/model.rknn" in facts["rk3588.models.rknn"]
|
||
assert "29G" in facts["rk3588.disk"]
|
||
assert "rknn-toolkit-lite2" in facts["rk3588.packages"]
|
||
|
||
|
||
def test_collect_device_offline():
|
||
"""设备离线时应写入 error key，不抛异常。"""
|
||
dev = DeviceConfig(host="1.2.3.4", user="pi", password="pi",
|
||
workspace="/home/pi", type="linux", connect="ssh")
|
||
col = EnvCollector()
|
||
col._run_on_device = MagicMock(side_effect=Exception("Connection refused"))
|
||
facts = col.collect_device("rk3588", dev)
|
||
assert "offline" in facts.get("rk3588.error", "")
|
||
|
||
|
||
def test_save_writes_to_memory(tmp_path):
|
||
"""save() 应将 facts 写入 ProjectMemory __env__ 命名空间。"""
|
||
mem = ProjectMemory(db_path=tmp_path / "m.db")
|
||
col = EnvCollector()
|
||
col.save({"local.models.rknn": "/data/a.rknn"}, mem)
|
||
stored = mem.get_facts("__env__")
|
||
assert stored["local.models.rknn"] == "/data/a.rknn"
|
||
assert "collected_at" in stored
|
||
|
||
|
||
def test_env_collector_role_can_write_env_facts(tmp_path):
|
||
"""env-collector 是合法的写入角色，不应抛 PermissionError。"""
|
||
mem = ProjectMemory(db_path=tmp_path / "m.db")
|
||
# set_fact 对所有 role 开放，env-collector 可正常写入
|
||
mem.set_fact("__env__", "test.key", "val", role="env-collector")
|
||
assert mem.get_facts("__env__")["test.key"] == "val"
|
||
# set_goal 仅限 WRITE_ROLES，developer 不在其中应被拒绝
|
||
with pytest.raises(PermissionError):
|
||
mem.set_goal("__env__", "some goal", role="developer")
|
||
|
||
|
||
def test_run_collection_local_only(tmp_path):
|
||
"""run_collection 无设备时应收集本机信息并保存。"""
|
||
cfg = AgentsConfig(
|
||
projects={},
|
||
scheduler=SchedulerConfig(),
|
||
claude=ClaudeConfig(api_key="fake"),
|
||
feishu=FeishuConfig(),
|
||
devices={},
|
||
)
|
||
mem = ProjectMemory(db_path=tmp_path / "m.db")
|
||
with patch.dict(os.environ, {"RKNN_MODEL_PATH": "/data/m"}, clear=False):
|
||
facts = run_collection(cfg, mem)
|
||
assert "local.envvar.RKNN_MODEL_PATH" in facts
|
||
assert "collected_at" in mem.get_facts("__env__")
|