« back to all changes in this revision
Viewing changes to mojo/tests/test_mojo.py
- browse files at revision 546
- compare with another revision
- download diff
- download tarball
- view history from revision 546
- Committer: mergebot at canonical
- Author(s): "Stuart Bishop"
- Date: 2020-04-22 13:04:02 UTC
- mfrom: (545.1.6 blacken)
- Revision ID: mergebot@juju-139df4-prod-is-toolbox-0.canonical.com-20200422130402-55dc5d92dj2lx14h
Reformat code with black
Reviewed-on: https://code.launchpad.net/~stub/mojo/blacken/+merge/382726
Reviewed-by: Tom Haddon <tom.haddon@canonical.com>
Reviewed-on: https://code.launchpad.net/~stub/mojo/blacken/+merge/382726
Reviewed-by: Tom Haddon <tom.haddon@canonical.com>
- files modified:
- Makefile
added
removed
mojo/tests/test_mojo.py
27
27
from unittest import mock
28
28
29
29
30
local_copy = os.path.join(
31
os.path.dirname(os.path.dirname(__file__)), "..",
32
"mojo")
30
local_copy = os.path.join(os.path.dirname(os.path.dirname(__file__)), "..", "mojo")
33
31
if os.path.exists(local_copy) and os.path.isdir(local_copy):
34
32
sys.path.insert(0, local_copy)
35
33
40
38
41
39
42
40
class WorkspaceTest(TestCase):
43
44
41
def setUp(self):
45
42
super(WorkspaceTest, self).setUp()
46
43
self.bad_path = "/INVALID/PATH"
53
50
self.local_dir_name = "local"
54
51
self.log_dir_name = "log"
55
52
self.spec_dir_name = "spec"
56
self.all_names = [
57
self.build_dir_name,
58
self.local_dir_name,
59
self.log_dir_name,
60
self.spec_dir_name,
61
]
53
self.all_names = [self.build_dir_name, self.local_dir_name, self.log_dir_name, self.spec_dir_name]
62
54
63
55
def test_workspace_sets_path(self):
64
56
w = mojo.workspace.Workspace("workspace_name", self.good_path)
78
70
79
71
all_subdirs = [os.path.join(self.good_path, d) for d in self.all_names]
80
72
for subdir in all_subdirs:
81
if os.path.basename(subdir) == 'spec': # Skip the spec dir
73
if os.path.basename(subdir) == "spec": # Skip the spec dir
82
74
continue
83
75
self.assertTrue(os.path.exists(subdir))
84
76
88
80
89
81
all_subdirs = [os.path.join(self.good_path, d) for d in self.all_names]
90
82
for subdir in all_subdirs:
91
if os.path.basename(subdir) == 'spec': # Skip the spec dir
83
if os.path.basename(subdir) == "spec": # Skip the spec dir
92
84
continue
93
85
self.assertTrue(os.path.exists(subdir))
94
86
112
104
spec_dir = os.path.join(self.good_path, self.spec_dir_name)
113
105
self.assertEqual(w.spec_dir, spec_dir)
114
106
115
@mock.patch('logging.info')
107
@mock.patch("logging.info")
116
108
def test_workspace_spec_update_from_source(self, _logging_info):
117
109
w = mojo.workspace.Workspace("workspace_name", self.good_path, spec_url=self.working_spec_path)
118
110
w.spec.update_from_source()
119
111
# Confirm the working spec dir is empty and source for spec only
120
112
# contains the auto-generated codetree-collect-info yaml file.
121
113
self.assertEqual(os.listdir(self.working_spec_path), [])
122
self.assertEqual(os.listdir(w.spec_dir), ['codetree-collect-info.yaml'])
114
self.assertEqual(os.listdir(w.spec_dir), ["codetree-collect-info.yaml"])
123
115
# Now create a file in our source spec dir.
124
with open(os.path.join(self.working_spec_path, 'testfile'), 'w') as f:
116
with open(os.path.join(self.working_spec_path, "testfile"), "w") as f:
125
117
f.write("testcontents")
126
118
# Refresh the working spec directory from source.
127
119
w.spec.update_from_source()
128
120
# And now confirm our working spec directory contains the file.
129
self.assertEqual(sorted(os.listdir(w.spec_dir)), ['codetree-collect-info.yaml', 'testfile'])
130
with open(os.path.join(w.spec_dir, 'testfile'), 'r') as f:
131
self.assertEqual(f.read(), 'testcontents')
121
self.assertEqual(sorted(os.listdir(w.spec_dir)), ["codetree-collect-info.yaml", "testfile"])
122
with open(os.path.join(w.spec_dir, "testfile"), "r") as f:
123
self.assertEqual(f.read(), "testcontents")
132
124
133
125
134
126
class SpecTest(TestCase, MojoTestCaseMixin):
135
136
127
def setUp(self):
137
128
self.git_daemon_pid = None
138
129
139
130
def test_get_config(self):
140
cases = [{
141
"config_files": ['staging/service.yml'],
142
"input": dict(base="service.yml", stage='staging'),
143
"expected": "staging/service.yml",
144
}, {
145
"config_files": ['service.yml', 'staging/service.yml'],
146
"input": dict(base="service.yml", stage='staging'),
147
"expected": "staging/service.yml",
148
}, {
149
"config_files": ['service.yml', 'service.yml-staging'],
150
"input": dict(base="service.yml", stage=None),
151
"expected": "service.yml",
152
}, {
153
"config_files": [
154
'service.yml',
155
'pes/service.yml',
156
'pes/staging/service.yml',
157
],
158
"input": dict(base="service.yml", stage='pes/staging'),
159
"expected": "pes/staging/service.yml",
160
}, {
161
"config_files": ['service.yml', 'pes/service.yml'],
162
"input": dict(base="service.yml", stage='pes/staging'),
163
"expected": "pes/service.yml",
164
}]
131
cases = [
132
{
133
"config_files": ["staging/service.yml"],
134
"input": dict(base="service.yml", stage="staging"),
135
"expected": "staging/service.yml",
136
},
137
{
138
"config_files": ["service.yml", "staging/service.yml"],
139
"input": dict(base="service.yml", stage="staging"),
140
"expected": "staging/service.yml",
141
},
142
{
143
"config_files": ["service.yml", "service.yml-staging"],
144
"input": dict(base="service.yml", stage=None),
145
"expected": "service.yml",
146
},
147
{
148
"config_files": ["service.yml", "pes/service.yml", "pes/staging/service.yml"],
149
"input": dict(base="service.yml", stage="pes/staging"),
150
"expected": "pes/staging/service.yml",
151
},
152
{
153
"config_files": ["service.yml", "pes/service.yml"],
154
"input": dict(base="service.yml", stage="pes/staging"),
155
"expected": "pes/service.yml",
156
},
157
]
165
158
for test_case in cases:
166
spec_dir = self.create_spec_dir(test_case['config_files'],
167
stage=test_case['input']['stage'])
159
spec_dir = self.create_spec_dir(test_case["config_files"], stage=test_case["input"]["stage"])
168
160
169
161
spec = mojo.spec.Spec(spec_dir, spec_dir) # Tests spec_dir and spec_url being the same
170
162
171
163
expected_config = os.path.join(spec_dir, test_case["expected"])
172
actual_config = spec.get_config(**test_case['input'])
164
actual_config = spec.get_config(**test_case["input"])
173
165
self.assertEqual(expected_config, actual_config)
174
166
175
167
def test_get_configs(self):
176
cases = [{
177
"config_files": ['staging/service.yml'],
178
"input": dict(base="service.yml", stage='staging'),
179
"expected": ["staging/service.yml"],
180
}, {
181
"config_files": ['service.yml', 'staging/service.yml'],
182
"input": dict(base="service.yml", stage='staging'),
183
"expected": ["service.yml", "staging/service.yml"],
184
}, {
185
"config_files": [
186
'staging/service.yml',
187
'service.yml',
188
],
189
"input": dict(base="service.yml", stage='staging'),
190
"expected": [
191
"service.yml",
192
'staging/service.yml',
193
],
194
}, {
195
"config_files": ['service.yml', 'service.yml-staging'],
196
"input": dict(base="service.yml", stage=None),
197
"expected": ["service.yml"],
198
}, {
199
"config_files": [
200
'service.yml',
201
'pes/service.yml',
202
'pes/staging/service.yml',
203
],
204
"input": dict(base="service.yml", stage='pes/staging'),
205
"expected": [
206
"service.yml",
207
"pes/service.yml",
208
"pes/staging/service.yml",
209
],
210
}, {
211
"config_files": ['service.yml', 'pes/service.yml'],
212
"input": dict(base="service.yml", stage='pes/staging'),
213
"expected": ["service.yml", "pes/service.yml"],
214
}]
168
cases = [
169
{
170
"config_files": ["staging/service.yml"],
171
"input": dict(base="service.yml", stage="staging"),
172
"expected": ["staging/service.yml"],
173
},
174
{
175
"config_files": ["service.yml", "staging/service.yml"],
176
"input": dict(base="service.yml", stage="staging"),
177
"expected": ["service.yml", "staging/service.yml"],
178
},
179
{
180
"config_files": ["staging/service.yml", "service.yml"],
181
"input": dict(base="service.yml", stage="staging"),
182
"expected": ["service.yml", "staging/service.yml"],
183
},
184
{
185
"config_files": ["service.yml", "service.yml-staging"],
186
"input": dict(base="service.yml", stage=None),
187
"expected": ["service.yml"],
188
},
189
{
190
"config_files": ["service.yml", "pes/service.yml", "pes/staging/service.yml"],
191
"input": dict(base="service.yml", stage="pes/staging"),
192
"expected": ["service.yml", "pes/service.yml", "pes/staging/service.yml"],
193
},
194
{
195
"config_files": ["service.yml", "pes/service.yml"],
196
"input": dict(base="service.yml", stage="pes/staging"),
197
"expected": ["service.yml", "pes/service.yml"],
198
},
199
]
215
200
for test_case in cases:
216
spec_dir = self.create_spec_dir(test_case['config_files'],
217
stage=test_case['input']['stage'])
201
spec_dir = self.create_spec_dir(test_case["config_files"], stage=test_case["input"]["stage"])
218
202
219
203
spec = mojo.spec.Spec(spec_dir) # Tests spec_url being None
220
204
221
actual_configs = spec.get_configs(**test_case['input'])
205
actual_configs = spec.get_configs(**test_case["input"])
222
206
self.assertEqual(test_case["expected"], actual_configs)
223
207
224
208
def _wait_for_socket(self, addr, port):
225
socket = '{}:{}'.format(addr, port)
226
cmd = ['ss', 'state', 'listening', 'src', socket]
209
socket = "{}:{}".format(addr, port)
210
cmd = ["ss", "state", "listening", "src", socket]
227
211
waited = 1
228
212
229
213
while waited < 10:
237
221
238
222
def test_git_spec(self):
239
223
# Do the spec_dir setup like other tests as our baseline
240
bzr_spec_dir = self.create_spec_dir(['service.yml', 'staging/service.yml'], stage='staging')
224
bzr_spec_dir = self.create_spec_dir(["service.yml", "staging/service.yml"], stage="staging")
241
225
bzr_spec = mojo.spec.Spec(bzr_spec_dir)
242
226
bzr_configs = bzr_spec.get_configs(base="service.yml", stage="staging")
243
227
249
233
git_dir = mkdtemp()
250
234
self.addCleanup(rmtree, git_dir)
251
235
os.rmdir(git_dir)
252
subprocess.check_call(['cp', '-r', bzr_spec_dir, git_dir])
236
subprocess.check_call(["cp", "-r", bzr_spec_dir, git_dir])
253
237
254
238
# give git an identity if none exists
255
if subprocess.call(['git', 'config', '--global', '--get', 'user.email']) == 1:
256
subprocess.call(['git', 'config', '--global', 'user.email', "you@example.com"])
257
if subprocess.call(['git', 'config', '--global', '--get', 'user.name']) == 1:
258
subprocess.call(['git', 'config', '--global', 'user.name', "Your Name"])
239
if subprocess.call(["git", "config", "--global", "--get", "user.email"]) == 1:
240
subprocess.call(["git", "config", "--global", "user.email", "you@example.com"])
241
if subprocess.call(["git", "config", "--global", "--get", "user.name"]) == 1:
242
subprocess.call(["git", "config", "--global", "user.name", "Your Name"])
259
243
260
subprocess.check_call(['git', '-C', git_dir, 'init', '-q'])
261
subprocess.check_call(('git', '-C', git_dir, 'add', '*'))
262
subprocess.check_call(('git', '-C', git_dir, 'commit', '-q', '-a', '-m', 'initial commit'))
244
subprocess.check_call(["git", "-C", git_dir, "init", "-q"])
245
subprocess.check_call(("git", "-C", git_dir, "add", "*"))
246
subprocess.check_call(("git", "-C", git_dir, "commit", "-q", "-a", "-m", "initial commit"))
263
247
264
248
# Daemon servers do not stay alive via subprocess.Popen
265
249
# self.git_daemon_pid = subprocess.Popen(('git', '-C', git_dir, 'daemon', '--reuseaddr', '--export-all',
278
262
279
263
280
264
class UtilsTest(TestCase):
281
282
265
def test_bicommand_environment_expansion(self):
283
266
"""Environment variables in bicommand invocations are expanded."""
284
267
response = "I've come to talk with you again."
285
268
status, output = mojo.utils.bicommand(
286
"echo ${HELLO_DARKNESS_MY_OLD_FRIEND}",
287
env={"HELLO_DARKNESS_MY_OLD_FRIEND": response},
288
)
269
"echo ${HELLO_DARKNESS_MY_OLD_FRIEND}", env={"HELLO_DARKNESS_MY_OLD_FRIEND": response}
270
)
289
271
self.assertEqual(status, 0)
290
272
self.assertEqual(output, response)
291
273
292
274
def test_is_config_line(self):
293
275
"""Test various types of lines for whether they are valid config or not"""
294
self.assertEqual(True, is_config_line('This is some config'))
295
self.assertEqual(True, is_config_line(' This is some config '))
296
self.assertEqual(False, is_config_line(' '))
297
self.assertEqual(False, is_config_line('# This is not config'))
298
self.assertEqual(False, is_config_line(' # This is not config'))
276
self.assertEqual(True, is_config_line("This is some config"))
277
self.assertEqual(True, is_config_line(" This is some config "))
278
self.assertEqual(False, is_config_line(" "))
279
self.assertEqual(False, is_config_line("# This is not config"))
280
self.assertEqual(False, is_config_line(" # This is not config"))
299
281
300
282
301
283
class ProjectTest(TestCase):
302
303
284
def test_project_lxc_trusty(self):
304
project = mojo.project.Project('my-project', 'trusty', '/srv/mojo', container_class=mojo.contain.LXCContainer)
285
project = mojo.project.Project("my-project", "trusty", "/srv/mojo", container_class=mojo.contain.LXCContainer)
305
286
self.assertTrue(type(project.container) == mojo.contain.LXCContainer)
306
self.assertTrue(project.container.container_root == '/srv/mojo/my-project/trusty/ROOTFS')
287
self.assertTrue(project.container.container_root == "/srv/mojo/my-project/trusty/ROOTFS")
307
288
308
289
def test_project_lxc_trusty_exists(self):
309
290
"""Pre-existing project has lxc root at
310
291
/var/lib/lxc/my-project.trusty/rootfs"""
311
292
mojo_root = mkdtemp()
312
293
self.addCleanup(rmtree, mojo_root)
313
project_series_dir = os.path.join(mojo_root, 'my-project', 'trusty')
294
project_series_dir = os.path.join(mojo_root, "my-project", "trusty")
314
295
mojo.utils.mkdirs(project_series_dir)
315
project_info_file = os.path.join(mojo_root, 'my-project', 'trusty', '.project')
316
existing_project_info = {'container_root': '/var/lib/lxc/my-project.trusty/rootfs',
317
'container_class': 'lxc'}
318
with open(project_info_file, 'w') as info:
296
project_info_file = os.path.join(mojo_root, "my-project", "trusty", ".project")
297
existing_project_info = {"container_root": "/var/lib/lxc/my-project.trusty/rootfs", "container_class": "lxc"}
298
with open(project_info_file, "w") as info:
319
299
info.write(json.dumps(existing_project_info))
320
300
321
project = mojo.project.Project('my-project', 'trusty', mojo_root, container_class=mojo.contain.LXCContainer)
301
project = mojo.project.Project("my-project", "trusty", mojo_root, container_class=mojo.contain.LXCContainer)
322
302
self.assertTrue(type(project.container) == mojo.contain.LXCContainer)
323
self.assertTrue(project.container.container_root == '/var/lib/lxc/my-project.trusty/rootfs')
303
self.assertTrue(project.container.container_root == "/var/lib/lxc/my-project.trusty/rootfs")
324
304
325
305
def test_project_containerless(self):
326
project = mojo.project.Project('my-project', 'trusty', '/srv/mojo', container_class=mojo.contain.NoContainer)
306
project = mojo.project.Project("my-project", "trusty", "/srv/mojo", container_class=mojo.contain.NoContainer)
327
307
self.assertTrue(type(project.container) == mojo.contain.NoContainer)
328
self.assertTrue(project.container.container_root == '/srv/mojo/my-project/trusty')
308
self.assertTrue(project.container.container_root == "/srv/mojo/my-project/trusty")
329
309
330
310
def test_project_containerless_exists(self):
331
311
"""Pre-existing containerless project"""
332
312
mojo_root = mkdtemp()
333
313
self.addCleanup(rmtree, mojo_root)
334
project_series_dir = os.path.join(mojo_root, 'my-project', 'trusty')
314
project_series_dir = os.path.join(mojo_root, "my-project", "trusty")
335
315
mojo.utils.mkdirs(project_series_dir)
336
project_info_file = os.path.join(project_series_dir, '.project')
337
existing_project_info = {'container_root': project_series_dir,
338
'container_class': 'containerless'}
339
with open(project_info_file, 'w') as info:
316
project_info_file = os.path.join(project_series_dir, ".project")
317
existing_project_info = {"container_root": project_series_dir, "container_class": "containerless"}
318
with open(project_info_file, "w") as info:
340
319
info.write(json.dumps(existing_project_info))
341
320
342
321
# Recieves the default LXCContainer but is a pre-existing
343
322
# NoContainer project
344
project = mojo.project.Project('my-project', 'trusty', mojo_root, container_class=mojo.contain.LXCContainer)
323
project = mojo.project.Project("my-project", "trusty", mojo_root, container_class=mojo.contain.LXCContainer)
345
324
self.assertTrue(type(project.container) == mojo.contain.NoContainer)
346
325
self.assertTrue(project.container.container_root == project_series_dir)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1