1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
|
"""
Unit test for ClusterShell Task with all engines (local worker)
"""
import sys
import unittest
from ClusterShell.Defaults import DEFAULTS
from ClusterShell.Engine.Select import EngineSelect
from ClusterShell.Engine.Poll import EnginePoll
from ClusterShell.Engine.EPoll import EngineEPoll
from ClusterShell.Task import *
from TaskLocalMixin import TaskLocalMixin
ENGINE_SELECT_ID = EngineSelect.identifier
ENGINE_POLL_ID = EnginePoll.identifier
ENGINE_EPOLL_ID = EngineEPoll.identifier
class TaskLocalEngineSelectTest(TaskLocalMixin, unittest.TestCase):
def setUp(self):
# switch Engine
task_terminate()
self.engine_id_save = DEFAULTS.engine
DEFAULTS.engine = ENGINE_SELECT_ID
# select should be supported anywhere...
self.assertEqual(task_self().info('engine'), ENGINE_SELECT_ID)
# call base class setUp()
TaskLocalMixin.setUp(self)
def tearDown(self):
# call base class tearDown()
TaskLocalMixin.tearDown(self)
# restore Engine
DEFAULTS.engine = self.engine_id_save
task_terminate()
class TaskLocalEnginePollTest(TaskLocalMixin, unittest.TestCase):
def setUp(self):
# switch Engine
task_terminate()
self.engine_id_save = DEFAULTS.engine
DEFAULTS.engine = ENGINE_POLL_ID
if task_self().info('engine') != ENGINE_POLL_ID:
self.skipTest("engine %s not supported on this host"
% ENGINE_POLL_ID)
# call base class setUp()
TaskLocalMixin.setUp(self)
def tearDown(self):
# call base class tearDown()
TaskLocalMixin.tearDown(self)
# restore Engine
DEFAULTS.engine = self.engine_id_save
task_terminate()
# select.epoll is only available with Python 2.6 (if condition to be
# removed once we only support Py2.6+)
if sys.version_info >= (2, 6, 0):
class TaskLocalEngineEPollTest(TaskLocalMixin, unittest.TestCase):
def setUp(self):
# switch Engine
task_terminate()
self.engine_id_save = DEFAULTS.engine
DEFAULTS.engine = ENGINE_EPOLL_ID
if task_self().info('engine') != ENGINE_EPOLL_ID:
self.skipTest("engine %s not supported on this host"
% ENGINE_EPOLL_ID)
# call base class setUp()
TaskLocalMixin.setUp(self)
def tearDown(self):
# call base class tearDown()
TaskLocalMixin.tearDown(self)
# restore Engine
DEFAULTS.engine = self.engine_id_save
task_terminate()
|
