You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

189 lines
8.0 KiB

7 years ago
7 years ago
7 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
6 years ago
7 years ago
7 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
7 years ago
  1. # -*- coding: utf-8 -*-
  2. #
  3. # Copyright (C) 2017 Marcos Pereira <marcospereira.mpj@gmail.com>
  4. #
  5. # This program is free software: you can redistribute it and/or modify
  6. # it under the terms of the GNU Lesser General Public License as published by
  7. # the Free Software Foundation, either version 3 of the License, or
  8. # (at your option) any later version.
  9. #
  10. # This program is distributed in the hope that it will be useful,
  11. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. # GNU Lesser General Public License for more details.
  14. #
  15. # You should have received a copy of the GNU Lesser General Public License
  16. # along with this program. If not, see <http://www.gnu.org/licenses/>.
  17. from unittest import mock
  18. from httmock import urlmatch, response, HTTMock, all_requests
  19. from keycloak import KeycloakAdmin, KeycloakOpenID
  20. from ..connection import ConnectionManager
  21. try:
  22. import unittest
  23. except ImportError:
  24. import unittest2 as unittest
  25. class TestConnection(unittest.TestCase):
  26. def setUp(self):
  27. self._conn = ConnectionManager(
  28. base_url="http://localhost/",
  29. headers={},
  30. timeout=60)
  31. @all_requests
  32. def response_content_success(self, url, request):
  33. headers = {'content-type': 'application/json'}
  34. content = b'response_ok'
  35. return response(200, content, headers, None, 5, request)
  36. def test_raw_get(self):
  37. with HTTMock(self.response_content_success):
  38. resp = self._conn.raw_get("/known_path")
  39. self.assertEqual(resp.content, b'response_ok')
  40. self.assertEqual(resp.status_code, 200)
  41. def test_raw_post(self):
  42. @urlmatch(path="/known_path", method="post")
  43. def response_post_success(url, request):
  44. headers = {'content-type': 'application/json'}
  45. content = 'response'.encode("utf-8")
  46. return response(201, content, headers, None, 5, request)
  47. with HTTMock(response_post_success):
  48. resp = self._conn.raw_post("/known_path",
  49. {'field': 'value'})
  50. self.assertEqual(resp.content, b'response')
  51. self.assertEqual(resp.status_code, 201)
  52. def test_raw_put(self):
  53. @urlmatch(netloc="localhost", path="/known_path", method="put")
  54. def response_put_success(url, request):
  55. headers = {'content-type': 'application/json'}
  56. content = 'response'.encode("utf-8")
  57. return response(200, content, headers, None, 5, request)
  58. with HTTMock(response_put_success):
  59. resp = self._conn.raw_put("/known_path",
  60. {'field': 'value'})
  61. self.assertEqual(resp.content, b'response')
  62. self.assertEqual(resp.status_code, 200)
  63. def test_raw_get_fail(self):
  64. @urlmatch(netloc="localhost", path="/known_path", method="get")
  65. def response_get_fail(url, request):
  66. headers = {'content-type': 'application/json'}
  67. content = "404 page not found".encode("utf-8")
  68. return response(404, content, headers, None, 5, request)
  69. with HTTMock(response_get_fail):
  70. resp = self._conn.raw_get("/known_path")
  71. self.assertEqual(resp.content, b"404 page not found")
  72. self.assertEqual(resp.status_code, 404)
  73. def test_raw_post_fail(self):
  74. @urlmatch(netloc="localhost", path="/known_path", method="post")
  75. def response_post_fail(url, request):
  76. headers = {'content-type': 'application/json'}
  77. content = str(["Start can't be blank"]).encode("utf-8")
  78. return response(404, content, headers, None, 5, request)
  79. with HTTMock(response_post_fail):
  80. resp = self._conn.raw_post("/known_path",
  81. {'field': 'value'})
  82. self.assertEqual(resp.content, str(["Start can't be blank"]).encode("utf-8"))
  83. self.assertEqual(resp.status_code, 404)
  84. def test_raw_put_fail(self):
  85. @urlmatch(netloc="localhost", path="/known_path", method="put")
  86. def response_put_fail(url, request):
  87. headers = {'content-type': 'application/json'}
  88. content = str(["Start can't be blank"]).encode("utf-8")
  89. return response(404, content, headers, None, 5, request)
  90. with HTTMock(response_put_fail):
  91. resp = self._conn.raw_put("/known_path",
  92. {'field': 'value'})
  93. self.assertEqual(resp.content, str(["Start can't be blank"]).encode("utf-8"))
  94. self.assertEqual(resp.status_code, 404)
  95. def test_add_param_headers(self):
  96. self._conn.add_param_headers("test", "value")
  97. self.assertEqual(self._conn.headers,
  98. {"test": "value"})
  99. def test_del_param_headers(self):
  100. self._conn.add_param_headers("test", "value")
  101. self._conn.del_param_headers("test")
  102. self.assertEqual(self._conn.headers, {})
  103. def test_clean_param_headers(self):
  104. self._conn.add_param_headers("test", "value")
  105. self.assertEqual(self._conn.headers,
  106. {"test": "value"})
  107. self._conn.clean_headers()
  108. self.assertEqual(self._conn.headers, {})
  109. def test_exist_param_headers(self):
  110. self._conn.add_param_headers("test", "value")
  111. self.assertTrue(self._conn.exist_param_headers("test"))
  112. self.assertFalse(self._conn.exist_param_headers("test_no"))
  113. def test_get_param_headers(self):
  114. self._conn.add_param_headers("test", "value")
  115. self.assertTrue(self._conn.exist_param_headers("test"))
  116. self.assertFalse(self._conn.exist_param_headers("test_no"))
  117. def test_get_headers(self):
  118. self._conn.add_param_headers("test", "value")
  119. self.assertEqual(self._conn.headers,
  120. {"test": "value"})
  121. def test_KeycloakAdmin_custom_header(self):
  122. class FakeToken:
  123. @staticmethod
  124. def get(string_val):
  125. return "faketoken"
  126. fake_token = FakeToken()
  127. with mock.patch.object(KeycloakOpenID, "__init__", return_value=None) as mock_keycloak_open_id:
  128. with mock.patch("keycloak.keycloak_openid.KeycloakOpenID.token", return_value=fake_token):
  129. with mock.patch("keycloak.connection.ConnectionManager.__init__", return_value=None) as mock_connection_manager:
  130. server_url = "https://localhost/auth/"
  131. username = "admin"
  132. password = "secret"
  133. realm_name = "master"
  134. headers = {
  135. 'Custom': 'test-custom-header'
  136. }
  137. KeycloakAdmin(server_url=server_url,
  138. username=username,
  139. password=password,
  140. realm_name=realm_name,
  141. verify=False,
  142. custom_headers=headers)
  143. mock_keycloak_open_id.assert_called_with(server_url=server_url,
  144. realm_name=realm_name,
  145. client_id='admin-cli',
  146. client_secret_key=None,
  147. verify=False,
  148. custom_headers=headers)
  149. expected_header = {'Authorization': 'Bearer faketoken',
  150. 'Content-Type': 'application/json',
  151. 'Custom': 'test-custom-header'
  152. }
  153. mock_connection_manager.assert_called_with(base_url=server_url,
  154. headers=expected_header,
  155. timeout=60,
  156. verify=False)