mirror of
https://github.com/kubevirt/kubevirt.core.git
synced 2026-07-26 09:24:41 +00:00
cleanup: Migrate existing unit tests to pytest
Cleanup the existing unit tests and move them to pytest. By using pytest it becomes easier to add new tests for existing functionality in the collection. Signed-off-by: Felix Matouschek <fmatouschek@redhat.com>
This commit is contained in:
@@ -6,11 +6,9 @@ from __future__ import absolute_import, division, print_function
|
||||
|
||||
__metaclass__ = type
|
||||
|
||||
import unittest
|
||||
import pytest
|
||||
|
||||
from unittest.mock import patch, ANY
|
||||
|
||||
from ansible.module_utils import basic
|
||||
from ansible.module_utils.basic import AnsibleModule
|
||||
from ansible_collections.kubernetes.core.plugins.module_utils.k8s import runner
|
||||
from ansible_collections.kubevirt.core.plugins.modules import kubevirt_vm
|
||||
from ansible_collections.kubevirt.core.tests.unit.utils.ansible_module_mock import (
|
||||
@@ -19,47 +17,55 @@ from ansible_collections.kubevirt.core.tests.unit.utils.ansible_module_mock impo
|
||||
exit_json,
|
||||
fail_json,
|
||||
set_module_args,
|
||||
get_api_client,
|
||||
)
|
||||
|
||||
FIXTURE1 = {
|
||||
"apiVersion": "kubevirt.io/v1",
|
||||
"kind": "VirtualMachine",
|
||||
"metadata": {
|
||||
"name": "testvm",
|
||||
"namespace": "default",
|
||||
"labels": {"environment": "staging", "service": "loadbalancer"},
|
||||
},
|
||||
"spec": {
|
||||
"running": True,
|
||||
"instancetype": {"name": "u1.medium"},
|
||||
"preference": {"name": "fedora"},
|
||||
"dataVolumeTemplates": [
|
||||
{
|
||||
"metadata": {"name": "testdv"},
|
||||
"spec": {
|
||||
"source": {
|
||||
"registry": {
|
||||
"url": "docker://quay.io/containerdisks/fedora:latest"
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def vm_definition():
|
||||
return {
|
||||
"apiVersion": "kubevirt.io/v1",
|
||||
"kind": "VirtualMachine",
|
||||
"metadata": {
|
||||
"name": "testvm",
|
||||
"namespace": "default",
|
||||
"labels": {"environment": "staging", "service": "loadbalancer"},
|
||||
},
|
||||
"spec": {
|
||||
"running": True,
|
||||
"instancetype": {"name": "u1.medium"},
|
||||
"preference": {"name": "fedora"},
|
||||
"dataVolumeTemplates": [
|
||||
{
|
||||
"metadata": {"name": "testdv"},
|
||||
"spec": {
|
||||
"source": {
|
||||
"registry": {
|
||||
"url": "docker://quay.io/containerdisks/fedora:latest"
|
||||
},
|
||||
},
|
||||
"storage": {
|
||||
"accessModes": ["ReadWriteOnce"],
|
||||
"resources": {"requests": {"storage": "5Gi"}},
|
||||
},
|
||||
},
|
||||
"storage": {
|
||||
"accessModes": ["ReadWriteOnce"],
|
||||
"resources": {"requests": {"storage": "5Gi"}},
|
||||
},
|
||||
}
|
||||
],
|
||||
"template": {
|
||||
"metadata": {
|
||||
"labels": {"environment": "staging", "service": "loadbalancer"}
|
||||
},
|
||||
"spec": {
|
||||
"domain": {"devices": {}},
|
||||
"terminationGracePeriodSeconds": 180,
|
||||
},
|
||||
}
|
||||
],
|
||||
"template": {
|
||||
"metadata": {
|
||||
"labels": {"environment": "staging", "service": "loadbalancer"}
|
||||
},
|
||||
"spec": {"domain": {"devices": {}}, "terminationGracePeriodSeconds": 180},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
METADATA = """apiVersion: kubevirt.io/v1
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def vm_manifest():
|
||||
return """apiVersion: kubevirt.io/v1
|
||||
kind: VirtualMachine
|
||||
metadata:
|
||||
name: "testvm"
|
||||
@@ -97,121 +103,96 @@ spec:
|
||||
terminationGracePeriodSeconds: 180
|
||||
"""
|
||||
|
||||
FIXTURE2 = {
|
||||
"name": "testvm",
|
||||
"namespace": "default",
|
||||
"state": "present",
|
||||
"labels": {"service": "loadbalancer", "environment": "staging"},
|
||||
"instancetype": {"name": "u1.medium"},
|
||||
"preference": {"name": "fedora"},
|
||||
"data_volume_templates": [
|
||||
{
|
||||
"metadata": {"name": "testdv"},
|
||||
"spec": {
|
||||
"source": {
|
||||
"registry": {
|
||||
"url": "docker://quay.io/containerdisks/fedora:latest"
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def module_params_create():
|
||||
return {
|
||||
"name": "testvm",
|
||||
"namespace": "default",
|
||||
"state": "present",
|
||||
"labels": {"service": "loadbalancer", "environment": "staging"},
|
||||
"instancetype": {"name": "u1.medium"},
|
||||
"preference": {"name": "fedora"},
|
||||
"data_volume_templates": [
|
||||
{
|
||||
"metadata": {"name": "testdv"},
|
||||
"spec": {
|
||||
"source": {
|
||||
"registry": {
|
||||
"url": "docker://quay.io/containerdisks/fedora:latest"
|
||||
},
|
||||
},
|
||||
"storage": {
|
||||
"accessModes": ["ReadWriteOnce"],
|
||||
"resources": {"requests": {"storage": "5Gi"}},
|
||||
},
|
||||
},
|
||||
"storage": {
|
||||
"accessModes": ["ReadWriteOnce"],
|
||||
"resources": {"requests": {"storage": "5Gi"}},
|
||||
},
|
||||
},
|
||||
}
|
||||
],
|
||||
"spec": {"domain": {"devices": {}}, "terminationGracePeriodSeconds": 180},
|
||||
"api_version": "kubevirt.io/v1",
|
||||
"running": True,
|
||||
"wait": False,
|
||||
"wait_sleep": 5,
|
||||
"wait_timeout": 120,
|
||||
"force": False,
|
||||
"generate_name": None,
|
||||
"annotations": None,
|
||||
"kubeconfig": None,
|
||||
"context": None,
|
||||
"host": None,
|
||||
"api_key": None,
|
||||
"username": None,
|
||||
"password": None,
|
||||
"validate_certs": None,
|
||||
"ca_cert": None,
|
||||
"client_cert": None,
|
||||
"client_key": None,
|
||||
"proxy": None,
|
||||
"no_proxy": None,
|
||||
"proxy_headers": None,
|
||||
"persist_config": None,
|
||||
"impersonate_user": None,
|
||||
"impersonate_groups": None,
|
||||
"delete_options": None,
|
||||
"resource_definition": METADATA,
|
||||
"wait_condition": {"type": "Ready", "status": True},
|
||||
}
|
||||
|
||||
|
||||
class TestCreateVM(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mock_module_helper = patch.multiple(
|
||||
basic.AnsibleModule, exit_json=exit_json, fail_json=fail_json
|
||||
)
|
||||
self.mock_module_helper.start()
|
||||
|
||||
self.mock_runner = patch.multiple(runner, get_api_client=get_api_client)
|
||||
self.mock_runner.start()
|
||||
|
||||
# Stop the patch after test execution
|
||||
# like tearDown but executed also when the setup failed
|
||||
self.addCleanup(self.mock_module_helper.stop)
|
||||
self.addCleanup(self.mock_runner.stop)
|
||||
|
||||
def test_module_fail_when_required_args_missing(self):
|
||||
with self.assertRaises(AnsibleFailJson):
|
||||
set_module_args({})
|
||||
kubevirt_vm.main()
|
||||
|
||||
def test_create(self):
|
||||
set_module_args(
|
||||
{
|
||||
"name": "testvm",
|
||||
"namespace": "default",
|
||||
"state": "present",
|
||||
"labels": {"service": "loadbalancer", "environment": "staging"},
|
||||
"instancetype": {"name": "u1.medium"},
|
||||
"preference": {"name": "fedora"},
|
||||
"data_volume_templates": [
|
||||
{
|
||||
"metadata": {"name": "testdv"},
|
||||
"spec": {
|
||||
"source": {
|
||||
"registry": {
|
||||
"url": "docker://quay.io/containerdisks/fedora:latest"
|
||||
},
|
||||
},
|
||||
"storage": {
|
||||
"accessModes": ["ReadWriteOnce"],
|
||||
"resources": {"requests": {"storage": "5Gi"}},
|
||||
},
|
||||
},
|
||||
}
|
||||
],
|
||||
"spec": {
|
||||
"domain": {"devices": {}},
|
||||
"terminationGracePeriodSeconds": 180,
|
||||
},
|
||||
}
|
||||
)
|
||||
with patch.object(runner, "perform_action") as mock_run_command:
|
||||
mock_run_command.return_value = {
|
||||
"method": "create",
|
||||
"changed": True,
|
||||
"result": "success",
|
||||
} # successful execution
|
||||
with self.assertRaises(AnsibleExitJson) as result:
|
||||
kubevirt_vm.main()
|
||||
mock_run_command.assert_called_once_with(
|
||||
ANY,
|
||||
FIXTURE1,
|
||||
FIXTURE2,
|
||||
)
|
||||
],
|
||||
"spec": {
|
||||
"domain": {"devices": {}},
|
||||
"terminationGracePeriodSeconds": 180,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def k8s_module_params_create(module_params_create, vm_manifest):
|
||||
return module_params_create | {
|
||||
"api_version": "kubevirt.io/v1",
|
||||
"running": True,
|
||||
"wait": False,
|
||||
"wait_sleep": 5,
|
||||
"wait_timeout": 120,
|
||||
"force": False,
|
||||
"generate_name": None,
|
||||
"annotations": None,
|
||||
"kubeconfig": None,
|
||||
"context": None,
|
||||
"host": None,
|
||||
"api_key": None,
|
||||
"username": None,
|
||||
"password": None,
|
||||
"validate_certs": None,
|
||||
"ca_cert": None,
|
||||
"client_cert": None,
|
||||
"client_key": None,
|
||||
"proxy": None,
|
||||
"no_proxy": None,
|
||||
"proxy_headers": None,
|
||||
"persist_config": None,
|
||||
"impersonate_user": None,
|
||||
"impersonate_groups": None,
|
||||
"delete_options": None,
|
||||
"resource_definition": vm_manifest,
|
||||
"wait_condition": {"type": "Ready", "status": True},
|
||||
}
|
||||
|
||||
|
||||
def test_module_fails_when_required_args_missing(monkeypatch):
|
||||
monkeypatch.setattr(AnsibleModule, "fail_json", fail_json)
|
||||
with pytest.raises(AnsibleFailJson):
|
||||
set_module_args({})
|
||||
kubevirt_vm.main()
|
||||
|
||||
|
||||
def test_module_create(
|
||||
monkeypatch, mocker, module_params_create, k8s_module_params_create, vm_definition
|
||||
):
|
||||
monkeypatch.setattr(AnsibleModule, "exit_json", exit_json)
|
||||
monkeypatch.setattr(runner, "get_api_client", lambda _: None)
|
||||
|
||||
set_module_args(module_params_create)
|
||||
|
||||
perform_action = mocker.patch.object(runner, "perform_action")
|
||||
perform_action.return_value = {
|
||||
"method": "create",
|
||||
"changed": True,
|
||||
"result": "success",
|
||||
}
|
||||
|
||||
with pytest.raises(AnsibleExitJson):
|
||||
kubevirt_vm.main()
|
||||
perform_action.assert_called_once_with(
|
||||
mocker.ANY, vm_definition, k8s_module_params_create
|
||||
)
|
||||
|
||||
@@ -6,8 +6,7 @@ from __future__ import absolute_import, division, print_function
|
||||
|
||||
__metaclass__ = type
|
||||
|
||||
from unittest import TestCase
|
||||
from unittest.mock import patch
|
||||
import pytest
|
||||
|
||||
from ansible.module_utils.basic import AnsibleModule
|
||||
from ansible_collections.kubernetes.core.plugins.module_utils.k8s.service import (
|
||||
@@ -19,12 +18,10 @@ from ansible_collections.kubevirt.core.plugins.modules import (
|
||||
from ansible_collections.kubevirt.core.tests.unit.utils.ansible_module_mock import (
|
||||
AnsibleExitJson,
|
||||
exit_json,
|
||||
fail_json,
|
||||
set_module_args,
|
||||
get_api_client,
|
||||
)
|
||||
|
||||
FIXTURE1 = {
|
||||
FIND_ARGS_DEFAULT = {
|
||||
"kind": "VirtualMachine",
|
||||
"api_version": "kubevirt.io/v1",
|
||||
"name": None,
|
||||
@@ -37,7 +34,7 @@ FIXTURE1 = {
|
||||
"condition": {"type": "Ready", "status": True},
|
||||
}
|
||||
|
||||
FIXTURE2 = {
|
||||
FIND_ARGS_NAME_NAMESPACE = {
|
||||
"kind": "VirtualMachine",
|
||||
"api_version": "kubevirt.io/v1",
|
||||
"name": "testvm",
|
||||
@@ -51,43 +48,26 @@ FIXTURE2 = {
|
||||
}
|
||||
|
||||
|
||||
class TestDescribeVM(TestCase):
|
||||
def setUp(self):
|
||||
self.mock_module_helper = patch.multiple(
|
||||
AnsibleModule, exit_json=exit_json, fail_json=fail_json
|
||||
)
|
||||
self.mock_module_helper.start()
|
||||
@pytest.mark.parametrize(
|
||||
"module_args,find_args",
|
||||
[
|
||||
({}, FIND_ARGS_DEFAULT),
|
||||
({"name": "testvm", "namespace": "default"}, FIND_ARGS_NAME_NAMESPACE),
|
||||
],
|
||||
)
|
||||
def test_module(monkeypatch, mocker, module_args, find_args):
|
||||
monkeypatch.setattr(AnsibleModule, "exit_json", exit_json)
|
||||
monkeypatch.setattr(kubevirt_vm_info, "get_api_client", lambda _: None)
|
||||
|
||||
self.mock_main = patch.multiple(kubevirt_vm_info, get_api_client=get_api_client)
|
||||
self.mock_main.start()
|
||||
set_module_args(module_args)
|
||||
|
||||
# Stop the patch after test execution
|
||||
# like tearDown but executed also when the setup failed
|
||||
self.addCleanup(self.mock_module_helper.stop)
|
||||
self.addCleanup(self.mock_main.stop)
|
||||
find = mocker.patch.object(K8sService, "find")
|
||||
find.return_value = {
|
||||
"api_found": True,
|
||||
"failed": False,
|
||||
"resources": [],
|
||||
}
|
||||
|
||||
def run_module(self, fixture):
|
||||
with patch.object(K8sService, "find") as mock_find_command:
|
||||
mock_find_command.return_value = {
|
||||
"api_found": True,
|
||||
"failed": False,
|
||||
"resources": [],
|
||||
} # successful execution
|
||||
with self.assertRaises(AnsibleExitJson):
|
||||
kubevirt_vm_info.main()
|
||||
mock_find_command.assert_called_once_with(
|
||||
**fixture,
|
||||
)
|
||||
|
||||
def test_describe_without_args(self):
|
||||
set_module_args({})
|
||||
self.run_module(FIXTURE1)
|
||||
|
||||
def test_describe_with_args(self):
|
||||
set_module_args(
|
||||
{
|
||||
"name": "testvm",
|
||||
"namespace": "default",
|
||||
}
|
||||
)
|
||||
self.run_module(FIXTURE2)
|
||||
with pytest.raises(AnsibleExitJson):
|
||||
kubevirt_vm_info.main()
|
||||
find.assert_called_once_with(**find_args)
|
||||
|
||||
Reference in New Issue
Block a user