Check pull requests for undefined names and run unit tests

.github/workflows/deploy.yml

@@ -21,6 +21,15 @@ jobs:
           python-version: "3.12"
       - name: Byte-compile all Python files
         run: python -m compileall -q .
+      # Syntax errors and names used but never defined (e.g. a variable
+      # deleted by mistake), which compiling alone does not catch
+      - name: Look for undefined names
+        run: |
+          python -m pip install --quiet "flake8>=7,<8"
+          python -m flake8 . --select=E9,F63,F7,F82 --show-source --statistics \
+            --exclude=.git,__pycache__,TeamTalkPy,TeamTalk_DLL
+      - name: Run the tests
+        run: python -m unittest discover -s tests -t . -v
 
   deploy:
     name: Deploy to the server

CHANGELOG.md

@@ -6,6 +6,7 @@ The format is based on [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 ### Added
+- Automated checks on every pull request: flake8 looks for undefined names (such as the `chan_name` that broke every `@ai` answer) and `tests/` holds unit tests for the parts that run without the TeamTalk SDK (AI chat and its tool calls, AI review, badword filter, abuse tracker, help texts). Run them locally with `python -m unittest discover -s tests -t .`.
 - `@ai <question>` in a channel: the AI (Cloudflare Workers AI, `ai_chat.py`) answers in that channel, as short plain text for screen readers, followed by `AI_CHAT_DISCLAIMER` ("AI bisa salah. Periksa kembali info penting."). It remembers the channel's last few questions for follow-ups, never gets nicknames, and has a per-person cooldown and a daily limit; answers with clear badwords are replaced by a refusal. Off by default (`/abt aichat`); users can read `/help ai`.
 - Natural language bot commands via Function Calling for `@ai`: users can speak commands naturally (e.g. `@ai matiin word filter dong`, `@ai matiin moderation`, `@ai kick orang yang namanya kak fian dong`, `@ai siapa aja yang lagi online?`). Features 18 distinct tools covering moderation toggling (`toggle_moderation_feature`, `get_moderation_status`, `forgive_user`, `temp_ban_user`, `unban_user`), user management, badwords list management, and server inspection. Sensitive operations strictly verify admin privileges (`USERTYPE_ADMIN`), rejecting unauthorized calls with a Gen Z refusal. If an unsupported action is requested, the AI informs the user that the function is not yet available. Includes safe cross-thread tool dispatching and screen reader sanitized plain-text output.
 - Guest accounts (`tamu`, `hadirin`) are blocked from using `@ai <question>` in channels by default (`AI_CHAT_BLOCKED_USERS`), with a private notification (`AI_CHAT_BLOCKED_MESSAGE`).

docs/deploy.md

@@ -1,7 +1,9 @@
 # Automatic deployment
 
-Work happens on branches. A pull request into `main` runs the **Check** job
-(every Python file must compile). When the pull request is merged, the
+Work happens on branches. A pull request into `main` runs the **Check** job:
+every Python file must compile, flake8 looks for names that are used but never
+defined (a typo or a variable deleted by mistake), and the tests in `tests/`
+must pass. When the pull request is merged, the
 **Deploy to the server** job connects to the server over SSH, and
 `tools/deploy.sh` there pulls the new `main`, restarts the bot and checks
 that it is running. Both jobs live in `.github/workflows/deploy.yml`; their
@@ -111,7 +113,12 @@ server prints for `ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub`.
 ## Day-to-day
 
 1. Create a branch, commit, push, open a pull request into `main`.
-2. Wait for **Check** to pass, then merge.
+   Before pushing you can run the same checks locally:
+   `python -m unittest discover -s tests -t .` and
+   `python -m flake8 . --select=E9,F63,F7,F82 --exclude=TeamTalkPy,TeamTalk_DLL`
+   (`pip install flake8` once).
+2. Wait for **Check** to pass, then merge. If it fails, the run's log names
+   the file, line and test.
 3. The bot restarts with the new code within a minute or two; the run's log
    shows `Deployed <commit>; TeamTalkBot is running.`
 

tests/__init__.py

@@ -0,0 +1,4 @@
+"""Unit tests for the parts of the bot that run without the TeamTalk SDK.
+
+Run from the repository root: python -m unittest discover -s tests -t .
+"""

tests/test_abuse_tracker.py

@@ -0,0 +1,59 @@
+"""AbuseTracker: counting events per person and escalating warn -> kick -> ban."""
+
+import unittest
+from unittest import mock
+
+import abuse_tracker
+from abuse_tracker import AbuseTracker
+
+
+class TrackerTest(unittest.TestCase):
+    def setUp(self):
+        self.now = [1000.0]
+        patcher = mock.patch.object(abuse_tracker.time, "time", lambda: self.now[0])
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        self.tracker = AbuseTracker(login_count=3, join_count=3, badword_count=2, window_sec=60)
+
+    def record(self, times, kind="badword", key="budi@10.0.0.1"):
+        return [self.tracker.record(kind, key) for _ in range(times)]
+
+    def test_stages_escalate_up_to_three(self):
+        self.assertEqual(self.record(2), [0, 1])
+        self.assertEqual(self.record(2), [0, 2])
+        self.assertEqual(self.record(2), [0, 3])
+        self.assertEqual(self.record(2), [0, 0])  # already at the last stage
+        self.assertEqual(self.tracker.get_stage("badword", "budi@10.0.0.1"), 3)
+
+    def test_people_are_counted_separately(self):
+        self.assertEqual(self.record(1, key="a"), [0])
+        self.assertEqual(self.record(1, key="b"), [0])
+        self.assertEqual(self.record(1, key="a"), [1])
+
+    def test_events_outside_the_window_are_forgotten(self):
+        self.record(1)
+        self.now[0] += 61
+        self.assertEqual(self.record(1), [0])
+
+    def test_stage_clears_after_the_window(self):
+        self.record(2)
+        self.now[0] += 61
+        self.assertEqual(self.record(2), [0, 1])
+
+    def test_reset(self):
+        self.record(2)
+        self.tracker.reset("badword", "budi@10.0.0.1")
+        self.assertEqual(self.tracker.get_stage("badword", "budi@10.0.0.1"), 0)
+        self.assertNotIn(("badword", "budi@10.0.0.1"), self.tracker.tracked_keys())
+
+    def test_empty_key_is_ignored(self):
+        self.assertEqual(self.record(5, key=""), [0] * 5)
+
+    def test_active_stages(self):
+        self.record(2)
+        self.now[0] += 10
+        self.assertEqual(self.tracker.active_stages(), [("badword", "budi@10.0.0.1", 1, 50.0)])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ai_chat.py

@@ -0,0 +1,175 @@
+"""AIChat: what is sent to Workers AI and how answers come back."""
+
+import copy
+import time
+import unittest
+from unittest import mock
+
+import ai_chat
+from ai_chat import TOOLS, AIChat, clean_answer
+
+
+class FakeWorkersAI:
+    """Stands in for call_workers_ai: records request bodies, returns replies in order.
+
+    A reply that is an exception is raised instead; the last reply repeats.
+    """
+
+    def __init__(self, *replies):
+        self.bodies = []
+        self.replies = list(replies)
+
+    def __call__(self, api_url, account, token, model, body, timeout):
+        self.bodies.append(copy.deepcopy(body))  # answer() edits the body to retry
+        reply = self.replies.pop(0) if len(self.replies) > 1 else self.replies[0]
+        if isinstance(reply, Exception):
+            raise reply
+        return {"success": True, "result": reply}
+
+
+def asker_line(body):
+    system = body["messages"][0]["content"]
+    return next(line for line in system.splitlines() if line.startswith("Penanya"))
+
+
+class AnswerTest(unittest.TestCase):
+    def setUp(self):
+        self.chat = AIChat("account", "token")
+
+    def ask(self, fake, question="halo", history=(), context=None):
+        with mock.patch.object(ai_chat, "call_workers_ai", fake):
+            return self.chat.answer(question, list(history), context)
+
+    def test_channel_question(self):
+        fake = FakeWorkersAI({"response": "Halo semua"})
+        context = {"nickname": "Siti", "channel": 17, "channel_name": "/Kelas 7/", "is_pm": False}
+        self.assertEqual(self.ask(fake, context=context), "Halo semua")
+        self.assertEqual(asker_line(fake.bodies[0]), "Penanya: Siti (berada di channel: /Kelas 7/).")
+
+    def test_pm_question(self):
+        fake = FakeWorkersAI({"response": "Halo"})
+        context = {"nickname": "Siti", "channel": 17, "channel_name": "/Kelas 7/", "is_pm": True}
+        self.ask(fake, context=context)
+        line = asker_line(fake.bodies[0])
+        self.assertIn("Private Message", line)
+        self.assertIn("/Kelas 7/", line)
+
+    def test_context_without_channel_name(self):
+        fake = FakeWorkersAI({"response": "Halo"})
+        self.ask(fake, context={"nickname": "Siti", "channel": 17})
+        self.assertEqual(asker_line(fake.bodies[0]), "Penanya: Siti (berada di channel: channel 17).")
+
+    def test_without_context(self):
+        fake = FakeWorkersAI({"response": "Halo"})
+        self.assertEqual(self.ask(fake), "Halo")
+
+    def test_history_comes_before_the_question(self):
+        fake = FakeWorkersAI({"response": "Lebih simpel: ..."})
+        self.ask(fake, "jelasin lebih simpel", [("apa itu fotosintesis?", "Fotosintesis adalah ...")])
+        self.assertEqual(
+            fake.bodies[0]["messages"][1:],
+            [
+                {"role": "user", "content": "apa itu fotosintesis?"},
+                {"role": "assistant", "content": "Fotosintesis adalah ..."},
+                {"role": "user", "content": "jelasin lebih simpel"},
+            ],
+        )
+
+    def test_answer_is_cleaned(self):
+        fake = FakeWorkersAI({"response": "**Halo** semua"})
+        self.assertEqual(self.ask(fake), "Halo semua")
+
+    def test_retries_without_tools_when_the_model_rejects_them(self):
+        fake = FakeWorkersAI(RuntimeError("tools not supported"), {"response": "Halo"})
+        self.assertEqual(self.ask(fake), "Halo")
+        self.assertIn("tools", fake.bodies[0])
+        self.assertNotIn("tools", fake.bodies[1])
+
+    def test_failed_request_raises(self):
+        def failing(*args):
+            return {"success": False, "errors": [{"code": 10000, "message": "Authentication error"}]}
+
+        with self.assertRaises(RuntimeError):
+            self.ask(failing)
+
+    def test_tool_call_runs_the_tool_and_returns_the_follow_up(self):
+        fake = FakeWorkersAI(
+            {"tool_calls": [{"name": "list_channels", "arguments": {}}]},
+            {"response": "Ada 3 channel."},
+        )
+        context = {"nickname": "Guru", "channel": 1, "channel_name": "/", "is_admin": True}
+        with mock.patch.object(
+            self.chat, "request_tool", return_value={"status": "success", "message": "/, /Kelas 7/, /Musik/"}
+        ) as request_tool:
+            answer = self.ask(fake, "channel apa saja?", context=context)
+        request_tool.assert_called_once_with("list_channels", {}, context)
+        self.assertEqual(answer, "Ada 3 channel.")
+        self.assertEqual(fake.bodies[1]["messages"][-1]["role"], "tool")
+
+
+class ExtractToolCallsTest(unittest.TestCase):
+    def setUp(self):
+        self.extract = AIChat("account", "token")._extract_tool_calls
+
+    def test_native_tool_call_with_string_arguments(self):
+        calls = self.extract({"tool_calls": [{"name": "kick_user", "arguments": '{"nickname": "budi"}'}]})
+        self.assertEqual(calls, [{"name": "kick_user", "arguments": {"nickname": "budi"}, "id": "call_1"}])
+
+    def test_alias(self):
+        calls = self.extract({"tool_calls": [{"name": "pm_user", "arguments": {}}]})
+        self.assertEqual(calls[0]["name"], "send_private_message")
+
+    def test_unknown_tool_is_ignored(self):
+        self.assertIsNone(self.extract({"tool_calls": [{"name": "format_disk", "arguments": {}}]}))
+
+    def test_tool_call_written_as_text(self):
+        calls = self.extract({"response": 'Oke: {"name": "list_channels", "arguments": {}}'})
+        self.assertEqual(calls[0]["name"], "list_channels")
+
+    def test_plain_answer_has_no_tool_call(self):
+        self.assertIsNone(self.extract({"response": "Fotosintesis adalah proses ..."}))
+
+
+class BackgroundTest(unittest.TestCase):
+    def drain(self, chat, count):
+        done, deadline = [], time.time() + 5
+        while len(done) < count and time.time() < deadline:
+            done += chat.drain()
+            time.sleep(0.01)
+        return done
+
+    def test_answers_come_back_through_drain(self):
+        chat = AIChat("account", "token")
+        answers = {"halo": "Halo juga"}
+
+        def answer(question, history, context=None):
+            if question not in answers:
+                raise RuntimeError("down")
+            return answers[question]
+
+        chat.answer = answer
+        self.assertTrue(chat.ask("halo", [], {"id": 1}))
+        self.assertTrue(chat.ask("gagal", [], {"id": 2}))
+        self.assertEqual(sorted(self.drain(chat, 2), key=lambda r: r[0]["id"]),
+                         [({"id": 1}, "Halo juga"), ({"id": 2}, None)])
+
+    def test_not_configured(self):
+        self.assertFalse(AIChat("", "").ask("halo", [], {}))
+
+
+class HelpersTest(unittest.TestCase):
+    def test_clean_answer(self):
+        self.assertEqual(clean_answer("**Fotosintesis** adalah proses `x`"), "Fotosintesis adalah proses x")
+        self.assertEqual(clean_answer("Langkah:\n\n- satu\n• dua\n"), "Langkah:\nsatu\ndua")
+
+    def test_tools_are_well_formed(self):
+        names = [tool["name"] for tool in TOOLS]
+        self.assertEqual(len(names), len(set(names)), "duplicate tool names")
+        for tool in TOOLS:
+            with self.subTest(tool=tool["name"]):
+                self.assertTrue(tool.get("description"))
+                self.assertEqual(tool["parameters"]["type"], "object")
+
+
+if __name__ == "__main__":
+    unittest.main()