胡绍康andCopilot
cf7817bb95
UI 改进: - 设置页面内嵌显示（不再弹窗） - 临时邮箱卡片添加验证码提取按钮 - 导入账号弹窗 textarea 样式优化 - 账号头像 8 色渐变轮转 - 临时邮箱页顶栏邮箱点击复制 - 邮箱管理页自动选中默认分组 - 切换分组时重置邮件列 UI - 修复收件箱/垃圾邮件 Tab 切换（.folder-tab → .email-tab） - 修复从临时邮箱返回后账号列数据残留 - 仪表盘分组列表可点击导航 恢复: - 恢复 .github/ CI 工作流 - 恢复 tests/ 测试目录 - 恢复 docs/ 文档目录 - 恢复 CONTRIBUTING.md, SECURITY.md Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
65 lines
2.1 KiB
Python
65 lines
2.1 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()
|