JWeBot/tests/test_sdk_contracts.py

#!/usr/bin/env python3

import sys
import unittest
from pathlib import Path


PROJECT_ROOT = Path(__file__).resolve().parents[1]
SDK_SRC = PROJECT_ROOT / "packages" / "jwebot-sdk" / "src"
sys.path.insert(0, str(SDK_SRC))

from jwebot_sdk import (  # noqa: E402
    ContractViolation,
    Event,
    ExtensionResult,
    Fact,
    PlanKind,
    ReplyPlan,
    ServiceCommandAdapter,
)


class EventContractTests(unittest.TestCase):
    def event(self, **overrides):
        value = {
            "version": 1,
            "event_id": "event-1",
            "observed_at": "2026-10-14T10:00:00+08:00",
            "sender": "Alice",
            "text": "@JWeBot hello",
            "confidence": 0.95,
            "sender_confidence": 0.96,
            "body_confidence": 0.97,
            "side": "incoming",
            "mention_bot": True,
            "group_fingerprint": "group-a",
            "source": "rapidocr-local",
        }
        value.update(overrides)
        return value

    def test_event_round_trip_preserves_normalized_fields(self):
        event = Event.from_mapping(self.event())
        self.assertEqual(event.to_mapping(), self.event())

    def test_event_rejects_outgoing_messages(self):
        with self.assertRaisesRegex(ContractViolation, "incoming"):
            Event.from_mapping(self.event(side="outgoing"))

    def test_event_rejects_out_of_range_confidence(self):
        with self.assertRaisesRegex(ContractViolation, "between 0 and 1"):
            Event.from_mapping(self.event(confidence=1.1))

    def test_event_rejects_unsupported_version(self):
        with self.assertRaisesRegex(ContractViolation, "unsupported event version"):
            Event.from_mapping(self.event(version=2))


class PlanContractTests(unittest.TestCase):
    def test_text_plan_has_a_bounded_serialized_shape(self):
        plan = ReplyPlan.text("hello")
        self.assertEqual(plan.kind, PlanKind.TEXT)
        self.assertEqual(plan.to_mapping(), {"kind": "text", "text": "hello"})

    def test_multi_text_plan_has_bounded_parts(self):
        plan = ReplyPlan.multi_text(["one", "two"])
        self.assertEqual(plan.kind, PlanKind.MULTI_TEXT)
        self.assertEqual(
            plan.to_mapping(),
            {"kind": "multi_text", "texts": ["one", "two"]},
        )
        restored = ReplyPlan.from_mapping(plan.to_mapping())
        self.assertEqual(restored, plan)

    def test_no_reply_plan_cannot_contain_text(self):
        with self.assertRaisesRegex(ContractViolation, "cannot contain text"):
            ReplyPlan(kind=PlanKind.NO_REPLY, message="unexpected")

    def test_result_freezes_metadata_and_serializes_facts(self):
        result = ExtensionResult(
            reply=ReplyPlan.text("hello"),
            facts=(Fact(kind="topic", text="likes servers", confidence=0.8),),
            metadata={"source": "test"},
        )
        self.assertEqual(result.to_mapping()["metadata"], {"source": "test"})
        with self.assertRaises(TypeError):
            result.metadata["source"] = "changed"  # type: ignore[index]

    def test_result_rejects_unbounded_fact_count(self):
        facts = tuple(Fact(kind="topic", text=f"fact-{index}") for index in range(13))
        with self.assertRaisesRegex(ContractViolation, "exceeds 12 facts"):
            ExtensionResult(facts=facts)

    def test_result_round_trip_preserves_reply_facts_and_metadata(self):
        result = ExtensionResult(
            reply=ReplyPlan.text("hello"),
            facts=(Fact(kind="topic", text="likes servers", confidence=0.8),),
            metadata={"source": "test"},
        )
        restored = ExtensionResult.from_mapping(result.to_mapping())
        self.assertEqual(restored.reply, result.reply)
        self.assertEqual(restored.facts, result.facts)
        self.assertEqual(dict(restored.metadata), dict(result.metadata))

    def test_result_rejects_wrong_api_version(self):
        value = ExtensionResult(reply=ReplyPlan.text("hello")).to_mapping()
        value["api_version"] = "2"
        with self.assertRaisesRegex(ContractViolation, "API version"):
            ExtensionResult.from_mapping(value)


class ServiceCommandAdapterTests(unittest.TestCase):
    def event(self):
        return Event(
            event_id="event-1",
            observed_at="2026-10-14T10:00:00+08:00",
            sender="Alice",
            text="@JWeBot /coin",
            group_fingerprint="group-a",
            mention_bot=True,
        )

    def context(self, response):
        class Commands:
            def __init__(self):
                self.handlers = {}

            def register(self, command, handler):
                self.handlers[command] = handler

        class Services:
            def __init__(self):
                self.calls = []

            def call(self, profile, *, operation, payload):
                self.calls.append((profile, operation, payload))
                return response

        class Context:
            pass

        context = Context()
        context.commands = Commands()
        context.services = Services()
        return context

    def test_adapter_delegates_to_declared_service_profile(self):
        context = self.context({"texts": ["第一条", "第二条"]})
        adapter = ServiceCommandAdapter(
            "coin",
            metadata={"local_extension": "coin"},
        )

        adapter.register(context, ["/翻硬币"])
        result = context.commands.handlers["/翻硬币"](self.event(), "")

        self.assertEqual(result.reply.kind, PlanKind.MULTI_TEXT)
        self.assertEqual(result.reply.messages, ("第一条", "第二条"))
        self.assertEqual(dict(result.metadata), {"local_extension": "coin"})
        self.assertEqual(context.services.calls[0][0], "coin")
        self.assertEqual(context.services.calls[0][1], "execute")
        self.assertEqual(context.services.calls[0][2]["command"], "/翻硬币")
        self.assertEqual(
            context.services.calls[0][2]["event"]["event_id"],
            "event-1",
        )

    def test_adapter_rejects_invalid_service_text(self):
        context = self.context({"texts": ["ok", ""]})
        adapter = ServiceCommandAdapter("coin")

        adapter.register(context, ["/翻硬币"])
        with self.assertRaisesRegex(ValueError, "invalid text parts"):
            context.commands.handlers["/翻硬币"](self.event(), "")


if __name__ == "__main__":
    unittest.main()