import sys import unittest import numpy as np sys.path.insert(0, "/home/program/NextAgent/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()