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_message_routing_route import ( 20 ApiParameters, ModuleParameters, ModuleManager, GenericModuleManager, 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 name='foo', 53 partition='foobar', 54 description='my description', 55 dst_address='some_destination', 56 src_address='some_address', 57 peer_selection_mode='ratio', 58 peers=['/Common/example'] 59 ) 60 61 p = ModuleParameters(params=args) 62 assert p.name == 'foo' 63 assert p.partition == 'foobar' 64 assert p.description == 'my description' 65 assert p.dst_address == 'some_destination' 66 assert p.src_address == 'some_address' 67 assert p.peer_selection_mode == 'ratio' 68 assert p.peers == ['/Common/example'] 69 70 def test_api_parameters(self): 71 args = load_fixture('load_generic_route.json') 72 73 p = ApiParameters(params=args) 74 assert p.name == 'some' 75 assert p.partition == 'Common' 76 assert p.dst_address == 'annoying_user' 77 assert p.src_address == '99.99.99.99' 78 assert p.peer_selection_mode == 'sequential' 79 assert p.peers == ['/Common/testy'] 80 81 82class TestManager(unittest.TestCase): 83 def setUp(self): 84 self.spec = ArgumentSpec() 85 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_message_routing_route.tmos_version') 86 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_message_routing_route.send_teem') 87 self.m2 = self.p2.start() 88 self.m2.return_value = '14.1.0' 89 self.m3 = self.p3.start() 90 self.m3.return_value = True 91 92 def tearDown(self): 93 self.p2.stop() 94 self.p3.stop() 95 96 def test_create_generic_route(self, *args): 97 set_module_args(dict( 98 name='some', 99 partition='foobar', 100 description='my description', 101 dst_address='some_destination', 102 src_address='some_address', 103 peer_selection_mode='ratio', 104 peers=['/Common/example'], 105 provider=dict( 106 server='localhost', 107 password='password', 108 user='admin' 109 ) 110 )) 111 112 module = AnsibleModule( 113 argument_spec=self.spec.argument_spec, 114 supports_check_mode=self.spec.supports_check_mode 115 ) 116 117 # Override methods in the specific type of manager 118 gm = GenericModuleManager(module=module) 119 gm.exists = Mock(return_value=False) 120 gm.create_on_device = Mock(return_value=True) 121 122 mm = ModuleManager(module=module) 123 mm.version_less_than_14 = Mock(return_value=False) 124 mm.get_manager = Mock(return_value=gm) 125 126 results = mm.exec_module() 127 128 assert results['changed'] is True 129 assert results['description'] == 'my description' 130 assert results['dst_address'] == 'some_destination' 131 assert results['src_address'] == 'some_address' 132 assert results['peer_selection_mode'] == 'ratio' 133 assert results['peers'] == ['/Common/example'] 134 135 def test_update_generic_peer(self, *args): 136 set_module_args(dict( 137 name='some', 138 dst_address="blackhole", 139 peer_selection_mode='ratio', 140 peers=['/Common/example'], 141 provider=dict( 142 server='localhost', 143 password='password', 144 user='admin' 145 ) 146 )) 147 148 current = ApiParameters(params=load_fixture('load_generic_route.json')) 149 150 module = AnsibleModule( 151 argument_spec=self.spec.argument_spec, 152 supports_check_mode=self.spec.supports_check_mode 153 ) 154 155 # Override methods in the specific type of manager 156 gm = GenericModuleManager(module=module) 157 gm.exists = Mock(return_value=True) 158 gm.update_on_device = Mock(return_value=True) 159 gm.read_current_from_device = Mock(return_value=current) 160 161 mm = ModuleManager(module=module) 162 mm.version_less_than_14 = Mock(return_value=False) 163 mm.get_manager = Mock(return_value=gm) 164 165 results = mm.exec_module() 166 167 assert results['changed'] is True 168 assert results['dst_address'] == 'blackhole' 169 assert results['peer_selection_mode'] == 'ratio' 170 assert results['peers'] == ['/Common/example'] 171