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
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
|
# pylint: disable=missing-docstring
import os
import yaml
import ooinstall.cli_installer as cli
from test.oo_config_tests import OOInstallFixture
from click.testing import CliRunner
# Substitute in a product name before use:
SAMPLE_CONFIG = """
variant: %s
ansible_ssh_user: root
hosts:
- connect_to: 10.0.0.1
ip: 10.0.0.1
hostname: master-private.example.com
public_ip: 24.222.0.1
public_hostname: master.example.com
master: true
node: true
- connect_to: 10.0.0.2
ip: 10.0.0.2
hostname: node1-private.example.com
public_ip: 24.222.0.2
public_hostname: node1.example.com
node: true
- connect_to: 10.0.0.3
ip: 10.0.0.3
hostname: node2-private.example.com
public_ip: 24.222.0.3
public_hostname: node2.example.com
node: true
"""
def read_yaml(config_file_path):
cfg_f = open(config_file_path, 'r')
config = yaml.safe_load(cfg_f.read())
cfg_f.close()
return config
class OOCliFixture(OOInstallFixture):
def setUp(self):
OOInstallFixture.setUp(self)
self.runner = CliRunner()
# Add any arguments you would like to test here, the defaults ensure
# we only do unattended invocations here, and using temporary files/dirs.
self.cli_args = ["-a", self.work_dir]
def run_cli(self):
return self.runner.invoke(cli.cli, self.cli_args)
def assert_result(self, result, exit_code):
if result.exception is not None or result.exit_code != exit_code:
print "Unexpected result from CLI execution"
print "Exit code: %s" % result.exit_code
print "Exception: %s" % result.exception
print result.exc_info
import traceback
traceback.print_exception(*result.exc_info)
print "Output:\n%s" % result.output
self.fail("Exception during CLI execution")
def _verify_load_facts(self, load_facts_mock):
""" Check that we ran load facts with expected inputs. """
load_facts_args = load_facts_mock.call_args[0]
self.assertEquals(os.path.join(self.work_dir, ".ansible/hosts"),
load_facts_args[0])
self.assertEquals(os.path.join(self.work_dir,
"playbooks/byo/openshift_facts.yml"),
load_facts_args[1])
env_vars = load_facts_args[2]
self.assertEquals(os.path.join(self.work_dir,
'.ansible/callback_facts.yaml'),
env_vars['OO_INSTALL_CALLBACK_FACTS_YAML'])
self.assertEqual('/tmp/ansible.log', env_vars['ANSIBLE_LOG_PATH'])
def _verify_run_playbook(self, run_playbook_mock, exp_hosts_len, exp_hosts_to_run_on_len):
""" Check that we ran playbook with expected inputs. """
hosts = run_playbook_mock.call_args[0][0]
hosts_to_run_on = run_playbook_mock.call_args[0][1]
self.assertEquals(exp_hosts_len, len(hosts))
self.assertEquals(exp_hosts_to_run_on_len, len(hosts_to_run_on))
def _verify_config_hosts(self, written_config, host_count):
self.assertEquals(host_count, len(written_config['hosts']))
for host in written_config['hosts']:
self.assertTrue('hostname' in host)
self.assertTrue('public_hostname' in host)
if 'preconfigured' not in host:
self.assertTrue(host['node'])
self.assertTrue('ip' in host)
self.assertTrue('public_ip' in host)
#pylint: disable=too-many-arguments
def _verify_get_hosts_to_run_on(self, mock_facts, load_facts_mock,
run_playbook_mock, cli_input,
exp_hosts_len=None, exp_hosts_to_run_on_len=None,
force=None):
"""
Tests cli_installer.py:get_hosts_to_run_on. That method has quite a
few subtle branches in the logic. The goal with this method is simply
to handle all the messy stuff here and allow the main test cases to be
easily read. The basic idea is to modify mock_facts to return a
version indicating OpenShift is already installed on particular hosts.
"""
load_facts_mock.return_value = (mock_facts, 0)
run_playbook_mock.return_value = 0
if cli_input:
self.cli_args.append("install")
result = self.runner.invoke(cli.cli,
self.cli_args,
input=cli_input)
else:
config_file = self.write_config(
os.path.join(self.work_dir,
'ooinstall.conf'), SAMPLE_CONFIG % 'openshift-enterprise')
self.cli_args.extend(["-c", config_file, "install"])
if force:
self.cli_args.append("--force")
result = self.runner.invoke(cli.cli, self.cli_args)
written_config = read_yaml(config_file)
self._verify_config_hosts(written_config, exp_hosts_len)
self.assert_result(result, 0)
self._verify_load_facts(load_facts_mock)
self._verify_run_playbook(run_playbook_mock, exp_hosts_len, exp_hosts_to_run_on_len)
# Make sure we ran on the expected masters and nodes:
hosts = run_playbook_mock.call_args[0][0]
hosts_to_run_on = run_playbook_mock.call_args[0][1]
self.assertEquals(exp_hosts_len, len(hosts))
self.assertEquals(exp_hosts_to_run_on_len, len(hosts_to_run_on))
#pylint: disable=too-many-arguments,too-many-branches
def build_input(ssh_user=None, hosts=None, variant_num=None,
add_nodes=None, confirm_facts=None, scheduleable_masters_ok=None,
master_lb=None):
"""
Build an input string simulating a user entering values in an interactive
attended install.
This is intended to give us one place to update when the CLI prompts change.
We should aim to keep this dependent on optional keyword arguments with
sensible defaults to keep things from getting too fragile.
"""
inputs = [
'y', # let's proceed
]
if ssh_user:
inputs.append(ssh_user)
if variant_num:
inputs.append(str(variant_num)) # Choose variant + version
num_masters = 0
if hosts:
i = 0
min_masters_for_ha = 3
for (host, is_master) in hosts:
inputs.append(host)
if is_master:
inputs.append('y')
num_masters += 1
else:
inputs.append('n')
#inputs.append('rpm')
if i < len(hosts) - 1:
if num_masters <= 1 or num_masters >= min_masters_for_ha:
inputs.append('y') # Add more hosts
else:
inputs.append('n') # Done adding hosts
i += 1
# You can pass a single master_lb or a list if you intend for one to get rejected:
if master_lb:
if isinstance(master_lb[0], list) or isinstance(master_lb[0], tuple):
inputs.extend(master_lb[0])
else:
inputs.append(master_lb[0])
inputs.append('y' if master_lb[1] else 'n')
# TODO: support option 2, fresh install
if add_nodes:
if scheduleable_masters_ok:
inputs.append('y')
inputs.append('1') # Add more nodes
i = 0
for (host, is_master) in add_nodes:
inputs.append(host)
#inputs.append('rpm')
if i < len(add_nodes) - 1:
inputs.append('y') # Add more hosts
else:
inputs.append('n') # Done adding hosts
i += 1
if add_nodes is None:
total_hosts = hosts
else:
total_hosts = hosts + add_nodes
if total_hosts is not None and num_masters == len(total_hosts):
inputs.append('y')
inputs.extend([
confirm_facts,
'y', # lets do this
])
return '\n'.join(inputs)
|
