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
|
# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import fixtures
from tempest.cmd import init
from tempest.tests import base
class TestTempestInit(base.TestCase):
def test_generate_stestr_conf(self):
# Create fake conf dir
conf_dir = self.useFixture(fixtures.TempDir())
init_cmd = init.TempestInit(None, None)
init_cmd.generate_stestr_conf(conf_dir.path)
# Generate expected file contents
top_level_path = os.path.dirname(os.path.dirname(init.__file__))
discover_path = os.path.join(top_level_path, 'test_discover')
stestr_conf_file = init.STESTR_CONF % (discover_path, top_level_path)
conf_path = conf_dir.join('.stestr.conf')
with open(conf_path, 'r') as conf_file:
self.assertEqual(conf_file.read(), stestr_conf_file)
def test_generate_sample_config(self):
local_dir = self.useFixture(fixtures.TempDir())
etc_dir_path = os.path.join(local_dir.path, 'etc')
os.mkdir(etc_dir_path)
init_cmd = init.TempestInit(None, None)
local_sample_conf_file = os.path.join(etc_dir_path,
'tempest.conf.sample')
# Verify no sample config file exist
self.assertFalse(os.path.isfile(local_sample_conf_file))
init_cmd.generate_sample_config(local_dir.path)
# Verify sample config file exist with some content
self.assertTrue(os.path.isfile(local_sample_conf_file))
self.assertGreater(os.path.getsize(local_sample_conf_file), 0)
def test_update_local_conf(self):
local_dir = self.useFixture(fixtures.TempDir())
etc_dir_path = os.path.join(local_dir.path, 'etc')
os.mkdir(etc_dir_path)
lock_dir = os.path.join(local_dir.path, 'tempest_lock')
config_path = os.path.join(etc_dir_path, 'tempest.conf')
log_dir = os.path.join(local_dir.path, 'logs')
init_cmd = init.TempestInit(None, None)
# Generate the config file
init_cmd.generate_sample_config(local_dir.path)
# Create a conf file with populated values
config_parser_pre = init_cmd.get_configparser(config_path)
with open(config_path, 'w+') as conf_file:
# create the same section init will check for and add values to
config_parser_pre.add_section('oslo_concurrency')
config_parser_pre.set('oslo_concurrency', 'TEST', local_dir.path)
# create a new section
config_parser_pre.add_section('TEST')
config_parser_pre.set('TEST', 'foo', "bar")
config_parser_pre.write(conf_file)
# Update the config file the same way tempest init does
init_cmd.update_local_conf(config_path, lock_dir, log_dir)
# parse the new config file to verify it
config_parser_post = init_cmd.get_configparser(config_path)
# check that our value in oslo_concurrency wasn't overwritten
self.assertTrue(config_parser_post.has_section('oslo_concurrency'))
self.assertEqual(config_parser_post.get('oslo_concurrency', 'TEST'),
local_dir.path)
# check that the lock directory was set correctly
self.assertEqual(config_parser_post.get('oslo_concurrency',
'lock_path'), lock_dir)
# check that our new section still exists and wasn't modified
self.assertTrue(config_parser_post.has_section('TEST'))
self.assertEqual(config_parser_post.get('TEST', 'foo'), 'bar')
# check that the DEFAULT values are correct
# NOTE(auggy): has_section ignores DEFAULT
self.assertEqual(config_parser_post.get('DEFAULT', 'log_dir'), log_dir)
def test_create_working_dir_with_existing_local_dir_non_empty(self):
fake_local_dir = self.useFixture(fixtures.TempDir())
fake_local_conf_dir = self.useFixture(fixtures.TempDir())
open("%s/foo" % fake_local_dir.path, 'w').close()
_init = init.TempestInit(None, None)
self.assertRaises(OSError,
_init.create_working_dir,
fake_local_dir.path,
fake_local_conf_dir.path)
def test_create_working_dir(self):
fake_local_dir = self.useFixture(fixtures.TempDir())
fake_local_conf_dir = self.useFixture(fixtures.TempDir())
os.rmdir(fake_local_dir.path)
# Create a fake conf file
fake_file = fake_local_conf_dir.join('conf_file.conf')
open(fake_file, 'w').close()
init_cmd = init.TempestInit(None, None)
init_cmd.create_working_dir(fake_local_dir.path,
fake_local_conf_dir.path)
# Assert directories are created
lock_path = os.path.join(fake_local_dir.path, 'tempest_lock')
etc_dir = os.path.join(fake_local_dir.path, 'etc')
log_dir = os.path.join(fake_local_dir.path, 'logs')
stestr_dir = os.path.join(fake_local_dir.path, '.stestr')
self.assertTrue(os.path.isdir(lock_path))
self.assertTrue(os.path.isdir(etc_dir))
self.assertTrue(os.path.isdir(log_dir))
self.assertTrue(os.path.isdir(stestr_dir))
# Assert file creation
fake_file_moved = os.path.join(etc_dir, 'conf_file.conf')
local_conf_file = os.path.join(etc_dir, 'tempest.conf')
local_stestr_conf = os.path.join(fake_local_dir.path, '.stestr.conf')
self.assertTrue(os.path.isfile(fake_file_moved))
self.assertTrue(os.path.isfile(local_conf_file))
self.assertTrue(os.path.isfile(local_stestr_conf))
def test_take_action_fails(self):
class ParsedArgs(object):
workspace_dir = self.useFixture(fixtures.TempDir()).path
workspace_path = os.path.join(workspace_dir, 'workspace.yaml')
name = 'test'
dir_base = self.useFixture(fixtures.TempDir()).path
dir = os.path.join(dir_base, 'foo', 'bar')
config_dir = self.useFixture(fixtures.TempDir()).path
show_global_dir = False
pa = ParsedArgs()
init_cmd = init.TempestInit(None, None)
self.assertRaises(OSError, init_cmd.take_action, pa)
# one more trying should be a same error not "workspace already exists"
self.assertRaises(OSError, init_cmd.take_action, pa)
|
