Unit tests for openvswitch_db module (#23974)

pull/24055/head
Ricardo Carrillo Cruz 8 years ago committed by GitHub
parent 09b8badc5e
commit a3a970180a

@ -129,7 +129,6 @@ def map_config_to_obj(module):
templatized_command = "%(ovs-vsctl)s -t %(timeout)s list %(table)s %(record)s" templatized_command = "%(ovs-vsctl)s -t %(timeout)s list %(table)s %(record)s"
command = templatized_command % module.params command = templatized_command % module.params
rc, out, err = module.run_command(command, check_rc=True) rc, out, err = module.run_command(command, check_rc=True)
if rc != 0: if rc != 0:
module.fail_json(msg=err) module.fail_json(msg=err)
@ -137,7 +136,7 @@ def map_config_to_obj(module):
col_value = match.group(3) col_value = match.group(3)
col_value_to_dict = {} col_value_to_dict = {}
if match.group(3): if col_value and col_value != '{}':
for kv in col_value[1:-1].split(','): for kv in col_value[1:-1].split(','):
k, v = kv.split('=') k, v = kv.split('=')
col_value_to_dict[k.strip()] = v.strip() col_value_to_dict[k.strip()] = v.strip()

@ -0,0 +1,23 @@
_uuid : 64f04422-d510-4258-9648-ee0a982f58c1
auto_attach : []
controller : []
datapath_id : "00002244f0645842"
datapath_type : ""
datapath_version : "<unknown>"
external_ids : {}
fail_mode : []
flood_vlans : []
flow_tables : {}
ipfix : []
mcast_snooping_enable: false
mirrors : []
name : test-br
netflow : []
other_config : {}
ports : [2c9c1b35-a304-4dee-bb7a-092d656543c7]
protocols : []
rstp_enable : false
rstp_status : {}
sflow : []
status : {}
stp_enable : false

@ -0,0 +1,23 @@
_uuid : 64f04422-d510-4258-9648-ee0a982f58c1
auto_attach : []
controller : []
datapath_id : "00002244f0645842"
datapath_type : ""
datapath_version : "<unknown>"
external_ids : {}
fail_mode : []
flood_vlans : []
flow_tables : {}
ipfix : []
mcast_snooping_enable: false
mirrors : []
name : test-br
netflow : []
other_config : {disable-in-band=True}
ports : [2c9c1b35-a304-4dee-bb7a-092d656543c7]
protocols : []
rstp_enable : false
rstp_status : {}
sflow : []
status : {}
stp_enable : false

@ -0,0 +1,112 @@
# (c) 2017 Red Hat Inc.
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import json
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import patch
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
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:
pass
fixture_data[path] = data
return data
class AnsibleExitJson(Exception):
pass
class AnsibleFailJson(Exception):
pass
class TestOpenVSwitchModule(unittest.TestCase):
def execute_module(self, failed=False, changed=False,
command=None, fixture_name=None):
self.load_fixtures(fixture_name)
if failed:
result = self.failed()
self.assertTrue(result['failed'], result)
else:
result = self.changed(changed)
self.assertEqual(result['changed'], changed, result)
if command:
self.assertEqual(command, result['command'], result['command'])
return result
def failed(self):
def fail_json(*args, **kwargs):
kwargs['failed'] = True
raise AnsibleFailJson(kwargs)
with patch.object(basic.AnsibleModule, 'fail_json', fail_json):
with self.assertRaises(AnsibleFailJson) as exc:
self.module.main()
result = exc.exception.args[0]
self.assertTrue(result['failed'], result)
return result
def changed(self, changed=False):
def exit_json(*args, **kwargs):
if 'changed' not in kwargs:
kwargs['changed'] = False
raise AnsibleExitJson(kwargs)
with patch.object(basic.AnsibleModule, 'exit_json', exit_json):
with self.assertRaises(AnsibleExitJson) as exc:
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], changed, result)
return result
def load_fixtures(self, fixture_name):
pass

@ -0,0 +1,97 @@
#
# (c) 2016 Red Hat Inc.
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import json
from ansible.compat.tests.mock import patch
from ansible.modules.network.ovs import openvswitch_db
from .ovs_module import TestOpenVSwitchModule, load_fixture, set_module_args
class TestOpenVSwitchDBModule(TestOpenVSwitchModule):
module = openvswitch_db
def setUp(self):
self.mock_run_command = (
patch('ansible.module_utils.basic.AnsibleModule.run_command'))
self.run_command = self.mock_run_command.start()
self.mock_get_bin_path = (
patch('ansible.module_utils.basic.AnsibleModule.get_bin_path'))
self.get_bin_path = self.mock_get_bin_path.start()
def tearDown(self):
self.mock_run_command.stop()
self.mock_get_bin_path.stop()
def load_fixtures(self, fixture_name):
self.run_command.side_effect = [
(0, load_fixture(fixture_name + '.cfg'), None),
(0, None, None)]
self.get_bin_path.return_value = '/usr/bin/ovs-vsctl'
def test_openvswitch_db_absent_idempotent(self):
set_module_args(dict(state='absent',
table='Bridge', record='test-br',
col='other_config', key='disable-in-band',
value='True'))
self.execute_module(fixture_name='openvswitch_db_disable_in_band_missing')
def test_openvswitch_db_absent_removes_key(self):
set_module_args(dict(state='absent',
table='Bridge', record='test-br',
col='other_config', key='disable-in-band',
value='True'))
self.execute_module(
changed=True,
command='/usr/bin/ovs-vsctl -t 5 remove Bridge test-br other_config'
' disable-in-band=True',
fixture_name='openvswitch_db_disable_in_band_true')
def test_openvswitch_db_present_idempotent(self):
set_module_args(dict(state='present',
table='Bridge', record='test-br',
col='other_config', key='disable-in-band',
value='True'))
self.execute_module(fixture_name='openvswitch_db_disable_in_band_true')
def test_openvswitch_db_present_adds_key(self):
set_module_args(dict(state='present',
table='Bridge', record='test-br',
col='other_config', key='disable-in-band',
value='True'))
self.execute_module(
changed=True,
command='/usr/bin/ovs-vsctl -t 5 add Bridge test-br other_config'
' disable-in-band=True',
fixture_name='openvswitch_db_disable_in_band_missing')
def test_openvswitch_db_present_updates_key(self):
set_module_args(dict(state='present',
table='Bridge', record='test-br',
col='other_config', key='disable-in-band',
value='False'))
self.execute_module(
changed=True,
command='/usr/bin/ovs-vsctl -t 5 set Bridge test-br other_config'
':disable-in-band=False',
fixture_name='openvswitch_db_disable_in_band_true')
Loading…
Cancel
Save