1
0

test_vault.py 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205
  1. """
  2. Test case for the vault SDB module
  3. """
  4. # Import python libs
  5. # Import Salt libs
  6. import salt.sdb.vault as vault
  7. from tests.support.mixins import LoaderModuleMockMixin
  8. from tests.support.mock import MagicMock, call, patch
  9. # Import Salt Testing libs
  10. from tests.support.unit import TestCase
  11. class TestVaultSDB(LoaderModuleMockMixin, TestCase):
  12. """
  13. Test case for the vault SDB module
  14. """
  15. def setup_loader_modules(self):
  16. return {
  17. vault: {
  18. "__opts__": {
  19. "vault": {
  20. "url": "http://127.0.0.1",
  21. "auth": {"token": "test", "method": "token"},
  22. }
  23. }
  24. }
  25. }
  26. def test_set(self):
  27. """
  28. Test salt.sdb.vault.set function
  29. """
  30. version = {"v2": False, "data": None, "metadata": None, "type": None}
  31. mock_version = MagicMock(return_value=version)
  32. mock_vault = MagicMock()
  33. mock_vault.return_value.status_code = 200
  34. with patch.dict(
  35. vault.__utils__, {"vault.make_request": mock_vault}
  36. ), patch.dict(vault.__utils__, {"vault.is_v2": mock_version}):
  37. vault.set_("sdb://myvault/path/to/foo/bar", "super awesome")
  38. self.assertEqual(
  39. mock_vault.call_args_list,
  40. [
  41. call(
  42. "POST",
  43. "v1/sdb://myvault/path/to/foo",
  44. json={"bar": "super awesome"},
  45. )
  46. ],
  47. )
  48. def test_set_v2(self):
  49. """
  50. Test salt.sdb.vault.set function with kv v2 backend
  51. """
  52. version = {
  53. "v2": True,
  54. "data": "path/data/to/foo",
  55. "metadata": "path/metadata/to/foo",
  56. "type": "kv",
  57. }
  58. mock_version = MagicMock(return_value=version)
  59. mock_vault = MagicMock()
  60. mock_vault.return_value.status_code = 200
  61. with patch.dict(
  62. vault.__utils__, {"vault.make_request": mock_vault}
  63. ), patch.dict(vault.__utils__, {"vault.is_v2": mock_version}):
  64. vault.set_("sdb://myvault/path/to/foo/bar", "super awesome")
  65. self.assertEqual(
  66. mock_vault.call_args_list,
  67. [
  68. call(
  69. "POST",
  70. "v1/path/data/to/foo",
  71. json={"data": {"bar": "super awesome"}},
  72. )
  73. ],
  74. )
  75. def test_set_question_mark(self):
  76. """
  77. Test salt.sdb.vault.set_ while using the old
  78. deprecated solution with a question mark.
  79. """
  80. version = {"v2": False, "data": None, "metadata": None, "type": None}
  81. mock_version = MagicMock(return_value=version)
  82. mock_vault = MagicMock()
  83. mock_vault.return_value.status_code = 200
  84. with patch.dict(
  85. vault.__utils__, {"vault.make_request": mock_vault}
  86. ), patch.dict(vault.__utils__, {"vault.is_v2": mock_version}):
  87. vault.set_("sdb://myvault/path/to/foo?bar", "super awesome")
  88. self.assertEqual(
  89. mock_vault.call_args_list,
  90. [
  91. call(
  92. "POST",
  93. "v1/sdb://myvault/path/to/foo",
  94. json={"bar": "super awesome"},
  95. )
  96. ],
  97. )
  98. def test_get(self):
  99. """
  100. Test salt.sdb.vault.get function
  101. """
  102. version = {"v2": False, "data": None, "metadata": None, "type": None}
  103. mock_version = MagicMock(return_value=version)
  104. mock_vault = MagicMock()
  105. mock_vault.return_value.status_code = 200
  106. mock_vault.return_value.json.return_value = {"data": {"bar": "test"}}
  107. with patch.dict(
  108. vault.__utils__, {"vault.make_request": mock_vault}
  109. ), patch.dict(vault.__utils__, {"vault.is_v2": mock_version}):
  110. self.assertEqual(vault.get("sdb://myvault/path/to/foo/bar"), "test")
  111. self.assertEqual(
  112. mock_vault.call_args_list, [call("GET", "v1/sdb://myvault/path/to/foo")],
  113. )
  114. def test_get_v2(self):
  115. """
  116. Test salt.sdb.vault.get function with kv v2 backend
  117. """
  118. version = {
  119. "v2": True,
  120. "data": "path/data/to/foo",
  121. "metadata": "path/metadata/to/foo",
  122. "type": "kv",
  123. }
  124. mock_version = MagicMock(return_value=version)
  125. mock_vault = MagicMock()
  126. mock_vault.return_value.status_code = 200
  127. mock_vault.return_value.json.return_value = {"data": {"data": {"bar": "test"}}}
  128. with patch.dict(
  129. vault.__utils__, {"vault.make_request": mock_vault}
  130. ), patch.dict(vault.__utils__, {"vault.is_v2": mock_version}):
  131. self.assertEqual(vault.get("sdb://myvault/path/to/foo/bar"), "test")
  132. self.assertEqual(
  133. mock_vault.call_args_list, [call("GET", "v1/path/data/to/foo")]
  134. )
  135. def test_get_question_mark(self):
  136. """
  137. Test salt.sdb.vault.get while using the old
  138. deprecated solution with a question mark.
  139. """
  140. version = {"v2": False, "data": None, "metadata": None, "type": None}
  141. mock_version = MagicMock(return_value=version)
  142. mock_vault = MagicMock()
  143. mock_vault.return_value.status_code = 200
  144. mock_vault.return_value.json.return_value = {"data": {"bar": "test"}}
  145. with patch.dict(
  146. vault.__utils__, {"vault.make_request": mock_vault}
  147. ), patch.dict(vault.__utils__, {"vault.is_v2": mock_version}):
  148. self.assertEqual(vault.get("sdb://myvault/path/to/foo?bar"), "test")
  149. self.assertEqual(
  150. mock_vault.call_args_list, [call("GET", "v1/sdb://myvault/path/to/foo")],
  151. )
  152. def test_get_missing(self):
  153. """
  154. Test salt.sdb.vault.get function returns None
  155. if vault does not have an entry
  156. """
  157. version = {"v2": False, "data": None, "metadata": None, "type": None}
  158. mock_version = MagicMock(return_value=version)
  159. mock_vault = MagicMock()
  160. mock_vault.return_value.status_code = 404
  161. with patch.dict(
  162. vault.__utils__, {"vault.make_request": mock_vault}
  163. ), patch.dict(vault.__utils__, {"vault.is_v2": mock_version}):
  164. self.assertIsNone(vault.get("sdb://myvault/path/to/foo/bar"))
  165. assert mock_vault.call_args_list == [
  166. call("GET", "v1/sdb://myvault/path/to/foo")
  167. ]
  168. def test_get_missing_key(self):
  169. """
  170. Test salt.sdb.vault.get function returns None
  171. if vault does not have the key but does have the entry
  172. """
  173. version = {"v2": False, "data": None, "metadata": None, "type": None}
  174. mock_version = MagicMock(return_value=version)
  175. mock_vault = MagicMock()
  176. mock_vault.return_value.status_code = 200
  177. mock_vault.return_value.json.return_value = {"data": {"bar": "test"}}
  178. with patch.dict(
  179. vault.__utils__, {"vault.make_request": mock_vault}
  180. ), patch.dict(vault.__utils__, {"vault.is_v2": mock_version}):
  181. self.assertIsNone(vault.get("sdb://myvault/path/to/foo/foo"))
  182. assert mock_vault.call_args_list == [
  183. call("GET", "v1/sdb://myvault/path/to/foo")
  184. ]