1# -*- coding: utf-8 -*- 2# 3# Copyright: (c) 2019, 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_network_globals import ( 20 ApiParameters, ModuleParameters, ModuleManager, ArgumentSpec 21) 22from ansible_collections.f5networks.f5_modules.tests.unit.compat import unittest 23from ansible_collections.f5networks.f5_modules.tests.unit.compat.mock import Mock, patch 24from ansible_collections.f5networks.f5_modules.tests.unit.modules.utils import set_module_args 25 26 27fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures') 28fixture_data = {} 29 30 31def load_fixture(name): 32 path = os.path.join(fixture_path, name) 33 34 if path in fixture_data: 35 return fixture_data[path] 36 37 with open(path) as f: 38 data = f.read() 39 40 try: 41 data = json.loads(data) 42 except Exception: 43 pass 44 45 fixture_data[path] = data 46 return data 47 48 49class TestParameters(unittest.TestCase): 50 def test_module_parameters(self): 51 args = dict( 52 stp=dict( 53 config_name='foobar', 54 config_revision=1, 55 description='description', 56 fwd_delay=16, 57 hello_time=2, 58 max_age=6, 59 max_hops=2, 60 mode='mstp', 61 transmit_hold=1 62 ), 63 multicast=dict( 64 max_pending_packets=2000, 65 max_pending_routes=200, 66 rate_limit=True, 67 route_lookup_timeout=80000, 68 ), 69 dag=dict( 70 round_robin_mode='local', 71 dag_ipv6_prefix_len=64, 72 icmp_hash='ipicmp', 73 ), 74 lldp=dict( 75 enabled=True, 76 max_neighbors_per_port=200, 77 reinit_delay=1000, 78 tx_delay=1500, 79 tx_hold=200, 80 tx_interval=100 81 ) 82 ) 83 p = ModuleParameters(params=args) 84 assert p.stp_config_name == 'foobar' 85 assert p.stp_config_revision == 1 86 assert p.stp_description == 'description' 87 assert p.stp_fwd_delay == 16 88 assert p.stp_hello_time == 2 89 assert p.stp_max_age == 6 90 assert p.stp_max_hops == 2 91 assert p.stp_mode == 'mstp' 92 assert p.stp_transmit_hold == 1 93 assert p.mcast_max_pending_packets == 2000 94 assert p.mcast_max_pending_routes == 200 95 assert p.mcast_rate_limit == 'enabled' 96 assert p.mcast_route_lookup_timeout == 80000 97 assert p.dag_round_robin_mode == 'local' 98 assert p.dag_ipv6_prefix_len == 64 99 assert p.dag_icmp_hash == 'ipicmp' 100 assert p.lldp_enabled is True 101 assert p.lldp_disabled is None 102 assert p.lldp_max_neighbors_per_port == 200 103 assert p.lldp_reinit_delay == 1000 104 assert p.lldp_tx_delay == 1500 105 assert p.lldp_tx_hold == 200 106 assert p.lldp_tx_interval == 100 107 108 def test_api_parameters(self): 109 args = dict( 110 stp=load_fixture('load_stp_globals.json'), 111 multicast=load_fixture('load_mcast_globals.json'), 112 dag=load_fixture('load_dag_globals.json'), 113 lldp=load_fixture('load_lldp_globals.json') 114 ) 115 116 p = ApiParameters(params=args) 117 118 assert p.stp_config_name is None 119 assert p.stp_config_revision == 0 120 assert p.stp_description is None 121 assert p.stp_fwd_delay == 15 122 assert p.stp_hello_time == 2 123 assert p.stp_max_age == 20 124 assert p.stp_max_hops == 20 125 assert p.stp_mode == 'passthru' 126 assert p.stp_transmit_hold == 6 127 assert p.mcast_max_pending_packets == 16 128 assert p.mcast_max_pending_routes == 256 129 assert p.mcast_rate_limit == 'enabled' 130 assert p.mcast_route_lookup_timeout == 2 131 assert p.dag_round_robin_mode == 'global' 132 assert p.dag_ipv6_prefix_len == 128 133 assert p.dag_icmp_hash == 'icmp' 134 assert p.lldp_enabled is None 135 assert p.lldp_disabled is True 136 assert p.lldp_max_neighbors_per_port == 10 137 assert p.lldp_reinit_delay == 2 138 assert p.lldp_tx_delay == 0 139 assert p.lldp_tx_hold == 5 140 assert p.lldp_tx_interval == 0 141 142 143class TestManager(unittest.TestCase): 144 def setUp(self): 145 self.spec = ArgumentSpec() 146 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_network_globals.tmos_version') 147 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_network_globals.send_teem') 148 self.m2 = self.p2.start() 149 self.m2.return_value = '14.1.0' 150 self.m3 = self.p3.start() 151 self.m3.return_value = True 152 153 def tearDown(self): 154 self.p2.stop() 155 self.p3.stop() 156 157 def test_update_all_settings(self, *args): 158 set_module_args(dict( 159 stp=dict( 160 config_name='foobar', 161 config_revision=1, 162 max_hops=2, 163 mode='mstp', 164 transmit_hold=1 165 ), 166 multicast=dict( 167 max_pending_routes=200, 168 rate_limit='no', 169 ), 170 dag=dict( 171 round_robin_mode='local', 172 dag_ipv6_prefix_len=64, 173 icmp_hash='ipicmp', 174 ), 175 lldp=dict( 176 enabled='yes', 177 tx_delay=1500, 178 tx_hold=200, 179 tx_interval=100 180 ), 181 provider=dict( 182 server='localhost', 183 password='password', 184 user='admin' 185 ) 186 )) 187 188 # Configure the parameters that would be returned by querying the 189 # remote device 190 current = dict( 191 stp=load_fixture('load_stp_globals.json'), 192 multicast=load_fixture('load_mcast_globals.json'), 193 dag=load_fixture('load_dag_globals.json'), 194 lldp=load_fixture('load_lldp_globals.json') 195 ) 196 197 module = AnsibleModule( 198 argument_spec=self.spec.argument_spec, 199 supports_check_mode=self.spec.supports_check_mode, 200 required_one_of=self.spec.required_one_of 201 ) 202 mm = ModuleManager(module=module) 203 204 # Override methods to force specific logic in the module to happen 205 mm.update_on_device = Mock(return_value=True) 206 mm.read_current_from_device = Mock(return_value=ApiParameters(params=current)) 207 208 results = mm.exec_module() 209 210 assert results['changed'] is True 211 assert results['stp']['config_name'] == 'foobar' 212 assert results['stp']['config_revision'] == 1 213 assert results['stp']['max_hops'] == 2 214 assert results['stp']['mode'] == 'mstp' 215 assert results['stp']['transmit_hold'] == 1 216 assert results['dag']['dag_ipv6_prefix_len'] == 64 217 assert results['dag']['icmp_hash'] == 'ipicmp' 218 assert results['multicast']['rate_limit'] == 'no' 219 assert results['multicast']['max_pending_routes'] == 200 220 assert results['lldp']['enabled'] == 'yes' 221 assert results['lldp']['tx_delay'] == 1500 222 assert results['lldp']['tx_hold'] == 200 223 assert results['lldp']['tx_interval'] == 100 224 225 def test_update_one_group(self, *args): 226 set_module_args(dict( 227 stp=dict( 228 hello_time=1, 229 mode='rstp', 230 ), 231 provider=dict( 232 server='localhost', 233 password='password', 234 user='admin' 235 ) 236 )) 237 238 # Configure the parameters that would be returned by querying the 239 # remote device 240 current = dict( 241 stp=load_fixture('load_stp_globals.json'), 242 multicast=None, 243 dag=None, 244 lldp=None, 245 ) 246 247 module = AnsibleModule( 248 argument_spec=self.spec.argument_spec, 249 supports_check_mode=self.spec.supports_check_mode, 250 required_one_of=self.spec.required_one_of 251 ) 252 mm = ModuleManager(module=module) 253 254 # Override methods to force specific logic in the module to happen 255 mm.update_on_device = Mock(return_value=True) 256 mm.read_current_from_device = Mock(return_value=ApiParameters(params=current)) 257 258 results = mm.exec_module() 259 260 assert results['changed'] is True 261 assert results['stp']['hello_time'] == 1 262 assert results['stp']['mode'] == 'rstp' 263