No reply if permission denied

mmpy_bot/function.py

@@ -63,6 +63,7 @@ def __init__(
         *args,
         direct_only: bool = False,
         needs_mention: bool = False,
+        silence_fail_msg: bool = False,
         allowed_users: Optional[Sequence[str]] = None,
         allowed_channels: Optional[Sequence[str]] = None,
         **kwargs,
@@ -72,6 +73,7 @@ def __init__(
         self.is_click_function = isinstance(self.function, click.Command)
         self.direct_only = direct_only
         self.needs_mention = needs_mention
+        self.silence_fail_msg = silence_fail_msg
 
         if allowed_users is None:
             self.allowed_users = []
@@ -127,15 +129,17 @@ def __call__(self, message: Message, *args):
             return return_value
 
         if self.allowed_users and message.sender_name not in self.allowed_users:
-            self.plugin.driver.reply_to(
-                message, "You do not have permission to perform this action!"
-            )
+            if self.silence_fail_msg is False:
+                self.plugin.driver.reply_to(
+                    message, "You do not have permission to perform this action!"
+                )
             return return_value
 
         if self.allowed_channels and message.channel_name not in self.allowed_channels:
-            self.plugin.driver.reply_to(
-                message, "You do not have permission to perform this action!"
-            )
+            if self.silence_fail_msg is False:
+                self.plugin.driver.reply_to(
+                    message, "You do not have permission to perform this action!"
+                )
             return return_value
 
         if self.is_click_function:
@@ -164,6 +168,7 @@ def get_help_string(self):
                 self.direct_only,
                 self.allowed_users,
                 self.allowed_channels,
+                self.silence_fail_msg,
             ]
         ):
             # Print some information describing the usage settings.
@@ -182,6 +187,9 @@ def get_help_string(self):
             if self.allowed_channels:
                 string += f"{spaces(4)}- Restricted to certain channels.\n"
 
+            if self.silence_fail_msg:
+                string += f"{spaces(4)}- If it should reply to a non privileged user / in a non privileged channel.\n"
+
         return string
 
 
@@ -193,6 +201,7 @@ def listen_to(
     needs_mention=False,
     allowed_users=None,
     allowed_channels=None,
+    silence_fail_msg=False,
     **metadata,
 ):
     """Wrap the given function in a MessageFunction class so we can register some
@@ -227,6 +236,7 @@ def wrapped_func(func):
             needs_mention=needs_mention,
             allowed_users=allowed_users,
             allowed_channels=allowed_channels,
+            silence_fail_msg=silence_fail_msg,
             **metadata,
         )
 

mmpy_bot/plugins/example.py

@@ -19,6 +19,11 @@ async def users_access(self, message: Message):
         """Showcases a function with restricted access."""
         self.driver.reply_to(message, "Access allowed!")
 
+    @listen_to("^offtopic_channel$", allowed_channels=["off-topic"])
+    async def channels_access(self, message: Message):
+        """Showcases a function which can only be used in specific channels."""
+        self.driver.reply_to(message, "Access allowed!")
+
     @listen_to("^busy|jobs$", re.IGNORECASE, needs_mention=True)
     async def busy_reply(self, message: Message):
         """Show the number of busy worker threads."""

tests/integration_tests/docker-compose.yml

@@ -5,6 +5,7 @@ services:
     container_name: "mattermost-bot-test"
     build: .
     command: ./mm/docker-entry.sh
-    network_mode: host
+    ports:
+      - 8065:8065
     extra_hosts:
       - "dockerhost:127.0.0.1"

tests/unit_tests/event_handler_test.py

@@ -13,13 +13,14 @@ def create_message(
     mentions=["qmw86q7qsjriura9jos75i4why"],
     channel_type="O",
     sender_name="betty",
+    channel_name="off-topic",
 ):
     return Message(
         {
             "event": "posted",
             "data": {
                 "channel_display_name": "Off-Topic",
-                "channel_name": "off-topic",
+                "channel_name": channel_name,
                 "channel_type": channel_type,
                 "mentions": mentions,
                 "post": {

tests/unit_tests/function_test.py

@@ -188,6 +188,56 @@ def fake_reply(message, text):
         wrapped.assert_not_called()
         driver.reply_to.assert_called_once()
 
+    def test_allowed_channels(self):
+        wrapped = mock.create_autospec(example_listener)
+        wrapped.__qualname__ = "wrapped"
+        # Create a driver with a mocked reply function
+        driver = Driver()
+
+        def fake_reply(message, text):
+            assert "you do not have permission" in text.lower()
+
+        driver.reply_to = mock.Mock(wraps=fake_reply)
+
+        f = listen_to("", allowed_channels=["off-topic"])(wrapped)
+        f.plugin = ExamplePlugin().initialize(driver)
+
+        # This is fine, the names are not caps sensitive
+        f(create_message(channel_name="off-topic"))
+        wrapped.assert_called_once()
+        wrapped.reset_mock()
+
+        # This is not fine, and we expect the fake reply to be called.
+        f(create_message(channel_name="town-square"))
+        wrapped.assert_not_called()
+        driver.reply_to.assert_called_once()
+
+    def test_allowed_channels_silence_fail_msg(self):
+        wrapped = mock.create_autospec(example_listener)
+        wrapped.__qualname__ = "wrapped"
+        # Create a driver with a mocked reply function
+        driver = Driver()
+
+        def fake_reply(message, text):
+            assert "you do not have permission" in text.lower()
+
+        driver.reply_to = mock.Mock(wraps=fake_reply)
+
+        f = listen_to("", allowed_channels=["off-topic"], silence_fail_msg=True)(
+            wrapped
+        )
+        f.plugin = ExamplePlugin().initialize(driver)
+
+        # This is fine, the names are not caps sensitive
+        f(create_message(channel_name="off-topic"))
+        wrapped.assert_called_once()
+        wrapped.reset_mock()
+
+        # This is not fine, and we expect the fake reply not to be called.
+        f(create_message(channel_name="town-square"))
+        wrapped.assert_not_called()
+        driver.reply_to.assert_not_called()
+
 
 def example_webhook_listener(self, event):
     # Used to copy the arg specs to mock.Mock functions.