mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
27 lines
614 B
Python
27 lines
614 B
Python
import unittest
|
|
|
|
from mock import Mock, patch
|
|
from patroni.async_executor import AsyncExecutor, CriticalTask
|
|
from threading import Thread
|
|
|
|
|
|
class TestAsyncExecutor(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.a = AsyncExecutor(Mock())
|
|
|
|
@patch.object(Thread, 'start', Mock())
|
|
def test_run_async(self):
|
|
self.a.run_async(Mock(return_value=True))
|
|
|
|
def test_run(self):
|
|
self.a.run(Mock(side_effect=Exception()))
|
|
|
|
|
|
class TestCriticalTask(unittest.TestCase):
|
|
|
|
def test_completed_task(self):
|
|
ct = CriticalTask()
|
|
ct.complete(1)
|
|
self.assertFalse(ct.cancel())
|