bea8aefd8d
tests/test_protocol.py 原本把 "/home/program/NextAgent/src" 硬编码进 sys.path，指向宿主机上另一份（工作区为脏的）NextAgent 检出。结果测试 import 的是那份副本而非本仓库的 src，导致两个用例失败： - test_decode_raises_on_wrong_fmt_id → DecodingError not raised - test_register_and_send → ValueError: inhomogeneous shape 改为基于 __file__ 推导仓库内的 src/ 路径，测试不再依赖仓库外的任何 绝对路径，任何检出位置都能跑通。实现代码未做改动——这是测试耦合问题， 不是实现 bug。 同时取消跟踪 7 个误提交的 .pyc（.gitignore 规则此前加晚了）。 仅 git rm --cached，工作区文件保留；.gitignore 的 __pycache__/ 与 *.pyc 规则继续生效。 python3 -m unittest discover -s tests → Ran 31 tests, OK git ls-files '*.pyc' | wc -l → 0
214 lines
7.2 KiB
Python
214 lines
7.2 KiB
Python
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
|
||
# 仅依赖本仓库布局：tests/ 的上一级的 src/，不引用仓库外任何绝对路径
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
|
||
|
||
from protocol.registry import PROTOCOL_REGISTRY, FMT_NAMES, PARSER_NAMES
|
||
from protocol.codec import CodecAdapter, IdentityCodec, EncodingError, DecodingError
|
||
from protocol.tensor_bus import TensorBus
|
||
from parsers.rule_based import RuleBasedParser
|
||
|
||
|
||
class _TestCodec(CodecAdapter):
|
||
"""测试用编解码器：fmt_id=0 (ref_tensor), parser_id=0 (ref_parser)"""
|
||
def __init__(self):
|
||
super().__init__(subnet_id="core", fmt_id=0, parser_id=0)
|
||
|
||
|
||
class _LaxCodec(CodecAdapter):
|
||
"""宽松编解码器（类似 IdentityCodec 但使用合法 id）"""
|
||
def __init__(self):
|
||
super().__init__(subnet_id="core", fmt_id=0, parser_id=0)
|
||
|
||
def _validate_prefix(self, tensor):
|
||
if tensor.shape[0] < self.prefix_len:
|
||
raise DecodingError(f"张量过短: {tensor.shape[0]} < {self.prefix_len}")
|
||
|
||
def _validate_internal(self, tensor):
|
||
if not isinstance(tensor, np.ndarray):
|
||
raise EncodingError(f"内部类型必须是 np.ndarray, 得到 {type(tensor)}")
|
||
|
||
|
||
class TestProtocolRegistry(unittest.TestCase):
|
||
|
||
def test_validate_fmt_id_valid(self):
|
||
for fid in FMT_NAMES:
|
||
PROTOCOL_REGISTRY.validate_fmt_id(fid)
|
||
|
||
def test_validate_fmt_id_invalid(self):
|
||
with self.assertRaises(ValueError):
|
||
PROTOCOL_REGISTRY.validate_fmt_id(999)
|
||
|
||
def test_validate_parser_id_valid(self):
|
||
for pid in PARSER_NAMES:
|
||
PROTOCOL_REGISTRY.validate_parser_id(pid)
|
||
|
||
def test_validate_parser_id_invalid(self):
|
||
with self.assertRaises(ValueError):
|
||
PROTOCOL_REGISTRY.validate_parser_id(999)
|
||
|
||
def test_validate_subnet_id_valid(self):
|
||
for sid in PROTOCOL_REGISTRY.subnet_ids:
|
||
PROTOCOL_REGISTRY.validate_subnet_id(sid)
|
||
|
||
def test_validate_subnet_id_invalid(self):
|
||
with self.assertRaises(ValueError):
|
||
PROTOCOL_REGISTRY.validate_subnet_id("nonexistent")
|
||
|
||
def test_fmt_name_known(self):
|
||
self.assertEqual(PROTOCOL_REGISTRY.fmt_name(0), "ref_tensor")
|
||
|
||
def test_fmt_name_unknown(self):
|
||
self.assertEqual(PROTOCOL_REGISTRY.fmt_name(999), "unknown_fmt_999")
|
||
|
||
def test_parser_name_known(self):
|
||
self.assertEqual(PROTOCOL_REGISTRY.parser_name(0), "ref_parser")
|
||
|
||
def test_parser_name_unknown(self):
|
||
self.assertEqual(PROTOCOL_REGISTRY.parser_name(999),
|
||
"unknown_parser_999")
|
||
|
||
|
||
class TestCodecAdapter(unittest.TestCase):
|
||
|
||
def test_encode_decode_roundtrip(self):
|
||
codec = _TestCodec()
|
||
payload = np.array([1.0, 2.0, 3.0, 4.0])
|
||
comm = codec.encode(payload, target="routing", reply_req=True)
|
||
decoded = codec.decode(comm)
|
||
np.testing.assert_array_equal(decoded, payload)
|
||
|
||
def test_encode_raises_on_non_ndarray(self):
|
||
codec = _TestCodec()
|
||
with self.assertRaises(EncodingError):
|
||
codec.encode([1, 2, 3], target="routing", reply_req=True)
|
||
|
||
def test_decode_raises_on_short_tensor(self):
|
||
codec = _TestCodec()
|
||
short = np.array([0.0])
|
||
with self.assertRaises(DecodingError):
|
||
codec.decode(short)
|
||
|
||
def test_decode_raises_on_wrong_fmt_id(self):
|
||
codec = _TestCodec()
|
||
wrong = np.array([99.0, 0.0, 1.0, 2.0])
|
||
with self.assertRaises(DecodingError):
|
||
codec.decode(wrong)
|
||
|
||
|
||
class TestLaxCodec(unittest.TestCase):
|
||
|
||
def test_encode_decode_roundtrip(self):
|
||
codec = _LaxCodec()
|
||
payload = np.array([10.0, 20.0, 30.0])
|
||
comm = codec.encode(payload, target="core", reply_req=False)
|
||
decoded = codec.decode(comm)
|
||
np.testing.assert_array_equal(decoded, payload)
|
||
|
||
def test_decode_allows_any_prefix(self):
|
||
codec = _LaxCodec()
|
||
comm = np.array([99.0, 99.0, 42.0])
|
||
decoded = codec.decode(comm)
|
||
np.testing.assert_array_equal(decoded, np.array([42.0]))
|
||
|
||
def test_decode_raises_on_short_tensor(self):
|
||
codec = _LaxCodec()
|
||
with self.assertRaises(DecodingError):
|
||
codec.decode(np.array([0.0]))
|
||
|
||
def test_encode_raises_on_non_ndarray(self):
|
||
codec = _LaxCodec()
|
||
with self.assertRaises(EncodingError):
|
||
codec.encode("bad", target="core", reply_req=True)
|
||
|
||
|
||
class TestRuleBasedParser(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.parser = RuleBasedParser()
|
||
|
||
def test_ref_tensor(self):
|
||
tensor = np.array([0.0, 0.0, 5.0, 1.0, 2.0, 3.0])
|
||
result = self.parser.parse(tensor)
|
||
self.assertEqual(result, "引用张量: 5个标签")
|
||
|
||
def test_memory_result(self):
|
||
tensor = np.array([1.0, 1.0, 3.0, 7.0])
|
||
result = self.parser.parse(tensor)
|
||
self.assertEqual(result, "记忆结果: 3个实体, 7条关系")
|
||
|
||
def test_memory_query(self):
|
||
tensor = np.array([2.0, 2.0, 1.0, 2.0, 3.0])
|
||
result = self.parser.parse(tensor)
|
||
self.assertEqual(result, "记忆查询")
|
||
|
||
def test_memory_update(self):
|
||
tensor = np.array([3.0, 3.0, 2.0, 4.0])
|
||
result = self.parser.parse(tensor)
|
||
self.assertEqual(result, "记忆更新: 2个实体, 4条关系")
|
||
|
||
def test_reasoning_path(self):
|
||
tensor = np.array([4.0, 4.0, 6.0])
|
||
result = self.parser.parse(tensor)
|
||
self.assertEqual(result, "推理路径: 6步")
|
||
|
||
def test_expression_verify(self):
|
||
tensor = np.zeros(2 + 769)
|
||
tensor[0] = 5.0
|
||
tensor[1] = 5.0
|
||
tensor[2 + 768] = 0.85
|
||
result = self.parser.parse(tensor)
|
||
self.assertEqual(result, "表达验证: 意图对齐度=0.85")
|
||
|
||
def test_control(self):
|
||
tensor = np.array([6.0, 6.0])
|
||
result = self.parser.parse(tensor)
|
||
self.assertEqual(result, "控制信号")
|
||
|
||
def test_short_tensor(self):
|
||
tensor = np.array([0.0])
|
||
result = self.parser.parse(tensor)
|
||
self.assertIn("过短", result)
|
||
|
||
|
||
class TestTensorBus(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.bus = TensorBus()
|
||
self.codec_a = _TestCodec()
|
||
self.codec_b = _LaxCodec()
|
||
self.bus.register("routing", self.codec_a)
|
||
self.bus.register("core", self.codec_b)
|
||
|
||
def test_register_and_send(self):
|
||
payload = np.array([7.0, 8.0, 9.0])
|
||
result = self.bus.send(payload, source="routing", target="core")
|
||
np.testing.assert_array_equal(result, payload)
|
||
|
||
def test_recv(self):
|
||
payload = np.array([1.0, 2.0])
|
||
comm = self.codec_a.encode(payload, target="routing", reply_req=True)
|
||
result = self.bus.recv("routing", comm)
|
||
np.testing.assert_array_equal(result, payload)
|
||
|
||
def test_send_unregistered_source(self):
|
||
with self.assertRaises(KeyError):
|
||
self.bus.send(np.array([1.0]), source="nonexistent", target="core")
|
||
|
||
def test_send_unregistered_target(self):
|
||
with self.assertRaises(KeyError):
|
||
self.bus.send(np.array([1.0]), source="routing",
|
||
target="nonexistent")
|
||
|
||
def test_recv_unregistered_source(self):
|
||
with self.assertRaises(KeyError):
|
||
self.bus.recv("nonexistent", np.array([1.0, 2.0]))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|