test_boto_cognitoidentity.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740
  1. # -*- coding: utf-8 -*-
  2. # Import Python libs
  3. from __future__ import absolute_import, print_function, unicode_literals
  4. import logging
  5. import random
  6. import string
  7. # Import Salt libs
  8. import salt.config
  9. import salt.loader
  10. import salt.modules.boto_cognitoidentity as boto_cognitoidentity
  11. from salt.ext.six.moves import range # pylint: disable=import-error
  12. from salt.utils.versions import LooseVersion
  13. # Import Salt Testing libs
  14. from tests.support.mixins import LoaderModuleMockMixin
  15. from tests.support.mock import MagicMock, patch
  16. from tests.support.unit import TestCase, skipIf
  17. # Import 3rd-party libs
  18. # pylint: disable=import-error,no-name-in-module
  19. try:
  20. import boto3
  21. from botocore.exceptions import ClientError
  22. HAS_BOTO = True
  23. except ImportError:
  24. HAS_BOTO = False
  25. # pylint: enable=import-error,no-name-in-module
  26. # the boto_lambda module relies on the connect_to_region() method
  27. # which was added in boto 2.8.0
  28. # https://github.com/boto/boto/commit/33ac26b416fbb48a60602542b4ce15dcc7029f12
  29. required_boto3_version = "1.2.1"
  30. region = "us-east-1"
  31. access_key = "GKTADJGHEIQSXMKKRBJ08H"
  32. secret_key = "askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs"
  33. conn_parameters = {
  34. "region": region,
  35. "key": access_key,
  36. "keyid": secret_key,
  37. "profile": {},
  38. }
  39. error_message = (
  40. "An error occurred (101) when calling the {0} operation: Test-defined error"
  41. )
  42. error_content = {"Error": {"Code": 101, "Message": "Test-defined error"}}
  43. first_pool_id = "first_pool_id"
  44. first_pool_name = "first_pool"
  45. second_pool_id = "second_pool_id"
  46. second_pool_name = "second_pool"
  47. second_pool_name_updated = "second_pool_updated"
  48. third_pool_id = "third_pool_id"
  49. third_pool_name = first_pool_name
  50. identity_pools_ret = dict(
  51. IdentityPools=[
  52. dict(IdentityPoolId=first_pool_id, IdentityPoolName=first_pool_name),
  53. dict(IdentityPoolId=second_pool_id, IdentityPoolName=second_pool_name),
  54. dict(IdentityPoolId=third_pool_id, IdentityPoolName=third_pool_name),
  55. ]
  56. )
  57. first_pool_ret = dict(
  58. IdentityPoolId=first_pool_id,
  59. IdentityPoolName=first_pool_name,
  60. AllowUnauthenticatedIdentities=False,
  61. SupportedLoginProviders={
  62. "accounts.google.com": "testing123",
  63. "api.twitter.com": "testing123",
  64. "graph.facebook.com": "testing123",
  65. "www.amazon.com": "testing123",
  66. },
  67. DeveloperProviderName="test_provider",
  68. OpenIdConnectProviderARNs=["some_provider_arn", "another_provider_arn"],
  69. )
  70. first_pool_role_ret = dict(
  71. IdentityPoolId=first_pool_id,
  72. Roles=dict(
  73. authenticated="first_pool_auth_role", unauthenticated="first_pool_unauth_role"
  74. ),
  75. )
  76. second_pool_ret = dict(
  77. IdentityPoolId=second_pool_id,
  78. IdentityPoolName=second_pool_name,
  79. AllowUnauthenticatedIdentities=False,
  80. )
  81. third_pool_ret = dict(
  82. IdentityPoolId=third_pool_id,
  83. IdentityPoolName=third_pool_name,
  84. AllowUnauthenticatedIdentities=False,
  85. DeveloperProviderName="test_provider2",
  86. )
  87. third_pool_role_ret = dict(IdentityPoolId=third_pool_id)
  88. default_pool_ret = dict(
  89. IdentityPoolId="default_pool_id",
  90. IdentityPoolName="default_pool_name",
  91. AllowUnauthenticatedIdentities=False,
  92. DeveloperProviderName="test_provider_default",
  93. )
  94. default_pool_role_ret = dict(IdentityPoolId="default_pool_id")
  95. log = logging.getLogger(__name__)
  96. def _has_required_boto():
  97. """
  98. Returns True/False boolean depending on if Boto is installed and correct
  99. version.
  100. """
  101. if not HAS_BOTO:
  102. return False
  103. elif LooseVersion(boto3.__version__) < LooseVersion(required_boto3_version):
  104. return False
  105. else:
  106. return True
  107. class BotoCognitoIdentityTestCaseBase(TestCase, LoaderModuleMockMixin):
  108. conn = None
  109. def setup_loader_modules(self):
  110. self.opts = opts = salt.config.DEFAULT_MINION_OPTS.copy()
  111. utils = salt.loader.utils(
  112. opts, whitelist=["boto3", "args", "systemd", "path", "platform"], context={}
  113. )
  114. return {boto_cognitoidentity: {"__utils__": utils}}
  115. def setUp(self):
  116. super(BotoCognitoIdentityTestCaseBase, self).setUp()
  117. boto_cognitoidentity.__init__(self.opts)
  118. del self.opts
  119. # Set up MagicMock to replace the boto3 session
  120. # connections keep getting cached from prior tests, can't find the
  121. # correct context object to clear it. So randomize the cache key, to prevent any
  122. # cache hits
  123. conn_parameters["key"] = "".join(
  124. random.choice(string.ascii_lowercase + string.digits) for _ in range(50)
  125. )
  126. self.patcher = patch("boto3.session.Session")
  127. self.addCleanup(self.patcher.stop)
  128. self.addCleanup(delattr, self, "patcher")
  129. mock_session = self.patcher.start()
  130. session_instance = mock_session.return_value
  131. self.conn = MagicMock()
  132. self.addCleanup(delattr, self, "conn")
  133. session_instance.client.return_value = self.conn
  134. class BotoCognitoIdentityTestCaseMixin(object):
  135. pass
  136. @skipIf(True, "Skip these tests while investigating failures")
  137. @skipIf(HAS_BOTO is False, "The boto module must be installed.")
  138. @skipIf(
  139. _has_required_boto() is False,
  140. "The boto3 module must be greater than"
  141. " or equal to version {0}".format(required_boto3_version),
  142. )
  143. class BotoCognitoIdentityTestCase(
  144. BotoCognitoIdentityTestCaseBase, BotoCognitoIdentityTestCaseMixin
  145. ):
  146. """
  147. TestCase for salt.modules.boto_cognitoidentity module
  148. """
  149. def _describe_identity_pool_side_effect(self, *args, **kwargs):
  150. if kwargs.get("IdentityPoolId") == first_pool_id:
  151. return first_pool_ret
  152. elif kwargs.get("IdentityPoolId") == third_pool_id:
  153. return third_pool_ret
  154. else:
  155. return default_pool_ret
  156. def test_that_when_describing_a_named_identity_pool_and_pool_exists_the_describe_identity_pool_method_returns_pools_properties(
  157. self,
  158. ):
  159. """
  160. Tests describing identity pool when the pool's name exists
  161. """
  162. self.conn.list_identity_pools.return_value = identity_pools_ret
  163. self.conn.describe_identity_pool.side_effect = (
  164. self._describe_identity_pool_side_effect
  165. )
  166. result = boto_cognitoidentity.describe_identity_pools(
  167. IdentityPoolName=first_pool_name, **conn_parameters
  168. )
  169. self.assertEqual(result.get("identity_pools"), [first_pool_ret, third_pool_ret])
  170. def test_that_when_describing_a_identity_pool_by_its_id_and_pool_exists_the_desribe_identity_pool_method_returns_pools_properties(
  171. self,
  172. ):
  173. """
  174. Tests describing identity pool when the given pool's id exists
  175. """
  176. self.conn.describe_identity_pool.return_value = third_pool_ret
  177. result = boto_cognitoidentity.describe_identity_pools(
  178. IdentityPoolName="", IdentityPoolId=third_pool_id, **conn_parameters
  179. )
  180. self.assertEqual(result.get("identity_pools"), [third_pool_ret])
  181. self.assertTrue(self.conn.list_identity_pools.call_count == 0)
  182. def test_that_when_describing_a_named_identity_pool_and_pool_does_not_exist_the_describe_identity_pool_method_returns_none(
  183. self,
  184. ):
  185. """
  186. Tests describing identity pool when the pool's name doesn't exist
  187. """
  188. self.conn.list_identity_pools.return_value = identity_pools_ret
  189. self.conn.describe_identity_pool.return_value = first_pool_ret
  190. result = boto_cognitoidentity.describe_identity_pools(
  191. IdentityPoolName="no_such_pool", **conn_parameters
  192. )
  193. self.assertEqual(result.get("identity_pools", "no such key"), None)
  194. def test_that_when_describing_a_named_identity_pool_and_error_thrown_the_describe_identity_pool_method_returns_error(
  195. self,
  196. ):
  197. """
  198. Tests describing identity pool returns error when there is an exception to boto3 calls
  199. """
  200. self.conn.list_identity_pools.return_value = identity_pools_ret
  201. self.conn.describe_identity_pool.side_effect = ClientError(
  202. error_content, "error on describe identity pool"
  203. )
  204. result = boto_cognitoidentity.describe_identity_pools(
  205. IdentityPoolName=first_pool_name, **conn_parameters
  206. )
  207. self.assertEqual(
  208. result.get("error", {}).get("message"),
  209. error_message.format("error on describe identity pool"),
  210. )
  211. def test_that_when_create_identity_pool_the_create_identity_pool_method_returns_created_identity_pool(
  212. self,
  213. ):
  214. """
  215. Tests the positive case where create identity pool succeeds
  216. """
  217. return_val = default_pool_ret.copy()
  218. return_val.pop("DeveloperProviderName", None)
  219. self.conn.create_identity_pool.return_value = return_val
  220. result = boto_cognitoidentity.create_identity_pool(
  221. IdentityPoolName="default_pool_name", **conn_parameters
  222. )
  223. mock_calls = self.conn.mock_calls
  224. # name, args, kwargs = mock_calls[0]
  225. self.assertTrue(result.get("created"))
  226. self.assertEqual(len(mock_calls), 1)
  227. self.assertEqual(mock_calls[0][0], "create_identity_pool")
  228. self.assertNotIn("DeveloperProviderName", mock_calls[0][2])
  229. def test_that_when_create_identity_pool_and_error_thrown_the_create_identity_pool_method_returns_error(
  230. self,
  231. ):
  232. """
  233. Tests the negative case where create identity pool has a boto3 client error
  234. """
  235. self.conn.create_identity_pool.side_effect = ClientError(
  236. error_content, "create_identity_pool"
  237. )
  238. result = boto_cognitoidentity.create_identity_pool(
  239. IdentityPoolName="default_pool_name", **conn_parameters
  240. )
  241. self.assertIs(result.get("created"), False)
  242. self.assertEqual(
  243. result.get("error", {}).get("message"),
  244. error_message.format("create_identity_pool"),
  245. )
  246. def test_that_when_delete_identity_pools_with_multiple_matching_pool_names_the_delete_identity_pools_methos_returns_true_and_deleted_count(
  247. self,
  248. ):
  249. """
  250. Tests that given 2 matching pool ids, the operation returns deleted status of true and
  251. count 2
  252. """
  253. self.conn.list_identity_pools.return_value = identity_pools_ret
  254. self.conn.delete_identity_pool.return_value = None
  255. result = boto_cognitoidentity.delete_identity_pools(
  256. IdentityPoolName=first_pool_name, **conn_parameters
  257. )
  258. mock_calls = self.conn.mock_calls
  259. self.assertTrue(result.get("deleted"))
  260. self.assertEqual(result.get("count"), 2)
  261. self.assertEqual(len(mock_calls), 3)
  262. self.assertEqual(mock_calls[1][0], "delete_identity_pool")
  263. self.assertEqual(mock_calls[2][0], "delete_identity_pool")
  264. self.assertEqual(mock_calls[1][2].get("IdentityPoolId"), first_pool_id)
  265. self.assertEqual(mock_calls[2][2].get("IdentityPoolId"), third_pool_id)
  266. def test_that_when_delete_identity_pools_with_no_matching_pool_names_the_delete_identity_pools_method_returns_false(
  267. self,
  268. ):
  269. """
  270. Tests that the given pool name does not exist, the operation returns deleted status of false
  271. and count 0
  272. """
  273. self.conn.list_identity_pools.return_value = identity_pools_ret
  274. result = boto_cognitoidentity.delete_identity_pools(
  275. IdentityPoolName="no_such_pool_name", **conn_parameters
  276. )
  277. mock_calls = self.conn.mock_calls
  278. self.assertIs(result.get("deleted"), False)
  279. self.assertEqual(result.get("count"), 0)
  280. self.assertEqual(len(mock_calls), 1)
  281. def test_that_when_delete_identity_pools_and_error_thrown_the_delete_identity_pools_method_returns_false_and_the_error(
  282. self,
  283. ):
  284. """
  285. Tests that the delete_identity_pool method throws an exception
  286. """
  287. self.conn.delete_identity_pool.side_effect = ClientError(
  288. error_content, "delete_identity_pool"
  289. )
  290. # try to delete an unexistent pool id "no_such_pool_id"
  291. result = boto_cognitoidentity.delete_identity_pools(
  292. IdentityPoolName=first_pool_name,
  293. IdentityPoolId="no_such_pool_id",
  294. **conn_parameters
  295. )
  296. mock_calls = self.conn.mock_calls
  297. self.assertIs(result.get("deleted"), False)
  298. self.assertIs(result.get("count"), None)
  299. self.assertEqual(
  300. result.get("error", {}).get("message"),
  301. error_message.format("delete_identity_pool"),
  302. )
  303. self.assertEqual(len(mock_calls), 1)
  304. def _get_identity_pool_roles_side_effect(self, *args, **kwargs):
  305. if kwargs.get("IdentityPoolId") == first_pool_id:
  306. return first_pool_role_ret
  307. elif kwargs.get("IdentityPoolId") == third_pool_id:
  308. return third_pool_role_ret
  309. else:
  310. return default_pool_role_ret
  311. def test_that_when_get_identity_pool_roles_with_matching_pool_names_the_get_identity_pool_roles_method_returns_pools_role_properties(
  312. self,
  313. ):
  314. """
  315. Tests that the given 2 pool id's matching the given pool name, the results are
  316. passed through as a list of identity_pool_roles
  317. """
  318. self.conn.list_identity_pools.return_value = identity_pools_ret
  319. self.conn.get_identity_pool_roles.side_effect = (
  320. self._get_identity_pool_roles_side_effect
  321. )
  322. result = boto_cognitoidentity.get_identity_pool_roles(
  323. IdentityPoolName=first_pool_name, **conn_parameters
  324. )
  325. id_pool_roles = result.get("identity_pool_roles")
  326. self.assertIsNot(id_pool_roles, None)
  327. self.assertEqual(
  328. result.get("error", "key_should_not_be_there"), "key_should_not_be_there"
  329. )
  330. self.assertEqual(len(id_pool_roles), 2)
  331. self.assertEqual(id_pool_roles[0], first_pool_role_ret)
  332. self.assertEqual(id_pool_roles[1], third_pool_role_ret)
  333. def test_that_when_get_identity_pool_roles_with_no_matching_pool_names_the_get_identity_pool_roles_method_returns_none(
  334. self,
  335. ):
  336. """
  337. Tests that the given no pool id's matching the given pool name, the results returned is
  338. None and get_identity_pool_roles should never be called
  339. """
  340. self.conn.list_identity_pools.return_value = identity_pools_ret
  341. result = boto_cognitoidentity.get_identity_pool_roles(
  342. IdentityPoolName="no_such_pool_name", **conn_parameters
  343. )
  344. mock_calls = self.conn.mock_calls
  345. self.assertIs(result.get("identity_pool_roles", "key_should_be_there"), None)
  346. self.assertEqual(len(mock_calls), 1)
  347. self.assertEqual(
  348. result.get("error", "key_should_not_be_there"), "key_should_not_be_there"
  349. )
  350. def test_that_when_get_identity_pool_roles_and_error_thrown_due_to_invalid_pool_id_the_get_identity_pool_roles_method_returns_error(
  351. self,
  352. ):
  353. """
  354. Tests that given an invalid pool id, we properly handle error generated from get_identity_pool_roles
  355. """
  356. self.conn.get_identity_pool_roles.side_effect = ClientError(
  357. error_content, "get_identity_pool_roles"
  358. )
  359. # try to delete an unexistent pool id "no_such_pool_id"
  360. result = boto_cognitoidentity.get_identity_pool_roles(
  361. IdentityPoolName="", IdentityPoolId="no_such_pool_id", **conn_parameters
  362. )
  363. mock_calls = self.conn.mock_calls
  364. self.assertEqual(
  365. result.get("error", {}).get("message"),
  366. error_message.format("get_identity_pool_roles"),
  367. )
  368. self.assertEqual(len(mock_calls), 1)
  369. def test_that_when_set_identity_pool_roles_with_invalid_pool_id_the_set_identity_pool_roles_method_returns_set_false_and_error(
  370. self,
  371. ):
  372. """
  373. Tests that given an invalid pool id, we properly handle error generated from set_identity_pool_roles
  374. """
  375. self.conn.set_identity_pool_roles.side_effect = ClientError(
  376. error_content, "set_identity_pool_roles"
  377. )
  378. result = boto_cognitoidentity.set_identity_pool_roles(
  379. IdentityPoolId="no_such_pool_id", **conn_parameters
  380. )
  381. mock_calls = self.conn.mock_calls
  382. self.assertIs(result.get("set"), False)
  383. self.assertEqual(
  384. result.get("error", {}).get("message"),
  385. error_message.format("set_identity_pool_roles"),
  386. )
  387. self.assertEqual(len(mock_calls), 1)
  388. def test_that_when_set_identity_pool_roles_with_no_roles_specified_the_set_identity_pool_roles_method_unset_the_roles(
  389. self,
  390. ):
  391. """
  392. Tests that given a valid pool id, and no other roles given, the role for the pool is cleared.
  393. """
  394. self.conn.set_identity_pool_roles.return_value = None
  395. result = boto_cognitoidentity.set_identity_pool_roles(
  396. IdentityPoolId="some_id", **conn_parameters
  397. )
  398. mock_calls = self.conn.mock_calls
  399. self.assertTrue(result.get("set"))
  400. self.assertEqual(len(mock_calls), 1)
  401. self.assertEqual(mock_calls[0][2].get("Roles"), {})
  402. def test_that_when_set_identity_pool_roles_with_only_auth_role_specified_the_set_identity_pool_roles_method_only_set_the_auth_role(
  403. self,
  404. ):
  405. """
  406. Tests that given a valid pool id, and only other given role is the AuthenticatedRole, the auth role for the
  407. pool is set and unauth role is cleared.
  408. """
  409. self.conn.set_identity_pool_roles.return_value = None
  410. with patch.dict(
  411. boto_cognitoidentity.__salt__,
  412. {
  413. "boto_iam.describe_role": MagicMock(
  414. return_value={"arn": "my_auth_role_arn"}
  415. )
  416. },
  417. ):
  418. expected_roles = dict(authenticated="my_auth_role_arn")
  419. result = boto_cognitoidentity.set_identity_pool_roles(
  420. IdentityPoolId="some_id",
  421. AuthenticatedRole="my_auth_role",
  422. **conn_parameters
  423. )
  424. mock_calls = self.conn.mock_calls
  425. self.assertTrue(result.get("set"))
  426. self.assertEqual(len(mock_calls), 1)
  427. self.assertEqual(mock_calls[0][2].get("Roles"), expected_roles)
  428. def test_that_when_set_identity_pool_roles_with_only_unauth_role_specified_the_set_identity_pool_roles_method_only_set_the_unauth_role(
  429. self,
  430. ):
  431. """
  432. Tests that given a valid pool id, and only other given role is the UnauthenticatedRole, the unauth role for the
  433. pool is set and the auth role is cleared.
  434. """
  435. self.conn.set_identity_pool_roles.return_value = None
  436. with patch.dict(
  437. boto_cognitoidentity.__salt__,
  438. {
  439. "boto_iam.describe_role": MagicMock(
  440. return_value={"arn": "my_unauth_role_arn"}
  441. )
  442. },
  443. ):
  444. expected_roles = dict(unauthenticated="my_unauth_role_arn")
  445. result = boto_cognitoidentity.set_identity_pool_roles(
  446. IdentityPoolId="some_id",
  447. UnauthenticatedRole="my_unauth_role",
  448. **conn_parameters
  449. )
  450. mock_calls = self.conn.mock_calls
  451. self.assertTrue(result.get("set"))
  452. self.assertEqual(len(mock_calls), 1)
  453. self.assertEqual(mock_calls[0][2].get("Roles"), expected_roles)
  454. def test_that_when_set_identity_pool_roles_with_both_roles_specified_the_set_identity_pool_role_method_set_both_roles(
  455. self,
  456. ):
  457. """
  458. Tests setting of both roles to valid given roles
  459. """
  460. self.conn.set_identity_pool_roles.return_value = None
  461. with patch.dict(
  462. boto_cognitoidentity.__salt__,
  463. {
  464. "boto_iam.describe_role": MagicMock(
  465. return_value={"arn": "my_unauth_role_arn"}
  466. )
  467. },
  468. ):
  469. expected_roles = dict(
  470. authenticated="arn:aws:iam:my_auth_role",
  471. unauthenticated="my_unauth_role_arn",
  472. )
  473. result = boto_cognitoidentity.set_identity_pool_roles(
  474. IdentityPoolId="some_id",
  475. AuthenticatedRole="arn:aws:iam:my_auth_role",
  476. UnauthenticatedRole="my_unauth_role",
  477. **conn_parameters
  478. )
  479. mock_calls = self.conn.mock_calls
  480. self.assertTrue(result.get("set"))
  481. self.assertEqual(len(mock_calls), 1)
  482. self.assertEqual(mock_calls[0][2].get("Roles"), expected_roles)
  483. def test_that_when_set_identity_pool_roles_given_invalid_auth_role_the_set_identity_pool_method_returns_set_false_and_error(
  484. self,
  485. ):
  486. """
  487. Tests error handling for invalid auth role
  488. """
  489. with patch.dict(
  490. boto_cognitoidentity.__salt__,
  491. {"boto_iam.describe_role": MagicMock(return_value=False)},
  492. ):
  493. result = boto_cognitoidentity.set_identity_pool_roles(
  494. IdentityPoolId="some_id",
  495. AuthenticatedRole="no_such_auth_role",
  496. **conn_parameters
  497. )
  498. mock_calls = self.conn.mock_calls
  499. self.assertIs(result.get("set"), False)
  500. self.assertIn("no_such_auth_role", result.get("error", ""))
  501. self.assertEqual(len(mock_calls), 0)
  502. def test_that_when_set_identity_pool_roles_given_invalid_unauth_role_the_set_identity_pool_method_returns_set_false_and_error(
  503. self,
  504. ):
  505. """
  506. Tests error handling for invalid unauth role
  507. """
  508. with patch.dict(
  509. boto_cognitoidentity.__salt__,
  510. {"boto_iam.describe_role": MagicMock(return_value=False)},
  511. ):
  512. result = boto_cognitoidentity.set_identity_pool_roles(
  513. IdentityPoolId="some_id",
  514. AuthenticatedRole="arn:aws:iam:my_auth_role",
  515. UnauthenticatedRole="no_such_unauth_role",
  516. **conn_parameters
  517. )
  518. mock_calls = self.conn.mock_calls
  519. self.assertIs(result.get("set"), False)
  520. self.assertIn("no_such_unauth_role", result.get("error", ""))
  521. self.assertEqual(len(mock_calls), 0)
  522. def test_that_when_update_identity_pool_given_invalid_pool_id_the_update_identity_pool_method_returns_updated_false_and_error(
  523. self,
  524. ):
  525. """
  526. Tests error handling for invalid pool id
  527. """
  528. self.conn.describe_identity_pool.side_effect = ClientError(
  529. error_content, "error on describe identity pool with pool id"
  530. )
  531. result = boto_cognitoidentity.update_identity_pool(
  532. IdentityPoolId="no_such_pool_id", **conn_parameters
  533. )
  534. self.assertIs(result.get("updated"), False)
  535. self.assertEqual(
  536. result.get("error", {}).get("message"),
  537. error_message.format("error on describe identity pool with pool id"),
  538. )
  539. def test_that_when_update_identity_pool_given_only_valid_pool_id_the_update_identity_pool_method_returns_udpated_identity(
  540. self,
  541. ):
  542. """
  543. Tests the base case of calling update_identity_pool with only the pool id, verify
  544. that the passed parameters into boto3 update_identity_pool has at least all the
  545. expected required parameters in IdentityPoolId, IdentityPoolName and AllowUnauthenticatedIdentities
  546. """
  547. self.conn.describe_identity_pool.return_value = second_pool_ret
  548. self.conn.update_identity_pool.return_value = second_pool_ret
  549. expected_params = second_pool_ret
  550. result = boto_cognitoidentity.update_identity_pool(
  551. IdentityPoolId=second_pool_id, **conn_parameters
  552. )
  553. self.assertTrue(result.get("updated"))
  554. self.assertEqual(result.get("identity_pool"), second_pool_ret)
  555. self.conn.update_identity_pool.assert_called_with(**expected_params)
  556. def test_that_when_update_identity_pool_given_valid_pool_id_and_pool_name_the_update_identity_pool_method_returns_updated_identity_pool(
  557. self,
  558. ):
  559. """
  560. Tests successful update of a pool's name
  561. """
  562. self.conn.describe_identity_pool.return_value = second_pool_ret
  563. second_pool_updated_ret = second_pool_ret.copy()
  564. second_pool_updated_ret["IdentityPoolName"] = second_pool_name_updated
  565. self.conn.update_identity_pool.return_value = second_pool_updated_ret
  566. expected_params = second_pool_updated_ret
  567. result = boto_cognitoidentity.update_identity_pool(
  568. IdentityPoolId=second_pool_id,
  569. IdentityPoolName=second_pool_name_updated,
  570. **conn_parameters
  571. )
  572. self.assertTrue(result.get("updated"))
  573. self.assertEqual(result.get("identity_pool"), second_pool_updated_ret)
  574. self.conn.update_identity_pool.assert_called_with(**expected_params)
  575. def test_that_when_update_identity_pool_given_empty_dictionary_for_supported_login_providers_the_update_identity_pool_method_is_called_with_proper_request_params(
  576. self,
  577. ):
  578. """
  579. Tests the request parameters to boto3 update_identity_pool's AllowUnauthenticatedIdentities is {}
  580. """
  581. self.conn.describe_identity_pool.return_value = first_pool_ret
  582. first_pool_updated_ret = first_pool_ret.copy()
  583. first_pool_updated_ret.pop("SupportedLoginProviders")
  584. self.conn.update_identity_pool.return_value = first_pool_updated_ret
  585. expected_params = first_pool_ret.copy()
  586. expected_params["SupportedLoginProviders"] = {}
  587. expected_params.pop("DeveloperProviderName")
  588. expected_params.pop("OpenIdConnectProviderARNs")
  589. result = boto_cognitoidentity.update_identity_pool(
  590. IdentityPoolId=first_pool_id, SupportedLoginProviders={}, **conn_parameters
  591. )
  592. self.assertTrue(result.get("updated"))
  593. self.assertEqual(result.get("identity_pool"), first_pool_updated_ret)
  594. self.conn.update_identity_pool.assert_called_with(**expected_params)
  595. def test_that_when_update_identity_pool_given_empty_list_for_openid_connect_provider_arns_the_update_identity_pool_method_is_called_with_proper_request_params(
  596. self,
  597. ):
  598. """
  599. Tests the request parameters to boto3 update_identity_pool's OpenIdConnectProviderARNs is []
  600. """
  601. self.conn.describe_identity_pool.return_value = first_pool_ret
  602. first_pool_updated_ret = first_pool_ret.copy()
  603. first_pool_updated_ret.pop("OpenIdConnectProviderARNs")
  604. self.conn.update_identity_pool.return_value = first_pool_updated_ret
  605. expected_params = first_pool_ret.copy()
  606. expected_params.pop("SupportedLoginProviders")
  607. expected_params.pop("DeveloperProviderName")
  608. expected_params["OpenIdConnectProviderARNs"] = []
  609. result = boto_cognitoidentity.update_identity_pool(
  610. IdentityPoolId=first_pool_id,
  611. OpenIdConnectProviderARNs=[],
  612. **conn_parameters
  613. )
  614. self.assertTrue(result.get("updated"))
  615. self.assertEqual(result.get("identity_pool"), first_pool_updated_ret)
  616. self.conn.update_identity_pool.assert_called_with(**expected_params)
  617. def test_that_when_update_identity_pool_given_developer_provider_name_when_developer_provider_name_was_set_previously_the_udpate_identity_pool_method_is_called_without_developer_provider_name_param(
  618. self,
  619. ):
  620. """
  621. Tests the request parameters do not include 'DeveloperProviderName' if this was previously set
  622. for the given pool id
  623. """
  624. self.conn.describe_identity_pool.return_value = first_pool_ret
  625. self.conn.update_identity_pool.return_value = first_pool_ret
  626. expected_params = first_pool_ret.copy()
  627. expected_params.pop("SupportedLoginProviders")
  628. expected_params.pop("DeveloperProviderName")
  629. expected_params.pop("OpenIdConnectProviderARNs")
  630. result = boto_cognitoidentity.update_identity_pool(
  631. IdentityPoolId=first_pool_id,
  632. DeveloperProviderName="this should not change",
  633. **conn_parameters
  634. )
  635. self.assertTrue(result.get("updated"))
  636. self.assertEqual(result.get("identity_pool"), first_pool_ret)
  637. self.conn.update_identity_pool.assert_called_with(**expected_params)
  638. def test_that_when_update_identity_pool_given_developer_provider_name_is_included_in_the_params_when_associated_for_the_first_time(
  639. self,
  640. ):
  641. """
  642. Tests the request parameters include 'DeveloperProviderName' when the pool did not have this
  643. property set previously.
  644. """
  645. self.conn.describe_identity_pool.return_value = second_pool_ret
  646. second_pool_updated_ret = second_pool_ret.copy()
  647. second_pool_updated_ret["DeveloperProviderName"] = "added_developer_provider"
  648. self.conn.update_identity_pool.return_value = second_pool_updated_ret
  649. expected_params = second_pool_updated_ret.copy()
  650. result = boto_cognitoidentity.update_identity_pool(
  651. IdentityPoolId=second_pool_id,
  652. DeveloperProviderName="added_developer_provider",
  653. **conn_parameters
  654. )
  655. self.assertTrue(result.get("updated"))
  656. self.assertEqual(result.get("identity_pool"), second_pool_updated_ret)
  657. self.conn.update_identity_pool.assert_called_with(**expected_params)
  658. def test_that_the_update_identity_pool_method_handles_exception_from_boto3(self):
  659. """
  660. Tests the error handling of exception generated by boto3 update_identity_pool
  661. """
  662. self.conn.describe_identity_pool.return_value = second_pool_ret
  663. second_pool_updated_ret = second_pool_ret.copy()
  664. second_pool_updated_ret["DeveloperProviderName"] = "added_developer_provider"
  665. self.conn.update_identity_pool.side_effect = ClientError(
  666. error_content, "update_identity_pool"
  667. )
  668. result = boto_cognitoidentity.update_identity_pool(
  669. IdentityPoolId=second_pool_id,
  670. DeveloperProviderName="added_developer_provider",
  671. **conn_parameters
  672. )
  673. self.assertIs(result.get("updated"), False)
  674. self.assertEqual(
  675. result.get("error", {}).get("message"),
  676. error_message.format("update_identity_pool"),
  677. )