1# -*- coding: utf-8 -*-
2#
3# Copyright: (c) 2017, F5 Networks Inc.
4# GNU General Public License v3.0 (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
5
6from __future__ import (absolute_import, division, print_function)
7__metaclass__ = type
8
9import os
10import json
11import pytest
12import sys
13
14if sys.version_info < (2, 7):
15    pytestmark = pytest.mark.skip("F5 Ansible modules require Python >= 2.7")
16
17from ansible.module_utils.basic import AnsibleModule
18
19from ansible_collections.f5networks.f5_modules.plugins.modules.bigip_snmp_community import (
20    ApiParameters, ModuleParameters, ModuleManager, V1Manager, V2Manager, ArgumentSpec
21)
22from ansible_collections.f5networks.f5_modules.plugins.module_utils.common import F5ModuleError
23from ansible_collections.f5networks.f5_modules.tests.unit.compat import unittest
24from ansible_collections.f5networks.f5_modules.tests.unit.compat.mock import Mock, patch
25from ansible_collections.f5networks.f5_modules.tests.unit.modules.utils import set_module_args
26
27
28fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
29fixture_data = {}
30
31
32def load_fixture(name):
33    path = os.path.join(fixture_path, name)
34
35    if path in fixture_data:
36        return fixture_data[path]
37
38    with open(path) as f:
39        data = f.read()
40
41    try:
42        data = json.loads(data)
43    except Exception:
44        pass
45
46    fixture_data[path] = data
47    return data
48
49
50class TestParameters(unittest.TestCase):
51    def test_module_parameters(self):
52        args = dict(
53            version='v2c',
54            community='foo',
55            source='1.1.1.1',
56            port='8080',
57            oid='.1',
58            access='ro',
59            ip_version=4,
60            snmp_username='admin',
61            snmp_auth_protocol='sha',
62            snmp_auth_password='secretsecret',
63            snmp_privacy_protocol='des',
64            snmp_privacy_password='secretsecret',
65            update_password='always',
66            state='present'
67        )
68
69        p = ModuleParameters(params=args)
70        assert p.version == 'v2c'
71        assert p.community == 'foo'
72        assert p.source == '1.1.1.1'
73        assert p.port == 8080
74        assert p.oid == '.1'
75        assert p.access == 'ro'
76        assert p.ip_version == 4
77        assert p.snmp_username == 'admin'
78        assert p.snmp_auth_protocol == 'sha'
79        assert p.snmp_auth_password == 'secretsecret'
80        assert p.snmp_privacy_protocol == 'des'
81        assert p.snmp_privacy_password == 'secretsecret'
82        assert p.update_password == 'always'
83        assert p.state == 'present'
84
85    def test_api_parameters_community_1(self):
86        args = load_fixture('load_sys_snmp_communities_1.json')
87
88        p = ApiParameters(params=args)
89        assert p.access == 'ro'
90        assert p.community == 'foo'
91        assert p.ip_version == 4
92
93    def test_api_parameters_community_2(self):
94        args = load_fixture('load_sys_snmp_communities_2.json')
95
96        p = ApiParameters(params=args)
97        assert p.access == 'rw'
98        assert p.community == 'foo'
99        assert p.ip_version == 4
100        assert p.oid == '.1'
101        assert p.source == '1.1.1.1'
102
103    def test_api_parameters_community_3(self):
104        args = load_fixture('load_sys_snmp_communities_3.json')
105
106        p = ApiParameters(params=args)
107        assert p.access == 'ro'
108        assert p.community == 'foo'
109        assert p.ip_version == 6
110        assert p.oid == '.1'
111        assert p.source == '2001:0db8:85a3:0000:0000:8a2e:0370:7334'
112
113    def test_api_parameters_community_4(self):
114        args = load_fixture('load_sys_snmp_communities_4.json')
115
116        p = ApiParameters(params=args)
117        assert p.access == 'ro'
118        assert p.community == 'foo'
119        assert p.ip_version == 6
120
121    def test_api_parameters_users_1(self):
122        args = load_fixture('load_sys_snmp_users_1.json')
123
124        p = ApiParameters(params=args)
125        assert p.access == 'ro'
126        assert p.snmp_auth_protocol == 'sha'
127        assert p.oid == '.1'
128        assert p.snmp_privacy_protocol == 'aes'
129        assert p.snmp_username == 'foo'
130
131
132class TestManager(unittest.TestCase):
133    def setUp(self):
134        self.spec = ArgumentSpec()
135        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_snmp_community.tmos_version')
136        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_snmp_community.send_teem')
137        self.m2 = self.p2.start()
138        self.m2.return_value = '14.1.0'
139        self.m3 = self.p3.start()
140        self.m3.return_value = True
141
142    def tearDown(self):
143        self.p2.stop()
144        self.p3.stop()
145
146    def test_create_v2c_community_1(self, *args):
147        set_module_args(dict(
148            version='v2c',
149            community='foo',
150            source='1.1.1.1',
151            port='8080',
152            oid='.1',
153            access='ro',
154            ip_version=4,
155            state='present',
156            partition='Common',
157            provider=dict(
158                server='localhost',
159                password='password',
160                user='admin'
161            )
162        ))
163
164        module = AnsibleModule(
165            argument_spec=self.spec.argument_spec,
166            supports_check_mode=self.spec.supports_check_mode,
167            required_if=self.spec.required_if
168        )
169        m1 = V1Manager(module=module)
170
171        # Override methods to force specific logic in the module to happen
172        m1.exists = Mock(side_effect=[False, True])
173        m1.create_on_device = Mock(return_value=True)
174
175        m0 = ModuleManager(module=module)
176        m0.get_manager = Mock(return_value=m1)
177
178        results = m0.exec_module()
179
180        assert results['changed'] is True
181
182    def test_create_v1_community_1(self, *args):
183        set_module_args(dict(
184            name='foo',
185            version='v1',
186            community='foo',
187            source='1.1.1.1',
188            port='8080',
189            oid='.1',
190            access='ro',
191            ip_version=4,
192            state='present',
193            partition='Common',
194            provider=dict(
195                server='localhost',
196                password='password',
197                user='admin'
198            )
199        ))
200
201        module = AnsibleModule(
202            argument_spec=self.spec.argument_spec,
203            supports_check_mode=self.spec.supports_check_mode,
204            required_if=self.spec.required_if
205        )
206        m1 = V1Manager(module=module)
207
208        # Override methods to force specific logic in the module to happen
209        m1.exists = Mock(side_effect=[False, True])
210        m1.create_on_device = Mock(return_value=True)
211
212        m0 = ModuleManager(module=module)
213        m0.get_manager = Mock(return_value=m1)
214
215        results = m0.exec_module()
216
217        assert results['changed'] is True
218
219    def test_create_v3_community_1(self, *args):
220        set_module_args(dict(
221            version='v3',
222            oid='.1',
223            access='ro',
224            snmp_username='admin',
225            snmp_auth_protocol='md5',
226            snmp_auth_password='secretsecret',
227            snmp_privacy_protocol='des',
228            snmp_privacy_password='secretsecret',
229            state='present',
230            partition='Common',
231            provider=dict(
232                server='localhost',
233                password='password',
234                user='admin'
235            )
236        ))
237
238        module = AnsibleModule(
239            argument_spec=self.spec.argument_spec,
240            supports_check_mode=self.spec.supports_check_mode,
241            required_if=self.spec.required_if
242        )
243        m1 = V2Manager(module=module)
244
245        # Override methods to force specific logic in the module to happen
246        m1.exists = Mock(side_effect=[False, True])
247        m1.create_on_device = Mock(return_value=True)
248
249        m0 = ModuleManager(module=module)
250        m0.get_manager = Mock(return_value=m1)
251
252        results = m0.exec_module()
253
254        assert results['changed'] is True
255
256    def test_create_v3_community_2(self, *args):
257        set_module_args(dict(
258            version='v3',
259            access='ro',
260            snmp_username='admin',
261            snmp_auth_protocol='md5',
262            snmp_auth_password='secretsecret',
263            snmp_privacy_protocol='des',
264            snmp_privacy_password='secretsecret',
265            state='present',
266            partition='Common',
267            provider=dict(
268                server='localhost',
269                password='password',
270                user='admin'
271            )
272        ))
273
274        module = AnsibleModule(
275            argument_spec=self.spec.argument_spec,
276            supports_check_mode=self.spec.supports_check_mode,
277            required_if=self.spec.required_if
278        )
279        m1 = V2Manager(module=module)
280
281        # Override methods to force specific logic in the module to happen
282        m1.exists = Mock(side_effect=[False, True])
283        m1.create_on_device = Mock(return_value=True)
284
285        m0 = ModuleManager(module=module)
286        m0.get_manager = Mock(return_value=m1)
287
288        with pytest.raises(F5ModuleError) as ex:
289            m0.exec_module()
290
291        assert 'oid must be specified when creating a new v3 community.' == str(ex.value)
292