胡绍康andClaude Opus 4.6
192240436d
- 架构重构：拆分为 routes/services/repositories/security 分层结构 - 新增测试：添加完整的单元测试套件（烟雾测试/核心功能/错误处理等） - 完善文档：添加开发者指南、运行验证、回归清单、性能基线等文档 - 前后端分离：静态资源独立，模板组件化 - CI/CD：添加 GitHub Actions 自动化测试 - Docker 优化：添加健康检查，修复调度器启动问题 - 配置增强：支持 SCHEDULER_AUTOSTART 环境变量 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
import time
|
|
import unittest
|
|
import uuid
|
|
|
|
from tests._import_app import import_web_app_module
|
|
|
|
|
|
class DistributedLockTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.module = import_web_app_module()
|
|
|
|
def test_acquire_and_release_distributed_lock(self):
|
|
conn = self.module.create_sqlite_connection()
|
|
try:
|
|
lock_name = "test_lock_" + uuid.uuid4().hex
|
|
owner_a = "a_" + uuid.uuid4().hex
|
|
owner_b = "b_" + uuid.uuid4().hex
|
|
|
|
ok, info = self.module.acquire_distributed_lock(conn, lock_name, owner_a, ttl_seconds=60)
|
|
self.assertTrue(ok)
|
|
self.assertIsNone(info)
|
|
|
|
ok2, info2 = self.module.acquire_distributed_lock(conn, lock_name, owner_b, ttl_seconds=60)
|
|
self.assertFalse(ok2)
|
|
self.assertIsInstance(info2, dict)
|
|
self.assertEqual(info2.get("owner_id"), owner_a)
|
|
|
|
released = self.module.release_distributed_lock(conn, lock_name, owner_a)
|
|
self.assertTrue(released)
|
|
|
|
ok3, info3 = self.module.acquire_distributed_lock(conn, lock_name, owner_b, ttl_seconds=60)
|
|
self.assertTrue(ok3)
|
|
self.assertIsNone(info3)
|
|
finally:
|
|
conn.close()
|
|
|
|
def test_expired_lock_can_be_taken_over(self):
|
|
conn = self.module.create_sqlite_connection()
|
|
try:
|
|
lock_name = "test_lock_expired_" + uuid.uuid4().hex
|
|
owner_a = "a_" + uuid.uuid4().hex
|
|
owner_b = "b_" + uuid.uuid4().hex
|
|
|
|
ok, _ = self.module.acquire_distributed_lock(conn, lock_name, owner_a, ttl_seconds=1)
|
|
self.assertTrue(ok)
|
|
time.sleep(1.2)
|
|
|
|
ok2, info2 = self.module.acquire_distributed_lock(conn, lock_name, owner_b, ttl_seconds=60)
|
|
self.assertTrue(ok2)
|
|
self.assertIsInstance(info2, dict)
|
|
self.assertEqual(info2.get("previous_owner_id"), owner_a)
|
|
finally:
|
|
conn.close()
|
|
|