import os import re import sys import tempfile import types import unittest from datetime import date from types import SimpleNamespace from unittest.mock import patch class FakeMessage: def __init__(self, text=None, **kwargs): self.text = text fake_models = types.ModuleType("zlapi.models") fake_models.Message = FakeMessage fake_models.ThreadType = object fake_zlapi = types.ModuleType("zlapi") fake_zlapi.models = fake_models sys.modules.setdefault("zlapi", fake_zlapi) sys.modules.setdefault("zlapi.models", fake_models) from modules import baohanh THREAD_TYPE = SimpleNamespace(USER="user", GROUP="group") TODAY = date(2026, 9, 23) NEXT_YEAR = date(2027, 9, 23) class FakeClient: def __init__(self): self.command_handler = SimpleNamespace(current_prefix=".") self.replies = [] self.direct_messages = [] def replyMessage(self, message, reply_target, thread_id, thread_type, ttl=0): self.replies.append(message.text) def sendMessage(self, message, thread_id, thread_type): self.direct_messages.append((message.text, thread_id, thread_type)) class BaoHanhTests(unittest.TestCase): def setUp(self): temp_dir = tempfile.TemporaryDirectory() self.addCleanup(temp_dir.cleanup) for target, value in ( ("STATE_FILE", os.path.join(temp_dir.name, "baohanh.json")), ("ThreadType", THREAD_TYPE), ("ADMIN", ["1"]), ("_today", lambda: TODAY), ("send_sales_contact", lambda *args: None), ): patcher = patch.object(baohanh, target, value) patcher.start() self.addCleanup(patcher.stop) def run_command(self, text, author_id, thread_type=THREAD_TYPE.GROUP, message_object=None): client = FakeClient() baohanh.handle_baohanh_command( text, message_object or SimpleNamespace(), "group-1", thread_type, author_id, client ) return client def test_shared_account_serves_max_uses_customers_then_runs_out(self): self.run_command(".baohanh them CapCut 2 mail@x.com | pass", "1") for uid in ("111111111", "222222222", "333333333"): baohanh.register_order(uid, "capcut", NEXT_YEAR, "1") self.assertEqual(baohanh.claim_account("111111111", "capcut")[0], "ok") self.assertEqual(baohanh.claim_account("222222222", "capcut")[0], "ok") # hỏi lại thì nhận lại đúng TK cũ, không tốn thêm lượt self.assertEqual(baohanh.claim_account("111111111", "capcut")[0], "ok") self.assertEqual(baohanh.claim_account("333333333", "capcut")[0], "out_of_stock") self.assertEqual(baohanh._load()["accounts"]["1"]["used_by"], ["111111111", "222222222"]) def test_unregistered_expired_and_unknown_products_are_refused(self): self.run_command(".baohanh them capcut 5 acc", "1") baohanh.register_order("111111111", "capcut", date(2026, 9, 1), "1") self.assertEqual(baohanh.claim_account("111111111", "capcut")[0], "expired") self.assertEqual(baohanh.claim_account("999999999", "capcut")[0], "no_order") baohanh.register_order("111111111", "chatgpt", NEXT_YEAR, "1") self.assertEqual(baohanh.claim_account("111111111", "chatgpt")[0], "out_of_stock") def test_disabled_account_is_replaced_on_next_request(self): self.run_command(".baohanh them capcut 5 old-acc", "1") baohanh.register_order("111111111", "capcut", NEXT_YEAR, "1") self.assertEqual(baohanh.claim_account("111111111", "capcut")[1]["credentials"], "old-acc") self.run_command(".baohanh xoa #1", "1") self.run_command(".baohanh them capcut 5 new-acc", "1") self.assertEqual(baohanh.claim_account("111111111", "capcut")[1]["credentials"], "new-acc") def test_sua_accepts_id_or_product_code(self): self.run_command(".baohanh them capcut 3 acc-a", "1") self.run_command(".baohanh them capcut 3 acc-b", "1") self.run_command(".baohanh sua capcut 5", "1") self.run_command(".baohanh sua 2 7", "1") accounts = baohanh._load()["accounts"] self.assertEqual((accounts["1"]["max_uses"], accounts["2"]["max_uses"]), (5, 7)) def test_trangthai_marks_one_account_broken_and_back(self): self.run_command(".baohanh them capcut 3 a@hotmail.com|123", "1", THREAD_TYPE.USER) self.run_command(".baohanh them capcut 3 b@hotmail.com|123", "1", THREAD_TYPE.USER) baohanh.register_order("111111111", "capcut", NEXT_YEAR, "1") accounts = lambda: baohanh._load()["accounts"] # khớp nhiều TK thì không đổi gì client = self.run_command(".baohanh trangthai loi hotmail", "1") self.assertIn("#1, #2", client.replies[0]) self.assertTrue(accounts()["1"]["active"] and accounts()["2"]["active"]) # reply đúng tin bot gửi tài khoản bot_message = SimpleNamespace(quote={"msg": "🛡 TÀI KHOẢN BẢO HÀNH CAPCUT\n\na@hotmail.com|123\n\n⏳ ..."}) self.run_command(".baohanh trangthai loi", "1", message_object=bot_message) self.assertFalse(accounts()["1"]["active"]) self.assertEqual(baohanh.claim_account("111111111", "capcut")[1]["credentials"], "b@hotmail.com|123") self.assertIn("#1 capcut — ⛔ lỗi", self.run_command(".baohanh kho", "1").replies[0]) self.run_command(".baohanh trangthai ok a@hotmail.com", "1") self.run_command(".baohanh trangthai loi B@HOTMAIL.COM", "1") self.assertEqual((accounts()["1"]["active"], accounts()["2"]["active"]), (True, False)) def test_xoa_deletes_by_id_only(self): for credentials in ("a@x.com|1", "b@x.com|1", "c@x.com|1"): self.run_command(f".baohanh them capcut 3 {credentials}", "1", THREAD_TYPE.USER) client = self.run_command(".baohanh xoa a@x.com", "1") self.assertIn("Sai cú pháp", client.replies[0]) self.assertEqual(len(baohanh._load()["accounts"]), 3) client = self.run_command(".baohanh xoa #1 3 9", "1") self.assertEqual(list(baohanh._load()["accounts"]), ["2"]) self.assertIn("#1 (capcut), #3 (capcut)", client.replies[0]) self.assertIn("Không có TK: #9", client.replies[0]) def test_admin_gui_sends_to_customer_private_chat(self): self.run_command(".baohanh them capcut 1 acc-gui", "1", THREAD_TYPE.USER) client = self.run_command(".baohanh gui capcut 111111111", "1") self.assertIn("chưa có đơn", client.replies[0]) self.assertEqual(client.direct_messages, []) # có ngày thì ghi đơn và gửi luôn, TK đi vào tin riêng của khách chứ không phải admin client = self.run_command(".baohanh gui capcut 30/12/2026 111111111", "1") self.assertEqual(client.direct_messages[0][1:], ("111111111", THREAD_TYPE.USER)) self.assertIn("acc-gui", client.direct_messages[0][0]) self.assertNotIn("acc-gui", client.replies[0]) self.assertEqual(baohanh._load()["accounts"]["1"]["used_by"], ["111111111"]) def test_group_replies_never_reveal_how_many_customers_share_an_account(self): baohanh.register_order("111111111", "capcut", NEXT_YEAR, "1") group_replies = [] for command in (".baohanh them capcut 3 acc-a", ".baohanh", ".baohanh sua capcut 5", ".baohanh gui capcut 111111111", ".baohanh kho", ".baohanh trangthai loi 1", ".baohanh trangthai ok 1"): group_replies += self.run_command(command, "1").replies leaked = [text for text in group_replies if re.search(r"\d+/\d+ đơn|khách đang dùng|dùng chung|lượt|tối đa \d+", text)] self.assertEqual(leaked, []) # nhắn riêng thì admin vẫn thấy đủ self.assertIn("1/5 đơn", self.run_command(".baohanh kho", "1", THREAD_TYPE.USER).replies[0]) self.assertNotRegex(self.run_command(".baohanh xoa 1", "1").replies[0], r"\d+ khách|/\d+ đơn") def test_group_request_sends_credentials_only_in_private(self): self.run_command(".baohanh them capcut 1 secret-acc", "1") self.run_command(".baohanh don capcut 30/12/2026 111111111", "1") client = self.run_command(".baohanh capcut", "111111111") self.assertEqual(client.direct_messages[0][1:], ("111111111", THREAD_TYPE.USER)) self.assertIn("secret-acc", client.direct_messages[0][0]) self.assertNotIn("secret-acc", "".join(client.replies)) def test_order_target_comes_from_reply_when_no_uid_given(self): quoted = SimpleNamespace(quote={"uidFrom": "555555555"}) self.run_command(".baohanh don capcut 30/12/2026", "1", message_object=quoted) self.assertEqual(baohanh._load()["orders"]["555555555"]["capcut"]["expires_on"], "2026-12-30") def test_customers_cannot_manage_stock(self): client = self.run_command(".baohanh them capcut 9 hacker-acc", "111111111") self.assertIn("🔒", client.replies[0]) self.assertEqual(baohanh._load()["accounts"], {}) if __name__ == "__main__": unittest.main()