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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
"""Tests for launchers
Doesn't actually start any subprocesses, but goes through the motions of constructing
objects, which should test basic config.
Authors:
* Min RK
"""
#-------------------------------------------------------------------------------
# Copyright (C) 2013 The IPython Development Team
#
# Distributed under the terms of the BSD License. The full license is in
# the file COPYING, distributed as part of this software.
#-------------------------------------------------------------------------------
#-------------------------------------------------------------------------------
# Imports
#-------------------------------------------------------------------------------
import logging
import os
import shutil
import sys
import tempfile
from unittest import TestCase
from nose import SkipTest
from IPython.config import Config
from IPython.parallel.apps import launcher
from IPython.testing import decorators as dec
from IPython.utils.py3compat import string_types
#-------------------------------------------------------------------------------
# TestCase Mixins
#-------------------------------------------------------------------------------
class LauncherTest:
"""Mixin for generic launcher tests"""
def setUp(self):
self.profile_dir = tempfile.mkdtemp(prefix="profile_")
def tearDown(self):
shutil.rmtree(self.profile_dir)
@property
def config(self):
return Config()
def build_launcher(self, **kwargs):
kw = dict(
work_dir=self.profile_dir,
profile_dir=self.profile_dir,
config=self.config,
cluster_id='',
log=logging.getLogger(),
)
kw.update(kwargs)
return self.launcher_class(**kw)
def test_profile_dir_arg(self):
launcher = self.build_launcher()
self.assertTrue("--profile-dir" in launcher.cluster_args)
self.assertTrue(self.profile_dir in launcher.cluster_args)
def test_cluster_id_arg(self):
launcher = self.build_launcher()
self.assertTrue("--cluster-id" in launcher.cluster_args)
idx = launcher.cluster_args.index("--cluster-id")
self.assertEqual(launcher.cluster_args[idx+1], '')
launcher.cluster_id = 'foo'
self.assertEqual(launcher.cluster_args[idx+1], 'foo')
def test_args(self):
launcher = self.build_launcher()
for arg in launcher.args:
self.assertTrue(isinstance(arg, string_types), str(arg))
class BatchTest:
"""Tests for batch-system launchers (LSF, SGE, PBS)"""
def test_batch_template(self):
launcher = self.build_launcher()
batch_file = os.path.join(self.profile_dir, launcher.batch_file_name)
self.assertEqual(launcher.batch_file, batch_file)
launcher.write_batch_script(1)
self.assertTrue(os.path.isfile(batch_file))
class SSHTest:
"""Tests for SSH launchers"""
def test_cluster_id_arg(self):
raise SkipTest("SSH Launchers don't support cluster ID")
def test_remote_profile_dir(self):
cfg = Config()
launcher_cfg = getattr(cfg, self.launcher_class.__name__)
launcher_cfg.remote_profile_dir = "foo"
launcher = self.build_launcher(config=cfg)
self.assertEqual(launcher.remote_profile_dir, "foo")
def test_remote_profile_dir_default(self):
launcher = self.build_launcher()
self.assertEqual(launcher.remote_profile_dir, self.profile_dir)
#-------------------------------------------------------------------------------
# Controller Launcher Tests
#-------------------------------------------------------------------------------
class ControllerLauncherTest(LauncherTest):
"""Tests for Controller Launchers"""
pass
class TestLocalControllerLauncher(ControllerLauncherTest, TestCase):
launcher_class = launcher.LocalControllerLauncher
class TestMPIControllerLauncher(ControllerLauncherTest, TestCase):
launcher_class = launcher.MPIControllerLauncher
class TestPBSControllerLauncher(BatchTest, ControllerLauncherTest, TestCase):
launcher_class = launcher.PBSControllerLauncher
class TestSGEControllerLauncher(BatchTest, ControllerLauncherTest, TestCase):
launcher_class = launcher.SGEControllerLauncher
class TestLSFControllerLauncher(BatchTest, ControllerLauncherTest, TestCase):
launcher_class = launcher.LSFControllerLauncher
class TestHTCondorControllerLauncher(BatchTest, ControllerLauncherTest, TestCase):
launcher_class = launcher.HTCondorControllerLauncher
class TestSSHControllerLauncher(SSHTest, ControllerLauncherTest, TestCase):
launcher_class = launcher.SSHControllerLauncher
#-------------------------------------------------------------------------------
# Engine Set Launcher Tests
#-------------------------------------------------------------------------------
class EngineSetLauncherTest(LauncherTest):
"""Tests for EngineSet launchers"""
pass
class TestLocalEngineSetLauncher(EngineSetLauncherTest, TestCase):
launcher_class = launcher.LocalEngineSetLauncher
class TestMPIEngineSetLauncher(EngineSetLauncherTest, TestCase):
launcher_class = launcher.MPIEngineSetLauncher
class TestPBSEngineSetLauncher(BatchTest, EngineSetLauncherTest, TestCase):
launcher_class = launcher.PBSEngineSetLauncher
class TestSGEEngineSetLauncher(BatchTest, EngineSetLauncherTest, TestCase):
launcher_class = launcher.SGEEngineSetLauncher
class TestLSFEngineSetLauncher(BatchTest, EngineSetLauncherTest, TestCase):
launcher_class = launcher.LSFEngineSetLauncher
class TestHTCondorEngineSetLauncher(BatchTest, EngineSetLauncherTest, TestCase):
launcher_class = launcher.HTCondorEngineSetLauncher
class TestSSHEngineSetLauncher(EngineSetLauncherTest, TestCase):
launcher_class = launcher.SSHEngineSetLauncher
def test_cluster_id_arg(self):
raise SkipTest("SSH Launchers don't support cluster ID")
class TestSSHProxyEngineSetLauncher(SSHTest, LauncherTest, TestCase):
launcher_class = launcher.SSHProxyEngineSetLauncher
class TestSSHEngineLauncher(SSHTest, LauncherTest, TestCase):
launcher_class = launcher.SSHEngineLauncher
#-------------------------------------------------------------------------------
# Windows Launcher Tests
#-------------------------------------------------------------------------------
class WinHPCTest:
"""Tests for WinHPC Launchers"""
def test_batch_template(self):
launcher = self.build_launcher()
job_file = os.path.join(self.profile_dir, launcher.job_file_name)
self.assertEqual(launcher.job_file, job_file)
launcher.write_job_file(1)
self.assertTrue(os.path.isfile(job_file))
class TestWinHPCControllerLauncher(WinHPCTest, ControllerLauncherTest, TestCase):
launcher_class = launcher.WindowsHPCControllerLauncher
class TestWinHPCEngineSetLauncher(WinHPCTest, EngineSetLauncherTest, TestCase):
launcher_class = launcher.WindowsHPCEngineSetLauncher
|
