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

New module: Add Pritunl VPN organization module (net_tools/pritunl/) (#804) (#2212)

Browse source 
(cherry picked from commit f0b7c6351e)

Co-authored-by: Florian Dambrine <Lowess@users.noreply.github.com>
This commit is contained in:
patchback[bot] 2021-04-09 05:35:07 +02:00 committed by GitHub
parent d5c8d7ddcc
commit ec78558559
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 1064 additions and 233 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
plugins/module_utils/net_tools/pritunl/api.py
View file

@ -57,6 +57,34 @@ def _get_pritunl_organizations(api_token, api_secret, base_url, validate_certs=T
)
def _delete_pritunl_organization(
api_token, api_secret, base_url, organization_id, validate_certs=True
):
return pritunl_auth_request(
base_url=base_url,
api_token=api_token,
api_secret=api_secret,
method="DELETE",
path="/organization/%s" % (organization_id),
validate_certs=validate_certs,
)
def _post_pritunl_organization(
api_token, api_secret, base_url, organization_data, validate_certs=True
):
return pritunl_auth_request(
api_token=api_token,
api_secret=api_secret,
base_url=base_url,
method="POST",
path="/organization/%s",
headers={"Content-Type": "application/json"},
data=json.dumps(organization_data),
validate_certs=validate_certs,
)
def _get_pritunl_users(
api_token, api_secret, base_url, organization_id, validate_certs=True
):
@ -179,6 +207,29 @@ def list_pritunl_users(
return users
def post_pritunl_organization(
api_token,
api_secret,
base_url,
organization_name,
validate_certs=True,
):
response = _post_pritunl_organization(
api_token=api_token,
api_secret=api_secret,
base_url=base_url,
organization_data={"name": organization_name},
validate_certs=True,
)
if response.getcode() != 200:
raise PritunlException(
"Could not add organization %s to Pritunl" % (organization_name)
)
# The user PUT request returns the updated user object
return json.loads(response.read())
def post_pritunl_user(
api_token,
api_secret,
@ -227,6 +278,25 @@ def post_pritunl_user(
return json.loads(response.read())
def delete_pritunl_organization(
api_token, api_secret, base_url, organization_id, validate_certs=True
):
response = _delete_pritunl_organization(
api_token=api_token,
api_secret=api_secret,
base_url=base_url,
organization_id=organization_id,
validate_certs=True,
)
if response.getcode() != 200:
raise PritunlException(
"Could not remove organization %s from Pritunl" % (organization_id)
)
return json.loads(response.read())
def delete_pritunl_user(
api_token, api_secret, base_url, organization_id, user_id, validate_certs=True
):

0
plugins/modules/net_tools/pritunl/__init__.py Normal file
View file

199
plugins/modules/net_tools/pritunl/pritunl_org.py Normal file
View file

@ -0,0 +1,199 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2021, Florian Dambrine <android.florian@gmail.com>
# 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
DOCUMENTATION = """
---
module: pritunl_org
author: Florian Dambrine (@Lowess)
version_added: 2.5.0
short_description: Manages Pritunl Organizations using the Pritunl API
description:
- A module to manage Pritunl organizations using the Pritunl API.
extends_documentation_fragment:
- community.general.pritunl
options:
name:
type: str
required: true
aliases:
- org
description:
- The name of the organization to manage in Pritunl.
force:
type: bool
default: false
description:
- If I(force) is C(true) and I(state) is C(absent), the module
will delete the organization, no matter if it contains users
or not. By default I(force) is C(false), which will cause the
module to fail the deletion of the organization when it contains
users.
state:
type: str
default: 'present'
choices:
- present
- absent
description:
- If C(present), the module adds organization I(name) to
Pritunl. If C(absent), attempt to delete the organization
from Pritunl (please read about I(force) usage).
"""
EXAMPLES = """
- name: Ensure the organization named MyOrg exists
community.general.pritunl_org:
state: present
name: MyOrg
- name: Ensure the organization named MyOrg does not exist
community.general.pritunl_org:
state: absent
name: MyOrg
"""
RETURN = """
response:
description: JSON representation of a Pritunl Organization.
returned: success
type: dict
sample:
{
"auth_api": False,
"name": "Foo",
"auth_token": None,
"user_count": 0,
"auth_secret": None,
"id": "csftwlu6uhralzi2dpmhekz3",
}
"""
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils._text import to_native
from ansible.module_utils.common.dict_transformations import dict_merge
from ansible_collections.community.general.plugins.module_utils.net_tools.pritunl.api import (
PritunlException,
delete_pritunl_organization,
post_pritunl_organization,
list_pritunl_organizations,
get_pritunl_settings,
pritunl_argument_spec,
)
def add_pritunl_organization(module):
result = {}
org_name = module.params.get("name")
org_obj_list = list_pritunl_organizations(
**dict_merge(
get_pritunl_settings(module),
{"filters": {"name": org_name}},
)
)
# If the organization already exists
if len(org_obj_list) > 0:
result["changed"] = False
result["response"] = org_obj_list[0]
else:
# Otherwise create it
response = post_pritunl_organization(
**dict_merge(
get_pritunl_settings(module),
{"organization_name": org_name},
)
)
result["changed"] = True
result["response"] = response
module.exit_json(**result)
def remove_pritunl_organization(module):
result = {}
org_name = module.params.get("name")
force = module.params.get("force")
org_obj_list = []
org_obj_list = list_pritunl_organizations(
**dict_merge(
get_pritunl_settings(module),
{
"filters": {"name": org_name},
},
)
)
# No organization found
if len(org_obj_list) == 0:
result["changed"] = False
result["response"] = {}
else:
# Otherwise attempt to delete it
org = org_obj_list[0]
# Only accept deletion under specific conditions
if force or org["user_count"] == 0:
response = delete_pritunl_organization(
**dict_merge(
get_pritunl_settings(module),
{"organization_id": org["id"]},
)
)
result["changed"] = True
result["response"] = response
else:
module.fail_json(
msg=(
"Can not remove organization '%s' with %d attached users. "
"Either set 'force' option to true or remove active users "
"from the organization"
)
% (org_name, org["user_count"])
)
module.exit_json(**result)
def main():
argument_spec = pritunl_argument_spec()
argument_spec.update(
dict(
name=dict(required=True, type="str", aliases=["org"]),
force=dict(required=False, type="bool", default=False),
state=dict(
required=False, choices=["present", "absent"], default="present"
),
)
),
module = AnsibleModule(argument_spec=argument_spec)
state = module.params.get("state")
try:
if state == "present":
add_pritunl_organization(module)
elif state == "absent":
remove_pritunl_organization(module)
except PritunlException as e:
module.fail_json(msg=to_native(e))
if __name__ == "__main__":
main()

129
plugins/modules/net_tools/pritunl/pritunl_org_info.py Normal file
View file

@ -0,0 +1,129 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2021, Florian Dambrine <android.florian@gmail.com>
# 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
DOCUMENTATION = """
---
module: pritunl_org_info
author: Florian Dambrine (@Lowess)
version_added: 2.5.0
short_description: List Pritunl Organizations using the Pritunl API
description:
- A module to list Pritunl organizations using the Pritunl API.
extends_documentation_fragment:
- community.general.pritunl
options:
organization:
type: str
required: false
aliases:
- org
default: null
description:
- Name of the Pritunl organization to search for.
If none provided, the module will return all Pritunl
organizations.
"""
EXAMPLES = """
- name: List all existing Pritunl organizations
community.general.pritunl_org_info:
- name: Search for an organization named MyOrg
community.general.pritunl_user_info:
organization: MyOrg
"""
RETURN = """
organizations:
description: List of Pritunl organizations.
returned: success
type: list
elements: dict
sample:
[
{
"auth_api": False,
"name": "FooOrg",
"auth_token": None,
"user_count": 0,
"auth_secret": None,
"id": "csftwlu6uhralzi2dpmhekz3",
},
{
"auth_api": False,
"name": "MyOrg",
"auth_token": None,
"user_count": 3,
"auth_secret": None,
"id": "58070daee63f3b2e6e472c36",
},
{
"auth_api": False,
"name": "BarOrg",
"auth_token": None,
"user_count": 0,
"auth_secret": None,
"id": "v1sncsxxybnsylc8gpqg85pg",
}
]
"""
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils._text import to_native
from ansible.module_utils.common.dict_transformations import dict_merge
from ansible_collections.community.general.plugins.module_utils.net_tools.pritunl.api import (
PritunlException,
get_pritunl_settings,
list_pritunl_organizations,
pritunl_argument_spec,
)
def get_pritunl_organizations(module):
org_name = module.params.get("organization")
organizations = []
organizations = list_pritunl_organizations(
**dict_merge(
get_pritunl_settings(module),
{"filters": {"name": org_name} if org_name else None},
)
)
if org_name and len(organizations) == 0:
# When an org_name is provided but no organization match return an error
module.fail_json(msg="Organization '%s' does not exist" % org_name)
result = {}
result["changed"] = False
result["organizations"] = organizations
module.exit_json(**result)
def main():
argument_spec = pritunl_argument_spec()
argument_spec.update(
dict(
organization=dict(required=False, type="str", default=None, aliases=["org"])
)
),
module = AnsibleModule(argument_spec=argument_spec, supports_check_mode=True)
try:
get_pritunl_organizations(module)
except PritunlException as e:
module.fail_json(msg=to_native(e))
if __name__ == "__main__":
main()

1
plugins/modules/pritunl_org.py Symbolic link
View file

@ -0,0 +1 @@
./net_tools/pritunl/pritunl_org.py

1
plugins/modules/pritunl_org_info.py Symbolic link
View file

@ -0,0 +1 @@
./net_tools/pritunl/pritunl_org_info.py

236
tests/unit/plugins/module_utils/net_tools/pritunl/test_api.py
View file

@ -9,7 +9,9 @@ import json
import pytest
from ansible.module_utils.common.dict_transformations import dict_merge
from ansible.module_utils.six import iteritems
from ansible_collections.community.general.plugins.module_utils.net_tools.pritunl import api
from ansible_collections.community.general.plugins.module_utils.net_tools.pritunl import (
api,
)
from mock import MagicMock
__metaclass__ = type
@ -17,16 +19,7 @@ __metaclass__ = type
# Pritunl Mocks
class PritunlListOrganizationMock(MagicMock):
"""Pritunl API Mock for organization GET API calls."""
def getcode(self):
return 200
def read(self):
return json.dumps(
[
PRITUNL_ORGS = [
{
"auth_api": False,
"name": "Foo",
@ -51,19 +44,18 @@ class PritunlListOrganizationMock(MagicMock):
"auth_secret": None,
"id": "v1sncsxxybnsylc8gpqg85pg",
},
]
)
]
NEW_PRITUNL_ORG = {
"auth_api": False,
"name": "NewOrg",
"auth_token": None,
"user_count": 0,
"auth_secret": None,
"id": "604a140ae63f3b36bc34c7bd",
}
class PritunlListUserMock(MagicMock):
"""Pritunl API Mock for user GET API calls."""
def getcode(self):
return 200
def read(self):
return json.dumps(
[
PRITUNL_USERS = [
{
"auth_type": "google",
"dns_servers": None,
@ -217,30 +209,9 @@ class PritunlListUserMock(MagicMock):
"otp_auth": True,
"organization": "58070daee63f3b2e6e472c36",
},
]
)
]
class PritunlErrorMock(MagicMock):
"""Pritunl API Mock for API call failures."""
def getcode(self):
return 500
def read(self):
return "{}"
class PritunlPostUserMock(MagicMock):
"""Pritunl API Mock for POST API calls."""
def getcode(self):
return 200
def read(self):
return json.dumps(
[
{
NEW_PRITUNL_USER = {
"auth_type": "local",
"disabled": False,
"dns_servers": None,
@ -257,9 +228,83 @@ class PritunlPostUserMock(MagicMock):
"port_forwarding": [],
"type": "client",
"id": "590add71e63f3b72d8bb951a",
}
]
)
}
NEW_PRITUNL_USER_UPDATED = dict_merge(
NEW_PRITUNL_USER,
{
"disabled": True,
"name": "bob",
"email": "bob@company.com",
"groups": ["c", "d"],
},
)
class PritunlEmptyOrganizationMock(MagicMock):
"""Pritunl API Mock for organization GET API calls."""
def getcode(self):
return 200
def read(self):
return json.dumps([])
class PritunlListOrganizationMock(MagicMock):
"""Pritunl API Mock for organization GET API calls."""
def getcode(self):
return 200
def read(self):
return json.dumps(PRITUNL_ORGS)
class PritunlListUserMock(MagicMock):
"""Pritunl API Mock for user GET API calls."""
def getcode(self):
return 200
def read(self):
return json.dumps(PRITUNL_USERS)
class PritunlErrorMock(MagicMock):
"""Pritunl API Mock for API call failures."""
def getcode(self):
return 500
def read(self):
return "{}"
class PritunlPostOrganizationMock(MagicMock):
def getcode(self):
return 200
def read(self):
return json.dumps(NEW_PRITUNL_ORG)
class PritunlListOrganizationAfterPostMock(MagicMock):
def getcode(self):
return 200
def read(self):
return json.dumps(PRITUNL_ORGS + [NEW_PRITUNL_ORG])
class PritunlPostUserMock(MagicMock):
"""Pritunl API Mock for POST API calls."""
def getcode(self):
return 200
def read(self):
return json.dumps([NEW_PRITUNL_USER])
class PritunlPutUserMock(MagicMock):
@ -269,26 +314,17 @@ class PritunlPutUserMock(MagicMock):
return 200
def read(self):
return json.dumps(
{
"auth_type": "local",
"disabled": True,
"dns_servers": None,
"otp_secret": "WEJANJYMF3Q2QSLG",
"name": "bob",
"pin": False,
"dns_suffix": False,
"client_to_client": False,
"email": "bob@company.com",
"organization_name": "GumGum",
"bypass_secondary": False,
"groups": ["c", "d"],
"organization": "58070daee63f3b2e6e472c36",
"port_forwarding": [],
"type": "client",
"id": "590add71e63f3b72d8bb951a",
}
)
return json.dumps(NEW_PRITUNL_USER_UPDATED)
class PritunlDeleteOrganizationMock(MagicMock):
"""Pritunl API Mock for DELETE API calls."""
def getcode(self):
return 200
def read(self):
return "{}"
class PritunlDeleteUserMock(MagicMock):
@ -321,14 +357,21 @@ def pritunl_settings():
}
@pytest.fixture
def pritunl_organization_data():
return {
"name": NEW_PRITUNL_ORG["name"],
}
@pytest.fixture
def pritunl_user_data():
return {
"name": "alice",
"email": "alice@company.com",
"groups": ["a", "b"],
"disabled": False,
"type": "client",
"name": NEW_PRITUNL_USER["name"],
"email": NEW_PRITUNL_USER["email"],
"groups": NEW_PRITUNL_USER["groups"],
"disabled": NEW_PRITUNL_USER["disabled"],
"type": NEW_PRITUNL_USER["type"],
}
@ -347,6 +390,11 @@ def get_pritunl_error_mock():
return PritunlErrorMock()
@pytest.fixture
def post_pritunl_organization_mock():
return PritunlPostOrganizationMock()
@pytest.fixture
def post_pritunl_user_mock():
return PritunlPostUserMock()
@ -357,6 +405,11 @@ def put_pritunl_user_mock():
return PritunlPutUserMock()
@pytest.fixture
def delete_pritunl_organization_mock():
return PritunlDeleteOrganizationMock()
@pytest.fixture
def delete_pritunl_user_mock():
return PritunlDeleteUserMock()
@ -460,6 +513,25 @@ class TestPritunlApi:
assert user["name"] == user_expected
# Test for POST operation on Pritunl API
def test_add_pritunl_organization(
self,
pritunl_settings,
pritunl_organization_data,
post_pritunl_organization_mock,
):
api._post_pritunl_organization = post_pritunl_organization_mock()
create_response = api.post_pritunl_organization(
**dict_merge(
pritunl_settings,
{"organization_name": pritunl_organization_data["name"]},
)
)
# Ensure provided settings match with the ones returned by Pritunl
for k, v in iteritems(pritunl_organization_data):
assert create_response[k] == v
@pytest.mark.parametrize("org_id", [("58070daee63f3b2e6e472c36")])
def test_add_and_update_pritunl_user(
self,
@ -513,6 +585,24 @@ class TestPritunlApi:
assert update_response[k] == create_response[k]
# Test for DELETE operation on Pritunl API
@pytest.mark.parametrize("org_id", [("58070daee63f3b2e6e472c36")])
def test_delete_pritunl_organization(
self, pritunl_settings, org_id, delete_pritunl_organization_mock
):
api._delete_pritunl_organization = delete_pritunl_organization_mock()
response = api.delete_pritunl_organization(
**dict_merge(
pritunl_settings,
{
"organization_id": org_id,
},
)
)
assert response == {}
@pytest.mark.parametrize(
"org_id,user_id", [("58070daee63f3b2e6e472c36", "590add71e63f3b72d8bb951a")]
)

204
tests/unit/plugins/modules/net_tools/pritunl/test_pritunl_org.py Normal file
View file

@ -0,0 +1,204 @@
# -*- coding: utf-8 -*-
# (c) 2021 Florian Dambrine <android.florian@gmail.com>
# 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
import sys
from ansible.module_utils.common.dict_transformations import dict_merge
from ansible.module_utils.six import iteritems
from ansible_collections.community.general.plugins.modules.net_tools.pritunl import (
pritunl_org,
)
from ansible_collections.community.general.tests.unit.compat.mock import patch
from ansible_collections.community.general.tests.unit.plugins.module_utils.net_tools.pritunl.test_api import (
PritunlDeleteOrganizationMock,
PritunlListOrganizationMock,
PritunlListOrganizationAfterPostMock,
PritunlPostOrganizationMock,
)
from ansible_collections.community.general.tests.unit.plugins.modules.utils import (
AnsibleExitJson,
AnsibleFailJson,
ModuleTestCase,
set_module_args,
)
__metaclass__ = type
class TestPritunlOrg(ModuleTestCase):
def setUp(self):
super(TestPritunlOrg, self).setUp()
self.module = pritunl_org
# Add backward compatibility
if sys.version_info < (3, 2):
self.assertRegex = self.assertRegexpMatches
def tearDown(self):
super(TestPritunlOrg, self).tearDown()
def patch_add_pritunl_organization(self, **kwds):
return patch(
"ansible_collections.community.general.plugins.module_utils.net_tools.pritunl.api._post_pritunl_organization",
autospec=True,
**kwds
)
def patch_delete_pritunl_organization(self, **kwds):
return patch(
"ansible_collections.community.general.plugins.module_utils.net_tools.pritunl.api._delete_pritunl_organization",
autospec=True,
**kwds
)
def patch_get_pritunl_organizations(self, **kwds):
return patch(
"ansible_collections.community.general.plugins.module_utils.net_tools.pritunl.api._get_pritunl_organizations",
autospec=True,
**kwds
)
def test_without_parameters(self):
"""Test without parameters"""
set_module_args({})
with self.assertRaises(AnsibleFailJson):
self.module.main()
def test_present(self):
"""Test Pritunl organization creation."""
org_params = {"name": "NewOrg"}
set_module_args(
dict_merge(
{
"pritunl_api_token": "token",
"pritunl_api_secret": "secret",
"pritunl_url": "https://pritunl.domain.com",
},
org_params,
)
)
# Test creation
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationMock
) as mock_get:
with self.patch_add_pritunl_organization(
side_effect=PritunlPostOrganizationMock
) as mock_add:
with self.assertRaises(AnsibleExitJson) as create_result:
self.module.main()
create_exc = create_result.exception.args[0]
self.assertTrue(create_exc["changed"])
self.assertEqual(create_exc["response"]["name"], org_params["name"])
self.assertEqual(create_exc["response"]["user_count"], 0)
# Test module idempotency
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationAfterPostMock
) as mock_get:
with self.patch_add_pritunl_organization(
side_effect=PritunlPostOrganizationMock
) as mock_add:
with self.assertRaises(AnsibleExitJson) as idempotent_result:
self.module.main()
idempotent_exc = idempotent_result.exception.args[0]
# Ensure both calls resulted in the same returned value
# except for changed which sould be false the second time
for k, v in iteritems(idempotent_exc):
if k == "changed":
self.assertFalse(idempotent_exc[k])
else:
self.assertEqual(create_exc[k], idempotent_exc[k])
def test_absent(self):
"""Test organization removal from Pritunl."""
org_params = {"name": "NewOrg"}
set_module_args(
dict_merge(
{
"state": "absent",
"pritunl_api_token": "token",
"pritunl_api_secret": "secret",
"pritunl_url": "https://pritunl.domain.com",
},
org_params,
)
)
# Test deletion
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationAfterPostMock
) as mock_get:
with self.patch_delete_pritunl_organization(
side_effect=PritunlDeleteOrganizationMock
) as mock_delete:
with self.assertRaises(AnsibleExitJson) as delete_result:
self.module.main()
delete_exc = delete_result.exception.args[0]
self.assertTrue(delete_exc["changed"])
self.assertEqual(delete_exc["response"], {})
# Test module idempotency
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationMock
) as mock_get:
with self.patch_delete_pritunl_organization(
side_effect=PritunlDeleteOrganizationMock
) as mock_add:
with self.assertRaises(AnsibleExitJson) as idempotent_result:
self.module.main()
idempotent_exc = idempotent_result.exception.args[0]
# Ensure both calls resulted in the same returned value
# except for changed which sould be false the second time
self.assertFalse(idempotent_exc["changed"])
self.assertEqual(idempotent_exc["response"], delete_exc["response"])
def test_absent_with_existing_users(self):
"""Test organization removal with attached users should fail except if force is true."""
module_args = {
"state": "absent",
"pritunl_api_token": "token",
"pritunl_api_secret": "secret",
"pritunl_url": "https://pritunl.domain.com",
"name": "GumGum",
}
set_module_args(module_args)
# Test deletion
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationMock
) as mock_get:
with self.patch_delete_pritunl_organization(
side_effect=PritunlDeleteOrganizationMock
) as mock_delete:
with self.assertRaises(AnsibleFailJson) as failure_result:
self.module.main()
failure_exc = failure_result.exception.args[0]
self.assertRegex(failure_exc["msg"], "Can not remove organization")
# Switch force=True which should run successfully
set_module_args(dict_merge(module_args, {"force": True}))
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationMock
) as mock_get:
with self.patch_delete_pritunl_organization(
side_effect=PritunlDeleteOrganizationMock
) as mock_delete:
with self.assertRaises(AnsibleExitJson) as delete_result:
self.module.main()
delete_exc = delete_result.exception.args[0]
self.assertTrue(delete_exc["changed"])

