Spaces:
Sleeping
Sleeping
| """The registry is the extension point, so its contract is worth pinning down.""" | |
| from datetime import date | |
| from typing import Optional, Sequence | |
| import pytest | |
| from finbot.features.base import ( | |
| Attachment, | |
| BotContext, | |
| CommandSpec, | |
| Feature, | |
| Reply, | |
| Services, | |
| SimpleFeature, | |
| ) | |
| from finbot.registry import DuplicateCommand, FeatureRegistry | |
| TODAY = date(2026, 8, 15) | |
| class Greeter(SimpleFeature): | |
| name = "greeter" | |
| description = "Greetings" | |
| priority = 10 | |
| def commands(self) -> Sequence[CommandSpec]: | |
| return ( | |
| CommandSpec("hello", "Say hello", aliases=("hi",)), | |
| CommandSpec("secret", "Hidden thing", hidden=True), | |
| ) | |
| def cmd_hello(self, ctx: BotContext) -> Reply: | |
| return Reply(f"hello {ctx.args or 'world'}") | |
| def cmd_secret(self, ctx: BotContext) -> Reply: | |
| return Reply("shh") | |
| class Echo(Feature): | |
| name = "echo" | |
| description = "Echo" | |
| priority = 50 | |
| def handle_message(self, ctx: BotContext) -> Optional[Reply]: | |
| if ctx.text.startswith("say "): | |
| return Reply(ctx.text[4:]) | |
| return None | |
| class Filer(Feature): | |
| name = "filer" | |
| description = "Files" | |
| priority = 5 | |
| def handle_attachment(self, ctx: BotContext) -> Optional[Reply]: | |
| return Reply(f"got {ctx.attachment.filename}") | |
| class Exploder(Feature): | |
| name = "exploder" | |
| priority = 1 | |
| def commands(self) -> Sequence[CommandSpec]: | |
| return (CommandSpec("boom", "Raises"),) | |
| def handle_command(self, ctx: BotContext) -> Reply: | |
| raise RuntimeError("intentional") | |
| def make_ctx(registry, text="", command="", args="", attachment=None) -> BotContext: | |
| services = Services( | |
| settings=None, store=None, llm=None, ingestor=None, advisor=None, registry=registry | |
| ) | |
| return BotContext( | |
| user_id=1, | |
| chat_id=1, | |
| services=services, | |
| today=TODAY, | |
| text=text, | |
| command=command, | |
| args=args, | |
| attachment=attachment, | |
| ) | |
| def registry(): | |
| reg = FeatureRegistry() | |
| reg.register_all([Greeter(), Echo(), Filer()]) | |
| return reg | |
| class TestRegistration: | |
| def test_features_sorted_by_priority(self, registry): | |
| assert [f.name for f in registry.features] == ["filer", "greeter", "echo"] | |
| def test_duplicate_command_is_rejected(self, registry): | |
| class Clash(SimpleFeature): | |
| name = "clash" | |
| def commands(self): | |
| return (CommandSpec("hello", "clashing"),) | |
| with pytest.raises(DuplicateCommand, match="greeter"): | |
| registry.register(Clash()) | |
| def test_duplicate_alias_is_rejected(self, registry): | |
| class Clash(SimpleFeature): | |
| name = "clash" | |
| def commands(self): | |
| return (CommandSpec("other", "x", aliases=("hi",)),) | |
| with pytest.raises(DuplicateCommand): | |
| registry.register(Clash()) | |
| def test_known_command(self, registry): | |
| assert registry.known_command("hello") | |
| assert registry.known_command("hi") | |
| assert not registry.known_command("nope") | |
| class TestDispatch: | |
| def test_command(self, registry): | |
| reply = registry.dispatch(make_ctx(registry, command="hello", args="there")) | |
| assert reply.text == "hello there" | |
| def test_alias(self, registry): | |
| reply = registry.dispatch(make_ctx(registry, command="hi")) | |
| assert reply.text == "hello world" | |
| def test_unknown_command_is_reported(self, registry): | |
| reply = registry.dispatch(make_ctx(registry, command="nonsense")) | |
| assert "don't know" in reply.text | |
| def test_plain_message_routes_to_a_message_handler(self, registry): | |
| reply = registry.dispatch(make_ctx(registry, text="say cheese")) | |
| assert reply.text == "cheese" | |
| def test_declined_message_returns_none(self, registry): | |
| assert registry.dispatch(make_ctx(registry, text="unhandled")) is None | |
| def test_attachment_routes_to_attachment_handler(self, registry): | |
| attachment = Attachment(data=b"x", filename="a.csv") | |
| reply = registry.dispatch(make_ctx(registry, attachment=attachment)) | |
| assert reply.text == "got a.csv" | |
| def test_a_failing_feature_does_not_crash_the_bot(self, registry): | |
| registry.register(Exploder()) | |
| reply = registry.dispatch(make_ctx(registry, command="boom")) | |
| assert "went wrong" in reply.text | |
| class TestHelpGeneration: | |
| def test_help_lists_visible_commands(self, registry): | |
| text = registry.help_text() | |
| assert "/hello" in text | |
| assert "Greetings" in text | |
| def test_help_omits_hidden_commands(self, registry): | |
| assert "/secret" not in registry.help_text() | |
| def test_telegram_commands_payload(self, registry): | |
| payload = registry.telegram_commands() | |
| names = {c["command"] for c in payload} | |
| assert "hello" in names | |
| assert "secret" not in names | |
| assert all(len(c["description"]) <= 256 for c in payload) | |
| def test_help_escapes_html(self): | |
| class Sneaky(SimpleFeature): | |
| name = "sneaky" | |
| description = "<b>bold</b>" | |
| def commands(self): | |
| return (CommandSpec("x", "a & b"),) | |
| reg = FeatureRegistry() | |
| reg.register(Sneaky()) | |
| text = reg.help_text() | |
| assert "<b>bold</b>" in text | |
| assert "a & b" in text | |