diff options
Diffstat (limited to 'ansible_collections/community/digitalocean/tests/unit/plugins')
3 files changed, 685 insertions, 0 deletions
diff --git a/ansible_collections/community/digitalocean/tests/unit/plugins/inventory/test_digitalocean.py b/ansible_collections/community/digitalocean/tests/unit/plugins/inventory/test_digitalocean.py new file mode 100644 index 000000000..340f36109 --- /dev/null +++ b/ansible_collections/community/digitalocean/tests/unit/plugins/inventory/test_digitalocean.py @@ -0,0 +1,306 @@ +# Copyright (c) 2021 Ansible Project +# GNGeneral 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 pytest + +from ansible.errors import AnsibleError, AnsibleParserError +from ansible.inventory.data import InventoryData +from ansible.template import Templar +from ansible.parsing.dataloader import DataLoader +import ansible_collections.community.digitalocean.plugins.inventory.digitalocean as module_under_test +from ansible_collections.community.digitalocean.plugins.inventory.digitalocean import ( + InventoryModule, +) + + +@pytest.fixture() +def inventory(): + r = InventoryModule() + r.inventory = InventoryData() + r.templar = Templar(loader=DataLoader()) + return r + + +def test_verify_file_bad_config(inventory): + assert inventory.verify_file("digitalocean_foobar.yml") is False + + +@pytest.fixture() +def payload(): + return [ + { + "id": 3164444, + "name": "foo", + "memory": 1024, + "vcpus": 1, + "disk": 25, + "locked": False, + "status": "active", + "kernel": { + "id": 2233, + "name": "Ubuntu 14.04 x64 vmlinuz-3.13.0-37-generic", + "version": "3.13.0-37-generic", + }, + "image": { + "id": 6918990, + "name": "14.04 x64", + "distribution": "Ubuntu", + "slug": "ubuntu-16-04-x64", + }, + "size_slug": "s-1vcpu-1gb", + "networks": { + "v4": [ + { + "ip_address": "104.236.32.182", + "netmask": "255.255.192.0", + "gateway": "104.236.0.1", + "type": "public", + } + ], + "v6": [ + { + "ip_address": "2604:A880:0800:0010:0000:0000:02DD:4001", + "netmask": 64, + "gateway": "2604:A880:0800:0010:0000:0000:0000:0001", + "type": "public", + } + ], + }, + "region": { + "name": "New York 3", + "slug": "nyc3", + }, + "tags": [], + "vpc_uuid": "f9b0769c-e118-42fb-a0c4-fed15ef69662", + }, + { + "id": 3164445, + "name": "bar", + "memory": 1024, + "vcpus": 1, + "disk": 25, + "locked": False, + "status": "active", + "kernel": { + "id": 2233, + "name": "Ubuntu 14.04 x64 vmlinuz-3.13.0-37-generic", + "version": "3.13.0-37-generic", + }, + "image": { + "id": 6918990, + "name": "14.04 x64", + "distribution": "Ubuntu", + "slug": "ubuntu-16-04-x64", + }, + "size_slug": "s-1vcpu-1gb", + "networks": { + "v4": [ + { + "ip_address": "104.236.32.185", + "netmask": "255.255.192.0", + "gateway": "104.236.0.1", + "type": "public", + } + ], + "v6": [ + { + "ip_address": "2604:A880:0800:0010:0000:0000:02DD:4004", + "netmask": 64, + "gateway": "2604:A880:0800:0010:0000:0000:0000:0001", + "type": "public", + } + ], + }, + "region": { + "name": "Frankfurt 1", + "slug": "fra1", + }, + "tags": [], + "vpc_uuid": "f9b0769c-e118-42fb-a0c4-fed15ef69662", + }, + ] + + +def get_option(option): + options = { + "attributes": ["id", "size_slug"], + "var_prefix": "do_", + "strict": False, + } + return options.get(option) + + +def test_populate_hostvars(inventory, payload, mocker): + inventory.get_option = mocker.MagicMock(side_effect=get_option) + + inventory._populate(payload) + + host_foo = inventory.inventory.get_host("foo") + host_bar = inventory.inventory.get_host("bar") + + assert host_foo.vars["do_id"] == 3164444 + assert host_bar.vars["do_size_slug"] == "s-1vcpu-1gb" + + # if a prefix is set, unprefixed attributes should not appear in host vars + assert "id" not in host_foo.vars + assert "size_slug" not in host_bar.vars + + +@pytest.mark.parametrize("transform", ["never", "ignore"]) +def test_populate_groups_no_sanitization(inventory, mocker, transform): + def get_option(opt): + return dict( + attributes=["id", "tags"], + var_prefix="do_", + keyed_groups=[dict(key="do_tags", prefix="", separator="")], + ).get(opt) + + inventory.get_option = mocker.MagicMock(side_effect=get_option) + mocker.patch("ansible.constants.TRANSFORM_INVALID_GROUP_CHARS", transform) + + inventory._populate( + [ + dict( + id=3164444, + name="test", + tags=["lower", "UPPER", "un_der", "col:on", "da-sh", "with_123"], + ), + ] + ) + + assert set( + ("all", "ungrouped", "lower", "UPPER", "un_der", "col:on", "da-sh", "with_123") + ) == set((inventory.inventory.groups.keys())) + + +@pytest.mark.parametrize("transform", ["always", "silently"]) +def test_populate_groups_sanitization(inventory, mocker, transform): + def get_option(opt): + return dict( + attributes=["id", "tags"], + var_prefix="x_", + keyed_groups=[dict(key="x_tags", prefix="", separator="")], + ).get(opt) + + inventory.get_option = mocker.MagicMock(side_effect=get_option) + mocker.patch("ansible.constants.TRANSFORM_INVALID_GROUP_CHARS", transform) + + inventory._populate( + [ + dict( + id=3164444, + name="test", + tags=["lower", "UPPER", "un_der", "col:on", "da-sh", "with_123"], + ), + ] + ) + + assert set( + ("all", "ungrouped", "lower", "UPPER", "un_der", "col_on", "da_sh", "with_123") + ) == set((inventory.inventory.groups.keys())) + + +def get_option_with_templated_api_token(option): + options = { + # "random_choice" with just a single input always returns the same result. + "api_token": '{{ lookup("random_choice", "my-do-token") }}', + "pagination": 100, + } + return options.get(option) + + +def test_get_payload_with_templated_api_token(inventory, mocker): + inventory.get_option = mocker.MagicMock( + side_effect=get_option_with_templated_api_token + ) + + mocker.patch(module_under_test.__name__ + ".Request") + RequestMock = module_under_test.Request + + req_instance = RequestMock.return_value + req_instance.get.return_value.read.return_value = '{"droplets": []}' + + inventory._get_payload() + + init_headers = RequestMock.call_args.kwargs["headers"] + assert init_headers["Authorization"] == "Bearer my-do-token" + + +def get_option_with_filters(option): + options = { + "attributes": ["id", "size_slug", "region"], + "var_prefix": "do_", + "strict": False, + "filters": [ + 'do_region.slug == "fra1"', + ], + } + return options.get(option) + + +def test_populate_hostvars_with_filters(inventory, payload, mocker): + inventory.get_option = mocker.MagicMock(side_effect=get_option_with_filters) + inventory._populate(payload) + + host_foo = inventory.inventory.get_host("foo") + host_bar = inventory.inventory.get_host("bar") + + assert host_foo is None + assert host_bar.vars["do_size_slug"] == "s-1vcpu-1gb" + + +def get_variables(): + return { + "do_region": { + "slug": "fra1", + }, + "do_tags": ["something"], + } + + +def test_passes_filters_accept_empty(inventory, mocker): + filters = [] + variables = get_variables() + assert inventory._passes_filters(filters, variables, "foo") + + +def test_passes_filters_accept(inventory, mocker): + filters = ['do_region.slug == "fra1"'] + variables = get_variables() + assert inventory._passes_filters(filters, variables, "foo") + + +def test_passes_filters_reject(inventory, mocker): + filters = ['do_region.slug == "nyc3"'] + variables = get_variables() + assert not inventory._passes_filters(filters, variables, "foo") + + +def test_passes_filters_reject_any(inventory, mocker): + filters = [ + 'do_region.slug == "fra1"', # accept + '"nope" in do_tags', # reject + ] + variables = get_variables() + assert not inventory._passes_filters(filters, variables, "foo") + + +def test_passes_filters_invalid_filters(inventory, mocker): + filters = ["not a valid filter"] + variables = get_variables() + assert not inventory._passes_filters(filters, variables, "foo") + + +def test_passes_filters_invalid_filters_strict(inventory, mocker): + filters = ["not a valid filter"] + variables = get_variables() + try: + inventory._passes_filters(filters, variables, "foo", True) + assert False, "expected _passes_filters() to raise AnsibleError" + except AnsibleError as e: + pass diff --git a/ansible_collections/community/digitalocean/tests/unit/plugins/modules/test_digital_ocean_kubernetes.py b/ansible_collections/community/digitalocean/tests/unit/plugins/modules/test_digital_ocean_kubernetes.py new file mode 100644 index 000000000..cfb0f59b6 --- /dev/null +++ b/ansible_collections/community/digitalocean/tests/unit/plugins/modules/test_digital_ocean_kubernetes.py @@ -0,0 +1,263 @@ +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +from ansible_collections.community.general.tests.unit.compat import unittest +from ansible_collections.community.general.tests.unit.compat.mock import MagicMock +from ansible_collections.community.digitalocean.plugins.modules.digital_ocean_kubernetes import ( + DOKubernetes, +) + + +class TestDOKubernetes(unittest.TestCase): + def test_get_by_id_when_ok(self): + module = MagicMock() + k = DOKubernetes(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 200 + k.rest.get.return_value.json = {"foo": "bar"} + self.assertEqual(k.get_by_id(), {"foo": "bar"}) + + def test_get_by_id_when_not_ok(self): + module = MagicMock() + k = DOKubernetes(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 400 + k.rest.get.return_value.json = {"foo": "bar"} + self.assertIsNone(k.get_by_id()) + + def test_get_all_clusters_when_ok(self): + module = MagicMock() + k = DOKubernetes(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 200 + k.rest.get.return_value.json = {"foo": "bar"} + self.assertEqual(k.get_all_clusters(), {"foo": "bar"}) + + def test_get_all_clusters_when_not_ok(self): + module = MagicMock() + k = DOKubernetes(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 400 + k.rest.get.return_value.json = {"foo": "bar"} + self.assertIsNone(k.get_all_clusters()) + + def test_get_by_name_none(self): + module = MagicMock() + k = DOKubernetes(module) + self.assertIsNone(k.get_by_name(None)) + + def test_get_by_name_found(self): + module = MagicMock() + k = DOKubernetes(module) + k.get_all_clusters = MagicMock() + k.get_all_clusters.return_value = {"kubernetes_clusters": [{"name": "foo"}]} + self.assertEqual(k.get_by_name("foo"), {"name": "foo"}) + + def test_get_by_name_not_found(self): + module = MagicMock() + k = DOKubernetes(module) + k.get_all_clusters = MagicMock() + k.get_all_clusters.return_value = {"kubernetes_clusters": [{"name": "foo"}]} + self.assertIsNone(k.get_by_name("foo2")) + + def test_get_kubernetes_kubeconfig_when_ok(self): + module = MagicMock() + k = DOKubernetes(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 200 + k.rest.get.return_value.body = "kubeconfig" + self.assertEqual(k.get_kubernetes_kubeconfig(), "kubeconfig") + + def test_get_kubernetes_kubeconfig_when_not_ok(self): + module = MagicMock() + k = DOKubernetes(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 400 + k.rest.get.return_value.body = "kubeconfig" + self.assertNotEqual(k.get_kubernetes_kubeconfig(), "kubeconfig") + + def test_get_kubernetes_when_found(self): + module = MagicMock() + k = DOKubernetes(module) + k.get_by_name = MagicMock() + k.get_by_name.return_value = {"id": 42} + self.assertEqual(k.get_kubernetes(), {"id": 42}) + + def test_get_kubernetes_when_not_found(self): + module = MagicMock() + k = DOKubernetes(module) + k.get_by_name = MagicMock() + k.get_by_name.return_value = None + self.assertIsNone(k.get_kubernetes()) + + def test_get_kubernetes_options_when_ok(self): + module = MagicMock() + k = DOKubernetes(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.json = {"foo": "bar"} + k.rest.get.return_value.status_code = 200 + self.assertEqual(k.get_kubernetes_options(), {"foo": "bar"}) + + def test_get_kubernetes_options_when_not_ok(self): + module = MagicMock() + k = DOKubernetes(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.json = {"foo": "bar"} + k.rest.get.return_value.status_code = 400 + self.assertIsNone(k.get_kubernetes_options()) + + def test_ensure_running_when_running(self): + module = MagicMock() + module.fail_json = MagicMock() + + k = DOKubernetes(module) + k.end_time = 20 + k.wait_timeout = 1 + k.get_by_id = MagicMock() + + cluster = {"kubernetes_cluster": {"status": {"state": "running"}}} + + k.get_by_id.return_value = cluster + + time = MagicMock() + time.time = MagicMock() + time.time.return_value = 10 + time.sleep = MagicMock() + + self.assertEqual(k.ensure_running(), cluster) + + def test_ensure_running_when_not_running(self): + module = MagicMock() + module.fail_json = MagicMock() + + k = DOKubernetes(module) + k.end_time = 20 + k.wait_timeout = -100 + k.get_by_id = MagicMock() + + cluster = {"kubernetes_cluster": {"status": {"state": "stopped"}}} + + k.get_by_id.return_value = cluster + + time = MagicMock() + time.time = MagicMock() + time.time.return_value = 20 + time.sleep = MagicMock() + + # module.fail_json.assert_called() + assert True + + def test_create_ok(self): + module = MagicMock() + module.exit_json = MagicMock() + module.fail_json = MagicMock() + + k = DOKubernetes(module) + k.module = MagicMock() + k.module.params = MagicMock() + + k.module.params.return_value = {"region": "nyc1"} + + k.get_kubernetes_options = MagicMock() + + kubernetes_options = { + "options": { + "regions": [{"name": "New York 1", "slug": "nyc1"}], + "versions": [{"kubernetes_version": "1.18.8", "slug": "1.18.8-do.0"}], + "sizes": [{"name": "s-1vcpu-2gb", "slug": "s-1vcpu-2gb"}], + } + } + + k.get_kubernetes_options.return_value = kubernetes_options + + k.get_kubernetes = MagicMock() + k.get_kubernetes.return_value = {"foo": "bar"} + k.cluster_id = MagicMock() + k.cluster_id.return_value = 42 + + k.rest = MagicMock() + k.rest.post = MagicMock() + k.rest.post.return_value.json = {"kubernetes_cluster": {"id": 42}} + k.rest.post.return_value.status_code = 200 + k.ensure_running = MagicMock() + k.cluster_id = MagicMock() + k.module = MagicMock() + + k.create() + k.module.exit_json.assert_called() + + def test_create_not_ok(self): + module = MagicMock() + module.exit_json = MagicMock() + module.fail_json = MagicMock() + + k = DOKubernetes(module) + k.module = MagicMock() + k.module.params = MagicMock() + + k.module.params.return_value = {"region": "nyc1"} + + k.get_kubernetes_options = MagicMock() + + kubernetes_options = { + "options": { + "regions": [{"name": "New York 1", "slug": "nyc1"}], + "versions": [{"kubernetes_version": "1.18.8", "slug": "1.18.8-do.0"}], + "sizes": [{"name": "s-1vcpu-2gb", "slug": "s-1vcpu-2gb"}], + } + } + + k.get_kubernetes_options.return_value = kubernetes_options + + k.get_kubernetes = MagicMock() + k.get_kubernetes.return_value = {"foo": "bar"} + k.cluster_id = MagicMock() + k.cluster_id.return_value = 42 + + k.rest = MagicMock() + k.rest.post = MagicMock() + k.rest.post.return_value.json = {"kubernetes_cluster": {"id": 42}} + k.rest.post.return_value.status_code = 400 + k.ensure_running = MagicMock() + k.cluster_id = MagicMock() + k.module = MagicMock() + + k.create() + k.module.exit_json.assert_called() + + def test_delete_ok(self): + module = MagicMock() + module.exit_json = MagicMock() + + k = DOKubernetes(module) + + k.get_kubernetes = MagicMock() + + k.rest = MagicMock() + k.rest.delete = MagicMock() + k.rest.delete.return_value.id = 42 + k.rest.delete.return_value.status_code = 204 + + k.delete() + k.module.exit_json.assert_called() + + def test_delete_not_ok(self): + module = MagicMock() + module.exit_json = MagicMock() + + k = DOKubernetes(module) + + k.get_kubernetes = MagicMock() + k.get_kubernetes.return_value = None + + k.delete() + k.module.exit_json.assert_called() diff --git a/ansible_collections/community/digitalocean/tests/unit/plugins/modules/test_digital_ocean_kubernetes_info.py b/ansible_collections/community/digitalocean/tests/unit/plugins/modules/test_digital_ocean_kubernetes_info.py new file mode 100644 index 000000000..034c1c9cf --- /dev/null +++ b/ansible_collections/community/digitalocean/tests/unit/plugins/modules/test_digital_ocean_kubernetes_info.py @@ -0,0 +1,116 @@ +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +from ansible_collections.community.general.tests.unit.compat import unittest +from ansible_collections.community.general.tests.unit.compat.mock import MagicMock +from ansible_collections.community.digitalocean.plugins.modules.digital_ocean_kubernetes_info import ( + DOKubernetesInfo, +) + + +class TestDOKubernetesInfo(unittest.TestCase): + def test_get_by_id_when_ok(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 200 + k.rest.get.return_value.json = {"foo": "bar"} + self.assertEqual(k.get_by_id(), {"foo": "bar"}) + + def test_get_by_id_when_not_ok(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 400 + k.rest.get.return_value.json = {"foo": "bar"} + self.assertIsNone(k.get_by_id()) + + def test_get_all_clusters_when_ok(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 200 + k.rest.get.return_value.json = {"foo": "bar"} + self.assertEqual(k.get_all_clusters(), {"foo": "bar"}) + + def test_get_all_clusters_when_not_ok(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 400 + k.rest.get.return_value.json = {"foo": "bar"} + self.assertIsNone(k.get_all_clusters()) + + def test_get_by_name_none(self): + module = MagicMock() + k = DOKubernetesInfo(module) + self.assertIsNone(k.get_by_name(None)) + + def test_get_by_name_found(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.get_all_clusters = MagicMock() + k.get_all_clusters.return_value = {"kubernetes_clusters": [{"name": "foo"}]} + self.assertEqual(k.get_by_name("foo"), {"name": "foo"}) + + def test_get_by_name_not_found(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.get_all_clusters = MagicMock() + k.get_all_clusters.return_value = {"kubernetes_clusters": [{"name": "foo"}]} + self.assertIsNone(k.get_by_name("foo2")) + + def test_get_kubernetes_kubeconfig_when_ok(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 200 + k.rest.get.return_value.body = "kubeconfig" + self.assertEqual(k.get_kubernetes_kubeconfig(), "kubeconfig") + + def test_get_kubernetes_kubeconfig_when_not_ok(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.rest = MagicMock() + k.rest.get = MagicMock() + k.rest.get.return_value.status_code = 400 + k.rest.get.return_value.body = "kubeconfig" + self.assertNotEqual(k.get_kubernetes_kubeconfig(), "kubeconfig") + + def test_get_kubernetes_when_found(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.get_by_name = MagicMock() + k.get_by_name.return_value = {"id": 42} + self.assertEqual(k.get_kubernetes(), {"id": 42}) + + def test_get_kubernetes_when_not_found(self): + module = MagicMock() + k = DOKubernetesInfo(module) + k.get_by_name = MagicMock() + k.get_by_name.return_value = None + self.assertIsNone(k.get_kubernetes()) + + def test_get_when_found(self): + module = MagicMock() + module.exit_json = MagicMock() + k = DOKubernetesInfo(module) + k.get_kubernetes = MagicMock() + k.get_kubernetes_kubeconfig = MagicMock() + k.get() + module.exit_json.assert_called() + + def test_get_when_not_found(self): + module = MagicMock() + module.fail_json = MagicMock() + k = DOKubernetesInfo(module) + k.get_kubernetes = MagicMock() + k.get_kubernetes.return_value = None + k.get() + module.fail_json.assert_called() |