AnabasisChatRemove/tests/test_main_contracts.py

import ast
import unittest
from pathlib import Path


class MainContractsTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.main_source = Path("main.py").read_text(encoding="utf-8-sig")
        cls.module = ast.parse(cls.main_source)
        cls.vk_chat_manager = cls._find_class("VkChatManager")

    @classmethod
    def _find_class(cls, class_name):
        for node in cls.module.body:
            if isinstance(node, ast.ClassDef) and node.name == class_name:
                return node
        raise AssertionError(f"Class {class_name} not found")

    def _find_method(self, method_name):
        for node in self.vk_chat_manager.body:
            if isinstance(node, ast.FunctionDef) and node.name == method_name:
                return node
        self.fail(f"Method {method_name} not found")

    def _iter_nodes(self, node):
        return ast.walk(node)

    def test_auth_error_contexts_contains_only_supported_contexts(self):
        expected_contexts = {"load_chats", "execute_user_action", "set_user_admin", "unset_user_admin"}
        for node in self.module.body:
            if isinstance(node, ast.Assign):
                for target in node.targets:
                    if isinstance(target, ast.Name) and target.id == "AUTH_ERROR_CONTEXTS":
                        actual = set(ast.literal_eval(node.value))
                        self.assertSetEqual(actual, expected_contexts)
                        return
        self.fail("AUTH_ERROR_CONTEXTS assignment not found")

    def test_unset_user_admin_method_exists(self):
        self._find_method("unset_user_admin")

    def test_uses_custom_standalone_vk_app(self):
        constants = {}
        for node in self.module.body:
            if not isinstance(node, ast.Assign) or len(node.targets) != 1:
                continue
            target = node.targets[0]
            if isinstance(target, ast.Name):
                try:
                    constants[target.id] = ast.literal_eval(node.value)
                except Exception:
                    continue

        self.assertEqual(constants.get("VK_APP_ID"), "54454043")
        self.assertIn("ANABASIS_VK_REDIRECT_URI", self.main_source)
        self.assertIn("https://vk.daemonlord.ru/vk/callback", self.main_source)

    def test_remove_action_demotes_member_before_removing(self):
        bulk_worker = None
        for node in self.module.body:
            if isinstance(node, ast.ClassDef) and node.name == "BulkActionWorker":
                bulk_worker = node
                break
        self.assertIsNotNone(bulk_worker, "BulkActionWorker class not found")

        run_method = None
        for node in bulk_worker.body:
            if isinstance(node, ast.FunctionDef) and node.name == "run":
                run_method = node
                break
        self.assertIsNotNone(run_method, "BulkActionWorker.run method not found")

        has_member_role = False
        has_remove_call = False
        for node in ast.walk(run_method):
            if isinstance(node, ast.keyword) and node.arg == "role":
                if isinstance(node.value, ast.Constant) and node.value.value == "member":
                    has_member_role = True
            if isinstance(node, ast.Attribute) and node.attr == "removeChatUser":
                has_remove_call = True

        self.assertTrue(has_member_role, "remove action must demote admins with role='member'")
        self.assertTrue(has_remove_call, "remove action must still call removeChatUser")

    def test_check_for_updates_has_reentry_guard(self):
        method = self._find_method("check_for_updates")
        has_guard = False
        for node in method.body:
            if not isinstance(node, ast.If):
                continue
            test = node.test
            if (
                isinstance(test, ast.Attribute)
                and isinstance(test.value, ast.Name)
                and test.value.id == "self"
                and test.attr == "_update_in_progress"
            ):
                has_guard = any(isinstance(stmt, ast.Return) for stmt in node.body)
                if has_guard:
                    break
        self.assertTrue(has_guard, "check_for_updates must return when update is already in progress")

    def test_check_for_updates_connects_thread_finish_handler(self):
        method = self._find_method("check_for_updates")
        for node in self._iter_nodes(method):
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            if not (isinstance(func, ast.Attribute) and func.attr == "connect"):
                continue
            value = func.value
            if not (
                isinstance(value, ast.Attribute)
                and value.attr == "finished"
                and isinstance(value.value, ast.Attribute)
                and value.value.attr == "update_thread"
                and isinstance(value.value.value, ast.Name)
                and value.value.value.id == "self"
            ):
                continue
            if len(node.args) != 1:
                continue
            arg = node.args[0]
            if (
                isinstance(arg, ast.Attribute)
                and arg.attr == "_on_update_thread_finished"
                and isinstance(arg.value, ast.Name)
                and arg.value.id == "self"
            ):
                return
        self.fail("update_thread.finished must be connected to _on_update_thread_finished")

    def test_on_update_thread_finished_clears_update_state(self):
        method = self._find_method("_on_update_thread_finished")
        assignments = {}
        for node in method.body:
            if not isinstance(node, ast.Assign) or len(node.targets) != 1:
                continue
            target = node.targets[0]
            if (
                isinstance(target, ast.Attribute)
                and isinstance(target.value, ast.Name)
                and target.value.id == "self"
            ):
                assignments[target.attr] = node.value

        self.assertIn("_update_in_progress", assignments)
        self.assertIn("update_checker", assignments)
        self.assertIn("update_thread", assignments)
        self.assertIsInstance(assignments["_update_in_progress"], ast.Constant)
        self.assertIs(assignments["_update_in_progress"].value, False)
        self.assertIsInstance(assignments["update_checker"], ast.Constant)
        self.assertIsNone(assignments["update_checker"].value)
        self.assertIsInstance(assignments["update_thread"], ast.Constant)
        self.assertIsNone(assignments["update_thread"].value)


if __name__ == "__main__":
    unittest.main()