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
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
|
#
# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
#
from ipatests.test_xmlrpc import objectclasses
from ipatests.test_xmlrpc.xmlrpc_test import fuzzy_uuid
from ipatests.test_xmlrpc.tracker.base import Tracker
from ipatests.util import assert_deepequal
from ipalib import api
from ipapython.dn import DN
class HostGroupTracker(Tracker):
""" Class for tracking hostgroups """
retrieve_keys = {u'dn', u'cn', u'member_host', u'description',
u'member_hostgroup', u'memberindirect_host'}
retrieve_all_keys = retrieve_keys | {u'ipauniqueid', u'objectclass',
u'mepmanagedentry'}
create_keys = retrieve_all_keys
update_keys = retrieve_keys - {u'dn'}
add_member_keys = retrieve_keys | {u'member_host'}
find_keys = {
u'dn', u'cn', u'description',
}
find_all_keys = {
u'dn', u'cn', u'member_host', u'description', u'member_hostgroup',
u'memberindirect_host', u'ipauniqueid', u'objectclass',
u'mepmanagedentry',
}
def __init__(self, name, description=u'HostGroup desc'):
super(HostGroupTracker, self).__init__(default_version=None)
self.cn = name
self.description = description
self.dn = DN(('cn', self.cn), ('cn', 'hostgroups'),
('cn', 'accounts'), api.env.basedn)
def make_create_command(self,
force=True, *args, **kwargs):
""" Make function that creates a hostgroup using 'hostgroup-add' """
return self.make_command('hostgroup_add', self.cn,
description=self.description,
*args, **kwargs)
def make_delete_command(self):
""" Make function that deletes a hostgroup using 'hostgroup-del' """
return self.make_command('hostgroup_del', self.cn)
def make_retrieve_command(self, all=False, raw=False):
""" Make function that retrieves a hostgroup using 'hostgroup-show' """
return self.make_command('hostgroup_show', self.cn, all=all)
def make_find_command(self, *args, **kwargs):
""" Make function that searches for a hostgroup
using 'hostgroup-find' """
return self.make_command('hostgroup_find', *args, **kwargs)
def make_update_command(self, updates):
""" Make function that updates a hostgroup using 'hostgroup-mod' """
return self.make_command('hostgroup_mod', self.cn, **updates)
def make_add_member_command(self, options={}):
""" Make function that adds a member to a hostgroup """
return self.make_command('hostgroup_add_member', self.cn, **options)
def make_remove_member_command(self, options={}):
""" Make function that removes a member from a hostgroup """
return self.make_command('hostgroup_remove_member', self.cn, **options)
def track_create(self):
""" Updates expected state for hostgroup creation"""
self.attrs = dict(
dn=self.dn,
mepmanagedentry=[DN(('cn', self.cn), ('cn', 'ng'),
('cn', 'alt'), api.env.basedn)],
cn=[self.cn],
description=[self.description],
ipauniqueid=[fuzzy_uuid],
objectclass=objectclasses.hostgroup,
)
self.exists = True
def add_member(self, options):
""" Add a member host to hostgroup and perform check """
if u'host' in options:
try:
self.attrs[u'member_host'] =\
self.attrs[u'member_host'] + [options[u'host']]
except KeyError:
self.attrs[u'member_host'] = [options[u'host']]
# search for hosts in the target hostgroup and
# add them as memberindirect hosts
elif u'hostgroup' in options:
try:
self.attrs[u'member_hostgroup'] =\
self.attrs[u'member_hostgroup'] + [options[u'hostgroup']]
except KeyError:
self.attrs[u'member_hostgroup'] = [options[u'hostgroup']]
command = self.make_add_member_command(options)
result = command()
self.check_add_member(result)
def remove_member(self, options):
""" Remove a member host from hostgroup and perform check """
if u'host' in options:
self.attrs[u'member_host'].remove(options[u'host'])
elif u'hostgroup' in options:
self.attrs[u'member_hostgroup'].remove(options[u'hostgroup'])
try:
if not self.attrs[u'member_host']:
del self.attrs[u'member_host']
except KeyError:
pass
try:
if not self.attrs[u'member_hostgroup']:
del self.attrs[u'member_hostgroup']
except KeyError:
pass
command = self.make_remove_member_command(options)
result = command()
self.check_remove_member(result)
def update(self, updates, expected_updates=None):
"""Helper function to update this user and check the result
Overriding Tracker method for setting self.attrs correctly;
* most attributes stores its value in list
* the rest can be overridden by expected_updates
* allow deleting parametrs if update value is None
"""
if expected_updates is None:
expected_updates = {}
self.ensure_exists()
command = self.make_update_command(updates)
result = command()
for key, value in updates.items():
if value is None:
del self.attrs[key]
else:
self.attrs[key] = [value]
for key, value in expected_updates.items():
if value is None:
del self.attrs[key]
else:
self.attrs[key] = value
self.check_update(
result,
extra_keys=set(updates.keys()) | set(expected_updates.keys())
)
def check_create(self, result):
""" Checks 'hostgroup_add' command result """
assert_deepequal(dict(
value=self.cn,
summary=u'Added hostgroup "%s"' % self.cn,
result=self.filter_attrs(self.create_keys)
), result)
def check_delete(self, result):
""" Checks 'hostgroup_del' command result """
assert_deepequal(dict(
value=[self.cn],
summary=u'Deleted hostgroup "%s"' % self.cn,
result=dict(failed=[]),
), result)
def check_retrieve(self, result, all=False, raw=False):
""" Checks 'hostgroup_show' command result """
if all:
expected = self.filter_attrs(self.retrieve_all_keys)
else:
expected = self.filter_attrs(self.retrieve_keys)
assert_deepequal(dict(
value=self.cn,
summary=None,
result=expected
), result)
def check_find(self, result, all=False, raw=False):
""" Checks 'hostgroup_find' command result """
if all:
expected = self.filter_attrs(self.find_all_keys)
else:
expected = self.filter_attrs(self.find_keys)
assert_deepequal(dict(
count=1,
truncated=False,
summary=u'1 hostgroup matched',
result=[expected],
), result)
def check_update(self, result, extra_keys={}):
""" Checks 'hostgroup_mod' command result """
assert_deepequal(dict(
value=self.cn,
summary=u'Modified hostgroup "%s"' % self.cn,
result=self.filter_attrs(self.update_keys | set(extra_keys))
), result)
def check_add_member(self, result):
""" Checks 'hostgroup_add_member' command result """
assert_deepequal(dict(
completed=1,
failed={u'member': {u'host': (), u'hostgroup': ()}},
result=self.filter_attrs(self.add_member_keys)
), result)
def check_add_member_negative(self, result, options):
""" Checks 'hostgroup_add_member' command result
when expected result is failure of the operation"""
expected = dict(
completed=0,
failed={u'member': {u'hostgroup': (), u'user': ()}},
result=self.filter_attrs(self.add_member_keys)
)
if u'host' in options:
expected[u'failed'][u'member'][u'host'] = [(
options[u'host'], u'no such entry')]
elif u'hostgroup' in options:
expected[u'failed'][u'member'][u'hostgroup'] = [(
options[u'hostgroup'], u'no such entry')]
assert_deepequal(expected, result)
def check_remove_member_negative(self, result, options):
""" Checks 'hostgroup_remove_member' command result
when expected result is failure of the operation"""
expected = dict(
completed=0,
failed={u'member': {u'hostgroup': (), u'host': ()}},
result=self.filter_attrs(self.add_member_keys)
)
if u'user' in options:
expected[u'failed'][u'member'][u'host'] = [(
options[u'user'], u'This entry is not a member')]
elif u'hostgroup' in options:
expected[u'failed'][u'member'][u'hostgroup'] = [(
options[u'hostgroup'], u'This entry is not a member')]
assert_deepequal(expected, result)
def check_remove_member(self, result):
""" Checks 'hostgroup_remove_member' command result """
self.check_add_member(result)
|
