1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00

Refactors bigip_sys_db (#33490)

Bringing it up to speed with the other f5 modules.
This commit is contained in:
Tim Rupp 2017-12-02 19:08:01 -08:00 committed by GitHub
parent 277d416b5b
commit 008e23dcd1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 318 additions and 107 deletions

View file

@ -1,14 +1,18 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (c) 2017 F5 Networks Inc.
# Copyright (c) 2016 F5 Networks Inc.
# GNU General Public License v3.0 (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = '''
DOCUMENTATION = r'''
---
module: bigip_sys_db
short_description: Manage BIG-IP system database variables
@ -19,14 +23,14 @@ options:
key:
description:
- The database variable to manipulate.
required: true
required: True
state:
description:
- The state of the variable on the system. When C(present), guarantees
that an existing variable is set to C(value). When C(reset) sets the
variable back to the default value. At least one of value and state
C(reset) are required.
required: false
required: False
default: present
choices:
- present
@ -35,7 +39,7 @@ options:
description:
- The value to set the key to. At least one of value and state C(reset)
are required.
required: false
required: False
notes:
- Requires the f5-sdk Python package on the host. This is as easy as pip
install f5-sdk.
@ -47,167 +51,240 @@ author:
- Tim Rupp (@caphrim007)
'''
EXAMPLES = '''
EXAMPLES = r'''
- name: Set the boot.quiet DB variable on the BIG-IP
bigip_sys_db:
user: "admin"
password: "secret"
server: "lb.mydomain.com"
key: "boot.quiet"
value: "disable"
user: admin
password: secret
server: lb.mydomain.com
key: boot.quiet
value: disable
delegate_to: localhost
- name: Disable the initial setup screen
bigip_sys_db:
user: "admin"
password: "secret"
server: "lb.mydomain.com"
key: "setup.run"
value: "false"
user: admin
password: secret
server: lb.mydomain.com
key: setup.run
value: false
delegate_to: localhost
- name: Reset the initial setup screen
bigip_sys_db:
user: "admin"
password: "secret"
server: "lb.mydomain.com"
key: "setup.run"
state: "reset"
user: admin
password: secret
server: lb.mydomain.com
key: setup.run
state: reset
delegate_to: localhost
'''
RETURN = '''
RETURN = r'''
name:
description: The key in the system database that was specified
returned: changed and success
type: string
sample: "setup.run"
description: The key in the system database that was specified
returned: changed and success
type: string
sample: setup.run
default_value:
description: The default value of the key
returned: changed and success
type: string
sample: "true"
description: The default value of the key
returned: changed and success
type: string
sample: true
value:
description: The value that you set the key to
returned: changed and success
type: string
sample: "false"
description: The value that you set the key to
returned: changed and success
type: string
sample: false
'''
from ansible.module_utils.f5_utils import AnsibleF5Client
from ansible.module_utils.f5_utils import AnsibleF5Parameters
from ansible.module_utils.f5_utils import HAS_F5SDK
from ansible.module_utils.f5_utils import F5ModuleError
try:
from f5.bigip import ManagementRoot
HAS_F5SDK = True
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
except ImportError:
HAS_F5SDK = False
class BigIpSysDb(object):
def __init__(self, *args, **kwargs):
if not HAS_F5SDK:
raise F5ModuleError("The python f5-sdk module is required")
class Parameters(AnsibleF5Parameters):
api_map = {
'defaultValue': 'default_value'
}
api_attributes = ['value']
updatables = ['value']
returnables = ['name', 'value', 'default_value']
self.params = kwargs
self.api = ManagementRoot(kwargs['server'],
kwargs['user'],
kwargs['password'],
port=kwargs['server_port'])
def to_return(self):
result = {}
for returnable in self.returnables:
result[returnable] = getattr(self, returnable)
result = self._filter_params(result)
return result
def flush(self):
result = dict()
state = self.params['state']
value = self.params['value']
if not state == 'reset' and not value:
raise F5ModuleError(
"When setting a key, a value must be supplied"
)
current = self.read()
if self.params['check_mode']:
if value == current:
changed = False
def api_params(self):
result = {}
for api_attribute in self.api_attributes:
if self.api_map is not None and api_attribute in self.api_map:
result[api_attribute] = getattr(self, self.api_map[api_attribute])
else:
changed = True
else:
result[api_attribute] = getattr(self, api_attribute)
result = self._filter_params(result)
return result
@property
def name(self):
return self._values['key']
@name.setter
def name(self, value):
self._values['key'] = value
class ModuleManager(object):
def __init__(self, client):
self.client = client
self.have = None
self.want = Parameters(self.client.module.params)
self.changes = Parameters()
def _update_changed_options(self):
changed = {}
for key in Parameters.updatables:
if getattr(self.want, key) is not None:
attr1 = getattr(self.want, key)
attr2 = getattr(self.have, key)
if attr1 != attr2:
changed[key] = attr1
if self.want.state == 'reset':
if str(self.want.value) == str(self.want.default_value):
changed[self.want.key] = self.want.value
if changed:
self.changes = Parameters(changed)
return True
return False
def exec_module(self):
changed = False
result = dict()
state = self.want.state
try:
if state == "present":
changed = self.present()
elif state == "reset":
changed = self.reset()
current = self.read()
result.update(
name=current.name,
default_value=current.defaultValue,
value=current.value
)
except iControlUnexpectedHTTPError as e:
raise F5ModuleError(str(e))
changes = self.changes.to_return()
result.update(**changes)
result.update(dict(changed=changed))
return result
def read(self):
dbs = self.api.tm.sys.dbs.db.load(
name=self.params['key']
def read_current_from_device(self):
resource = self.client.api.tm.sys.dbs.db.load(
name=self.want.key
)
return dbs
result = resource.attrs
return Parameters(result)
def exists(self):
resource = self.client.api.tm.sys.dbs.db.load(
name=self.want.key
)
if str(resource.value) == str(self.want.value):
return True
return False
def present(self):
current = self.read()
if current.value == self.params['value']:
if self.exists():
return False
else:
return self.update()
current.update(value=self.params['value'])
current.refresh()
if current.value != self.params['value']:
def update(self):
if self.want.value is None:
raise F5ModuleError(
"Failed to set the DB variable"
"When setting a key, a value must be supplied"
)
self.have = self.read_current_from_device()
if not self.should_update():
return False
if self.client.check_mode:
return True
self.update_on_device()
return True
def should_update(self):
result = self._update_changed_options()
if result:
return True
return False
def update_on_device(self):
params = self.want.api_params()
resource = self.client.api.tm.sys.dbs.db.load(
name=self.want.key
)
resource.update(**params)
def reset(self):
current = self.read()
default = current.defaultValue
if current.value == default:
self.have = self.read_current_from_device()
if not self.should_update():
return False
current.update(value=default)
current.refresh()
if current.value != current.defaultValue:
if self.client.check_mode:
return True
self.update_on_device()
if self.exists():
return True
else:
raise F5ModuleError(
"Failed to reset the DB variable"
)
return True
def reset_on_device(self):
resource = self.client.api.tm.sys.dbs.db.load(
name=self.want.key
)
resource.update(value=self.want.default_value)
class ArgumentSpec(object):
def __init__(self):
self.supports_check_mode = True
self.argument_spec = dict(
key=dict(required=True),
state=dict(
default='present',
choices=['present', 'reset']
),
value=dict()
)
self.f5_product_name = 'bigip'
def main():
argument_spec = f5_argument_spec()
if not HAS_F5SDK:
raise F5ModuleError("The python f5-sdk module is required")
meta_args = dict(
key=dict(required=True),
state=dict(default='present', choices=['present', 'reset']),
value=dict(required=False, default=None)
)
argument_spec.update(meta_args)
spec = ArgumentSpec()
module = AnsibleModule(
argument_spec=argument_spec,
supports_check_mode=True
client = AnsibleF5Client(
argument_spec=spec.argument_spec,
supports_check_mode=spec.supports_check_mode,
f5_product_name=spec.f5_product_name
)
try:
obj = BigIpSysDb(check_mode=module.check_mode, **module.params)
result = obj.flush()
module.exit_json(**result)
mm = ModuleManager(client)
results = mm.exec_module()
client.module.exit_json(**results)
except F5ModuleError as e:
module.fail_json(msg=str(e))
client.module.fail_json(msg=str(e))
from ansible.module_utils.basic import *
from ansible.module_utils.f5_utils import *
if __name__ == '__main__':
main()

View file

@ -0,0 +1,134 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2017 F5 Networks Inc.
# GNU General Public License v3.0 (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import json
import sys
from nose.plugins.skip import SkipTest
if sys.version_info < (2, 7):
raise SkipTest("F5 Ansible modules require Python >= 2.7")
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import Mock
from ansible.compat.tests.mock import patch
from ansible.module_utils.f5_utils import AnsibleF5Client
try:
from library.bigip_sys_db import Parameters
from library.bigip_sys_db import ModuleManager
from library.bigip_sys_db import ArgumentSpec
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
from test.unit.modules.utils import set_module_args
except ImportError:
try:
from ansible.modules.network.f5.bigip_sys_db import Parameters
from ansible.modules.network.f5.bigip_sys_db import ModuleManager
from ansible.modules.network.f5.bigip_sys_db import ArgumentSpec
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
from units.modules.utils import set_module_args
except ImportError:
raise SkipTest("F5 Ansible modules require the f5-sdk Python library")
fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
fixture_data = {}
def load_fixture(name):
path = os.path.join(fixture_path, name)
if path in fixture_data:
return fixture_data[path]
with open(path) as f:
data = f.read()
try:
data = json.loads(data)
except Exception:
pass
fixture_data[path] = data
return data
class TestParameters(unittest.TestCase):
def test_module_parameters(self):
args = dict(
key='foo',
value='bar',
password='password',
server='localhost',
user='admin'
)
p = Parameters(args)
assert p.key == 'foo'
assert p.value == 'bar'
def test_api_parameters(self):
args = dict(
key='foo',
value='bar',
password='password',
server='localhost',
defaultValue='baz',
user='admin'
)
p = Parameters(args)
assert p.key == 'foo'
assert p.value == 'bar'
@patch('ansible.module_utils.f5_utils.AnsibleF5Client._get_mgmt_root',
return_value=True)
class TestManager(unittest.TestCase):
def setUp(self):
self.spec = ArgumentSpec()
def test_create_blackhole(self, *args):
set_module_args(dict(
key='provision.cpu.afm',
value='1',
password='admin',
server='localhost',
user='admin',
state='present'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(
dict(
kind="tm:sys:db:dbstate",
name="provision.cpu.afm",
fullPath="provision.cpu.afm",
generation=1,
selfLink="https://localhost/mgmt/tm/sys/db/provision.cpu.afm?ver=11.6.1",
defaultValue="0",
scfConfig="false",
value="0",
valueRange="integer min:0 max:100"
)
)
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.exists = Mock(return_value=False)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
results = mm.exec_module()
assert results['changed'] is True