1from __future__ import (absolute_import, division, print_function) 2 3from unittest import TestCase 4from unittest.mock import patch, MagicMock 5from ansible_collections.community.grafana.plugins.modules import grafana_team 6from ansible.module_utils._text import to_bytes 7from ansible.module_utils import basic 8from ansible.module_utils.urls import basic_auth_header 9import json 10 11__metaclass__ = type 12 13 14class MockedReponse(object): 15 def __init__(self, data): 16 self.data = data 17 18 def read(self): 19 return self.data 20 21 22def exit_json(*args, **kwargs): 23 """function to patch over exit_json; package return data into an exception""" 24 if 'changed' not in kwargs: 25 kwargs['changed'] = False 26 raise AnsibleExitJson(kwargs) 27 28 29def fail_json(*args, **kwargs): 30 """function to patch over fail_json; package return data into an exception""" 31 kwargs['failed'] = True 32 raise AnsibleFailJson(kwargs) 33 34 35class AnsibleExitJson(Exception): 36 """Exception class to be raised by module.exit_json and caught by the test case""" 37 pass 38 39 40class AnsibleFailJson(Exception): 41 """Exception class to be raised by module.fail_json and caught by the test case""" 42 pass 43 44 45def set_module_args(args): 46 """prepare arguments so that they will be picked up during module creation""" 47 args = json.dumps({'ANSIBLE_MODULE_ARGS': args}) 48 basic._ANSIBLE_ARGS = to_bytes(args) 49 50 51def unauthorized_resp(): 52 return (None, {"status": 401}) 53 54 55def permission_denied_resp(): 56 return (None, {"status": 403}) 57 58 59def get_version_resp(): 60 return {"major": 6, "minor": 0, "rev": 0} 61 62 63def get_low_version_resp(): 64 return {"major": 4, "minor": 6, "rev": 0} 65 66 67def team_exists_resp(): 68 server_response = json.dumps({"totalCount": 1, "teams": [{"name": "MyTestTeam", "email": "email@test.com"}]}, sort_keys=True) 69 return (MockedReponse(server_response), {"status": 200}) 70 71 72def team_not_found_resp(): 73 server_response = json.dumps({"totalCount": 0, "teams": []}) 74 return (MockedReponse(server_response), {"status": 200}) 75 76 77def team_created_resp(): 78 server_response = json.dumps({"message": "Team created", "teamId": 2}) 79 return (MockedReponse(server_response), {"status": 200}) 80 81 82def team_updated_resp(): 83 server_response = json.dumps({"message": "Team updated"}) 84 return (MockedReponse(server_response), {"status": 200}) 85 86 87def team_deleted_resp(): 88 server_response = json.dumps({"message": "Team deleted"}) 89 return (MockedReponse(server_response), {"status": 200}) 90 91 92def team_members_resp(): 93 server_response = json.dumps([{ 94 "orgId": 1, 95 "teamId": 2, 96 "userId": 3, 97 "email": "user1@email.com", 98 "login": "user1", 99 "avatarUrl": r"\/avatar\/1b3c32f6386b0185c40d359cdc733a79" 100 }, { 101 "orgId": 1, 102 "teamId": 2, 103 "userId": 2, 104 "email": "user2@email.com", 105 "login": "user2", 106 "avatarUrl": r"\/avatar\/cad3c68da76e45d10269e8ef02f8e73e" 107 }]) 108 return (MockedReponse(server_response), {"status": 200}) 109 110 111def team_members_no_members_resp(): 112 server_response = json.dumps([]) 113 return (MockedReponse(server_response), {"status": 200}) 114 115 116def add_team_member_resp(): 117 server_response = json.dumps({"message": "Member added to Team"}) 118 return (MockedReponse(server_response), {"status": 200}) 119 120 121def delete_team_member_resp(): 122 server_response = json.dumps({"message": "Team Member removed"}) 123 return (MockedReponse(server_response), {"status": 200}) 124 125 126class GrafanaTeamsTest(TestCase): 127 128 def setUp(self): 129 self.authorization = basic_auth_header("admin", "admin") 130 self.mock_module_helper = patch.multiple(basic.AnsibleModule, 131 exit_json=exit_json, 132 fail_json=fail_json) 133 134 self.mock_module_helper.start() 135 self.addCleanup(self.mock_module_helper.stop) 136 137 def test_module_setup_fails_without_params(self): 138 set_module_args({}) 139 140 with self.assertRaises(AnsibleFailJson) as result: 141 grafana_team.main() 142 err, arg_list = result.exception.args[0]['msg'].split(':') 143 missing_args = [item.strip() for item in arg_list.split(',')] 144 self.assertEqual(err, 'missing required arguments') 145 self.assertEqual(arg_list, ["name", "email", "url"]) 146 147 def test_module_setup_fails_without_name(self): 148 set_module_args({ 149 'email': 'email@test.com', 150 'url': 'http://grafana.example.com' 151 }) 152 153 with self.assertRaises(AnsibleFailJson) as result: 154 grafana_team.main() 155 self.assertEqual(result.exception.args[0]['msg'], 'missing required arguments: name') 156 157 def test_module_setup_fails_without_email(self): 158 set_module_args({ 159 'name': 'MyTestTeam', 160 'url': 'http://grafana.example.com' 161 }) 162 163 with self.assertRaises(AnsibleFailJson) as result: 164 grafana_team.main() 165 self.assertEqual(result.exception.args[0]['msg'], 'missing required arguments: email') 166 167 def test_module_setup_fails_without_url(self): 168 set_module_args({ 169 'name': 'MyTestTeam', 170 'email': 'email@test.com', 171 }) 172 173 with self.assertRaises(AnsibleFailJson) as result: 174 grafana_team.main() 175 self.assertEqual(result.exception.args[0]['msg'], 'missing required arguments: url') 176 177 def test_module_setup_fails_with_mutually_exclusive_auth_methods(self): 178 set_module_args({ 179 'name': 'MyTestTeam', 180 'email': 'email@test.com', 181 'url': 'http://grafana.example.com', 182 'grafana_user': 'admin', 183 'grafana_api_key': 'random_api_key', 184 }) 185 186 with self.assertRaises(AnsibleFailJson) as result: 187 grafana_team.main() 188 self.assertEqual(result.exception.args[0]['msg'], 'parameters are mutually exclusive: url_username|grafana_api_key') 189 190 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 191 def test_module_fails_with_low_grafana_version(self, mock_get_version): 192 set_module_args({ 193 'name': 'MyTestTeam', 194 'email': 'email@test.com', 195 'url': 'http://grafana.example.com', 196 'grafana_user': 'admin', 197 'grafana_password': 'admin', 198 }) 199 200 module = grafana_team.setup_module_object() 201 mock_get_version.return_value = get_low_version_resp() 202 203 with self.assertRaises(AnsibleFailJson) as result: 204 grafana_team.main() 205 self.assertEqual(result.exception.args[0]['msg'], 'Teams API is available starting Grafana v5') 206 207 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 208 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 209 def test_module_failure_with_unauthorized_resp(self, mock_fetch_url, mock_get_version): 210 set_module_args({ 211 'name': 'MyTestTeam', 212 'email': 'email@test.com', 213 'url': 'http://grafana.example.com', 214 }) 215 module = grafana_team.setup_module_object() 216 mock_fetch_url.return_value = unauthorized_resp() 217 mock_get_version.return_value = get_version_resp() 218 219 with self.assertRaises(AnsibleFailJson) as result: 220 grafana_team.main() 221 self.assertTrue(result.exception.args[0]['msg'].startswith('Unauthorized to perform action')) 222 223 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 224 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 225 def test_module_failure_with_permission_denied_resp(self, mock_fetch_url, mock_get_version): 226 set_module_args({ 227 'name': 'MyTestTeam', 228 'email': 'email@test.com', 229 'url': 'http://grafana.example.com', 230 }) 231 module = grafana_team.setup_module_object() 232 mock_fetch_url.return_value = permission_denied_resp() 233 mock_get_version.return_value = get_version_resp() 234 235 with self.assertRaises(AnsibleFailJson) as result: 236 grafana_team.main() 237 self.assertTrue(result.exception.args[0]['msg'].startswith('Permission Denied')) 238 239 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 240 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 241 def test_get_team_method_with_existing_team(self, mock_fetch_url, mock_get_version): 242 set_module_args({ 243 'state': 'present', 244 'name': 'MyTestTeam', 245 'email': 'email@test.com', 246 'url': 'http://grafana.example.com' 247 }) 248 module = grafana_team.setup_module_object() 249 mock_fetch_url.return_value = team_exists_resp() 250 mock_get_version.return_value = get_version_resp() 251 252 grafana_iface = grafana_team.GrafanaTeamInterface(module) 253 res = grafana_iface.get_team("MyTestTeam") 254 mock_fetch_url.assert_called_once_with( 255 module, 'http://grafana.example.com/api/teams/search?name=MyTestTeam', 256 data=None, 257 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 258 method='GET') 259 self.assertEquals(res, {"email": "email@test.com", "name": "MyTestTeam"}) 260 261 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 262 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 263 def test_get_team_method_with_non_existing_team(self, mock_fetch_url, mock_get_version): 264 set_module_args({ 265 'state': 'present', 266 'name': 'MyTestTeam', 267 'email': 'email@test.com', 268 'url': 'http://grafana.example.com' 269 }) 270 module = grafana_team.setup_module_object() 271 mock_fetch_url.return_value = team_not_found_resp() 272 mock_get_version.return_value = get_version_resp() 273 274 grafana_iface = grafana_team.GrafanaTeamInterface(module) 275 res = grafana_iface.get_team("MyTestTeam") 276 mock_fetch_url.assert_called_once_with( 277 module, 'http://grafana.example.com/api/teams/search?name=MyTestTeam', 278 data=None, 279 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 280 method='GET') 281 self.assertEquals(res, None) 282 283 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 284 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 285 def test_create_team_method(self, mock_fetch_url, mock_get_version): 286 set_module_args({ 287 'state': 'present', 288 'name': 'MyTestTeam', 289 'email': 'email@test.com', 290 'url': 'http://grafana.example.com' 291 }) 292 module = grafana_team.setup_module_object() 293 mock_fetch_url.return_value = team_created_resp() 294 mock_get_version.return_value = get_version_resp() 295 296 grafana_iface = grafana_team.GrafanaTeamInterface(module) 297 298 res = grafana_iface.create_team("MyTestTeam", "email@test.com") 299 mock_fetch_url.assert_called_once_with( 300 module, 'http://grafana.example.com/api/teams', 301 data=json.dumps({"email": "email@test.com", "name": "MyTestTeam"}, sort_keys=True), 302 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 303 method='POST') 304 self.assertEquals(res, {"message": "Team created", "teamId": 2}) 305 306 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 307 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 308 def test_update_team_method(self, mock_fetch_url, mock_get_version): 309 set_module_args({ 310 'state': 'present', 311 'name': 'MyTestTeam', 312 'email': 'email@test.com', 313 'url': 'http://grafana.example.com' 314 }) 315 module = grafana_team.setup_module_object() 316 mock_fetch_url.return_value = team_updated_resp() 317 mock_get_version.return_value = get_version_resp() 318 319 grafana_iface = grafana_team.GrafanaTeamInterface(module) 320 res = grafana_iface.update_team(2, "MyTestTeam", "email@test.com") 321 mock_fetch_url.assert_called_once_with( 322 module, 'http://grafana.example.com/api/teams/2', 323 data=json.dumps({"email": "email@test.com", "name": "MyTestTeam"}, sort_keys=True), 324 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 325 method='PUT') 326 self.assertEquals(res, {"message": "Team updated"}) 327 328 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 329 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 330 def test_delete_team_method(self, mock_fetch_url, mock_get_version): 331 set_module_args({ 332 'state': 'absent', 333 'name': 'MyTestTeam', 334 'email': 'email@test.com', 335 'url': 'http://grafana.example.com' 336 }) 337 module = grafana_team.setup_module_object() 338 mock_fetch_url.return_value = team_deleted_resp() 339 mock_get_version.return_value = get_version_resp() 340 341 grafana_iface = grafana_team.GrafanaTeamInterface(module) 342 res = grafana_iface.delete_team(2) 343 mock_fetch_url.assert_called_once_with( 344 module, 'http://grafana.example.com/api/teams/2', 345 data=None, 346 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 347 method='DELETE') 348 self.assertEquals(res, {"message": "Team deleted"}) 349 350 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 351 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 352 def test_get_team_members_method(self, mock_fetch_url, mock_get_version): 353 set_module_args({ 354 'state': 'present', 355 'name': 'MyTestTeam', 356 'email': 'email@test.com', 357 'url': 'http://grafana.example.com' 358 }) 359 module = grafana_team.setup_module_object() 360 mock_fetch_url.return_value = team_members_resp() 361 mock_get_version.return_value = get_version_resp() 362 363 grafana_iface = grafana_team.GrafanaTeamInterface(module) 364 res = grafana_iface.get_team_members(2) 365 mock_fetch_url.assert_called_once_with( 366 module, 'http://grafana.example.com/api/teams/2/members', 367 data=None, 368 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 369 method='GET') 370 self.assertEquals(res, ["user1@email.com", "user2@email.com"]) 371 372 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 373 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 374 def test_get_team_members_method_no_members_returned(self, mock_fetch_url, mock_get_version): 375 set_module_args({ 376 'state': 'present', 377 'name': 'MyTestTeam', 378 'email': 'email@test.com', 379 'url': 'http://grafana.example.com' 380 }) 381 module = grafana_team.setup_module_object() 382 mock_fetch_url.return_value = team_members_no_members_resp() 383 mock_get_version.return_value = get_version_resp() 384 385 grafana_iface = grafana_team.GrafanaTeamInterface(module) 386 res = grafana_iface.get_team_members(2) 387 mock_fetch_url.assert_called_once_with( 388 module, 'http://grafana.example.com/api/teams/2/members', 389 data=None, 390 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 391 method='GET') 392 self.assertEquals(res, []) 393 394 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 395 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 396 def test_add_team_member_method(self, mock_fetch_url, mock_get_version): 397 set_module_args({ 398 'state': 'present', 399 'name': 'MyTestTeam', 400 'email': 'email@test.com', 401 'url': 'http://grafana.example.com' 402 }) 403 module = grafana_team.setup_module_object() 404 mock_fetch_url.return_value = add_team_member_resp() 405 mock_get_version.return_value = get_version_resp() 406 407 grafana_iface = grafana_team.GrafanaTeamInterface(module) 408 with patch.object(grafana_team.GrafanaTeamInterface, 'get_user_id_from_mail') as mock_get_user_id_from_mail: 409 mock_get_user_id_from_mail.return_value = 42 410 res = grafana_iface.add_team_member(2, "another@test.com") 411 mock_fetch_url.assert_called_once_with( 412 module, 'http://grafana.example.com/api/teams/2/members', 413 data=json.dumps({'userId': 42}), 414 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 415 method='POST') 416 self.assertEquals(res, None) 417 418 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.GrafanaTeamInterface.get_version') 419 @patch('ansible_collections.community.grafana.plugins.modules.grafana_team.fetch_url') 420 def test_delete_team_member_method(self, mock_fetch_url, mock_get_version): 421 set_module_args({ 422 'state': 'present', 423 'name': 'MyTestTeam', 424 'email': 'email@test.com', 425 'url': 'http://grafana.example.com' 426 }) 427 module = grafana_team.setup_module_object() 428 mock_fetch_url.return_value = delete_team_member_resp() 429 mock_get_version.return_value = get_version_resp() 430 431 grafana_iface = grafana_team.GrafanaTeamInterface(module) 432 with patch.object(grafana_team.GrafanaTeamInterface, 'get_user_id_from_mail') as mock_get_user_id_from_mail: 433 mock_get_user_id_from_mail.return_value = 42 434 res = grafana_iface.delete_team_member(2, "another@test.com") 435 mock_fetch_url.assert_called_once_with( 436 module, 'http://grafana.example.com/api/teams/2/members/42', 437 data=None, 438 headers={'Content-Type': 'application/json', 'Authorization': self.authorization}, 439 method='DELETE') 440 self.assertEquals(res, None) 441 442 def test_diff_members_function(self): 443 list1 = ["foo@example.com", "bar@example.com"] 444 list2 = ["bar@example.com", "random@example.com"] 445 446 res = grafana_team.diff_members(list1, list2) 447 self.assertEquals(res, {"to_del": ["random@example.com"], "to_add": ["foo@example.com"]}) 448