Refactor integration test fixtures and implement direct_message integration test

docs/contributing.rst

@@ -42,7 +42,7 @@ tests and run tests, please install the dev requirements.
 
         $ pip install -r dev-requirements.txt
 
-All the tests are put in `mmpy_bot\tests`.
+All the tests are put in ``mmpy_bot\tests``.
 There are two test packages: :code:`unit_tests` and
 :code:`integration_tests`.
 
@@ -76,18 +76,18 @@ To run the unit tests (in parallel), simply execute:
 Adding integration tests
 ------------------------
 
-The integration tests are run on the `jneeven:mattermost-bot-test` docker
-image, for which dockerfiles are provided in the `tests/intergration_tests`
+The integration tests are run on the ``jneeven:mattermost-bot-test`` docker
+image, for which dockerfiles are provided in the ``tests/intergration_tests``
 folder. The tests are defined as interactions between a bot (the responder)
 and a driver (the one sending test messages), which live inside the docker
 image. Their respective tokens are available in
-`tests/integration_tests/utils.py`, and the two bots are available as pytest
+``tests/integration_tests/utils.py``, and the two bots are available as pytest
 fixtures so they can be easily re-used. Note that while the bot is also a
 fixture, it should not be used in any functions. It will simply be started
 whenever the integration tests are executed.
 
 An integration test might look like this (also have a look at the actual code
-in `tests/integration_tests/test_example_plugin.py`):
+in ``tests/integration_tests/test_example_plugin.py``):
 
     .. code-block:: python
 
@@ -108,7 +108,7 @@ in `tests/integration_tests/test_example_plugin.py`):
 In this test, the driver sends a message in the "off-topic" channel, and
 waits for the bot to reply 'Bring it on!'. If no reply occurs within a
 default response timeout (15 seconds by default, but this can be passed as an
-argument to `expect_reply`), an exception will be raised. The driver fixture
+argument to ``expect_reply``), an exception will be raised. The driver fixture
 is imported from the utils and can be re-used in every test function simply
 by adding it as a function argument.
 
@@ -117,9 +117,9 @@ by adding it as a function argument.
 Running the integration_tests
 -----------------------------
 
-Running the integration_tests is easy: simply `cd` into
-`tests/integration_tests`, and run `docker-compose up -d` to start a local
-mattermost server. Then run `pytest -n auto .` to start the tests! For more
+Running the integration_tests is easy: simply ``cd`` into
+``tests/integration_tests``, and run ``docker-compose up -d`` to start a local
+mattermost server. Then run ``pytest -n auto .`` to start the tests! For more
 info about the integration tests an the docker server, have a look at
 `tests/integration_tests/README.md`.
 

tests/integration_tests/test_direct_message_plugin.py

@@ -0,0 +1,73 @@
+import random
+import time
+from string import ascii_letters
+
+from .utils import start_bot  # noqa, only imported so that the bot is started
+from .utils import MAIN_BOT_ID, OFF_TOPIC_ID, RESPONSE_TIMEOUT, TEAM_ID
+from .utils import driver as driver_fixture
+from .utils import expect_reply
+
+# Hacky workaround to import the fixture without linting errors
+driver = driver_fixture
+
+
+# Verifies that the bot is running and listening to this non-targeted message
+def test_start(driver):
+    post = driver.create_post(OFF_TOPIC_ID, "starting direct tests!")
+    assert expect_reply(driver, post)["message"] == "Bring direct on!"
+
+
+# Verifies that the bot is running and listening to this non-targeted message
+class TestDirectPlugin:
+    def test_start_direct(self, driver):
+        def bot_and_user_direct_channel(channel):
+            """Find which channels are direct and have the user and bot as
+            participants."""
+            name = channel["name"]
+
+            user_chan = driver.user_id in name
+            bot_chan = MAIN_BOT_ID in name
+            # D = direct message channel
+            direct = channel["type"] == "D"
+
+            return user_chan and bot_chan and direct
+
+        # Create a random string of text so we can uniquely identify the bot reply
+        random_string = "".join(random.choices(ascii_letters, k=30))
+        trigger = f"direct reply {random_string}"
+        reply = f"Telling you privately! {random_string}"
+
+        # The bot should reply with a direct message
+        # which is implemented by mattermost as a channel
+        driver.create_post(OFF_TOPIC_ID, trigger)
+
+        user_channels = driver.channels.get_channels_for_user(driver.user_id, TEAM_ID)
+        channels = list(filter(bot_and_user_direct_channel, user_channels))
+
+        # We need to wait for the reply to be processed by mattermost
+        # and the private channel created
+        retries = 2
+
+        for _ in range(retries):
+            if len(channels) != 1:
+                time.sleep(RESPONSE_TIMEOUT)
+                user_channels = driver.channels.get_channels_for_user(
+                    driver.user_id, TEAM_ID
+                )
+                channels = list(filter(bot_and_user_direct_channel, user_channels))
+            else:
+                channel = channels.pop()
+                break
+        else:
+            raise ValueError("Couldn't find a direct channel between user and bot")
+
+        for _ in range(retries):
+            posts = driver.posts.get_posts_for_channel(channel["id"])
+
+            for post in posts["posts"].values():
+                if post["message"] == reply:
+                    return
+
+            time.sleep(RESPONSE_TIMEOUT)
+        else:
+            raise ValueError(f"Direct reply '{reply}' not found among direct messages")

tests/integration_tests/utils.py

@@ -46,6 +46,17 @@ async def reply_start(self, message: Message):
         self.driver.reply_to(message, "Bring it on!")
 
 
+# For direct message tests
+class DirectPlugin(Plugin):
+    @listen_to("^starting direct tests")
+    async def reply_start_direct(self, message: Message):
+        self.driver.reply_to(message, "Bring direct on!")
+
+    @listen_to("^direct reply (.*)")
+    async def reply_direct(self, message: Message, text):
+        self.driver.reply_to(message, f"Telling you privately! {text}", direct=True)
+
+
 @pytest.fixture(scope="session")
 def driver():
     return Bot(
@@ -81,7 +92,7 @@ def start_bot(request):
                 WEBHOOK_HOST_URL="http://127.0.0.1",
                 WEBHOOK_HOST_PORT=8579,
             ),
-            plugins=[TestPlugin(), ExamplePlugin(), WebHookExample()],
+            plugins=[DirectPlugin(), TestPlugin(), ExamplePlugin(), WebHookExample()],
         )
 
         def run_bot():