test(main): replace brittle smoke checks with AST contracts

2026-02-16 00:31:38 +03:00
parent cd5e6e1f6b
commit db5d901435
3 changed files with 115 additions and 61 deletions
--- a/tests/test_main_contracts.py
+++ b/tests/test_main_contracts.py
@@ -0,0 +1,114 @@
+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"}
+        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_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()