137
tests/unit/plugins/modules/net_tools/pritunl/test_pritunl_org_info.py Normal file
View file

@ -0,0 +1,137 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2021, Florian Dambrine <android.florian@gmail.com>
# 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
import sys
from ansible_collections.community.general.plugins.modules.net_tools.pritunl import (
pritunl_org_info,
)
from ansible_collections.community.general.tests.unit.compat.mock import patch
from ansible_collections.community.general.tests.unit.plugins.module_utils.net_tools.pritunl.test_api import (
PritunlListOrganizationMock,
PritunlEmptyOrganizationMock,
)
from ansible_collections.community.general.tests.unit.plugins.modules.utils import (
AnsibleExitJson,
AnsibleFailJson,
ModuleTestCase,
set_module_args,
)
__metaclass__ = type
class TestPritunlOrgInfo(ModuleTestCase):
def setUp(self):
super(TestPritunlOrgInfo, self).setUp()
self.module = pritunl_org_info
# Add backward compatibility
if sys.version_info < (3, 2):
self.assertRegex = self.assertRegexpMatches
def tearDown(self):
super(TestPritunlOrgInfo, self).tearDown()
def patch_get_pritunl_organizations(self, **kwds):
return patch(
"ansible_collections.community.general.plugins.module_utils.net_tools.pritunl.api._get_pritunl_organizations",
autospec=True,
**kwds
)
def test_without_parameters(self):
"""Test without parameters"""
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationMock
) as org_mock:
set_module_args({})
with self.assertRaises(AnsibleFailJson):
self.module.main()
self.assertEqual(org_mock.call_count, 0)
def test_list_empty_organizations(self):
"""Listing all organizations even when no org exists should be valid."""
with self.patch_get_pritunl_organizations(
side_effect=PritunlEmptyOrganizationMock
) as org_mock:
with self.assertRaises(AnsibleExitJson) as result:
set_module_args(
{
"pritunl_api_token": "token",
"pritunl_api_secret": "secret",
"pritunl_url": "https://pritunl.domain.com",
}
)
self.module.main()
self.assertEqual(org_mock.call_count, 1)
exc = result.exception.args[0]
self.assertEqual(len(exc["organizations"]), 0)
def test_list_specific_organization(self):
"""Listing a specific organization should be valid."""
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationMock
) as org_mock:
with self.assertRaises(AnsibleExitJson) as result:
set_module_args(
{
"pritunl_api_token": "token",
"pritunl_api_secret": "secret",
"pritunl_url": "https://pritunl.domain.com",
"org": "GumGum",
}
)
self.module.main()
self.assertEqual(org_mock.call_count, 1)
exc = result.exception.args[0]
self.assertEqual(len(exc["organizations"]), 1)
def test_list_unknown_organization(self):
"""Listing an unknown organization should result in a failure."""
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationMock
) as org_mock:
with self.assertRaises(AnsibleFailJson) as result:
set_module_args(
{
"pritunl_api_token": "token",
"pritunl_api_secret": "secret",
"pritunl_url": "https://pritunl.domain.com",
"org": "Unknown",
}
)
self.module.main()
self.assertEqual(org_mock.call_count, 1)
exc = result.exception.args[0]
self.assertRegex(exc["msg"], "does not exist")
def test_list_all_organizations(self):
"""Listing all organizations should be valid."""
with self.patch_get_pritunl_organizations(
side_effect=PritunlListOrganizationMock
) as org_mock:
with self.assertRaises(AnsibleExitJson) as result:
set_module_args(
{
"pritunl_api_token": "token",
"pritunl_api_secret": "secret",
"pritunl_url": "https://pritunl.domain.com",
}
)
self.module.main()
self.assertEqual(org_mock.call_count, 1)
exc = result.exception.args[0]
self.assertEqual(len(exc["organizations"]), 3)