forked from openstack-charmers/zaza-openstack-tests
-
Notifications
You must be signed in to change notification settings - Fork 0
/
tests.py
227 lines (194 loc) · 8.8 KB
/
tests.py
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
219
220
221
222
223
224
225
226
227
# Copyright 2020 Canonical Ltd.
#
# 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.
"""Ceph-mon Testing for cinder-ceph."""
import logging
import os
import zaza.model
from zaza.openstack.utilities import (
generic as generic_utils,
openstack as openstack_utils,
exceptions as zaza_exceptions
)
import zaza.openstack.charm_tests.test_utils as test_utils
class CinderCephMonTest(test_utils.BaseCharmTest):
"""Verify that the ceph mon units are healthy."""
@classmethod
def setUpClass(cls):
"""Run class setup for running ceph mon tests with cinder."""
super().setUpClass()
# ported from the cinder-ceph Amulet test
def test_499_ceph_cmds_exit_zero(self):
"""Verify expected state with security-checklist."""
logging.info("Checking exit values are 0 on ceph commands.")
units = zaza.model.get_units("ceph-mon", model_name=self.model_name)
current_release = openstack_utils.get_os_release(
application='ceph-mon')
bionic_train = openstack_utils.get_os_release('bionic_train')
if current_release < bionic_train:
units.extend(zaza.model.get_units("cinder-ceph",
model_name=self.model_name))
commands = [
'sudo ceph health',
'sudo ceph mds stat',
'sudo ceph pg stat',
'sudo ceph osd stat',
'sudo ceph mon stat',
]
for unit in units:
run_commands(unit.name, commands)
# ported from the cinder-ceph Amulet test
def test_500_ceph_alternatives_cleanup(self):
"""Check ceph alternatives removed when ceph-mon relation is broken."""
# Skip this test if release is less than xenial_ocata as in that case
# cinder HAS a relation with ceph directly and this test would fail
current_release = openstack_utils.get_os_release(
application='ceph-mon')
xenial_ocata = openstack_utils.get_os_release('xenial_ocata')
if current_release < xenial_ocata:
logging.info("Skipping test as release < xenial-ocata")
return
units = zaza.model.get_units("cinder-ceph",
model_name=self.model_name)
# check each unit prior to breaking relation
for unit in units:
dir_list = directory_listing(unit.name, "/etc/ceph")
if 'ceph.conf' in dir_list:
logging.debug(
"/etc/ceph/ceph.conf exists BEFORE relation-broken")
else:
raise zaza_exceptions.CephGenericError(
"unit: {} - /etc/ceph/ceph.conf does not exist "
"BEFORE relation-broken".format(unit.name))
# remove the relation so that /etc/ceph/ceph.conf is removed
logging.info("Removing ceph-mon:client <-> cinder-ceph:ceph relation")
zaza.model.remove_relation(
"ceph-mon", "ceph-mon:client", "cinder-ceph:ceph")
# zaza.model.wait_for_agent_status()
logging.info("Wait till relation is removed...")
ceph_mon_units = zaza.model.get_units("ceph-mon",
model_name=self.model_name)
conditions = [
invert_condition(
does_relation_exist(
u.name, "ceph-mon", "cinder-ceph", "ceph",
self.model_name))
for u in ceph_mon_units]
zaza.model.block_until(*conditions)
logging.info("Checking each unit after breaking relation...")
for unit in units:
dir_list = directory_listing(unit.name, "/etc/ceph")
if 'ceph.conf' not in dir_list:
logging.debug(
"/etc/ceph/ceph.conf removed AFTER relation-broken")
else:
raise zaza_exceptions.CephGenericError(
"unit: {} - /etc/ceph/ceph.conf still exists "
"AFTER relation-broken".format(unit.name))
# Restore cinder-ceph and ceph-mon relation to keep tests idempotent
logging.info("Restoring ceph-mon:client <-> cinder-ceph:ceph relation")
zaza.model.add_relation(
"ceph-mon", "ceph-mon:client", "cinder-ceph:ceph")
conditions = [
does_relation_exist(
u.name, "ceph-mon", "cinder-ceph", "ceph", self.model_name)
for u in ceph_mon_units]
logging.info("Wait till model is idle ...")
zaza.model.block_until(*conditions)
zaza.model.block_until_all_units_idle()
logging.info("... Done.")
class CephPermissionUpgradeTest(test_utils.OpenStackBaseTest):
"""Verify that the ceph mon units update permissions on upgrade."""
@classmethod
def setUpClass(cls):
"""Run class setup for running ceph mon tests."""
super().setUpClass()
def test_ceph_permission_upgrade(self):
"""Check that the charm updates permissions on charm upgrade."""
# Revoke 'osd blocklist' command
zaza.model.run_on_leader(
'ceph-mon',
'sudo -u ceph ceph auth caps client.cinder-ceph mon '
'"allow r; allow command \"osd blacklist\"" osd "allow rwx"')
charm = 'ceph-mon'
charm_path = os.getcwd() + '/' + charm + '.charm'
logging.debug("Upgrading {} to {}".format(charm, charm_path))
zaza.model.upgrade_charm(charm, path=charm_path)
auth = zaza.model.run_on_leader(
'ceph-mon', 'ceph auth get client.cinder-ceph')['Stdout']
self.assertIn('blocklist', auth)
def does_relation_exist(unit_name,
application_name,
remote_application_name,
remote_interface_name,
model_name):
"""For use in async blocking function, return True if it exists.
:param unit_name: the unit (by name) that to check on.
:type unit_name: str
:param application_name: Name of application on this side of relation
:type application_name: str
:param remote_application_name: the relation name at that unit to check for
:type relation_application_name: str
:param remote_interface_name: the interface name at that unit to check for
:type relation_interface_name: str
:param model_name: the model to check on
:type model_name: str
:returns: Corouting that returns True if the relation was found
:rtype: Coroutine[[], boolean]
"""
async def _async_does_relation_exist_closure():
async with zaza.model.run_in_model(model_name) as model:
spec = "{}:{}".format(
remote_application_name, remote_interface_name)
for rel in model.applications[application_name].relations:
if rel.matches(spec):
return True
return False
return _async_does_relation_exist_closure
def invert_condition(async_condition):
"""Invert the condition provided so it can be provided to the blocking fn.
:param async_condition: the async callable that is the test
:type async_condition: Callable[]
:returns: Corouting that returns not of the result of a the callable
:rtype: Coroutine[[], bool]
"""
async def _async_invert_condition_closure():
return not (await async_condition())
return _async_invert_condition_closure
def run_commands(unit_name, commands):
"""Run commands on unit.
Apply context to commands until all variables have been replaced, then
run the command on the given unit.
"""
errors = []
for cmd in commands:
try:
generic_utils.assertRemoteRunOK(zaza.model.run_on_unit(
unit_name,
cmd))
except Exception as e:
errors.append("unit: {}, command: {}, error: {}"
.format(unit_name, cmd, str(e)))
if errors:
raise zaza_exceptions.CephGenericError("\n".join(errors))
def directory_listing(unit_name, directory):
"""Return a list of files/directories from a directory on a unit.
:param unit_name: the unit to fetch the directory listing from
:type unit_name: str
:param directory: the directory to fetch the listing from
:type directory: str
:returns: A listing using "ls -1" on the unit
:rtype: List[str]
"""
result = zaza.model.run_on_unit(unit_name, "ls -1 {}".format(directory))
return result['Stdout'].splitlines()