# test_create_vm_form.py
from django.test import TestCase
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from virtualization.models import ClusterType, Cluster, VirtualMachine, VMInterface
from ipam.models import Service
from dcim.models import DeviceRole, Platform
from extras.models import Tag
from netbox_sys_plugin.models import ProviderTypeExtraConfig, VirtualMachineType
from netbox_sys_plugin.forms import CreateVmForm
# pylint: disable=W0201, R0902
class CreateVmFormTestCase(TestCase):
"""Test suite for the CreateVmForm."""
def setUp(self):
"""Set up the test data."""
self.cluster_type = ClusterType.objects.create(name="Test Cluster Type", slug="test-cluster-type")
self.cluster = Cluster.objects.create(name="Test Cluster", type=self.cluster_type)
self.role = DeviceRole.objects.create(name="Test Role", slug="test-role")
self.platform = Platform.objects.create(name="Test Platform", slug="test-platform")
self.tag = Tag.objects.create(name="Test Tag", slug="test-tag")
self.provider_extra_config = ProviderTypeExtraConfig.objects.create(
extra_config_name="Test Config",
extra_config_structure={
"config_structure": [{"field1": {"type": "string", "required": "True"}}]
},
extra_config_description="Test description"
)
self.vm_machine_type = VirtualMachineType.objects.create(
virtual_machine_type_name="BIG",
virtual_machine_type_desc="BIG",
assigned_object_type=ContentType.objects.get_for_model(ClusterType),
assigned_object_id=self.cluster_type.pk)
self.form_data_template = {
'cl_list_new-cluster_type': self.cluster_type.id,
'cl_list_new-cluster': self.cluster.id,
'vms_new-0-name': 'TEST_DATA2',
'vms_new-0-status': 'active',
'vms_new-0-role': str(self.role.id),
'vms_new-0-platform': str(self.platform.id),
'vms_new-0-description': 'TEST_DATA',
'vmassignedvmtype_new-virtual_machine_type': str(self.vm_machine_type.pk),
'vmis_new-0-name': 'TEST_DATA',
'ip_new-address': '100.0.0.1/24',
'ip_new-status': 'active',
'gateway_new-address': '100.2.3.4/24',
'gateway_new-status': 'active',
'domainnames_new-domain_names': '{"example_domain_name": "example_domain1","example_domain_name2": "example_domain2"}',
'vmassignedextraconfig_new-provider_type_extra_config': str(self.provider_extra_config.pk),
'vmassignedextraconfig_new-extra_config_values': '{"field1": "fieldvalue1"}',
'service_ntp-0-name': 'TEST_DATA1',
'service_ntp-0-protocol': 'tcp',
'service_ntp-0-ports': '440',
'service_dns-0-name': 'TEST_DATA2',
'service_dns-0-protocol': 'tcp',
'service_dns-0-ports': '44',
'service_syslog-0-name': 'TEST_DATA3',
'service_syslog-0-protocol': 'tcp',
'service_syslog-0-ports': '111',
'tag_new-tag':self.tag.id
}
def test_valid_form_submission_without_webhook(self):
"""Test a valid form submission."""
form_data = self.form_data_template
form = CreateVmForm(data=form_data)
vm = form.process_creation(form_data)
self.assertIsInstance(vm, VirtualMachine)
#Assert VM
self.assertEqual(VirtualMachine.objects.filter(id=vm.pk).count(), 1)
#Assert Interfaces
self.assertEqual(VMInterface.objects.filter(virtual_machine=vm).count(), 1)
#Assert Services
self.assertEqual(Service.objects.filter(virtual_machine=vm).count(), 3)
def test_invalid_form(self):
"""Test a valid form submission."""
#invalid json
self.invalid_ports_form_data = self.form_data_template.copy()
self.invalid_ports_form_data["domainnames_new-domain_names"] = "Invalid" # Invalid format
form_data = self.invalid_ports_form_data
form = CreateVmForm(data=form_data)
with self.assertRaises(ValueError) as context:
form.process_creation(form_data)
self.assertIn(
"Invalid JSON format: Please provide a valid JSON object.", str(context.exception))
#black ports
self.invalid_ports_form_data = self.form_data_template.copy()
self.invalid_ports_form_data["service_ntp-0-ports"] = "" # Invalid format
form_data = self.invalid_ports_form_data
form = CreateVmForm(data=form_data)
with self.assertRaises(ValidationError) as context:
form.process_creation(form_data)
self.assertIn(
"{'ports': ['This field cannot be blank.']}", str(context.exception))
#invalid/black ports
self.invalid_ports_form_data = self.form_data_template.copy()
self.invalid_ports_form_data["service_ntp-0-ports"] = "1,A," # Invalid format
form_data = self.invalid_ports_form_data
form = CreateVmForm(data=form_data)
with self.assertRaises(ValueError) as context:
form.process_creation(form_data)
self.assertIn(
"Invalid ports value:", str(context.exception))
#invalid tag
self.invalid_ports_form_data = self.form_data_template.copy()
self.invalid_ports_form_data["tag_new-tag"] = "10" # Invalid format
form_data = self.invalid_ports_form_data
form = CreateVmForm(data=form_data)
with self.assertRaises(ValueError) as context:
form.process_creation(form_data)
self.assertIn(
"Invalid tag", str(context.exception))
#invalid role/owner
self.invalid_ports_form_data = self.form_data_template.copy()
self.invalid_ports_form_data["vms_new-0-role"] = "2" # Invalid format
form_data = self.invalid_ports_form_data
form = CreateVmForm(data=form_data)
with self.assertRaises(ValueError) as context:
form.process_creation(form_data)
self.assertIn(
"Error during creation: Invalid Device Role ID", str(context.exception))
#invalid platform
self.invalid_ports_form_data = self.form_data_template.copy()
self.invalid_ports_form_data["vms_new-0-platform"] = "2" # Invalid format
form_data = self.invalid_ports_form_data
form = CreateVmForm(data=form_data)
with self.assertRaises(ValueError) as context:
form.process_creation(form_data)
self.assertIn(
"Invalid Platform ID", str(context.exception))
def tearDown(self):
"""Clean up after tests."""
VirtualMachine.objects.all().delete()
DeviceRole.objects.all().delete()
Platform.objects.all().delete()
ProviderTypeExtraConfig.objects.all().delete()
Service.objects.all().delete()
Cluster.objects.all().delete()
ClusterType.objects.all().delete()