diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index e359c24..d7944bb 100644 --- a/.github/workflows/deploy.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index d4c1d54..6f23b0f 100644 --- a/CHANGELOG.md +++ b/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 ` 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 ` in channels by default (`AI_CHAT_BLOCKED_USERS`), with a private notification (`AI_CHAT_BLOCKED_MESSAGE`). diff --git a/docs/deploy.md b/docs/deploy.md index 1bc4856..355f5fa 100644 --- a/docs/deploy.md +++ b/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 ; TeamTalkBot is running.` diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..01715ef --- /dev/null +++ b/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 . +""" diff --git a/tests/test_abuse_tracker.py b/tests/test_abuse_tracker.py new file mode 100644 index 0000000..67e3065 --- /dev/null +++ b/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() diff --git a/tests/test_ai_chat.py b/tests/test_ai_chat.py new file mode 100644 index 0000000..7228959 --- /dev/null +++ b/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() diff --git a/tests/test_ai_review.py b/tests/test_ai_review.py new file mode 100644 index 0000000..d452713 --- /dev/null +++ b/tests/test_ai_review.py @@ -0,0 +1,119 @@ +"""AIReviewer: judging ambiguous badwords with Workers AI.""" + +import time +import unittest +from unittest import mock + +import ai_review +from ai_review import ERROR, EXAMPLES, INSULT, OK, OTHER, SAME, AIReviewer, format_message, parse_verdict + + +def reply(response): + return {"success": True, "result": {"response": response}} + + +class ParseVerdictTest(unittest.TestCase): + def test_json_answer(self): + self.assertEqual(parse_verdict(reply('{"verdict": "insult"}')), INSULT) + self.assertEqual(parse_verdict(reply('Sure! {"verdict": "ok"}')), OK) + + def test_already_parsed_answer(self): + self.assertEqual(parse_verdict(reply({"verdict": "ok"})), OK) + + def test_plain_word_answer(self): + self.assertEqual(parse_verdict(reply("insult")), INSULT) + self.assertEqual(parse_verdict(reply("OK")), OK) + + def test_unusable_answer_raises(self): + with self.assertRaises(ValueError): + parse_verdict(reply("maybe")) + + def test_error_response_raises(self): + with self.assertRaises(RuntimeError): + parse_verdict({"success": False, "errors": ["bad token"]}) + + +class FormatMessageTest(unittest.TestCase): + def test_message_alone(self): + self.assertEqual(format_message("anjing lah"), "Message to judge:\nanjing lah") + + def test_with_conversation(self): + text = format_message("anjing", [(SAME, "aku punya binatang baru")], [(OTHER, "lucu")]) + self.assertEqual( + text.splitlines(), + [ + "Earlier messages (oldest first):", + "- [same person] aku punya binatang baru", + "Message to judge:", + "anjing", + "Later messages:", + "- [someone else] lucu", + ], + ) + + def test_question_to_the_assistant(self): + lines = format_message("@ai anjing", to_assistant=True).splitlines() + self.assertEqual(lines[0], "(This message is a question typed to the AI assistant.)") + self.assertEqual(lines[1:], ["Message to judge:", "@ai anjing"]) + + def test_long_text_is_cut(self): + self.assertEqual(len(format_message("a" * 1000).splitlines()[1]), ai_review.MAX_TEXT) + + +class ClassifyTest(unittest.TestCase): + def test_request(self): + bodies = [] + + def fake(api_url, account, token, model, body, timeout): + bodies.append(body) + return reply('{"verdict": "ok"}') + + reviewer = AIReviewer("account", "token") + with mock.patch.object(ai_review, "call_workers_ai", fake): + verdict = reviewer.classify("@ai anjing", ["anjing"], [(OTHER, "hewan apa?")], [], True) + self.assertEqual(verdict, OK) + messages = bodies[0]["messages"] + self.assertIn("anjing", messages[0]["content"]) + self.assertEqual(len(messages), 1 + 2 * len(EXAMPLES) + 1) + self.assertEqual( + messages[-1]["content"], format_message("@ai anjing", [(OTHER, "hewan apa?")], [], True) + ) + self.assertEqual(bodies[0]["temperature"], 0) + + def test_examples_are_complete(self): + for example in EXAMPLES: + with self.subTest(example=example[1]): + self.assertEqual(len(example), 5) + self.assertIn(example[3], (INSULT, OK)) + + +class BackgroundTest(unittest.TestCase): + def drain(self, reviewer, count): + done, deadline = [], time.time() + 5 + while len(done) < count and time.time() < deadline: + done += reviewer.drain() + time.sleep(0.01) + return done + + def test_verdicts_come_back_through_drain(self): + reviewer = AIReviewer("account", "token") + + def classify(text, words, before=(), after=(), to_assistant=False): + if text == "down": + raise RuntimeError("timeout") + return INSULT if to_assistant else OK + + reviewer.classify = classify + self.assertTrue(reviewer.submit("anjing", ["anjing"], 1)) + self.assertTrue(reviewer.submit("@ai x", ["anjing"], 2, to_assistant=True)) + self.assertTrue(reviewer.submit("down", ["anjing"], 3)) + self.assertEqual(sorted(self.drain(reviewer, 3)), [(1, OK), (2, INSULT), (3, ERROR)]) + + def test_not_configured(self): + reviewer = AIReviewer("", "") + self.assertFalse(reviewer.configured) + self.assertFalse(reviewer.submit("anjing", ["anjing"], 1)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_badwords.py b/tests/test_badwords.py new file mode 100644 index 0000000..28eda37 --- /dev/null +++ b/tests/test_badwords.py @@ -0,0 +1,83 @@ +"""BadWordsFilter: matching, wildcards, stretched spellings and the word file.""" + +import os +import shutil +import tempfile +import unittest + +from badwords import DEFAULT_LIST_NAME, BadWordsFilter + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + + +class FilterTest(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.dir) + self.path = os.path.join(self.dir, "words.txt") + with open(self.path, "w", encoding="utf-8") as f: + f.write("anjing\nkont*l\norang gila\ncok\nBabi, monyet\n") + self.filter = BadWordsFilter() + self.filter.load_file(self.path) + + def matches(self, text): + return self.filter.matching_entries(text) + + def test_plain_words(self): + self.assertEqual(self.matches("dasar ANJING!"), ["anjing"]) + self.assertEqual(self.matches("babi dan monyet"), ["babi", "monyet"]) + self.assertEqual(self.matches("halo semua"), []) + + def test_whole_words_only(self): + self.assertEqual(self.matches("cokelat enak"), []) + + def test_stretched_spellings(self): + self.assertEqual(self.matches("anjiiiiing"), ["anjing"]) + self.assertEqual(self.matches("annnnnjing"), ["anjing"]) + + def test_double_letters_are_not_collapsed(self): + self.assertEqual(self.matches("i cook rice"), []) + + def test_wildcards(self): + self.assertEqual(self.matches("kontol"), ["kont*l"]) + self.assertEqual(self.matches("kont0l"), ["kont*l"]) + self.assertEqual(self.matches("kont*l"), ["kont*l"]) + + def test_phrases(self): + self.assertEqual(self.matches("dasar orang gila"), ["orang gila"]) + self.assertEqual(self.matches("orang yang gila"), []) + + def test_too_broad_wildcards_are_refused(self): + self.assertTrue(self.filter.pattern_error("a*")) + self.assertEqual(self.filter.pattern_error("anj*ng"), "") + self.assertEqual(self.filter.add_words(["a*", "bangsat"]), ["bangsat"]) + + def test_add_and_remove_are_saved(self): + self.assertEqual(self.filter.add_words(["Bangsat", "anjing"]), ["bangsat"]) + self.assertEqual(self.filter.remove_words(["cok", "tidakada"]), ["cok"]) + reloaded = BadWordsFilter() + reloaded.load_file(self.path) + self.assertIn("bangsat", reloaded.list_words()) + self.assertNotIn("cok", reloaded.list_words()) + + def test_new_list_starts_from_the_default_list(self): + with open(os.path.join(self.dir, DEFAULT_LIST_NAME), "w", encoding="utf-8") as f: + f.write("tolol\n") + fresh = BadWordsFilter() + fresh.load_file(os.path.join(self.dir, "new_words.txt")) + self.assertEqual(fresh.list_words(), ["tolol"]) + + +class ShippedListTest(unittest.TestCase): + def test_every_entry_is_usable(self): + checker = BadWordsFilter() + with open(os.path.join(ROOT, "badwords", DEFAULT_LIST_NAME), encoding="utf-8") as f: + entries = [w.strip().lower() for line in f for w in line.split(",") if w.strip()] + self.assertTrue(entries) + for entry in entries: + with self.subTest(entry=entry): + self.assertEqual(checker.pattern_error(entry), "") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_help_texts.py b/tests/test_help_texts.py new file mode 100644 index 0000000..e9edfb1 --- /dev/null +++ b/tests/test_help_texts.py @@ -0,0 +1,23 @@ +"""Help texts: every topic can be shown.""" + +import unittest + +from help_texts import HELP_TOPICS, get_general_help, get_topic_help + + +class HelpTest(unittest.TestCase): + def test_every_topic(self): + for topic, (title, body) in HELP_TOPICS.items(): + with self.subTest(topic=topic): + self.assertTrue(title and body.strip()) + self.assertEqual(get_topic_help(topic), f"{title}\n\n{body}") + self.assertEqual(get_topic_help("/" + topic.upper()), get_topic_help(topic)) + + def test_unknown_topic_shows_the_overview(self): + self.assertEqual(get_topic_help("tidakada"), get_general_help()) + self.assertEqual(get_topic_help(None), get_general_help()) + self.assertIn("@ai", get_general_help()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tt_client.py b/tt_client.py index 8871f8e..ac2cad8 100644 --- a/tt_client.py +++ b/tt_client.py @@ -4048,11 +4048,8 @@ def _execute_ai_tool(self, tool_name: str, tool_args: dict, context: Any) -> Dic } elif tool_name == "get_bot_info": - ver = getattr(config, "VERSION", "1.1.2") - try: - tt_ver = tt5.getVersion() if hasattr(tt5, "getVersion") else "5.x" - except Exception: - tt_ver = "5.x" + info = version.collect_version_info() + ver, tt_ver = info["bot_version"], info["sdk_version"] return { "status": "success", "message": f"Aku bot TeamTalk untuk membantu teman-teman di server ini. Versi bot {ver}, TeamTalk SDK {tt_ver}. Fitur yang tersedia: moderasi otomatis, deteksi spam, filter kata kasar, manajemen channel, dan bantuan cerdas melalui @ai.",