diff --git a/tests_v4/test_capability_router.py b/tests_v4/test_capability_router.py new file mode 100644 index 000000000..aca27d9be --- /dev/null +++ b/tests_v4/test_capability_router.py @@ -0,0 +1,820 @@ +""" +Tests for runtime/capability_router.py - CapabilityRouter implementation. +""" +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from astrbot_sdk.context import CancelToken +from astrbot_sdk.errors import AstrBotError +from astrbot_sdk.protocol.descriptors import CapabilityDescriptor +from astrbot_sdk.runtime.capability_router import ( + CAPABILITY_NAME_PATTERN, + RESERVED_CAPABILITY_PREFIXES, + CallHandler, + FinalizeHandler, + StreamExecution, + StreamHandler, + _CapabilityRegistration, +) +from astrbot_sdk.runtime.capability_router import CapabilityRouter + + +class TestStreamExecution: + """Tests for StreamExecution dataclass.""" + + def test_init(self): + """StreamExecution should store iterator and finalize.""" + async def gen(): + yield {"text": "a"} + + def fin(chunks): + return {"count": len(chunks)} + + execution = StreamExecution(iterator=gen(), finalize=fin) + assert execution.iterator is not None + assert execution.finalize is fin + + +class TestCapabilityRegistration: + """Tests for _CapabilityRegistration dataclass.""" + + def test_init(self): + """_CapabilityRegistration should store all fields.""" + descriptor = CapabilityDescriptor(name="test.cap", description="Test") + call_handler = AsyncMock() + + reg = _CapabilityRegistration( + descriptor=descriptor, + call_handler=call_handler, + exposed=True, + ) + + assert reg.descriptor == descriptor + assert reg.call_handler == call_handler + assert reg.stream_handler is None + assert reg.finalize is None + assert reg.exposed is True + + def test_defaults(self): + """_CapabilityRegistration should have correct defaults.""" + descriptor = CapabilityDescriptor(name="test.cap", description="Test") + + reg = _CapabilityRegistration(descriptor=descriptor) + + assert reg.call_handler is None + assert reg.stream_handler is None + assert reg.finalize is None + assert reg.exposed is True + + +class TestCapabilityNamePattern: + """Tests for capability name validation pattern.""" + + def test_valid_names(self): + """Valid capability names should match pattern.""" + valid_names = [ + "llm.chat", + "db.get", + "memory.search", + "platform.send", + "a.b", + "my_module.my_method", + "ns123.method456", + ] + for name in valid_names: + assert CAPABILITY_NAME_PATTERN.fullmatch(name), f"{name} should be valid" + + def test_invalid_names(self): + """Invalid capability names should not match pattern.""" + invalid_names = [ + "llm", # No dot + "LLM.chat", # Uppercase + "llm.Chat", # Uppercase method + "llm.chat.extra", # Too many parts + "1llm.chat", # Starts with number + "llm.1chat", # Method starts with number + ".chat", # Empty namespace + "llm.", # Empty method + "llm-chat", # Hyphen instead of dot + ] + for name in invalid_names: + assert not CAPABILITY_NAME_PATTERN.fullmatch(name), f"{name} should be invalid" + + +class TestReservedCapabilityPrefixes: + """Tests for reserved capability prefixes.""" + + def test_reserved_prefixes(self): + """Reserved prefixes should be defined.""" + assert "handler." in RESERVED_CAPABILITY_PREFIXES + assert "system." in RESERVED_CAPABILITY_PREFIXES + assert "internal." in RESERVED_CAPABILITY_PREFIXES + + def test_reserved_names_are_detected(self): + """Reserved names should be detected by startswith.""" + reserved_names = [ + "handler.demo", + "system.health", + "internal.trace", + ] + for name in reserved_names: + assert any( + name.startswith(prefix) for prefix in RESERVED_CAPABILITY_PREFIXES + ), f"{name} should be reserved" + + +class TestCapabilityRouterInit: + """Tests for CapabilityRouter initialization.""" + + def test_init_creates_empty_registrations(self): + """CapabilityRouter should start with empty registrations.""" + router = CapabilityRouter() + assert router._registrations == {} + assert router.db_store == {} + assert router.memory_store == {} + assert router.sent_messages == [] + + def test_init_registers_builtin_capabilities(self): + """CapabilityRouter should register built-in capabilities on init.""" + router = CapabilityRouter() + descriptors = router.descriptors() + + capability_names = [d.name for d in descriptors] + + # LLM capabilities + assert "llm.chat" in capability_names + assert "llm.chat_raw" in capability_names + assert "llm.stream_chat" in capability_names + + # Memory capabilities + assert "memory.search" in capability_names + assert "memory.save" in capability_names + assert "memory.delete" in capability_names + + # DB capabilities + assert "db.get" in capability_names + assert "db.set" in capability_names + assert "db.delete" in capability_names + assert "db.list" in capability_names + + # Platform capabilities + assert "platform.send" in capability_names + assert "platform.send_image" in capability_names + assert "platform.get_members" in capability_names + + +class TestCapabilityRouterRegister: + """Tests for CapabilityRouter.register method.""" + + def test_register_adds_capability(self): + """register should add capability to registrations.""" + router = CapabilityRouter() + descriptor = CapabilityDescriptor(name="test.cap", description="Test") + + router.register(descriptor) + + assert "test.cap" in router._registrations + assert router._registrations["test.cap"].descriptor == descriptor + + def test_register_with_handlers(self): + """register should store handlers.""" + router = CapabilityRouter() + descriptor = CapabilityDescriptor(name="test.cap", description="Test") + + async def call_handler(req_id, payload, token): + return {"result": "ok"} + + async def stream_handler(req_id, payload, token): + yield {"chunk": 1} + + def finalize(chunks): + return {"count": len(chunks)} + + router.register( + descriptor, + call_handler=call_handler, + stream_handler=stream_handler, + finalize=finalize, + ) + + reg = router._registrations["test.cap"] + assert reg.call_handler == call_handler + assert reg.stream_handler == stream_handler + assert reg.finalize == finalize + + def test_register_invalid_name_raises(self): + """register should reject invalid capability names.""" + router = CapabilityRouter() + + with pytest.raises(ValueError, match="capability 名称必须匹配"): + router.register(CapabilityDescriptor(name="invalid", description="Bad")) + + def test_register_reserved_name_raises(self): + """register should reject reserved names for exposed registrations.""" + router = CapabilityRouter() + + with pytest.raises(ValueError, match="保留 capability"): + router.register( + CapabilityDescriptor(name="handler.demo", description="Reserved") + ) + + def test_register_reserved_name_allowed_for_internal(self): + """register should allow reserved names for internal (exposed=False).""" + router = CapabilityRouter() + + # Should not raise + router.register( + CapabilityDescriptor(name="handler.internal", description="Internal"), + exposed=False, + ) + + # Should not appear in descriptors + names = [d.name for d in router.descriptors()] + assert "handler.internal" not in names + + def test_descriptors_only_returns_exposed(self): + """descriptors should only return exposed capabilities.""" + router = CapabilityRouter() + + router.register( + CapabilityDescriptor(name="exposed.cap", description="Exposed"), + exposed=True, + ) + router.register( + CapabilityDescriptor(name="hidden.cap", description="Hidden"), + exposed=False, + ) + + names = [d.name for d in router.descriptors()] + assert "exposed.cap" in names + assert "hidden.cap" not in names + + +class TestCapabilityRouterExecute: + """Tests for CapabilityRouter.execute method.""" + + @pytest.mark.asyncio + async def test_execute_calls_handler(self): + """execute should call the registered handler.""" + router = CapabilityRouter() + router.register( + CapabilityDescriptor(name="test.cap", description="Test"), + call_handler=AsyncMock(return_value={"result": "ok"}), + ) + + token = CancelToken() + result = await router.execute( + "test.cap", + {}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + assert result == {"result": "ok"} + + @pytest.mark.asyncio + async def test_execute_validates_input_schema(self): + """execute should validate input against schema.""" + router = CapabilityRouter() + router.register( + CapabilityDescriptor( + name="test.cap", + description="Test", + input_schema={ + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, + ), + call_handler=AsyncMock(return_value={}), + ) + + token = CancelToken() + + # Missing required field + with pytest.raises(AstrBotError, match="缺少必填字段"): + await router.execute( + "test.cap", + {}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + @pytest.mark.asyncio + async def test_execute_validates_output_schema(self): + """execute should validate output against schema.""" + router = CapabilityRouter() + router.register( + CapabilityDescriptor( + name="test.cap", + description="Test", + output_schema={ + "type": "object", + "properties": {"result": {"type": "string"}}, + "required": ["result"], + }, + ), + call_handler=AsyncMock(return_value={}), # Missing required field + ) + + token = CancelToken() + + with pytest.raises(AstrBotError, match="缺少必填字段"): + await router.execute( + "test.cap", + {}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + @pytest.mark.asyncio + async def test_execute_missing_capability_raises(self): + """execute should raise for unknown capability.""" + router = CapabilityRouter() + token = CancelToken() + + with pytest.raises(AstrBotError, match="capability_not_found"): + await router.execute( + "unknown.cap", + {}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + @pytest.mark.asyncio + async def test_execute_stream_returns_stream_execution(self): + """execute with stream=True should return StreamExecution.""" + router = CapabilityRouter() + + async def stream_handler(req_id, payload, token): + yield {"chunk": 1} + yield {"chunk": 2} + + router.register( + CapabilityDescriptor( + name="test.stream", + description="Test", + supports_stream=True, + ), + stream_handler=stream_handler, + ) + + token = CancelToken() + result = await router.execute( + "test.stream", + {}, + stream=True, + cancel_token=token, + request_id="req-1", + ) + + assert isinstance(result, StreamExecution) + + @pytest.mark.asyncio + async def test_execute_stream_without_handler_raises(self): + """execute with stream=True and no stream_handler should raise.""" + router = CapabilityRouter() + router.register( + CapabilityDescriptor(name="test.cap", description="Test"), + call_handler=AsyncMock(return_value={}), + ) + + token = CancelToken() + + with pytest.raises(AstrBotError, match="不支持 stream=true"): + await router.execute( + "test.cap", + {}, + stream=True, + cancel_token=token, + request_id="req-1", + ) + + @pytest.mark.asyncio + async def test_execute_call_without_handler_raises(self): + """execute without stream and no call_handler should raise.""" + router = CapabilityRouter() + router.register( + CapabilityDescriptor( + name="test.stream_only", + description="Stream only", + supports_stream=True, + ), + stream_handler=AsyncMock(), + ) + + token = CancelToken() + + with pytest.raises(AstrBotError, match="只能以 stream=true 调用"): + await router.execute( + "test.stream_only", + {}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + +class TestBuiltinLlmCapabilities: + """Tests for built-in LLM capabilities.""" + + @pytest.mark.asyncio + async def test_llm_chat(self): + """llm.chat should return echo response.""" + router = CapabilityRouter() + token = CancelToken() + + result = await router.execute( + "llm.chat", + {"prompt": "hello"}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + assert result["text"] == "Echo: hello" + + @pytest.mark.asyncio + async def test_llm_chat_raw(self): + """llm.chat_raw should return full response.""" + router = CapabilityRouter() + token = CancelToken() + + result = await router.execute( + "llm.chat_raw", + {"prompt": "test"}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + assert result["text"] == "Echo: test" + assert "usage" in result + assert "finish_reason" in result + assert "tool_calls" in result + + @pytest.mark.asyncio + async def test_llm_stream_chat(self): + """llm.stream_chat should yield characters.""" + router = CapabilityRouter() + token = CancelToken() + + result = await router.execute( + "llm.stream_chat", + {"prompt": "hi"}, + stream=True, + cancel_token=token, + request_id="req-1", + ) + + assert isinstance(result, StreamExecution) + + chunks = [] + async for chunk in result.iterator: + chunks.append(chunk) + + # Should yield each character + text = "".join(c.get("text", "") for c in chunks) + assert text == "Echo: hi" + + +class TestBuiltinMemoryCapabilities: + """Tests for built-in memory capabilities.""" + + @pytest.mark.asyncio + async def test_memory_save_and_search(self): + """memory.save and memory.search should work together.""" + router = CapabilityRouter() + token = CancelToken() + + # Save + await router.execute( + "memory.save", + {"key": "test_key", "value": {"data": "test_value"}}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + # Search + result = await router.execute( + "memory.search", + {"query": "test"}, + stream=False, + cancel_token=token, + request_id="req-2", + ) + + assert len(result["items"]) == 1 + assert result["items"][0]["key"] == "test_key" + + @pytest.mark.asyncio + async def test_memory_delete(self): + """memory.delete should remove saved memory.""" + router = CapabilityRouter() + token = CancelToken() + + # Save + await router.execute( + "memory.save", + {"key": "to_delete", "value": {"data": "value"}}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + # Delete + await router.execute( + "memory.delete", + {"key": "to_delete"}, + stream=False, + cancel_token=token, + request_id="req-2", + ) + + # Search should return empty + result = await router.execute( + "memory.search", + {"query": "to_delete"}, + stream=False, + cancel_token=token, + request_id="req-3", + ) + + assert len(result["items"]) == 0 + + @pytest.mark.asyncio + async def test_memory_save_invalid_value(self): + """memory.save should reject non-object value.""" + router = CapabilityRouter() + token = CancelToken() + + with pytest.raises(AstrBotError, match="value 必须是 object"): + await router.execute( + "memory.save", + {"key": "test", "value": "not_an_object"}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + +class TestBuiltinDbCapabilities: + """Tests for built-in DB capabilities.""" + + @pytest.mark.asyncio + async def test_db_set_and_get(self): + """db.set and db.get should work together.""" + router = CapabilityRouter() + token = CancelToken() + + # Set + await router.execute( + "db.set", + {"key": "test_key", "value": {"data": "test_value"}}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + # Get + result = await router.execute( + "db.get", + {"key": "test_key"}, + stream=False, + cancel_token=token, + request_id="req-2", + ) + + assert result["value"] == {"data": "test_value"} + + @pytest.mark.asyncio + async def test_db_get_missing_key(self): + """db.get should return None for missing key.""" + router = CapabilityRouter() + token = CancelToken() + + result = await router.execute( + "db.get", + {"key": "nonexistent"}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + assert result["value"] is None + + @pytest.mark.asyncio + async def test_db_delete(self): + """db.delete should remove stored value.""" + router = CapabilityRouter() + token = CancelToken() + + # Set + await router.execute( + "db.set", + {"key": "to_delete", "value": {"data": "value"}}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + # Delete + await router.execute( + "db.delete", + {"key": "to_delete"}, + stream=False, + cancel_token=token, + request_id="req-2", + ) + + # Get should return None + result = await router.execute( + "db.get", + {"key": "to_delete"}, + stream=False, + cancel_token=token, + request_id="req-3", + ) + + assert result["value"] is None + + @pytest.mark.asyncio + async def test_db_list(self): + """db.list should return keys.""" + router = CapabilityRouter() + token = CancelToken() + + # Set multiple keys + await router.execute( + "db.set", + {"key": "prefix_a", "value": {}}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + await router.execute( + "db.set", + {"key": "prefix_b", "value": {}}, + stream=False, + cancel_token=token, + request_id="req-2", + ) + await router.execute( + "db.set", + {"key": "other", "value": {}}, + stream=False, + cancel_token=token, + request_id="req-3", + ) + + # List all + result = await router.execute( + "db.list", + {}, + stream=False, + cancel_token=token, + request_id="req-4", + ) + + assert "prefix_a" in result["keys"] + assert "prefix_b" in result["keys"] + assert "other" in result["keys"] + + # List with prefix + result = await router.execute( + "db.list", + {"prefix": "prefix_"}, + stream=False, + cancel_token=token, + request_id="req-5", + ) + + assert "prefix_a" in result["keys"] + assert "prefix_b" in result["keys"] + assert "other" not in result["keys"] + + @pytest.mark.asyncio + async def test_db_set_invalid_value(self): + """db.set should reject non-object value.""" + router = CapabilityRouter() + token = CancelToken() + + with pytest.raises(AstrBotError, match="value 必须是 object"): + await router.execute( + "db.set", + {"key": "test", "value": "not_an_object"}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + +class TestBuiltinPlatformCapabilities: + """Tests for built-in platform capabilities.""" + + @pytest.mark.asyncio + async def test_platform_send(self): + """platform.send should store message.""" + router = CapabilityRouter() + token = CancelToken() + + result = await router.execute( + "platform.send", + {"session": "session-1", "text": "Hello"}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + assert "message_id" in result + assert len(router.sent_messages) == 1 + assert router.sent_messages[0]["session"] == "session-1" + assert router.sent_messages[0]["text"] == "Hello" + + @pytest.mark.asyncio + async def test_platform_send_image(self): + """platform.send_image should store image message.""" + router = CapabilityRouter() + token = CancelToken() + + result = await router.execute( + "platform.send_image", + {"session": "session-1", "image_url": "http://example.com/image.png"}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + assert "message_id" in result + assert len(router.sent_messages) == 1 + assert router.sent_messages[0]["image_url"] == "http://example.com/image.png" + + @pytest.mark.asyncio + async def test_platform_get_members(self): + """platform.get_members should return mock members.""" + router = CapabilityRouter() + token = CancelToken() + + result = await router.execute( + "platform.get_members", + {"session": "session-1"}, + stream=False, + cancel_token=token, + request_id="req-1", + ) + + assert len(result["members"]) == 2 + assert result["members"][0]["user_id"] == "session-1:member-1" + + +class TestValidateSchema: + """Tests for _validate_schema method.""" + + @pytest.mark.asyncio + async def test_none_schema_passes(self): + """_validate_schema with None schema should pass.""" + router = CapabilityRouter() + + # Should not raise + router._validate_schema(None, {"any": "data"}) + + @pytest.mark.asyncio + async def test_non_object_payload_raises(self): + """_validate_schema should reject non-object when schema type is object.""" + router = CapabilityRouter() + + with pytest.raises(AstrBotError, match="输入必须是 object"): + router._validate_schema({"type": "object"}, "not an object") + + @pytest.mark.asyncio + async def test_missing_required_field_raises(self): + """_validate_schema should reject missing required fields.""" + router = CapabilityRouter() + + with pytest.raises(AstrBotError, match="缺少必填字段"): + router._validate_schema( + {"type": "object", "required": ["name"]}, + {}, + ) + + @pytest.mark.asyncio + async def test_none_required_field_raises(self): + """_validate_schema should reject None required fields.""" + router = CapabilityRouter() + + with pytest.raises(AstrBotError, match="缺少必填字段"): + router._validate_schema( + {"type": "object", "required": ["name"]}, + {"name": None}, + ) diff --git a/tests_v4/test_protocol_descriptors.py b/tests_v4/test_protocol_descriptors.py new file mode 100644 index 000000000..296a94f35 --- /dev/null +++ b/tests_v4/test_protocol_descriptors.py @@ -0,0 +1,341 @@ +""" +Tests for protocol/descriptors.py - Descriptor models. +""" +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from astrbot_sdk.protocol.descriptors import ( + CapabilityDescriptor, + CommandTrigger, + EventTrigger, + HandlerDescriptor, + MessageTrigger, + Permissions, + ScheduleTrigger, +) + + +class TestPermissions: + """Tests for Permissions model.""" + + def test_default_values(self): + """Permissions should have default values.""" + perms = Permissions() + assert perms.require_admin is False + assert perms.level == 0 + + def test_custom_values(self): + """Permissions should accept custom values.""" + perms = Permissions(require_admin=True, level=5) + assert perms.require_admin is True + assert perms.level == 5 + + def test_model_dump(self): + """Permissions should serialize correctly.""" + perms = Permissions(require_admin=True, level=10) + data = perms.model_dump() + assert data == {"require_admin": True, "level": 10} + + def test_extra_fields_forbidden(self): + """Permissions should forbid extra fields.""" + with pytest.raises(ValidationError): + Permissions(require_admin=True, unknown_field="value") + + +class TestCommandTrigger: + """Tests for CommandTrigger model.""" + + def test_required_command(self): + """CommandTrigger requires command field.""" + trigger = CommandTrigger(command="hello") + assert trigger.type == "command" + assert trigger.command == "hello" + assert trigger.aliases == [] + assert trigger.description is None + + def test_with_aliases_and_description(self): + """CommandTrigger should accept aliases and description.""" + trigger = CommandTrigger( + command="hello", + aliases=["hi", "hey"], + description="Say hello", + ) + assert trigger.command == "hello" + assert trigger.aliases == ["hi", "hey"] + assert trigger.description == "Say hello" + + def test_type_literal(self): + """CommandTrigger type should always be 'command'.""" + trigger = CommandTrigger(command="test") + assert trigger.type == "command" + + def test_extra_fields_forbidden(self): + """CommandTrigger should forbid extra fields.""" + with pytest.raises(ValidationError): + CommandTrigger(command="test", extra="field") + + +class TestMessageTrigger: + """Tests for MessageTrigger model.""" + + def test_default_values(self): + """MessageTrigger should have default values.""" + trigger = MessageTrigger() + assert trigger.type == "message" + assert trigger.regex is None + assert trigger.keywords == [] + assert trigger.platforms == [] + + def test_with_regex(self): + """MessageTrigger should accept regex pattern.""" + trigger = MessageTrigger(regex=r"^hello.*$") + assert trigger.regex == r"^hello.*$" + + def test_with_keywords(self): + """MessageTrigger should accept keywords.""" + trigger = MessageTrigger(keywords=["hello", "hi"]) + assert trigger.keywords == ["hello", "hi"] + + def test_with_platforms(self): + """MessageTrigger should accept platforms.""" + trigger = MessageTrigger(platforms=["wechat", "qq"]) + assert trigger.platforms == ["wechat", "qq"] + + def test_with_all_fields(self): + """MessageTrigger should accept all fields.""" + trigger = MessageTrigger( + regex=r"test", + keywords=["keyword"], + platforms=["platform"], + ) + assert trigger.regex == "test" + assert trigger.keywords == ["keyword"] + assert trigger.platforms == ["platform"] + + +class TestEventTrigger: + """Tests for EventTrigger model.""" + + def test_required_event_type(self): + """EventTrigger requires event_type field.""" + trigger = EventTrigger(event_type="message") + assert trigger.type == "event" + assert trigger.event_type == "message" + + def test_numeric_event_type(self): + """EventTrigger should accept numeric string event_type.""" + trigger = EventTrigger(event_type="3") + assert trigger.event_type == "3" + + def test_type_literal(self): + """EventTrigger type should always be 'event'.""" + trigger = EventTrigger(event_type="custom") + assert trigger.type == "event" + + +class TestScheduleTrigger: + """Tests for ScheduleTrigger model.""" + + def test_with_cron(self): + """ScheduleTrigger should accept cron expression.""" + trigger = ScheduleTrigger(cron="0 9 * * *") + assert trigger.type == "schedule" + assert trigger.cron == "0 9 * * *" + assert trigger.interval_seconds is None + + def test_with_interval_seconds(self): + """ScheduleTrigger should accept interval_seconds.""" + trigger = ScheduleTrigger(interval_seconds=60) + assert trigger.interval_seconds == 60 + assert trigger.cron is None + + def test_requires_exactly_one_strategy(self): + """ScheduleTrigger must have exactly one of cron or interval_seconds.""" + # Neither provided should raise + with pytest.raises(ValidationError) as exc_info: + ScheduleTrigger() + assert "必须且只能有一个非 null" in str(exc_info.value) + + # Both provided should raise + with pytest.raises(ValidationError) as exc_info: + ScheduleTrigger(cron="* * * * *", interval_seconds=10) + assert "必须且只能有一个非 null" in str(exc_info.value) + + def test_valid_cron_expressions(self): + """ScheduleTrigger should accept various cron expressions.""" + trigger1 = ScheduleTrigger(cron="*/5 * * * *") + assert trigger1.cron == "*/5 * * * *" + + trigger2 = ScheduleTrigger(cron="0 0 1 1 *") + assert trigger2.cron == "0 0 1 1 *" + + def test_valid_intervals(self): + """ScheduleTrigger should accept various intervals.""" + trigger1 = ScheduleTrigger(interval_seconds=30) + assert trigger1.interval_seconds == 30 + + trigger2 = ScheduleTrigger(interval_seconds=3600) + assert trigger2.interval_seconds == 3600 + + +class TestHandlerDescriptor: + """Tests for HandlerDescriptor model.""" + + def test_required_id_and_trigger(self): + """HandlerDescriptor requires id and trigger.""" + trigger = CommandTrigger(command="hello") + handler = HandlerDescriptor(id="test.handler", trigger=trigger) + assert handler.id == "test.handler" + assert handler.trigger == trigger + assert handler.priority == 0 + assert handler.permissions.require_admin is False + + def test_with_priority_and_permissions(self): + """HandlerDescriptor should accept priority and permissions.""" + trigger = CommandTrigger(command="admin") + perms = Permissions(require_admin=True, level=5) + handler = HandlerDescriptor( + id="admin.handler", + trigger=trigger, + priority=10, + permissions=perms, + ) + assert handler.priority == 10 + assert handler.permissions.require_admin is True + assert handler.permissions.level == 5 + + def test_with_event_trigger(self): + """HandlerDescriptor should work with EventTrigger.""" + trigger = EventTrigger(event_type="message") + handler = HandlerDescriptor(id="event.handler", trigger=trigger) + assert handler.trigger.type == "event" + assert handler.trigger.event_type == "message" + + def test_with_schedule_trigger(self): + """HandlerDescriptor should work with ScheduleTrigger.""" + trigger = ScheduleTrigger(cron="0 9 * * *") + handler = HandlerDescriptor(id="scheduled.handler", trigger=trigger) + assert handler.trigger.type == "schedule" + assert handler.trigger.cron == "0 9 * * *" + + def test_model_dump(self): + """HandlerDescriptor should serialize correctly.""" + trigger = CommandTrigger(command="test", aliases=["t"]) + perms = Permissions(require_admin=True, level=5) + handler = HandlerDescriptor( + id="test.handler", + trigger=trigger, + priority=10, + permissions=perms, + ) + data = handler.model_dump() + assert data["id"] == "test.handler" + assert data["priority"] == 10 + assert data["trigger"]["type"] == "command" + assert data["trigger"]["command"] == "test" + assert data["permissions"]["require_admin"] is True + + def test_extra_fields_forbidden(self): + """HandlerDescriptor should forbid extra fields.""" + trigger = CommandTrigger(command="test") + with pytest.raises(ValidationError): + HandlerDescriptor(id="test", trigger=trigger, extra="field") + + +class TestCapabilityDescriptor: + """Tests for CapabilityDescriptor model.""" + + def test_required_name_and_description(self): + """CapabilityDescriptor requires name and description.""" + cap = CapabilityDescriptor(name="llm.chat", description="Chat with LLM") + assert cap.name == "llm.chat" + assert cap.description == "Chat with LLM" + assert cap.input_schema is None + assert cap.output_schema is None + assert cap.supports_stream is False + assert cap.cancelable is False + + def test_with_schemas(self): + """CapabilityDescriptor should accept input/output schemas.""" + cap = CapabilityDescriptor( + name="db.query", + description="Query database", + input_schema={"type": "object", "properties": {"sql": {"type": "string"}}}, + output_schema={"type": "array"}, + ) + assert cap.input_schema["type"] == "object" + assert cap.output_schema["type"] == "array" + + def test_with_stream_and_cancelable(self): + """CapabilityDescriptor should accept stream and cancelable flags.""" + cap = CapabilityDescriptor( + name="llm.stream", + description="Stream chat", + supports_stream=True, + cancelable=True, + ) + assert cap.supports_stream is True + assert cap.cancelable is True + + def test_model_dump(self): + """CapabilityDescriptor should serialize correctly.""" + cap = CapabilityDescriptor( + name="test.cap", + description="Test capability", + supports_stream=True, + ) + data = cap.model_dump() + assert data["name"] == "test.cap" + assert data["description"] == "Test capability" + assert data["supports_stream"] is True + + def test_extra_fields_forbidden(self): + """CapabilityDescriptor should forbid extra fields.""" + with pytest.raises(ValidationError): + CapabilityDescriptor( + name="test", + description="Test", + extra="field", + ) + + +class TestTriggerDiscriminator: + """Tests for Trigger type discriminator.""" + + def test_command_trigger_discrimination(self): + """CommandTrigger should be correctly discriminated.""" + handler = HandlerDescriptor( + id="cmd.handler", + trigger={"type": "command", "command": "test"}, + ) + assert isinstance(handler.trigger, CommandTrigger) + assert handler.trigger.command == "test" + + def test_message_trigger_discrimination(self): + """MessageTrigger should be correctly discriminated.""" + handler = HandlerDescriptor( + id="msg.handler", + trigger={"type": "message", "keywords": ["hello"]}, + ) + assert isinstance(handler.trigger, MessageTrigger) + assert handler.trigger.keywords == ["hello"] + + def test_event_trigger_discrimination(self): + """EventTrigger should be correctly discriminated.""" + handler = HandlerDescriptor( + id="evt.handler", + trigger={"type": "event", "event_type": "message"}, + ) + assert isinstance(handler.trigger, EventTrigger) + assert handler.trigger.event_type == "message" + + def test_schedule_trigger_discrimination(self): + """ScheduleTrigger should be correctly discriminated.""" + handler = HandlerDescriptor( + id="sched.handler", + trigger={"type": "schedule", "cron": "0 9 * * *"}, + ) + assert isinstance(handler.trigger, ScheduleTrigger) + assert handler.trigger.cron == "0 9 * * *" diff --git a/tests_v4/test_protocol_legacy_adapter.py b/tests_v4/test_protocol_legacy_adapter.py new file mode 100644 index 000000000..8a02c04fc --- /dev/null +++ b/tests_v4/test_protocol_legacy_adapter.py @@ -0,0 +1,768 @@ +""" +Tests for protocol/legacy_adapter.py - Legacy protocol adapter. +""" +from __future__ import annotations + +import pytest + +from astrbot_sdk.protocol.descriptors import EventTrigger, HandlerDescriptor, Permissions +from astrbot_sdk.protocol.legacy_adapter import ( + LEGACY_ADAPTER_MESSAGE_EVENT, + LEGACY_CONTEXT_CAPABILITY, + LEGACY_HANDSHAKE_METADATA_KEY, + LEGACY_JSONRPC_VERSION, + LEGACY_PLUGIN_KEYS_METADATA_KEY, + LegacyAdapter, + LegacyErrorData, + LegacyErrorResponse, + LegacyRequest, + LegacySuccessResponse, + cancel_to_legacy_request, + event_to_legacy_notification, + initialize_to_legacy_handshake_response, + invoke_to_legacy_request, + legacy_message_to_v4, + legacy_request_to_invoke, + legacy_response_to_message, + parse_legacy_message, + result_to_legacy_response, +) +from astrbot_sdk.protocol.messages import ( + CancelMessage, + ErrorPayload, + EventMessage, + InitializeMessage, + InvokeMessage, + PeerInfo, + ResultMessage, +) + + +class TestLegacyRequest: + """Tests for LegacyRequest model.""" + + def test_default_values(self): + """LegacyRequest should have default values.""" + req = LegacyRequest(method="test_method") + assert req.jsonrpc == LEGACY_JSONRPC_VERSION + assert req.id is None + assert req.method == "test_method" + assert req.params == {} + + def test_with_all_fields(self): + """LegacyRequest should accept all fields.""" + req = LegacyRequest( + id="req_001", + method="handshake", + params={"key": "value"}, + ) + assert req.id == "req_001" + assert req.method == "handshake" + assert req.params["key"] == "value" + + +class TestLegacySuccessResponse: + """Tests for LegacySuccessResponse model.""" + + def test_with_result(self): + """LegacySuccessResponse should accept result.""" + resp = LegacySuccessResponse(id="req_001", result={"status": "ok"}) + assert resp.jsonrpc == LEGACY_JSONRPC_VERSION + assert resp.id == "req_001" + assert resp.result["status"] == "ok" + + +class TestLegacyErrorResponse: + """Tests for LegacyErrorResponse model.""" + + def test_with_error(self): + """LegacyErrorResponse should accept error.""" + error = LegacyErrorData(code=-32000, message="Server error") + resp = LegacyErrorResponse(id="req_001", error=error) + assert resp.id == "req_001" + assert resp.error.code == -32000 + assert resp.error.message == "Server error" + + +class TestLegacyErrorData: + """Tests for LegacyErrorData model.""" + + def test_default_code(self): + """LegacyErrorData should have default code.""" + error = LegacyErrorData(message="Error") + assert error.code == -32000 + assert error.message == "Error" + assert error.data is None + + def test_with_data(self): + """LegacyErrorData should accept data.""" + error = LegacyErrorData( + code=-32600, + message="Invalid Request", + data={"details": "Missing field"}, + ) + assert error.code == -32600 + assert error.data["details"] == "Missing field" + + +class TestParseLegacyMessage: + """Tests for parse_legacy_message function.""" + + def test_parse_request(self): + """parse_legacy_message should parse LegacyRequest.""" + payload = {"jsonrpc": "2.0", "id": "1", "method": "test", "params": {}} + msg = parse_legacy_message(payload) + assert isinstance(msg, LegacyRequest) + assert msg.method == "test" + + def test_parse_request_from_json(self): + """parse_legacy_message should parse request from JSON string.""" + json_str = '{"jsonrpc": "2.0", "id": "1", "method": "handshake"}' + msg = parse_legacy_message(json_str) + assert isinstance(msg, LegacyRequest) + assert msg.method == "handshake" + + def test_parse_request_from_bytes(self): + """parse_legacy_message should parse request from bytes.""" + json_bytes = b'{"jsonrpc": "2.0", "method": "call_handler"}' + msg = parse_legacy_message(json_bytes) + assert isinstance(msg, LegacyRequest) + assert msg.method == "call_handler" + + def test_parse_success_response(self): + """parse_legacy_message should parse LegacySuccessResponse.""" + payload = {"jsonrpc": "2.0", "id": "1", "result": {"status": "ok"}} + msg = parse_legacy_message(payload) + assert isinstance(msg, LegacySuccessResponse) + assert msg.result["status"] == "ok" + + def test_parse_error_response(self): + """parse_legacy_message should parse LegacyErrorResponse.""" + payload = { + "jsonrpc": "2.0", + "id": "1", + "error": {"code": -32000, "message": "Error"}, + } + msg = parse_legacy_message(payload) + assert isinstance(msg, LegacyErrorResponse) + assert msg.error.message == "Error" + + def test_parse_unknown_raises(self): + """parse_legacy_message should raise for unknown type.""" + with pytest.raises(ValueError) as exc_info: + parse_legacy_message({"jsonrpc": "2.0", "unknown": "field"}) + assert "未知" in str(exc_info.value) + + def test_pass_through_legacy_message(self): + """parse_legacy_message should pass through already-parsed messages.""" + req = LegacyRequest(method="test") + result = parse_legacy_message(req) + assert result is req + + +class TestLegacyAdapterInit: + """Tests for LegacyAdapter initialization.""" + + def test_default_values(self): + """LegacyAdapter should have default values.""" + adapter = LegacyAdapter() + assert adapter.protocol_version == "1.0" + assert adapter.legacy_peer_name == "legacy-peer" + assert adapter.legacy_peer_role == "plugin" + assert adapter.legacy_peer_version is None + + def test_custom_values(self): + """LegacyAdapter should accept custom values.""" + adapter = LegacyAdapter( + protocol_version="2.0", + legacy_peer_name="custom-peer", + legacy_peer_role="core", + legacy_peer_version="1.5.0", + ) + assert adapter.protocol_version == "2.0" + assert adapter.legacy_peer_name == "custom-peer" + assert adapter.legacy_peer_role == "core" + assert adapter.legacy_peer_version == "1.5.0" + + +class TestLegacyAdapterTrackHandler: + """Tests for LegacyAdapter.track_handler method.""" + + def test_track_handler(self): + """track_handler should store handler name by request ID.""" + adapter = LegacyAdapter() + adapter.track_handler("req_001", "module.handler") + assert adapter._handler_names_by_request_id["req_001"] == "module.handler" + + def test_track_handler_empty_id(self): + """track_handler should not store for empty request ID.""" + adapter = LegacyAdapter() + adapter.track_handler("", "module.handler") + assert "" not in adapter._handler_names_by_request_id + + +class TestLegacyAdapterHandshake: + """Tests for LegacyAdapter handshake handling.""" + + def test_legacy_request_to_handshake(self): + """legacy_request_to_message should convert handshake request.""" + adapter = LegacyAdapter() + req = LegacyRequest(id="req_001", method="handshake", params={}) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, InitializeMessage) + assert msg.protocol_version == "1.0" + assert msg.peer.name == "legacy-peer" + assert msg.peer.role == "plugin" + assert msg.metadata.get("legacy_handshake") is True + + def test_build_legacy_handshake_request(self): + """build_legacy_handshake_request should create handshake request.""" + adapter = LegacyAdapter() + result = adapter.build_legacy_handshake_request("req_001") + + assert result["jsonrpc"] == LEGACY_JSONRPC_VERSION + assert result["id"] == "req_001" + assert result["method"] == "handshake" + + +class TestLegacyAdapterCallHandler: + """Tests for LegacyAdapter call_handler handling.""" + + def test_legacy_request_to_call_handler(self): + """legacy_request_to_message should convert call_handler request.""" + adapter = LegacyAdapter() + req = LegacyRequest( + id="req_001", + method="call_handler", + params={ + "handler_full_name": "module.handler", + "event": {"type": "message"}, + "args": {"key": "value"}, + }, + ) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, InvokeMessage) + assert msg.capability == "handler.invoke" + assert msg.input["handler_id"] == "module.handler" + assert msg.input["event"]["type"] == "message" + + def test_call_handler_tracks_handler(self): + """call_handler should track handler name.""" + adapter = LegacyAdapter() + req = LegacyRequest( + id="req_001", + method="call_handler", + params={"handler_full_name": "test.handler"}, + ) + adapter.legacy_request_to_message(req) + assert adapter._handler_names_by_request_id["req_001"] == "test.handler" + + +class TestLegacyAdapterContextFunction: + """Tests for LegacyAdapter call_context_function handling.""" + + def test_legacy_request_to_context_function(self): + """legacy_request_to_message should convert call_context_function.""" + adapter = LegacyAdapter() + req = LegacyRequest( + id="req_001", + method="call_context_function", + params={"name": "get_user", "args": {"user_id": 123}}, + ) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, InvokeMessage) + assert msg.capability == LEGACY_CONTEXT_CAPABILITY + assert msg.input["name"] == "get_user" + assert msg.input["args"]["user_id"] == 123 + + +class TestLegacyAdapterStreamMethods: + """Tests for LegacyAdapter stream handling.""" + + def test_handler_stream_start(self): + """legacy_request_to_message should convert handler_stream_start.""" + adapter = LegacyAdapter() + req = LegacyRequest( + method="handler_stream_start", + params={"id": "stream_001", "handler_full_name": "module.handler"}, + ) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, EventMessage) + assert msg.phase == "started" + + def test_handler_stream_update(self): + """legacy_request_to_message should convert handler_stream_update.""" + adapter = LegacyAdapter() + req = LegacyRequest( + method="handler_stream_update", + params={ + "id": "stream_001", + "handler_full_name": "module.handler", + "data": {"text": "chunk"}, + }, + ) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, EventMessage) + assert msg.phase == "delta" + assert msg.data["text"] == "chunk" + + def test_handler_stream_end_completed(self): + """legacy_request_to_message should convert handler_stream_end (completed).""" + adapter = LegacyAdapter() + req = LegacyRequest( + method="handler_stream_end", + params={"id": "stream_001", "handler_full_name": "module.handler"}, + ) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, EventMessage) + assert msg.phase == "completed" + + def test_handler_stream_end_failed(self): + """legacy_request_to_message should convert handler_stream_end (failed).""" + adapter = LegacyAdapter() + req = LegacyRequest( + method="handler_stream_end", + params={ + "id": "stream_001", + "handler_full_name": "module.handler", + "error": {"message": "Something went wrong"}, + }, + ) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, EventMessage) + assert msg.phase == "failed" + assert msg.error is not None + assert msg.error.message == "Something went wrong" + + +class TestLegacyAdapterCancel: + """Tests for LegacyAdapter cancel handling.""" + + def test_legacy_request_to_cancel(self): + """legacy_request_to_message should convert cancel request.""" + adapter = LegacyAdapter() + req = LegacyRequest( + id="req_001", + method="cancel", + params={"reason": "user_cancelled"}, + ) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, CancelMessage) + assert msg.reason == "user_cancelled" + + +class TestLegacyAdapterGenericMethod: + """Tests for LegacyAdapter generic method handling.""" + + def test_unknown_method_becomes_invoke(self): + """Unknown methods should become InvokeMessage with capability=method.""" + adapter = LegacyAdapter() + req = LegacyRequest( + id="req_001", + method="custom.capability", + params={"key": "value"}, + ) + msg = adapter.legacy_request_to_message(req) + + assert isinstance(msg, InvokeMessage) + assert msg.capability == "custom.capability" + assert msg.input["key"] == "value" + + +class TestLegacyAdapterResponseHandling: + """Tests for LegacyAdapter response handling.""" + + def test_success_response_to_result(self): + """legacy_response_to_message should convert success response.""" + adapter = LegacyAdapter() + resp = LegacySuccessResponse(id="req_001", result={"status": "ok"}) + msg = adapter.legacy_response_to_message(resp) + + assert isinstance(msg, ResultMessage) + assert msg.success is True + assert msg.output["status"] == "ok" + + def test_error_response_to_result(self): + """legacy_error_to_result should convert error response.""" + adapter = LegacyAdapter() + error = LegacyErrorData(code=-32000, message="Server error") + resp = LegacyErrorResponse(id="req_001", error=error) + msg = adapter.legacy_error_to_result(resp) + + assert isinstance(msg, ResultMessage) + assert msg.success is False + assert msg.error.code == "legacy_rpc_error" + assert msg.error.message == "Server error" + + def test_handshake_response_to_initialize(self): + """legacy_response_to_message should detect handshake response.""" + adapter = LegacyAdapter() + resp = LegacySuccessResponse( + id="req_001", + result={ + "module.path": { + "name": "test-plugin", + "version": "1.0.0", + "handlers": [], + } + }, + ) + msg = adapter.legacy_response_to_message(resp) + + assert isinstance(msg, InitializeMessage) + assert msg.peer.name == "test-plugin" + assert msg.peer.version == "1.0.0" + + +class TestLegacyAdapterV4ToLegacy: + """Tests for LegacyAdapter V4 to legacy conversion.""" + + def test_initialize_to_legacy_handshake_response(self): + """initialize_to_legacy_handshake_response should convert InitializeMessage.""" + adapter = LegacyAdapter() + init_msg = InitializeMessage( + id="msg_001", + protocol_version="1.0", + peer=PeerInfo(name="test", role="plugin", version="1.0.0"), + handlers=[], + metadata={ + "plugin_id": "test-plugin", + "display_name": "Test Plugin", + }, + ) + result = adapter.initialize_to_legacy_handshake_response(init_msg) + + assert result["jsonrpc"] == LEGACY_JSONRPC_VERSION + assert result["id"] == "msg_001" + assert "result" in result + + def test_initialize_with_legacy_payload(self): + """initialize_to_legacy_handshake_response should preserve legacy payload.""" + adapter = LegacyAdapter() + legacy_payload = { + "module.path": { + "name": "test-plugin", + "version": "1.0.0", + "handlers": [], + } + } + init_msg = InitializeMessage( + id="msg_001", + protocol_version="1.0", + peer=PeerInfo(name="test", role="plugin", version="1.0.0"), + handlers=[], + metadata={LEGACY_HANDSHAKE_METADATA_KEY: legacy_payload}, + ) + result = adapter.initialize_to_legacy_handshake_response(init_msg) + + assert result["result"] == legacy_payload + + def test_invoke_to_legacy_request_handler(self): + """invoke_to_legacy_request should convert handler.invoke.""" + adapter = LegacyAdapter() + invoke_msg = InvokeMessage( + id="msg_001", + capability="handler.invoke", + input={ + "handler_id": "module.handler", + "event": {"type": "message"}, + "args": {}, + }, + ) + result = adapter.invoke_to_legacy_request(invoke_msg) + + assert result["method"] == "call_handler" + assert result["params"]["handler_full_name"] == "module.handler" + + def test_invoke_to_legacy_request_context_function(self): + """invoke_to_legacy_request should convert context function.""" + adapter = LegacyAdapter() + invoke_msg = InvokeMessage( + id="msg_001", + capability=LEGACY_CONTEXT_CAPABILITY, + input={"name": "get_user", "args": {}}, + ) + result = adapter.invoke_to_legacy_request(invoke_msg) + + assert result["method"] == "call_context_function" + assert result["params"]["name"] == "get_user" + + def test_invoke_to_legacy_request_generic(self): + """invoke_to_legacy_request should convert generic capability.""" + adapter = LegacyAdapter() + invoke_msg = InvokeMessage( + id="msg_001", + capability="custom.capability", + input={"key": "value"}, + ) + result = adapter.invoke_to_legacy_request(invoke_msg) + + assert result["method"] == "custom.capability" + assert result["params"]["key"] == "value" + + def test_result_to_legacy_response_success(self): + """result_to_legacy_response should convert success result.""" + adapter = LegacyAdapter() + result_msg = ResultMessage( + id="msg_001", + success=True, + output={"status": "ok"}, + ) + result = adapter.result_to_legacy_response(result_msg) + + assert "result" in result + assert result["result"]["status"] == "ok" + + def test_result_to_legacy_response_error(self): + """result_to_legacy_response should convert error result.""" + adapter = LegacyAdapter() + result_msg = ResultMessage( + id="msg_001", + success=False, + error=ErrorPayload(code="error", message="Failed"), + ) + result = adapter.result_to_legacy_response(result_msg) + + assert "error" in result + assert result["error"]["message"] == "Failed" + + def test_event_to_legacy_notification_started(self): + """event_to_legacy_notification should convert started event.""" + adapter = LegacyAdapter() + adapter.track_handler("msg_001", "module.handler") + event_msg = EventMessage(id="msg_001", phase="started") + result = adapter.event_to_legacy_notification(event_msg) + + assert result["method"] == "handler_stream_start" + assert result["params"]["handler_full_name"] == "module.handler" + + def test_event_to_legacy_notification_delta(self): + """event_to_legacy_notification should convert delta event.""" + adapter = LegacyAdapter() + event_msg = EventMessage( + id="msg_001", + phase="delta", + data={"text": "chunk"}, + ) + result = adapter.event_to_legacy_notification(event_msg) + + assert result["method"] == "handler_stream_update" + assert result["params"]["data"]["text"] == "chunk" + + def test_event_to_legacy_notification_completed(self): + """event_to_legacy_notification should convert completed event.""" + adapter = LegacyAdapter() + event_msg = EventMessage(id="msg_001", phase="completed") + result = adapter.event_to_legacy_notification(event_msg) + + assert result["method"] == "handler_stream_end" + + def test_event_to_legacy_notification_failed(self): + """event_to_legacy_notification should convert failed event.""" + adapter = LegacyAdapter() + event_msg = EventMessage( + id="msg_001", + phase="failed", + error=ErrorPayload(code="error", message="Failed"), + ) + result = adapter.event_to_legacy_notification(event_msg) + + assert result["method"] == "handler_stream_end" + assert result["params"]["error"]["message"] == "Failed" + + def test_cancel_to_legacy_request(self): + """cancel_to_legacy_request should convert cancel message.""" + adapter = LegacyAdapter() + cancel_msg = CancelMessage(id="msg_001", reason="user_request") + result = adapter.cancel_to_legacy_request(cancel_msg) + + assert result["method"] == "cancel" + assert result["params"]["reason"] == "user_request" + + +class TestLegacyAdapterHandlerDescriptors: + """Tests for LegacyAdapter handler descriptor conversion.""" + + def test_legacy_handlers_to_descriptors(self): + """_legacy_handlers_to_descriptors should convert handlers.""" + adapter = LegacyAdapter() + payload = { + "module.path": { + "handlers": [ + { + "handler_full_name": "module.handler", + "event_type": "3", + "extras_configs": { + "priority": 10, + "require_admin": True, + "level": 5, + }, + } + ] + } + } + handlers = adapter._legacy_handlers_to_descriptors(payload) + + assert len(handlers) == 1 + assert handlers[0].id == "module.handler" + assert handlers[0].priority == 10 + assert handlers[0].permissions.require_admin is True + assert handlers[0].permissions.level == 5 + + def test_descriptor_to_legacy_handler(self): + """_descriptor_to_legacy_handler should convert HandlerDescriptor.""" + descriptor = HandlerDescriptor( + id="module.handler", + trigger=EventTrigger(event_type="3"), + priority=10, + permissions=Permissions(require_admin=True, level=5), + ) + result = LegacyAdapter._descriptor_to_legacy_handler(descriptor) + + assert result["handler_full_name"] == "module.handler" + assert result["event_type"] == 3 + assert result["extras_configs"]["priority"] == 10 + assert result["extras_configs"]["require_admin"] is True + + +class TestLegacyAdapterHelpers: + """Tests for LegacyAdapter helper methods.""" + + def test_request_id_with_value(self): + """_request_id should return string value.""" + result = LegacyAdapter._request_id("req_001", "fallback") + assert result == "req_001" + + def test_request_id_with_none(self): + """_request_id should return fallback for None.""" + result = LegacyAdapter._request_id(None, "fallback") + assert result == "fallback" + + def test_request_id_with_empty_string(self): + """_request_id should return fallback for empty string.""" + result = LegacyAdapter._request_id("", "fallback") + assert result == "fallback" + + def test_as_dict_with_dict(self): + """_as_dict should pass through dict.""" + result = LegacyAdapter._as_dict({"key": "value"}, field_name="data") + assert result == {"key": "value"} + + def test_as_dict_with_none(self): + """_as_dict should return empty dict for None.""" + result = LegacyAdapter._as_dict(None, field_name="data") + assert result == {} + + def test_as_dict_with_other(self): + """_as_dict should wrap other values.""" + result = LegacyAdapter._as_dict("value", field_name="data") + assert result == {"data": "value"} + + def test_looks_like_handshake_payload_valid(self): + """_looks_like_handshake_payload should detect valid payload.""" + payload = {"module.path": {"handlers": []}} + assert LegacyAdapter._looks_like_handshake_payload(payload) is True + + def test_looks_like_handshake_payload_invalid(self): + """_looks_like_handshake_payload should reject invalid payload.""" + assert LegacyAdapter._looks_like_handshake_payload({}) is False + assert LegacyAdapter._looks_like_handshake_payload({"key": "value"}) is False + assert LegacyAdapter._looks_like_handshake_payload(None) is False + + +class TestLegacyConvenienceFunctions: + """Tests for module-level convenience functions.""" + + def test_legacy_message_to_v4(self): + """legacy_message_to_v4 should convert legacy message.""" + payload = {"jsonrpc": "2.0", "method": "handshake"} + msg = legacy_message_to_v4(payload) + assert isinstance(msg, InitializeMessage) + + def test_legacy_request_to_invoke(self): + """legacy_request_to_invoke should convert to InvokeMessage.""" + payload = { + "jsonrpc": "2.0", + "id": "1", + "method": "custom.capability", + "params": {}, + } + msg = legacy_request_to_invoke(payload) + assert isinstance(msg, InvokeMessage) + assert msg.capability == "custom.capability" + + def test_legacy_request_to_invoke_non_invoke_raises(self): + """legacy_request_to_invoke should raise for non-invoke messages.""" + payload = {"jsonrpc": "2.0", "method": "handshake"} + with pytest.raises(ValueError, match="不能直接映射为 invoke"): + legacy_request_to_invoke(payload) + + def test_legacy_response_to_message(self): + """legacy_response_to_message should convert response.""" + payload = {"jsonrpc": "2.0", "id": "1", "result": {"status": "ok"}} + msg = legacy_response_to_message(payload) + assert isinstance(msg, ResultMessage) + + def test_initialize_to_legacy_handshake_response(self): + """initialize_to_legacy_handshake_response should convert.""" + msg = InitializeMessage( + id="msg_001", + protocol_version="1.0", + peer=PeerInfo(name="test", role="plugin"), + handlers=[], + ) + result = initialize_to_legacy_handshake_response(msg) + assert result["jsonrpc"] == LEGACY_JSONRPC_VERSION + + def test_invoke_to_legacy_request(self): + """invoke_to_legacy_request should convert.""" + msg = InvokeMessage(id="msg_001", capability="test.cap", input={}) + result = invoke_to_legacy_request(msg) + assert result["method"] == "test.cap" + + def test_result_to_legacy_response(self): + """result_to_legacy_response should convert.""" + msg = ResultMessage(id="msg_001", success=True, output={"ok": True}) + result = result_to_legacy_response(msg) + assert result["result"]["ok"] is True + + def test_event_to_legacy_notification(self): + """event_to_legacy_notification should convert.""" + msg = EventMessage(id="msg_001", phase="started") + result = event_to_legacy_notification(msg, handler_full_name="test.handler") + assert result["method"] == "handler_stream_start" + + def test_cancel_to_legacy_request(self): + """cancel_to_legacy_request should convert.""" + msg = CancelMessage(id="msg_001", reason="test") + result = cancel_to_legacy_request(msg) + assert result["method"] == "cancel" + + +class TestLegacyConstants: + """Tests for legacy adapter constants.""" + + def test_jsonrpc_version(self): + """LEGACY_JSONRPC_VERSION should be 2.0.""" + assert LEGACY_JSONRPC_VERSION == "2.0" + + def test_context_capability(self): + """LEGACY_CONTEXT_CAPABILITY should be internal capability.""" + assert LEGACY_CONTEXT_CAPABILITY == "internal.legacy.call_context_function" + + def test_message_event(self): + """LEGACY_ADAPTER_MESSAGE_EVENT should be 3.""" + assert LEGACY_ADAPTER_MESSAGE_EVENT == 3 + + def test_metadata_keys(self): + """Metadata keys should be defined.""" + assert LEGACY_HANDSHAKE_METADATA_KEY == "legacy_handshake_payload" + assert LEGACY_PLUGIN_KEYS_METADATA_KEY == "legacy_plugin_keys" diff --git a/tests_v4/test_protocol_messages.py b/tests_v4/test_protocol_messages.py new file mode 100644 index 000000000..f4136f731 --- /dev/null +++ b/tests_v4/test_protocol_messages.py @@ -0,0 +1,509 @@ +""" +Tests for protocol/messages.py - Protocol message models. +""" +from __future__ import annotations + +import json + +import pytest +from pydantic import ValidationError + +from astrbot_sdk.protocol.descriptors import CommandTrigger, HandlerDescriptor +from astrbot_sdk.protocol.messages import ( + CancelMessage, + ErrorPayload, + EventMessage, + InitializeMessage, + InitializeOutput, + InvokeMessage, + PeerInfo, + ResultMessage, + parse_message, +) + + +class TestErrorPayload: + """Tests for ErrorPayload model.""" + + def test_required_code_and_message(self): + """ErrorPayload requires code and message.""" + error = ErrorPayload(code="test_error", message="Test error occurred") + assert error.code == "test_error" + assert error.message == "Test error occurred" + assert error.hint == "" + assert error.retryable is False + + def test_with_all_fields(self): + """ErrorPayload should accept all fields.""" + error = ErrorPayload( + code="server_error", + message="Internal server error", + hint="Try again later", + retryable=True, + ) + assert error.code == "server_error" + assert error.message == "Internal server error" + assert error.hint == "Try again later" + assert error.retryable is True + + def test_model_dump(self): + """ErrorPayload should serialize correctly.""" + error = ErrorPayload( + code="not_found", + message="Resource not found", + hint="Check the ID", + ) + data = error.model_dump() + assert data == { + "code": "not_found", + "message": "Resource not found", + "hint": "Check the ID", + "retryable": False, + } + + def test_extra_fields_forbidden(self): + """ErrorPayload should forbid extra fields.""" + with pytest.raises(ValidationError): + ErrorPayload(code="test", message="test", extra="field") + + +class TestPeerInfo: + """Tests for PeerInfo model.""" + + def test_required_name_and_role(self): + """PeerInfo requires name and role.""" + peer = PeerInfo(name="test-plugin", role="plugin") + assert peer.name == "test-plugin" + assert peer.role == "plugin" + assert peer.version is None + + def test_with_version(self): + """PeerInfo should accept version.""" + peer = PeerInfo(name="my-plugin", role="plugin", version="1.0.0") + assert peer.version == "1.0.0" + + def test_role_must_be_valid(self): + """PeerInfo role must be 'plugin' or 'core'.""" + peer1 = PeerInfo(name="p1", role="plugin") + assert peer1.role == "plugin" + + peer2 = PeerInfo(name="p2", role="core") + assert peer2.role == "core" + + with pytest.raises(ValidationError): + PeerInfo(name="p3", role="invalid") + + def test_model_dump(self): + """PeerInfo should serialize correctly.""" + peer = PeerInfo(name="test", role="plugin", version="2.0.0") + data = peer.model_dump() + assert data == {"name": "test", "role": "plugin", "version": "2.0.0"} + + def test_extra_fields_forbidden(self): + """PeerInfo should forbid extra fields.""" + with pytest.raises(ValidationError): + PeerInfo(name="test", role="plugin", extra="field") + + +class TestInitializeMessage: + """Tests for InitializeMessage model.""" + + def test_required_fields(self): + """InitializeMessage requires id, protocol_version, and peer.""" + peer = PeerInfo(name="test", role="plugin") + msg = InitializeMessage( + id="msg_001", + protocol_version="1.0", + peer=peer, + ) + assert msg.type == "initialize" + assert msg.id == "msg_001" + assert msg.protocol_version == "1.0" + assert msg.peer == peer + assert msg.handlers == [] + assert msg.metadata == {} + + def test_with_handlers(self): + """InitializeMessage should accept handlers.""" + peer = PeerInfo(name="test", role="plugin") + handler = HandlerDescriptor( + id="test.handler", + trigger=CommandTrigger(command="hello"), + ) + msg = InitializeMessage( + id="msg_002", + protocol_version="1.0", + peer=peer, + handlers=[handler], + ) + assert len(msg.handlers) == 1 + assert msg.handlers[0].id == "test.handler" + + def test_with_metadata(self): + """InitializeMessage should accept metadata.""" + peer = PeerInfo(name="test", role="plugin") + msg = InitializeMessage( + id="msg_003", + protocol_version="1.0", + peer=peer, + metadata={"author": "test", "version": "1.0.0"}, + ) + assert msg.metadata["author"] == "test" + assert msg.metadata["version"] == "1.0.0" + + def test_model_dump_json(self): + """InitializeMessage should serialize to JSON correctly.""" + peer = PeerInfo(name="test", role="plugin", version="1.0.0") + handler = HandlerDescriptor( + id="test.handler", + trigger=CommandTrigger(command="hello"), + ) + msg = InitializeMessage( + id="msg_004", + protocol_version="1.0", + peer=peer, + handlers=[handler], + metadata={"key": "value"}, + ) + json_str = msg.model_dump_json() + data = json.loads(json_str) + assert data["type"] == "initialize" + assert data["id"] == "msg_004" + assert data["peer"]["name"] == "test" + assert len(data["handlers"]) == 1 + + +class TestInitializeOutput: + """Tests for InitializeOutput model.""" + + def test_required_peer(self): + """InitializeOutput requires peer.""" + peer = PeerInfo(name="core", role="core") + output = InitializeOutput(peer=peer) + assert output.peer == peer + assert output.capabilities == [] + assert output.metadata == {} + + def test_with_capabilities(self): + """InitializeOutput should accept capabilities.""" + from astrbot_sdk.protocol.descriptors import CapabilityDescriptor + + peer = PeerInfo(name="core", role="core") + cap = CapabilityDescriptor(name="llm.chat", description="Chat capability") + output = InitializeOutput(peer=peer, capabilities=[cap]) + assert len(output.capabilities) == 1 + assert output.capabilities[0].name == "llm.chat" + + def test_with_metadata(self): + """InitializeOutput should accept metadata.""" + peer = PeerInfo(name="core", role="core") + output = InitializeOutput(peer=peer, metadata={"session": "abc"}) + assert output.metadata["session"] == "abc" + + +class TestResultMessage: + """Tests for ResultMessage model.""" + + def test_success_result(self): + """ResultMessage for success case.""" + msg = ResultMessage(id="msg_001", success=True, output={"text": "ok"}) + assert msg.type == "result" + assert msg.id == "msg_001" + assert msg.success is True + assert msg.output["text"] == "ok" + assert msg.error is None + assert msg.kind is None + + def test_error_result(self): + """ResultMessage for error case.""" + error = ErrorPayload(code="not_found", message="Resource not found") + msg = ResultMessage(id="msg_002", success=False, error=error) + assert msg.success is False + assert msg.error.code == "not_found" + assert msg.error.message == "Resource not found" + + def test_with_kind(self): + """ResultMessage should accept kind.""" + msg = ResultMessage( + id="msg_003", + kind="initialize_result", + success=True, + ) + assert msg.kind == "initialize_result" + + def test_default_output(self): + """ResultMessage should have empty dict as default output.""" + msg = ResultMessage(id="msg_004", success=True) + assert msg.output == {} + + +class TestInvokeMessage: + """Tests for InvokeMessage model.""" + + def test_required_fields(self): + """InvokeMessage requires id and capability.""" + msg = InvokeMessage(id="msg_001", capability="llm.chat") + assert msg.type == "invoke" + assert msg.id == "msg_001" + assert msg.capability == "llm.chat" + assert msg.input == {} + assert msg.stream is False + + def test_with_input(self): + """InvokeMessage should accept input payload.""" + msg = InvokeMessage( + id="msg_002", + capability="db.get", + input={"key": "user:123"}, + ) + assert msg.input["key"] == "user:123" + + def test_with_stream(self): + """InvokeMessage should accept stream flag.""" + msg = InvokeMessage( + id="msg_003", + capability="llm.stream", + input={"prompt": "hello"}, + stream=True, + ) + assert msg.stream is True + + def test_model_dump(self): + """InvokeMessage should serialize correctly.""" + msg = InvokeMessage( + id="msg_004", + capability="test.cap", + input={"data": "value"}, + stream=True, + ) + data = msg.model_dump() + assert data["type"] == "invoke" + assert data["capability"] == "test.cap" + assert data["input"] == {"data": "value"} + assert data["stream"] is True + + +class TestEventMessage: + """Tests for EventMessage model.""" + + def test_started_phase(self): + """EventMessage with started phase.""" + msg = EventMessage(id="msg_001", phase="started") + assert msg.type == "event" + assert msg.phase == "started" + assert msg.data == {} + assert msg.output == {} + assert msg.error is None + + def test_delta_phase(self): + """EventMessage with delta phase and data.""" + msg = EventMessage( + id="msg_002", + phase="delta", + data={"text": "chunk"}, + ) + assert msg.phase == "delta" + assert msg.data["text"] == "chunk" + + def test_completed_phase(self): + """EventMessage with completed phase.""" + msg = EventMessage( + id="msg_003", + phase="completed", + output={"result": "done"}, + ) + assert msg.phase == "completed" + assert msg.output["result"] == "done" + + def test_failed_phase(self): + """EventMessage with failed phase.""" + error = ErrorPayload(code="runtime_error", message="Failed") + msg = EventMessage( + id="msg_004", + phase="failed", + error=error, + ) + assert msg.phase == "failed" + assert msg.error.code == "runtime_error" + + def test_invalid_phase(self): + """EventMessage should reject invalid phase.""" + with pytest.raises(ValidationError): + EventMessage(id="msg_005", phase="invalid") + + +class TestCancelMessage: + """Tests for CancelMessage model.""" + + def test_default_reason(self): + """CancelMessage should have default reason.""" + msg = CancelMessage(id="msg_001") + assert msg.type == "cancel" + assert msg.id == "msg_001" + assert msg.reason == "user_cancelled" + + def test_custom_reason(self): + """CancelMessage should accept custom reason.""" + msg = CancelMessage(id="msg_002", reason="timeout") + assert msg.reason == "timeout" + + def test_model_dump(self): + """CancelMessage should serialize correctly.""" + msg = CancelMessage(id="msg_003", reason="user_request") + data = msg.model_dump() + assert data == { + "type": "cancel", + "id": "msg_003", + "reason": "user_request", + } + + +class TestParseMessage: + """Tests for parse_message function.""" + + def test_parse_initialize_from_dict(self): + """parse_message should parse InitializeMessage from dict.""" + data = { + "type": "initialize", + "id": "msg_001", + "protocol_version": "1.0", + "peer": {"name": "test", "role": "plugin"}, + } + msg = parse_message(data) + assert isinstance(msg, InitializeMessage) + assert msg.id == "msg_001" + assert msg.peer.name == "test" + + def test_parse_result_from_dict(self): + """parse_message should parse ResultMessage from dict.""" + data = { + "type": "result", + "id": "msg_002", + "success": True, + "output": {"text": "ok"}, + } + msg = parse_message(data) + assert isinstance(msg, ResultMessage) + assert msg.success is True + + def test_parse_invoke_from_dict(self): + """parse_message should parse InvokeMessage from dict.""" + data = { + "type": "invoke", + "id": "msg_003", + "capability": "test.cap", + "input": {"key": "value"}, + } + msg = parse_message(data) + assert isinstance(msg, InvokeMessage) + assert msg.capability == "test.cap" + + def test_parse_event_from_dict(self): + """parse_message should parse EventMessage from dict.""" + data = { + "type": "event", + "id": "msg_004", + "phase": "delta", + "data": {"text": "chunk"}, + } + msg = parse_message(data) + assert isinstance(msg, EventMessage) + assert msg.phase == "delta" + + def test_parse_cancel_from_dict(self): + """parse_message should parse CancelMessage from dict.""" + data = { + "type": "cancel", + "id": "msg_005", + "reason": "user_request", + } + msg = parse_message(data) + assert isinstance(msg, CancelMessage) + assert msg.reason == "user_request" + + def test_parse_from_json_string(self): + """parse_message should parse from JSON string.""" + json_str = '{"type": "invoke", "id": "msg_006", "capability": "test"}' + msg = parse_message(json_str) + assert isinstance(msg, InvokeMessage) + assert msg.capability == "test" + + def test_parse_from_bytes(self): + """parse_message should parse from bytes.""" + json_bytes = b'{"type": "result", "id": "msg_007", "success": true}' + msg = parse_message(json_bytes) + assert isinstance(msg, ResultMessage) + assert msg.success is True + + def test_parse_unknown_type_raises(self): + """parse_message should raise for unknown type.""" + with pytest.raises(ValueError) as exc_info: + parse_message({"type": "unknown"}) + assert "未知消息类型" in str(exc_info.value) + + def test_roundtrip_serialize_deserialize(self): + """Message should survive serialize/deserialize roundtrip.""" + original = InitializeMessage( + id="msg_008", + protocol_version="1.0", + peer=PeerInfo(name="test", role="plugin", version="1.0.0"), + handlers=[ + HandlerDescriptor( + id="test.handler", + trigger=CommandTrigger(command="hello"), + ) + ], + metadata={"key": "value"}, + ) + json_str = original.model_dump_json() + parsed = parse_message(json_str) + assert isinstance(parsed, InitializeMessage) + assert parsed.id == original.id + assert parsed.peer.name == original.peer.name + assert len(parsed.handlers) == 1 + + +class TestMessageExtraForbidden: + """Tests for extra field rejection across all message types.""" + + def test_initialize_extra_forbidden(self): + """InitializeMessage should reject extra fields.""" + with pytest.raises(ValidationError): + InitializeMessage( + id="msg_001", + protocol_version="1.0", + peer={"name": "test", "role": "plugin"}, + extra="field", + ) + + def test_result_extra_forbidden(self): + """ResultMessage should reject extra fields.""" + with pytest.raises(ValidationError): + ResultMessage( + id="msg_001", + success=True, + extra="field", + ) + + def test_invoke_extra_forbidden(self): + """InvokeMessage should reject extra fields.""" + with pytest.raises(ValidationError): + InvokeMessage( + id="msg_001", + capability="test", + extra="field", + ) + + def test_event_extra_forbidden(self): + """EventMessage should reject extra fields.""" + with pytest.raises(ValidationError): + EventMessage( + id="msg_001", + phase="started", + extra="field", + ) + + def test_cancel_extra_forbidden(self): + """CancelMessage should reject extra fields.""" + with pytest.raises(ValidationError): + CancelMessage(id="msg_001", extra="field") diff --git a/tests_v4/test_transport.py b/tests_v4/test_transport.py new file mode 100644 index 000000000..88d4599b3 --- /dev/null +++ b/tests_v4/test_transport.py @@ -0,0 +1,447 @@ +""" +Tests for runtime/transport.py - Transport implementations. +""" +from __future__ import annotations + +import asyncio +from io import StringIO +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from astrbot_sdk.runtime.transport import ( + MessageHandler, + StdioTransport, + Transport, + WebSocketClientTransport, + WebSocketServerTransport, +) + + +class TestTransportBase: + """Tests for Transport base class.""" + + def test_init_sets_handler_none(self): + """Transport should initialize with _handler as None.""" + transport = Transport() + assert transport._handler is None + + def test_set_message_handler(self): + """set_message_handler should store handler.""" + transport = Transport() + handler = MagicMock() + transport.set_message_handler(handler) + assert transport._handler is handler + + def test_start_not_implemented(self): + """Transport.start should raise NotImplementedError.""" + transport = Transport() + with pytest.raises(NotImplementedError): + asyncio.get_event_loop().run_until_complete(transport.start()) + + def test_stop_not_implemented(self): + """Transport.stop should raise NotImplementedError.""" + transport = Transport() + with pytest.raises(NotImplementedError): + asyncio.get_event_loop().run_until_complete(transport.stop()) + + def test_send_not_implemented(self): + """Transport.send should raise NotImplementedError.""" + transport = Transport() + with pytest.raises(NotImplementedError): + asyncio.get_event_loop().run_until_complete(transport.send("test")) + + @pytest.mark.asyncio + async def test_wait_closed(self): + """wait_closed should wait for _closed event.""" + transport = Transport() + transport._closed.set() + # Should return immediately since _closed is already set + await transport.wait_closed() + + @pytest.mark.asyncio + async def test_dispatch_calls_handler(self): + """_dispatch should call handler with payload.""" + transport = Transport() + handler = AsyncMock() + transport.set_message_handler(handler) + await transport._dispatch("test payload") + handler.assert_called_once_with("test payload") + + @pytest.mark.asyncio + async def test_dispatch_without_handler(self): + """_dispatch should work without handler.""" + transport = Transport() + # Should not raise + await transport._dispatch("test payload") + + +class TestStdioTransportInit: + """Tests for StdioTransport initialization.""" + + def test_default_init(self): + """StdioTransport should initialize with default values.""" + transport = StdioTransport() + assert transport._stdin is None + assert transport._stdout is None + assert transport._command is None + assert transport._cwd is None + assert transport._env is None + assert transport._process is None + assert transport._reader_task is None + + def test_with_custom_streams(self): + """StdioTransport should accept custom stdin/stdout.""" + stdin = StringIO() + stdout = StringIO() + transport = StdioTransport(stdin=stdin, stdout=stdout) + assert transport._stdin is stdin + assert transport._stdout is stdout + + def test_with_command(self): + """StdioTransport should accept command for subprocess.""" + transport = StdioTransport(command=["python", "-m", "module"]) + assert transport._command == ["python", "-m", "module"] + + def test_with_cwd_and_env(self): + """StdioTransport should accept cwd and env.""" + transport = StdioTransport(cwd="/tmp", env={"VAR": "value"}) + assert transport._cwd == "/tmp" + assert transport._env == {"VAR": "value"} + + +class TestStdioTransportFileMode: + """Tests for StdioTransport in file mode (no subprocess).""" + + @pytest.mark.asyncio + async def test_start_without_command(self): + """start() without command should use stdin/stdout.""" + transport = StdioTransport() + with patch("sys.stdin"), patch("sys.stdout"): + await transport.start() + assert transport._reader_task is not None + await transport.stop() + + @pytest.mark.asyncio + async def test_stop_cancels_reader_task(self): + """stop() should cancel reader task.""" + transport = StdioTransport() + with patch("sys.stdin"), patch("sys.stdout"): + await transport.start() + task = transport._reader_task + await transport.stop() + assert task.cancelled() or task.done() + + @pytest.mark.asyncio + async def test_send_without_process(self): + """send() without process should write to stdout.""" + stdout = MagicMock() + stdout.write = MagicMock() + stdout.flush = MagicMock() + transport = StdioTransport(stdout=stdout) + + with patch("sys.stdin"): + await transport.start() + + await transport.send("test message") + + # Should have written the message with newline + stdout.write.assert_called_once_with("test message\n") + stdout.flush.assert_called_once() + + await transport.stop() + + @pytest.mark.asyncio + async def test_send_adds_newline_if_missing(self): + """send() should add newline if not present.""" + stdout = MagicMock() + stdout.write = MagicMock() + stdout.flush = MagicMock() + transport = StdioTransport(stdout=stdout) + + with patch("sys.stdin"): + await transport.start() + + await transport.send("message") + stdout.write.assert_called_once_with("message\n") + + await transport.stop() + + @pytest.mark.asyncio + async def test_send_preserves_existing_newline(self): + """send() should not add extra newline.""" + stdout = MagicMock() + stdout.write = MagicMock() + stdout.flush = MagicMock() + transport = StdioTransport(stdout=stdout) + + with patch("sys.stdin"): + await transport.start() + + await transport.send("message\n") + stdout.write.assert_called_once_with("message\n") + + await transport.stop() + + @pytest.mark.asyncio + async def test_send_raises_without_stdout(self): + """send() should raise if stdout is None.""" + transport = StdioTransport(stdout=None) + transport._stdout = None + + with pytest.raises(RuntimeError, match="stdout"): + await transport.send("test") + + +class TestStdioTransportProcessMode: + """Tests for StdioTransport in subprocess mode.""" + + @pytest.mark.asyncio + async def test_start_with_command_creates_process(self): + """start() with command should create subprocess.""" + transport = StdioTransport(command=["echo", "test"]) + + await transport.start() + assert transport._process is not None + assert transport._reader_task is not None + + await transport.stop() + + @pytest.mark.asyncio + async def test_stop_terminates_process(self): + """stop() should terminate the subprocess.""" + transport = StdioTransport(command=["sleep", "100"]) + + await transport.start() + process = transport._process + assert process is not None + + await transport.stop() + assert process.returncode is not None + + @pytest.mark.asyncio + async def test_send_to_process(self): + """send() should write to process stdin.""" + transport = StdioTransport(command=["cat"]) + + await transport.start() + # Should not raise + await transport.send("test data") + + await transport.stop() + + @pytest.mark.asyncio + async def test_send_raises_if_process_stdin_none(self): + """send() should raise if process stdin is None.""" + transport = StdioTransport(command=["cat"]) + await transport.start() + + # Manually set stdin to None to simulate error condition + if transport._process: + transport._process.stdin = None # type: ignore + + with pytest.raises(RuntimeError, match="stdin"): + await transport.send("test") + + await transport.stop() + + +class TestWebSocketServerTransportInit: + """Tests for WebSocketServerTransport initialization.""" + + def test_default_init(self): + """WebSocketServerTransport should have default values.""" + transport = WebSocketServerTransport() + assert transport._host == "127.0.0.1" + assert transport._port == 8765 + assert transport._path == "/" + assert transport._heartbeat == 30.0 + assert transport._app is None + assert transport._ws is None + + def test_custom_values(self): + """WebSocketServerTransport should accept custom values.""" + transport = WebSocketServerTransport( + host="0.0.0.0", + port=9000, + path="/ws", + heartbeat=60.0, + ) + assert transport._host == "0.0.0.0" + assert transport._port == 9000 + assert transport._path == "/ws" + assert transport._heartbeat == 60.0 + + def test_port_property_returns_actual_port(self): + """port property should return actual port after start.""" + transport = WebSocketServerTransport(port=8765) + # Before start, should return configured port + assert transport.port == 8765 + + def test_url_property(self): + """url property should return WebSocket URL.""" + transport = WebSocketServerTransport(host="localhost", port=8080, path="/ws") + assert transport.url == "ws://localhost:8080/ws" + + +class TestWebSocketServerTransportLifecycle: + """Tests for WebSocketServerTransport lifecycle.""" + + @pytest.mark.asyncio + async def test_start_creates_app(self): + """start() should create aiohttp app.""" + transport = WebSocketServerTransport(port=0) + await transport.start() + + assert transport._app is not None + assert transport._runner is not None + assert transport._site is not None + + await transport.stop() + + @pytest.mark.asyncio + async def test_stop_closes_websocket(self): + """stop() should close the WebSocket.""" + transport = WebSocketServerTransport(port=0) + await transport.start() + await transport.stop() + + assert transport._ws is None + assert transport._runner is None + + @pytest.mark.asyncio + async def test_send_waits_for_connection(self): + """send() should wait for WebSocket connection.""" + transport = WebSocketServerTransport(port=0) + await transport.start() + + # Mock connected state + transport._connected.set() + transport._ws = MagicMock() + transport._ws.closed = False + transport._ws.send_str = AsyncMock() + + await transport.send("test") + transport._ws.send_str.assert_called_once_with("test") + + await transport.stop() + + @pytest.mark.asyncio + async def test_send_raises_if_not_connected(self): + """send() should raise if WebSocket not connected.""" + transport = WebSocketServerTransport(port=0, heartbeat=0) + await transport.start() + + # Set timeout to 0 for immediate failure + with pytest.raises((RuntimeError, asyncio.TimeoutError)): + await transport.send("test") + + await transport.stop() + + +class TestWebSocketClientTransportInit: + """Tests for WebSocketClientTransport initialization.""" + + def test_required_url(self): + """WebSocketClientTransport requires url.""" + transport = WebSocketClientTransport(url="ws://localhost:8080") + assert transport._url == "ws://localhost:8080" + assert transport._heartbeat == 30.0 + + def test_custom_heartbeat(self): + """WebSocketClientTransport should accept custom heartbeat.""" + transport = WebSocketClientTransport(url="ws://localhost:8080", heartbeat=60.0) + assert transport._heartbeat == 60.0 + + +class TestWebSocketClientTransportLifecycle: + """Tests for WebSocketClientTransport lifecycle.""" + + @pytest.mark.asyncio + async def test_start_creates_session(self): + """start() should create aiohttp session and connect.""" + server = WebSocketServerTransport(port=0) + await server.start() + + client = WebSocketClientTransport(url=server.url) + await client.start() + + assert client._session is not None + assert client._ws is not None + + await client.stop() + await server.stop() + + @pytest.mark.asyncio + async def test_stop_closes_session_and_websocket(self): + """stop() should close session and WebSocket.""" + server = WebSocketServerTransport(port=0) + await server.start() + + client = WebSocketClientTransport(url=server.url) + await client.start() + await client.stop() + + assert client._session is None + assert client._ws is None + + await server.stop() + + @pytest.mark.asyncio + async def test_send_after_start(self): + """send() should work after start().""" + server = WebSocketServerTransport(port=0) + await server.start() + + client = WebSocketClientTransport(url=server.url) + await client.start() + + # Should not raise + await client.send("test message") + + await client.stop() + await server.stop() + + @pytest.mark.asyncio + async def test_send_raises_if_not_connected(self): + """send() should raise if WebSocket not connected.""" + client = WebSocketClientTransport(url="ws://localhost:99999") + + with pytest.raises(RuntimeError, match="尚未连接"): + await client.send("test") + + +class TestTransportIntegration: + """Integration tests for transport pairs.""" + + @pytest.mark.asyncio + async def test_websocket_client_server_communication(self): + """WebSocket client and server should communicate.""" + server = WebSocketServerTransport(port=0) + client = WebSocketClientTransport(url="ws://invalid") + + received_messages = [] + + async def handle_message(payload: str): + received_messages.append(payload) + + server.set_message_handler(handle_message) + + await server.start() + + # Create new client with correct URL + client = WebSocketClientTransport(url=server.url) + await client.start() + + # Wait for connection + await asyncio.sleep(0.1) + + await client.send("hello from client") + + # Wait for message to be received + await asyncio.sleep(0.1) + + assert "hello from client" in received_messages + + await client.stop() + await server.stop()