WIP: Testing setup

pyproject.toml

@@ -31,7 +31,8 @@ test = [
     "black==23.9.1",
     "httpx==0.25.0",
     "pylint==2.17.5",
-    "pytest==7.4.2"
+    "pytest==7.4.2",
+    "pytest-asyncio==0.21.1"
 ]
 
 [project.urls]
@@ -44,3 +45,6 @@ pythonpath = [
   ".",
   "src"
 ]
+norecursedirs = [
+  "tests/helpers"
+]

tests/conftest.py

@@ -1,9 +1,11 @@
-import os
+"""Pytest Fixtures"""
+
 import pytest
 from fastapi.testclient import TestClient
 from threading import Thread
 
 from bot import API
+import database
 
 
 @pytest.fixture
@@ -16,3 +18,27 @@ def test_client():
 def threads_appear_dead(monkeypatch):
     """Include this fixture if you'd like for all your threads to be reported as dead."""
     monkeypatch.setattr(Thread, "is_alive", lambda x: False)
+
+
+@pytest.fixture
+def db_with_cleanup():
+    """
+    Fixture to provide a DB connection to tests and then ensure state doesn't bleed over
+    between them.
+    """
+    database.create_tables()
+
+    for conn in database.get_connection():
+        yield conn
+
+        cur = conn.cursor()
+
+        cur.executescript(
+            """
+            SELECT 'DELETE FROM ' || name
+            FROM sqlite_master
+            WHERE type = 'table';
+
+            VACUUM;
+            """
+        )

tests/helpers/__init__.py

@@ -0,0 +1 @@
+from .utils import *

tests/helpers/utils.py

@@ -0,0 +1,30 @@
+"""Pytest Helper Functions"""
+
+import urllib.parse
+
+
+# pylint: disable=too-many-arguments
+def create_slack_request_payload(
+    command: str,
+    token: str = "1CnbxdlkN3Ag2AafGvsp81za",
+    team_id: str = "LGPpTuQPsQx",
+    team_domain: str = "super_cool_domain",
+    channel_id: str = "jhVOsIAWtNW",
+    channel_name: str = "Testing",
+    user_id: str = "2xIIwe9Rs6y",
+    user_name: str = "thetester",
+    text: str = "",
+    api_app_id: str = "QpysuvDZwgb",
+    is_enterprise_install: str = "false",
+    response_url: str = "https://hooks.slack.com/commands/some-info",
+) -> bytes:
+    """Creates a representative payload that we would expect to receive from Slack's API."""
+    sample_payload = (
+        f"token={token}&team_id={team_id}&team_domain={team_domain}&channel_id{channel_id}&"
+        f"channel_name={channel_name}&user_id={user_id}"
+        f"&user_name={user_name}&command={urllib.parse.quote_plus(command)}&"
+        f"text={text}&api_app_id={api_app_id}&is_enterprise_install={is_enterprise_install}&"
+        f"response_url={urllib.parse.quote_plus(response_url)}"
+    )
+
+    return bytes(sample_payload, "utf-8")

tests/test_bot.py

@@ -1,9 +1,12 @@
 """
     Tests for the bot.py file.
 """
-
+import pytest
 import threading
 
+import database
+import helpers
+
 
 def test_health_check_healthy_threads(test_client):
     """Happy path scenario for the /healthz route where nothing is wrong."""
@@ -13,6 +16,7 @@ def test_health_check_healthy_threads(test_client):
     assert response.content == b'{"detail":"Everything is lookin\' good!"}'
 
 
+# pylint: disable=unused-argument
 def test_health_check_with_a_dead_thread(test_client, threads_appear_dead):
     """Tests what happens if a dead thread is found whenever this endpoint is hit."""
     response = test_client.get("healthz")
@@ -23,3 +27,76 @@ def test_health_check_with_a_dead_thread(test_client, threads_appear_dead):
     assert response.json() == {
         "detail": f"The {first_thread.name} thread has died. This container will soon restart."
     }
+
+
+TEAM_DOMAIN = "team_awesome"
+
+RATE_LIMIT_COPY = (
+    "This command has been run recently and is on a cooldown period. "
+    "Please try again in a little while!"
+)
+
+
+def test_check_api_whenever_someone_executes_it_for_first_time(
+    test_client, db_with_cleanup
+):
+    """Whenever an entity executes /check_api for the first time it should run successfully."""
+    response = test_client.post(
+        "/slack/events",
+        content=helpers.create_slack_request_payload(
+            command="/check_api", team_domain=TEAM_DOMAIN
+        ),
+    )
+
+    # Until the Slack Bolt client is properly mocked this is about as specific as we
+    # can get. Right now we will receive a 401 response in our test suite
+    # since there are checks being performed to validate our fake token on Slack's side.
+    #
+    # If we get some response other than RATE_LIMIT_COPY that means rate-limiting is not
+    # occurring, so this is at least targeting that aspect (though, sloppily).
+    assert response.content.decode("utf-8") != RATE_LIMIT_COPY
+
+
+@pytest.mark.asyncio
+async def test_check_api_whenever_someone_executes_it_after_expiry(
+    test_client, db_with_cleanup
+):
+    """
+    Whenever an entity has run /check_api before, and their cooldown window has expired,
+    then they should be able to run the command again.
+    """
+    # Create a cooldown that has expired 20 minutes ago.
+    await database.create_cooldown(TEAM_DOMAIN, "check_api", -20)
+
+    response = test_client.post(
+        "/slack/events",
+        content=helpers.create_slack_request_payload(
+            command="/check_api", team_domain=TEAM_DOMAIN
+        ),
+    )
+
+    #
+    assert response.content.decode("utf-8") != RATE_LIMIT_COPY
+
+
+@pytest.mark.asyncio
+async def test_check_api_whenever_someone_executes_it_before_expiry(
+    test_client, db_with_cleanup
+):
+    """
+    Whenever an entity has run /check_api before, and their cooldown window has NOT expired,
+    then they should receive a message telling them to try again later.
+    """
+    await database.create_cooldown(TEAM_DOMAIN, "check_api", 15)
+
+    response = test_client.post(
+        "/slack/events",
+        content=helpers.create_slack_request_payload(
+            command="/check_api", team_domain=TEAM_DOMAIN
+        ),
+    )
+    assert response.status_code == 200
+    assert response.content.decode("utf-8") == RATE_LIMIT_COPY
+
+
+# pylint: enable=unused-argument