1import copy
2import logging
3import random
4import string
5
6import salt.loader
7import salt.modules.boto_elasticsearch_domain as boto_elasticsearch_domain
8from salt.utils.versions import LooseVersion
9from tests.support.mixins import LoaderModuleMockMixin
10from tests.support.mock import MagicMock, patch
11from tests.support.unit import TestCase, skipIf
12
13# pylint: disable=import-error,no-name-in-module
14try:
15    import boto3
16    from botocore.exceptions import ClientError
17
18    HAS_BOTO = True
19except ImportError:
20    HAS_BOTO = False
21
22
23# pylint: enable=import-error,no-name-in-module
24
25# the boto_elasticsearch_domain module relies on the connect_to_region() method
26# which was added in boto 2.8.0
27# https://github.com/boto/boto/commit/33ac26b416fbb48a60602542b4ce15dcc7029f12
28required_boto3_version = "1.2.1"
29
30
31def _has_required_boto():
32    """
33    Returns True/False boolean depending on if Boto is installed and correct
34    version.
35    """
36    if not HAS_BOTO:
37        return False
38    elif LooseVersion(boto3.__version__) < LooseVersion(required_boto3_version):
39        return False
40    else:
41        return True
42
43
44if _has_required_boto():
45    region = "us-east-1"
46    access_key = "GKTADJGHEIQSXMKKRBJ08H"
47    secret_key = "askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs"
48    conn_parameters = {
49        "region": region,
50        "key": access_key,
51        "keyid": secret_key,
52        "profile": {},
53    }
54    error_message = (
55        "An error occurred (101) when calling the {0} operation: Test-defined error"
56    )
57    error_content = {"Error": {"Code": 101, "Message": "Test-defined error"}}
58    not_found_error = ClientError(
59        {
60            "Error": {
61                "Code": "ResourceNotFoundException",
62                "Message": "Test-defined error",
63            }
64        },
65        "msg",
66    )
67    domain_ret = dict(
68        DomainName="testdomain",
69        ElasticsearchClusterConfig={},
70        EBSOptions={},
71        AccessPolicies={},
72        SnapshotOptions={},
73        AdvancedOptions={},
74    )
75
76log = logging.getLogger(__name__)
77
78
79class BotoElasticsearchDomainTestCaseBase(TestCase, LoaderModuleMockMixin):
80    conn = None
81
82    def setup_loader_modules(self):
83        self.opts = salt.config.DEFAULT_MINION_OPTS.copy()
84        utils = salt.loader.utils(
85            self.opts,
86            whitelist=["boto3", "args", "systemd", "path", "platform"],
87            context={},
88        )
89        return {boto_elasticsearch_domain: {"__utils__": utils}}
90
91    def setUp(self):
92        super().setUp()
93        boto_elasticsearch_domain.__init__(self.opts)
94        del self.opts
95
96        # Set up MagicMock to replace the boto3 session
97        # connections keep getting cached from prior tests, can't find the
98        # correct context object to clear it. So randomize the cache key, to prevent any
99        # cache hits
100        conn_parameters["key"] = "".join(
101            random.choice(string.ascii_lowercase + string.digits) for _ in range(50)
102        )
103
104        self.patcher = patch("boto3.session.Session")
105        self.addCleanup(self.patcher.stop)
106        self.addCleanup(delattr, self, "patcher")
107        mock_session = self.patcher.start()
108
109        session_instance = mock_session.return_value
110        self.conn = MagicMock()
111        self.addCleanup(delattr, self, "conn")
112        session_instance.client.return_value = self.conn
113
114
115class BotoElasticsearchDomainTestCaseMixin:
116    pass
117
118
119# @skipIf(True, "Skip these tests while investigating failures")
120@skipIf(HAS_BOTO is False, "The boto module must be installed.")
121@skipIf(
122    _has_required_boto() is False,
123    "The boto3 module must be greater than or equal to version {}".format(
124        required_boto3_version
125    ),
126)
127class BotoElasticsearchDomainTestCase(
128    BotoElasticsearchDomainTestCaseBase, BotoElasticsearchDomainTestCaseMixin
129):
130    """
131    TestCase for salt.modules.boto_elasticsearch_domain module
132    """
133
134    def test_that_when_checking_if_a_domain_exists_and_a_domain_exists_the_domain_exists_method_returns_true(
135        self,
136    ):
137        """
138        Tests checking domain existence when the domain already exists
139        """
140        result = boto_elasticsearch_domain.exists(
141            DomainName="testdomain", **conn_parameters
142        )
143
144        self.assertTrue(result["exists"])
145
146    def test_that_when_checking_if_a_domain_exists_and_a_domain_does_not_exist_the_domain_exists_method_returns_false(
147        self,
148    ):
149        """
150        Tests checking domain existence when the domain does not exist
151        """
152        self.conn.describe_elasticsearch_domain.side_effect = not_found_error
153        result = boto_elasticsearch_domain.exists(
154            DomainName="mydomain", **conn_parameters
155        )
156
157        self.assertFalse(result["exists"])
158
159    def test_that_when_checking_if_a_domain_exists_and_boto3_returns_an_error_the_domain_exists_method_returns_error(
160        self,
161    ):
162        """
163        Tests checking domain existence when boto returns an error
164        """
165        self.conn.describe_elasticsearch_domain.side_effect = ClientError(
166            error_content, "list_domains"
167        )
168        result = boto_elasticsearch_domain.exists(
169            DomainName="mydomain", **conn_parameters
170        )
171
172        self.assertEqual(
173            result.get("error", {}).get("message"), error_message.format("list_domains")
174        )
175
176    def test_that_when_checking_domain_status_and_a_domain_exists_the_domain_status_method_returns_info(
177        self,
178    ):
179        """
180        Tests checking domain existence when the domain already exists
181        """
182        self.conn.describe_elasticsearch_domain.return_value = {
183            "DomainStatus": domain_ret
184        }
185        result = boto_elasticsearch_domain.status(
186            DomainName="testdomain", **conn_parameters
187        )
188
189        self.assertTrue(result["domain"])
190
191    def test_that_when_checking_domain_status_and_boto3_returns_an_error_the_domain_status_method_returns_error(
192        self,
193    ):
194        """
195        Tests checking domain existence when boto returns an error
196        """
197        self.conn.describe_elasticsearch_domain.side_effect = ClientError(
198            error_content, "list_domains"
199        )
200        result = boto_elasticsearch_domain.status(
201            DomainName="mydomain", **conn_parameters
202        )
203
204        self.assertEqual(
205            result.get("error", {}).get("message"), error_message.format("list_domains")
206        )
207
208    def test_that_when_describing_domain_it_returns_the_dict_of_properties_returns_true(
209        self,
210    ):
211        """
212        Tests describing parameters if domain exists
213        """
214        domainconfig = {}
215        for k, v in domain_ret.items():
216            if k == "DomainName":
217                continue
218            domainconfig[k] = {"Options": v}
219        self.conn.describe_elasticsearch_domain_config.return_value = {
220            "DomainConfig": domainconfig
221        }
222
223        result = boto_elasticsearch_domain.describe(
224            DomainName=domain_ret["DomainName"], **conn_parameters
225        )
226
227        log.warning(result)
228        desired_ret = copy.copy(domain_ret)
229        desired_ret.pop("DomainName")
230        self.assertEqual(result, {"domain": desired_ret})
231
232    def test_that_when_describing_domain_on_client_error_it_returns_error(self):
233        """
234        Tests describing parameters failure
235        """
236        self.conn.describe_elasticsearch_domain_config.side_effect = ClientError(
237            error_content, "list_domains"
238        )
239        result = boto_elasticsearch_domain.describe(
240            DomainName="testdomain", **conn_parameters
241        )
242        self.assertTrue("error" in result)
243
244    def test_that_when_creating_a_domain_succeeds_the_create_domain_method_returns_true(
245        self,
246    ):
247        """
248        tests True domain created.
249        """
250        self.conn.create_elasticsearch_domain.return_value = {
251            "DomainStatus": domain_ret
252        }
253        args = copy.copy(domain_ret)
254        args.update(conn_parameters)
255        result = boto_elasticsearch_domain.create(**args)
256
257        self.assertTrue(result["created"])
258
259    def test_that_when_creating_a_domain_fails_the_create_domain_method_returns_error(
260        self,
261    ):
262        """
263        tests False domain not created.
264        """
265        self.conn.create_elasticsearch_domain.side_effect = ClientError(
266            error_content, "create_domain"
267        )
268        args = copy.copy(domain_ret)
269        args.update(conn_parameters)
270        result = boto_elasticsearch_domain.create(**args)
271        self.assertEqual(
272            result.get("error", {}).get("message"),
273            error_message.format("create_domain"),
274        )
275
276    def test_that_when_deleting_a_domain_succeeds_the_delete_domain_method_returns_true(
277        self,
278    ):
279        """
280        tests True domain deleted.
281        """
282        result = boto_elasticsearch_domain.delete(
283            DomainName="testdomain", **conn_parameters
284        )
285        self.assertTrue(result["deleted"])
286
287    def test_that_when_deleting_a_domain_fails_the_delete_domain_method_returns_false(
288        self,
289    ):
290        """
291        tests False domain not deleted.
292        """
293        self.conn.delete_elasticsearch_domain.side_effect = ClientError(
294            error_content, "delete_domain"
295        )
296        result = boto_elasticsearch_domain.delete(
297            DomainName="testdomain", **conn_parameters
298        )
299        self.assertFalse(result["deleted"])
300
301    def test_that_when_updating_a_domain_succeeds_the_update_domain_method_returns_true(
302        self,
303    ):
304        """
305        tests True domain updated.
306        """
307        self.conn.update_elasticsearch_domain_config.return_value = {
308            "DomainConfig": domain_ret
309        }
310        args = copy.copy(domain_ret)
311        args.update(conn_parameters)
312        result = boto_elasticsearch_domain.update(**args)
313
314        self.assertTrue(result["updated"])
315
316    def test_that_when_updating_a_domain_fails_the_update_domain_method_returns_error(
317        self,
318    ):
319        """
320        tests False domain not updated.
321        """
322        self.conn.update_elasticsearch_domain_config.side_effect = ClientError(
323            error_content, "update_domain"
324        )
325        args = copy.copy(domain_ret)
326        args.update(conn_parameters)
327        result = boto_elasticsearch_domain.update(**args)
328        self.assertEqual(
329            result.get("error", {}).get("message"),
330            error_message.format("update_domain"),
331        )
332
333    def test_that_when_adding_tags_succeeds_the_add_tags_method_returns_true(self):
334        """
335        tests True tags added.
336        """
337        self.conn.describe_elasticsearch_domain.return_value = {
338            "DomainStatus": domain_ret
339        }
340        result = boto_elasticsearch_domain.add_tags(
341            DomainName="testdomain", a="b", **conn_parameters
342        )
343
344        self.assertTrue(result["tagged"])
345
346    def test_that_when_adding_tags_fails_the_add_tags_method_returns_false(self):
347        """
348        tests False tags not added.
349        """
350        self.conn.add_tags.side_effect = ClientError(error_content, "add_tags")
351        self.conn.describe_elasticsearch_domain.return_value = {
352            "DomainStatus": domain_ret
353        }
354        result = boto_elasticsearch_domain.add_tags(
355            DomainName=domain_ret["DomainName"], a="b", **conn_parameters
356        )
357        self.assertFalse(result["tagged"])
358
359    def test_that_when_removing_tags_succeeds_the_remove_tags_method_returns_true(self):
360        """
361        tests True tags removed.
362        """
363        self.conn.describe_elasticsearch_domain.return_value = {
364            "DomainStatus": domain_ret
365        }
366        result = boto_elasticsearch_domain.remove_tags(
367            DomainName=domain_ret["DomainName"], TagKeys=["a"], **conn_parameters
368        )
369
370        self.assertTrue(result["tagged"])
371
372    def test_that_when_removing_tags_fails_the_remove_tags_method_returns_false(self):
373        """
374        tests False tags not removed.
375        """
376        self.conn.remove_tags.side_effect = ClientError(error_content, "remove_tags")
377        self.conn.describe_elasticsearch_domain.return_value = {
378            "DomainStatus": domain_ret
379        }
380        result = boto_elasticsearch_domain.remove_tags(
381            DomainName=domain_ret["DomainName"], TagKeys=["b"], **conn_parameters
382        )
383        self.assertFalse(result["tagged"])
384
385    def test_that_when_listing_tags_succeeds_the_list_tags_method_returns_true(self):
386        """
387        tests True tags listed.
388        """
389        self.conn.describe_elasticsearch_domain.return_value = {
390            "DomainStatus": domain_ret
391        }
392        result = boto_elasticsearch_domain.list_tags(
393            DomainName=domain_ret["DomainName"], **conn_parameters
394        )
395
396        self.assertEqual(result["tags"], {})
397
398    def test_that_when_listing_tags_fails_the_list_tags_method_returns_false(self):
399        """
400        tests False tags not listed.
401        """
402        self.conn.list_tags.side_effect = ClientError(error_content, "list_tags")
403        self.conn.describe_elasticsearch_domain.return_value = {
404            "DomainStatus": domain_ret
405        }
406        result = boto_elasticsearch_domain.list_tags(
407            DomainName=domain_ret["DomainName"], **conn_parameters
408        )
409        self.assertTrue(result["error"])
410