ansible-collections.communi.../tests/unit/plugins/modules/test_api.py
Felix Fontein 30a79061f3
Use shared unit test utils from community.internal_test_tools (#353)
* Use shared unit test utils from community.internal_test_tools.

* Make sure community.internal_test_tools is installed in CI.
2025-03-12 21:34:12 +01:00

309 lines
13 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible_collections.community.internal_test_tools.tests.unit.compat.mock import patch, MagicMock
from ansible_collections.community.internal_test_tools.tests.unit.plugins.modules.utils import set_module_args, AnsibleExitJson, AnsibleFailJson, ModuleTestCase
from ansible_collections.community.routeros.tests.unit.plugins.modules.fake_api import FakeLibRouterosError, Key, Or, fake_ros_api
from ansible_collections.community.routeros.plugins.modules import api
class TestRouterosApiModule(ModuleTestCase):
def setUp(self):
super(TestRouterosApiModule, self).setUp()
self.module = api
self.module.LibRouterosError = FakeLibRouterosError
self.module.connect = MagicMock(new=fake_ros_api)
self.module.check_has_library = MagicMock()
self.patch_create_api = patch('ansible_collections.community.routeros.plugins.modules.api.create_api', MagicMock(new=fake_ros_api))
self.patch_create_api.start()
self.module.Key = MagicMock(new=Key)
self.module.Or = MagicMock(new=Or)
self.config_module_args = {"username": "admin",
"password": "pаss",
"hostname": "127.0.0.1",
"path": "interface bridge"}
def tearDown(self):
self.patch_create_api.stop()
def test_module_fail_when_required_args_missing(self):
with self.assertRaises(AnsibleFailJson) as exc:
with set_module_args({}):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['failed'], True)
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api.path)
def test_api_path(self):
with self.assertRaises(AnsibleExitJson) as exc:
with set_module_args(self.config_module_args.copy()):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_add(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['add'] = "name=unit_test_brige"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], True)
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_add_already_exist(self):
with self.assertRaises(AnsibleFailJson) as exc:
module_args = self.config_module_args.copy()
module_args['add'] = "name=unit_test_brige_exist"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['failed'], True)
self.assertEqual(result['msg'][0], 'failure: already have interface with such name')
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_remove(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['remove'] = "*A1"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], True)
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_remove_no_id(self):
with self.assertRaises(AnsibleFailJson) as exc:
module_args = self.config_module_args.copy()
module_args['remove'] = "*A2"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['failed'], True)
self.assertEqual(result['msg'][0], 'no such item (4)')
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api.arbitrary)
def test_api_cmd(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['cmd'] = "add name=unit_test_brige_arbitrary"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api.arbitrary)
def test_api_cmd_none_existing_cmd(self):
with self.assertRaises(AnsibleFailJson) as exc:
module_args = self.config_module_args.copy()
module_args['cmd'] = "add NONE_EXIST=unit_test_brige_arbitrary"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['failed'], True)
self.assertEqual(result['msg'][0], 'no such command')
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_update(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['update'] = ".id=*A1 name=unit_test_brige"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], True)
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_update_none_existing_id(self):
with self.assertRaises(AnsibleFailJson) as exc:
module_args = self.config_module_args.copy()
module_args['update'] = ".id=*A2 name=unit_test_brige"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['failed'], True)
self.assertEqual(result['msg'][0], 'no such item (4)')
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_query(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['query'] = ".id name"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], [
{'.id': '*A1', 'name': 'dummy_bridge_A1'},
{'.id': '*A2', 'name': 'dummy_bridge_A2'},
{'.id': '*A3', 'name': 'dummy_bridge_A3'},
])
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_query_missing_key(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['query'] = ".id other"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], ["no results for 'interface bridge 'query' .id other"])
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api.select_where)
def test_api_query_and_WHERE(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['query'] = ".id name WHERE name == dummy_bridge_A2"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], [
{'.id': '*A1', 'name': 'dummy_bridge_A1'},
])
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api.select_where)
def test_api_query_and_WHERE_no_cond(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['query'] = ".id name WHERE name != dummy_bridge_A2"
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], [
{'.id': '*A1', 'name': 'dummy_bridge_A1'},
])
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_extended_query(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['extended_query'] = {
'attributes': ['.id', 'name'],
}
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], [
{'.id': '*A1', 'name': 'dummy_bridge_A1'},
{'.id': '*A2', 'name': 'dummy_bridge_A2'},
{'.id': '*A3', 'name': 'dummy_bridge_A3'},
])
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api)
def test_api_extended_query_missing_key(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['extended_query'] = {
'attributes': ['.id', 'other'],
}
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], [])
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api.select_where)
def test_api_extended_query_and_WHERE(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['extended_query'] = {
'attributes': ['.id', 'name'],
'where': [
{
'attribute': 'name',
'is': '==',
'value': 'dummy_bridge_A2',
},
],
}
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], [
{'.id': '*A1', 'name': 'dummy_bridge_A1'},
])
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api.select_where)
def test_api_extended_query_and_WHERE_no_cond(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['extended_query'] = {
'attributes': ['.id', 'name'],
'where': [
{
'attribute': 'name',
'is': 'not',
'value': 'dummy_bridge_A2',
},
],
}
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], [
{'.id': '*A1', 'name': 'dummy_bridge_A1'},
])
@patch('ansible_collections.community.routeros.plugins.modules.api.ROS_api_module.api_add_path', new=fake_ros_api.select_where)
def test_api_extended_query_and_WHERE_or(self):
with self.assertRaises(AnsibleExitJson) as exc:
module_args = self.config_module_args.copy()
module_args['extended_query'] = {
'attributes': ['.id', 'name'],
'where': [
{
'or': [
{
'attribute': 'name',
'is': 'in',
'value': [1, 2],
},
{
'attribute': 'name',
'is': '!=',
'value': 5,
},
],
},
],
}
with set_module_args(module_args):
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], False)
self.assertEqual(result['msg'], [
{'.id': '*A1', 'name': 'dummy_bridge_A1'},
])