diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/tests/test_models_unittest.py | 255 | ||||
| -rw-r--r-- | src/booking/tests/test_quick_booking.py | 105 | ||||
| -rw-r--r-- | src/dashboard/testing_utils.py | 647 |
3 files changed, 416 insertions, 591 deletions
diff --git a/src/api/tests/test_models_unittest.py b/src/api/tests/test_models_unittest.py index 971f757..cabffc9 100644 --- a/src/api/tests/test_models_unittest.py +++ b/src/api/tests/test_models_unittest.py @@ -7,19 +7,14 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## - -from datetime import timedelta -from django.utils import timezone - -from booking.models import Booking from api.models import ( Job, JobStatus, JobFactory, - AccessRelation, HostNetworkRelation, HostHardwareRelation, SoftwareRelation, + AccessConfig, ) from resource_inventory.models import ( @@ -30,105 +25,122 @@ from resource_inventory.models import ( from django.test import TestCase, Client from dashboard.testing_utils import ( - instantiate_host, - instantiate_user, - instantiate_userprofile, - instantiate_lab, - instantiate_installer, - instantiate_image, - instantiate_scenario, - instantiate_os, - make_hostprofile_set, - instantiate_opnfvrole, - instantiate_publicnet, - instantiate_booking, + make_host, + make_user, + make_user_profile, + make_lab, + make_installer, + make_image, + make_scenario, + make_os, + make_complete_host_profile, + make_booking, ) class ValidBookingCreatesValidJob(TestCase): @classmethod def setUpTestData(cls): - cls.loginuser = instantiate_user(False, username="newtestuser", password="testpassword") - cls.userprofile = instantiate_userprofile(cls.loginuser) + cls.user = make_user(False, username="newtestuser", password="testpassword") + cls.userprofile = make_user_profile(cls.user) + cls.lab = make_lab() + + cls.host_profile = make_complete_host_profile(cls.lab) + cls.scenario = make_scenario() + cls.installer = make_installer([cls.scenario]) + os = make_os([cls.installer]) + cls.image = make_image(cls.lab, 1, cls.user, os, cls.host_profile) + for i in range(30): + make_host(cls.host_profile, cls.lab, name="host" + str(i), labid="host" + str(i)) + cls.client = Client() - lab_user = instantiate_user(True) - cls.lab = instantiate_lab(lab_user) + def setUp(self): + self.booking, self.compute_hostnames, self.jump_hostname = self.create_multinode_generic_booking() - cls.host_profile = make_hostprofile_set(cls.lab) - cls.scenario = instantiate_scenario() - cls.installer = instantiate_installer([cls.scenario]) - os = instantiate_os([cls.installer]) - cls.image = instantiate_image(cls.lab, 1, cls.loginuser, os, cls.host_profile) - for i in range(30): - instantiate_host(cls.host_profile, cls.lab, name="host" + str(i), labid="host" + str(i)) - cls.role = instantiate_opnfvrole("Jumphost") - cls.computerole = instantiate_opnfvrole("Compute") - instantiate_publicnet(10, cls.lab) - instantiate_publicnet(12, cls.lab) - instantiate_publicnet(14, cls.lab) + def create_multinode_generic_booking(self): + topology = {} - cls.lab_selected = 'lab_' + str(cls.lab.lab_user.id) + '_selected' - cls.host_selected = 'host_' + str(cls.host_profile.id) + '_selected' + compute_hostnames = ["cmp01", "cmp02", "cmp03"] - cls.post_data = cls.build_post_data() + host_type = HostProfile.objects.first() - cls.client = Client() + universal_networks = [ + {"name": "public", "tagged": False, "public": True}, + {"name": "admin", "tagged": True, "public": False}] + compute_networks = [{"name": "private", "tagged": True, "public": False}] + jumphost_networks = [{"name": "external", "tagged": True, "public": True}] - def setUp(self): - self.client.login( - username=self.loginuser.username, password="testpassword") - self.booking, self.compute_hostnames, self.jump_hostname = self.create_multinode_generic_booking() + # generate a bunch of extra networks + for i in range(10): + net = {"tagged": False, "public": False} + net["name"] = "net" + str(i) + universal_networks.append(net) - @classmethod - def build_post_data(cls): - post_data = {} - post_data['filter_field'] = '{"hosts":[{"host_' + str(cls.host_profile.id) + '":"true"}], "labs": [{"lab_' + str(cls.lab.lab_user.id) + '":"true"}]}' - post_data['purpose'] = 'purposefieldcontentstring' - post_data['project'] = 'projectfieldcontentstring' - post_data['length'] = '3' - post_data['ignore_this'] = 1 - post_data['users'] = '' - post_data['hostname'] = 'hostnamefieldcontentstring' - post_data['image'] = str(cls.image.id) - post_data['installer'] = str(cls.installer.id) - post_data['scenario'] = str(cls.scenario.id) - return post_data - - def post(self, changed_fields={}): - payload = self.post_data.copy() - payload.update(changed_fields) - response = self.client.post('/booking/quick/', payload) - return response - - def generate_booking(self): - self.post() - return Booking.objects.first() + jumphost_info = { + "type": host_type, + "role": OPNFVRole.objects.get_or_create(name="Jumphost")[0], + "nets": self.make_networks(host_type, jumphost_networks + universal_networks), + "image": self.image + } + topology["jump"] = jumphost_info + + for hostname in compute_hostnames: + host_info = { + "type": host_type, + "role": OPNFVRole.objects.get_or_create(name="Compute")[0], + "nets": self.make_networks(host_type, compute_networks + universal_networks), + "image": self.image + } + topology[hostname] = host_info + + booking = make_booking( + owner=self.user, + lab=self.lab, + topology=topology, + installer=self.installer, + scenario=self.scenario + ) + + if not booking.resource: + raise Exception("Booking does not have a resource when trying to pass to makeCompleteJob") + JobFactory.makeCompleteJob(booking) + + return booking, compute_hostnames, "jump" + + def make_networks(self, hostprofile, nets): + """ + distributes nets accross hostprofile's interfaces + returns a 2D array + """ + network_struct = [] + count = hostprofile.interfaceprofile.all().count() + for i in range(count): + network_struct.append([]) + while(nets): + index = len(nets) % count + network_struct[index].append(nets.pop()) + + return network_struct + + # begin tests def test_valid_access_configs(self): job = Job.objects.get(booking=self.booking) self.assertIsNotNone(job) - access_configs = [r.config for r in AccessRelation.objects.filter(job=job).all()] + access_configs = AccessConfig.objects.filter(accessrelation__job=job) - vpnconfigs = [] - sshconfigs = [] + vpn_configs = access_configs.filter(access_type="vpn") + ssh_configs = access_configs.filter(access_type="ssh") - for config in access_configs: - if config.access_type == "vpn": - vpnconfigs.append(config) - elif config.access_type == "ssh": - sshconfigs.append(config) - else: - self.fail(msg="Undefined accessconfig: " + config.access_type + " found") + self.assertFalse(AccessConfig.objects.exclude(access_type__in=["vpn", "ssh"]).exists()) - user_set = [] - user_set.append(self.booking.owner) - user_set += self.booking.collaborators.all() + all_users = list(self.booking.collaborators.all()) + all_users.append(self.booking.owner) - for configs in [vpnconfigs, sshconfigs]: - for user in user_set: - configusers = [c.user for c in configs] - self.assertTrue(user in configusers) + for user in all_users: + self.assertTrue(vpn_configs.filter(user=user).exists()) + self.assertTrue(ssh_configs.filter(user=user).exists()) def test_valid_network_configs(self): job = Job.objects.get(booking=self.booking) @@ -136,12 +148,12 @@ class ValidBookingCreatesValidJob(TestCase): booking_hosts = self.booking.resource.hosts.all() - netrelation_set = HostNetworkRelation.objects.filter(job=job) - netconfig_set = [r.config for r in netrelation_set] + netrelations = HostNetworkRelation.objects.filter(job=job) + netconfigs = [r.config for r in netrelations] - netrelation_hosts = [r.host for r in netrelation_set] + netrelation_hosts = [r.host for r in netrelations] - for config in netconfig_set: + for config in netconfigs: for interface in config.interfaces.all(): self.assertTrue(interface.host in booking_hosts) @@ -172,15 +184,15 @@ class ValidBookingCreatesValidJob(TestCase): job = Job.objects.get(booking=self.booking) self.assertIsNotNone(job) - hrelations = HostHardwareRelation.objects.filter(job=job).all() + hardware_relations = HostHardwareRelation.objects.filter(job=job) - job_hosts = [r.host for r in hrelations] + job_hosts = [r.host for r in hardware_relations] booking_hosts = self.booking.resource.hosts.all() self.assertEqual(len(booking_hosts), len(job_hosts)) - for relation in hrelations: + for relation in hardware_relations: self.assertTrue(relation.host in booking_hosts) self.assertEqual(relation.status, JobStatus.NEW) config = relation.config @@ -213,66 +225,3 @@ class ValidBookingCreatesValidJob(TestCase): self.assertTrue(host.template.resource.name in self.compute_hostnames) else: self.fail(msg="Host with non-configured role name related to job: " + str(role_name)) - - def create_multinode_generic_booking(self): - topology = {} - - compute_hostnames = ["cmp01", "cmp02", "cmp03"] - - host_type = HostProfile.objects.first() - - universal_networks = [ - {"name": "public", "tagged": False, "public": True}, - {"name": "admin", "tagged": True, "public": False}] - just_compute_networks = [{"name": "private", "tagged": True, "public": False}] - just_jumphost_networks = [{"name": "external", "tagged": True, "public": True}] - - # generate a bunch of extra networks - for i in range(10): - net = {"tagged": False, "public": False} - net["name"] = "u_net" + str(i) - universal_networks.append(net) - - jhost_info = {} - jhost_info["type"] = host_type - jhost_info["role"] = OPNFVRole.objects.get(name="Jumphost") - jhost_info["nets"] = self.make_networks(host_type, list(just_jumphost_networks + universal_networks)) - jhost_info["image"] = self.image - topology["jump"] = jhost_info - - for hostname in compute_hostnames: - host_info = {} - host_info["type"] = host_type - host_info["role"] = OPNFVRole.objects.get(name="Compute") - host_info["nets"] = self.make_networks(host_type, list(just_compute_networks + universal_networks)) - host_info["image"] = self.image - topology[hostname] = host_info - - booking = instantiate_booking(self.loginuser, - timezone.now(), - timezone.now() + timedelta(days=1), - "demobooking", - self.lab, - topology=topology, - installer=self.installer, - scenario=self.scenario) - - if not booking.resource: - raise Exception("Booking does not have a resource when trying to pass to makeCompleteJob") - JobFactory.makeCompleteJob(booking) - - return booking, compute_hostnames, "jump" - - """ - evenly distributes networks given across a given profile's interfaces - """ - def make_networks(self, hostprofile, nets): - network_struct = [] - count = hostprofile.interfaceprofile.all().count() - for i in range(count): - network_struct.append([]) - while(nets): - index = len(nets) % count - network_struct[index].append(nets.pop()) - - return network_struct diff --git a/src/booking/tests/test_quick_booking.py b/src/booking/tests/test_quick_booking.py index 936a9a5..e445860 100644 --- a/src/booking/tests/test_quick_booking.py +++ b/src/booking/tests/test_quick_booking.py @@ -13,60 +13,55 @@ from django.test import TestCase, Client from booking.models import Booking from dashboard.testing_utils import ( - instantiate_host, - instantiate_user, - instantiate_userprofile, - instantiate_lab, - instantiate_installer, - instantiate_image, - instantiate_scenario, - instantiate_os, - make_hostprofile_set, - instantiate_opnfvrole, - instantiate_publicnet, + make_host, + make_user, + make_user_profile, + make_lab, + make_installer, + make_image, + make_scenario, + make_os, + make_complete_host_profile, + make_opnfv_role, + make_public_net, ) -# from dashboard import test_utils class QuickBookingValidFormTestCase(TestCase): @classmethod def setUpTestData(cls): - cls.loginuser = instantiate_user(False, username="newtestuser", password="testpassword") - instantiate_userprofile(cls.loginuser, True) + cls.user = make_user(False, username="newtestuser", password="testpassword") + make_user_profile(cls.user, True) - lab_user = instantiate_user(True) - cls.lab = instantiate_lab(lab_user) + lab_user = make_user(True) + cls.lab = make_lab(lab_user) - cls.host_profile = make_hostprofile_set(cls.lab) - cls.scenario = instantiate_scenario() - cls.installer = instantiate_installer([cls.scenario]) - os = instantiate_os([cls.installer]) - cls.image = instantiate_image(cls.lab, 1, cls.loginuser, os, cls.host_profile) - cls.host = instantiate_host(cls.host_profile, cls.lab) - cls.role = instantiate_opnfvrole() - cls.pubnet = instantiate_publicnet(10, cls.lab) - - cls.lab_selected = 'lab_' + str(cls.lab.lab_user.id) + '_selected' - cls.host_selected = 'host_' + str(cls.host_profile.id) + '_selected' + cls.host_profile = make_complete_host_profile(cls.lab) + cls.scenario = make_scenario() + cls.installer = make_installer([cls.scenario]) + os = make_os([cls.installer]) + cls.image = make_image(cls.lab, 1, cls.user, os, cls.host_profile) + cls.host = make_host(cls.host_profile, cls.lab) + cls.role = make_opnfv_role() + cls.pubnet = make_public_net(10, cls.lab) cls.post_data = cls.build_post_data() - cls.client = Client() @classmethod def build_post_data(cls): - post_data = {} - post_data['filter_field'] = '{"hosts":[{"host_' + str(cls.host_profile.id) + '":"true"}], "labs": [{"lab_' + str(cls.lab.lab_user.id) + '":"true"}]}' - post_data['purpose'] = 'purposefieldcontentstring' - post_data['project'] = 'projectfieldcontentstring' - post_data['length'] = '3' - post_data['ignore_this'] = 1 - post_data['users'] = '' - post_data['hostname'] = 'hostnamefieldcontentstring' - post_data['image'] = str(cls.image.id) - post_data['installer'] = str(cls.installer.id) - post_data['scenario'] = str(cls.scenario.id) - return post_data + return { + 'filter_field': '{"hosts":[{"host_' + str(cls.host_profile.id) + '":"true"}], "labs": [{"lab_' + str(cls.lab.lab_user.id) + '":"true"}]}', + 'purpose': 'my_purpose', + 'project': 'my_project', + 'length': '3', + 'ignore_this': 1, + 'users': '', + 'hostname': 'my_host', + 'image': str(cls.image.id), + 'installer': str(cls.installer.id), + 'scenario': str(cls.scenario.id) + } def post(self, changed_fields={}): payload = self.post_data.copy() @@ -75,26 +70,26 @@ class QuickBookingValidFormTestCase(TestCase): return response def setUp(self): - self.client.login( - username=self.loginuser.username, password="testpassword") + self.client.login(username=self.user.username, password="testpassword") - def is_valid_booking(self, booking): - self.assertEqual(booking.owner, self.loginuser) - self.assertEqual(booking.purpose, 'purposefieldcontentstring') - self.assertEqual(booking.project, 'projectfieldcontentstring') + def assertValidBooking(self, booking): + self.assertEqual(booking.owner, self.user) + self.assertEqual(booking.purpose, 'my_purpose') + self.assertEqual(booking.project, 'my_project') delta = booking.end - booking.start delta -= datetime.timedelta(days=3) self.assertLess(delta, datetime.timedelta(minutes=1)) - resourcebundle = booking.resource - configbundle = booking.config_bundle + resource_bundle = booking.resource + config_bundle = booking.config_bundle - self.assertEqual(self.installer, configbundle.opnfv_config.first().installer) - self.assertEqual(self.scenario, configbundle.opnfv_config.first().scenario) - self.assertEqual(resourcebundle.template.getHosts()[0].profile, self.host_profile) - self.assertEqual(resourcebundle.template.getHosts()[0].resource.name, 'hostnamefieldcontentstring') + opnfv_config = config_bundle.opnfv_config.first() + self.assertEqual(self.installer, opnfv_config.installer) + self.assertEqual(self.scenario, opnfv_config.scenario) - return True + host = resource_bundle.hosts.first() + self.assertEqual(host.profile, self.host_profile) + self.assertEqual(host.template.resource.name, 'my_host') def test_with_too_long_length(self): response = self.post({'length': '22'}) @@ -144,7 +139,7 @@ class QuickBookingValidFormTestCase(TestCase): self.assertEqual(response.status_code, 200) booking = Booking.objects.first() self.assertIsNotNone(booking) - self.assertTrue(self.is_valid_booking(booking)) + self.assertValidBooking(booking) def test_with_valid_form(self): response = self.post() @@ -152,4 +147,4 @@ class QuickBookingValidFormTestCase(TestCase): self.assertEqual(response.status_code, 200) booking = Booking.objects.first() self.assertIsNotNone(booking) - self.assertTrue(self.is_valid_booking(booking)) + self.assertValidBooking(booking) diff --git a/src/dashboard/testing_utils.py b/src/dashboard/testing_utils.py index 558031d..1cca3e6 100644 --- a/src/dashboard/testing_utils.py +++ b/src/dashboard/testing_utils.py @@ -9,13 +9,13 @@ from django.contrib.auth.models import User from django.core.files.base import ContentFile +from django.utils import timezone import json import re +from datetime import timedelta -from dashboard.exceptions import ( - InvalidHostnameException -) +from dashboard.exceptions import InvalidHostnameException from booking.models import Booking from account.models import UserProfile, Lab, LabStatus, VlanManager, PublicNetwork from resource_inventory.models import ( @@ -31,7 +31,6 @@ from resource_inventory.models import ( OPNFVRole, RamProfile, Network, - Vlan, GenericResourceBundle, GenericResource, GenericHost, @@ -39,30 +38,13 @@ from resource_inventory.models import ( GenericInterface, HostConfiguration, OPNFVConfig, + NetworkConnection, + HostOPNFVConfig ) from resource_inventory.resource_manager import ResourceManager - -class BookingContextData(object): - def prepopulate(self, *args, **kwargs): - self.loginuser = instantiate_user(False, username=kwargs.get("login_username", "newtestuser"), password="testpassword") - instantiate_userprofile(self.loginuser) - - lab_user = kwargs.get("lab_user", instantiate_user(True)) - self.lab = instantiate_lab(lab_user) - - self.host_profile = make_hostprofile_set(self.lab) - self.scenario = instantiate_scenario() - self.installer = instantiate_installer([self.scenario]) - os = instantiate_os([self.installer]) - self.image = instantiate_image(self.lab, 1, self.loginuser, os, self.host_profile) - self.host = instantiate_host(self.host_profile, self.lab) - self.role = instantiate_opnfvrole() - self.pubnet = instantiate_publicnet(10, self.lab) - - """ -Info for instantiate_booking() function: +Info for make_booking() function: [topology] argument structure: the [topology] argument should describe the structure of the pod the top level should be a dictionary, with each key being a hostname @@ -91,424 +73,323 @@ Info for instantiate_booking() function: """ -def instantiate_booking(owner, - start, - end, - booking_identifier, - lab=Lab.objects.first(), - purpose="purposetext", - project="projecttext", - collaborators=[], - topology={}, - installer=None, - scenario=None): - (grb, host_set) = instantiate_grb(topology, owner, lab, booking_identifier) - cb = instantiate_cb(grb, owner, booking_identifier, topology, host_set, installer, scenario) - - resource = ResourceManager.getInstance().convertResourceBundle(grb, lab, cb) +def make_booking(owner=None, start=timezone.now(), + end=timezone.now() + timedelta(days=1), + lab=None, purpose="my_purpose", + project="my_project", collaborators=[], + topology={}, installer=None, scenario=None): - booking = Booking() - - booking.resource = resource + grb, host_set = make_grb(topology, owner, lab) + config_bundle = make_config_bundle(grb, owner, topology, host_set, installer, scenario) + resource = ResourceManager.getInstance().convertResourceBundle(grb, config=config_bundle) if not resource: raise Exception("Resource not created") - booking.config_bundle = cb - booking.start = start - booking.end = end - booking.owner = owner - booking.purpose = purpose - booking.project = project - booking.lab = lab - booking.save() - - return booking - - -def instantiate_cb(grb, - owner, - booking_identifier, - topology={}, - host_set={}, - installer=None, - scenario=None): - cb = ConfigBundle() - cb.owner = owner - cb.name = str(booking_identifier) + "_cb" - cb.description = "cb generated by instantiate_cb() method" - cb.save() - - opnfvconfig = OPNFVConfig() - opnfvconfig.installer = installer - opnfvconfig.scenario = scenario - opnfvconfig.bundle = cb - opnfvconfig.save() + + return Booking.objects.create( + resource=resource, + config_bundle=config_bundle, + start=start, + end=end, + owner=owner, + purpose=purpose, + project=project, + lab=lab + ) + + +def make_config_bundle(grb, owner, topology={}, host_set={}, + installer=None, scenario=None): + cb = ConfigBundle.objects.create( + owner=owner, + name="config bundle " + str(ConfigBundle.objects.count()), + description="cb generated by make_config_bundle() method" + ) + + opnfv_config = OPNFVConfig.objects.create( + installer=installer, + scenario=scenario, + bundle=cb + ) # generate host configurations based on topology and host set for hostname, host_info in topology.items(): - hconf = HostConfiguration() - hconf.bundle = cb - hconf.host = host_set[hostname] - hconf.image = host_info["image"] - hconf.opnfvRole = host_info["role"] - hconf.save() + host_config = HostConfiguration.objects.create( + host=host_set[hostname], + image=host_info["image"], + bundle=cb, + is_head_node=host_info['role'].name.lower() == "jumphost" + ) + HostOPNFVConfig.objects.create( + role=host_info["role"], + host_config=host_config, + opnfv_config=opnfv_config + ) return cb -def instantiate_grb(topology, - owner, - lab, - booking_identifier): +def make_network(name, lab, grb, public): + network = Network(name=name, bundle=grb, is_public=public) + if public: + public_net = lab.vlan_manager.get_public_vlan() + if not public_net: + raise Exception("No more public networks available") + lab.vlan_manager.reserve_public_vlan(public_net.vlan) + network.vlan_id = public_net.vlan + else: + private_net = lab.vlan_manager.get_vlan() + if not private_net: + raise Exception("No more generic vlans are available") + lab.vlan_manager.reserve_vlans([private_net]) + network.vlan_id = private_net + + network.save() + return network + - grb = GenericResourceBundle(owner=owner, lab=lab) - grb.name = str(booking_identifier) + "_grb" - grb.description = "grb generated by instantiate_grb() method" - grb.save() +def make_grb(topology, owner, lab): + + grb = GenericResourceBundle.objects.create( + owner=owner, + lab=lab, + name="Generic ResourceBundle " + str(GenericResourceBundle.objects.count()), + description="grb generated by make_grb() method" + ) networks = {} host_set = {} - for hostname in topology.keys(): - info = topology[hostname] + for hostname, info in topology.items(): host_profile = info["type"] # need to construct host from hostname and type - ghost = instantiate_ghost(grb, host_profile, hostname) - host_set[hostname] = ghost - - ghost.save() + generic_host = make_generic_host(grb, host_profile, hostname) + host_set[hostname] = generic_host # set up networks nets = info["nets"] for interface_index, interface_profile in enumerate(host_profile.interfaceprofile.all()): - generic_interface = GenericInterface() - generic_interface.host = ghost - generic_interface.profile = interface_profile - generic_interface.save() - + generic_interface = GenericInterface.objects.create(host=generic_host, profile=interface_profile) netconfig = nets[interface_index] - for network_index, network_info in enumerate(netconfig): + for network_info in netconfig: network_name = network_info["name"] - network = None - if network_name in networks: - network = networks[network_name] - else: - network = Network() - network.name = network_name - network.vlan_id = lab.vlan_manager.get_vlan() - network.save() - networks[network_name] = network - if network_info["public"]: - public_net = lab.vlan_manager.get_public_vlan() - if not public_net: - raise Exception("No more public networks available") - lab.vlan_manager.reserve_public_vlan(public_net.vlan) - network.vlan_id = public_net.vlan - else: - private_net = lab.vlan_manager.get_vlan() - if not private_net: - raise Exception("No more generic vlans are available") - lab.vlan_manager.reserve_vlans([private_net]) - network.vlan_id = private_net - - vlan = Vlan() - vlan.vlan_id = network.vlan_id - vlan.public = network_info["public"] - vlan.tagged = network_info["tagged"] - vlan.save() - generic_interface.vlans.add(vlan) - - return (grb, host_set) - - -def instantiate_ghost(grb, host_profile, hostname): - if not re.match(r"(?=^.{1,253}$)(^([A-Za-z0-9-_]{1,62}\.)*[A-Za-z0-9-_]{1,63})$", hostname): - raise InvalidHostnameException("Hostname must comply to RFC 952 and all extensions to it until this point") - gresource = GenericResource(bundle=grb, name=hostname) - gresource.save() + if network_name not in networks: + networks[network_name] = make_network(network_name, lab, grb, network_info['public']) + + generic_interface.connections.add(NetworkConnection.objects.create( + network=networks[network_name], + vlan_is_tagged=network_info["tagged"] + )) - ghost = GenericHost() - ghost.resource = gresource - ghost.profile = host_profile - ghost.save() + return grb, host_set - return ghost +def make_generic_host(grb, host_profile, hostname): + if not re.match(r"(?=^.{1,253}$)(^([A-Za-z0-9-_]{1,62}\.)*[A-Za-z0-9-_]{1,63})$", hostname): + raise InvalidHostnameException("Hostname must comply to RFC 952 and all extensions") + gresource = GenericResource.objects.create(bundle=grb, name=hostname) + + return GenericHost.objects.create(resource=gresource, profile=host_profile) -def instantiate_user(is_superuser, - username="testuser", - password="testpassword", - email="default_email@user.com" - ): + +def make_user(is_superuser=False, username="testuser", + password="testpassword", email="default_email@user.com"): user = User.objects.create_user(username=username, email=email, password=password) user.is_superuser = is_superuser - user.save() return user -def instantiate_userprofile(user, email_addr="email@email.com", company="company", full_name="John Doe", booking_privledge=True, ssh_file=None): - up = UserProfile() - up.email_address = email_addr - up.company = company - up.full_name = full_name - up.booking_privledge = booking_privledge - up.user = user - up.save() - up.ssh_public_key.save("user_ssh_key", ssh_file if ssh_file else ContentFile("public key content string")) +def make_user_profile(user=None, email_addr="email@email.com", + company="company", full_name="John Doe", + booking_privledge=True, ssh_file=None): + user = user or User.objects.first() or make_user() + profile = UserProfile.objects.create( + email_addr=email_addr, + company=company, + full_name=full_name, + booking_privledge=booking_privledge, + user=user + ) + profile.ssh_public_key.save("user_ssh_key", ssh_file if ssh_file else ContentFile("public key content string")) - return up + return profile -def instantiate_vlanmanager(vlans=None, - block_size=20, - allow_overlapping=False, - reserved_vlans=None - ): - vlanmanager = VlanManager() +def make_vlan_manager(vlans=None, block_size=20, allow_overlapping=False, reserved_vlans=None): if not vlans: - vlans = [] - for vlan in range(0, 4095): - vlans.append(vlan % 2) - vlanmanager.vlans = json.dumps(vlans) + vlans = [vlan % 2 for vlan in range(4095)] if not reserved_vlans: - reserved_vlans = [] - for vlan in range(0, 4095): - reserved_vlans.append(0) - vlanmanager.reserved_vlans = json.dumps(vlans) - vlanmanager.block_size = block_size - vlanmanager.allow_overlapping = allow_overlapping - - vlanmanager.save() + reserved_vlans = [0 for i in range(4095)] - return vlanmanager + return VlanManager.objects.create( + vlans=json.dumps(vlans), + reserved_vlans=json.dumps(vlans), + block_size=block_size, + allow_overlapping=allow_overlapping + ) -def instantiate_lab(user=None, - name="Test Lab Instance", - status=LabStatus.UP, - vlan_manager=None - ): +def make_lab(user=None, name="Test Lab Instance", + status=LabStatus.UP, vlan_manager=None, + pub_net_count=5): if not vlan_manager: - vlan_manager = instantiate_vlanmanager() + vlan_manager = make_vlan_manager() if not user: - user = instantiate_user(True, 'test_user', 'test_pass', 'test_user@test_site.org') - - lab = Lab() - lab.lab_user = user - lab.name = name - lab.contact_email = 'test_lab@test_site.org' - lab.contact_phone = '603 123 4567' - lab.status = status - lab.vlan_manager = vlan_manager - lab.description = 'test lab instantiation' - lab.api_token = '12345678' - - lab.save() + user = make_user() + + lab = Lab.objects.create( + lab_user=user, + name=name, + contact_email='test_lab@test_site.org', + contact_phone='603 123 4567', + status=status, + vlan_manager=vlan_manager, + description='test lab instantiation', + api_token='12345678' + ) + + for i in range(pub_net_count): + make_public_net(vlan=i * 2 + 1, lab=lab) return lab """ -resource_inventory instantiation section for permenant resources +resource_inventory instantiation section for permanent resources """ -def make_hostprofile_set(lab, name="test_hostprofile"): - hostprof = instantiate_hostprofile(lab, name=name) - instantiate_diskprofile(hostprof, 500, name=name) - instantiate_cpuprofile(hostprof) - instantiate_interfaceprofile(hostprof, name=name) - instantiate_ramprofile(hostprof) - - return hostprof - - -def instantiate_hostprofile(lab, - host_type=0, - name="test hostprofile instance" - ): - hostprof = HostProfile() - hostprof.host_type = host_type - hostprof.name = name - hostprof.description = 'test hostprofile instance' - hostprof.save() - hostprof.labs.add(lab) - - hostprof.save() - - return hostprof - - -def instantiate_ramprofile(host, - channels=4, - amount=256): - ramprof = RamProfile() - ramprof.host = host - ramprof.amount = amount - ramprof.channels = channels - ramprof.save() - - return ramprof - - -def instantiate_diskprofile(hostprofile, - size=0, - media_type="SSD", - name="test diskprofile", - rotation=0, - interface="sata"): - - diskprof = DiskProfile() - diskprof.name = name - diskprof.size = size - diskprof.media_type = media_type - diskprof.host = hostprofile - diskprof.rotation = rotation - diskprof.interface = interface - - diskprof.save() - - return diskprof - - -def instantiate_cpuprofile(hostprofile, - cores=4, - architecture="x86_64", - cpus=4, - ): - cpuprof = CpuProfile() - cpuprof.cores = cores - cpuprof.architecture = architecture - cpuprof.cpus = cpus - cpuprof.host = hostprofile - cpuprof.cflags = '' - - cpuprof.save() - - return cpuprof - - -def instantiate_interfaceprofile(hostprofile, - speed=1000, - name="test interface profile", - nic_type="pcie" - ): - intprof = InterfaceProfile() - intprof.host = hostprofile - intprof.name = name - intprof.speed = speed - intprof.nic_type = nic_type - - intprof.save() - - return intprof - - -def instantiate_image(lab, - lab_id, - owner, - os, - host_profile, - public=True, - name="default image", - description="default image" - ): - image = Image() - image.from_lab = lab - image.lab_id = lab_id - image.os = os - image.host_type = host_profile - image.public = public - image.name = name - image.description = description - - image.save() - - return image - - -def instantiate_scenario(name="test scenario"): - scenario = Scenario() - scenario.name = name - scenario.save() - return scenario - - -def instantiate_installer(supported_scenarios, - name="test installer" - ): - installer = Installer() - installer.name = name - installer.save() - for scenario in supported_scenarios: +def make_complete_host_profile(lab, name="test_hostprofile"): + host_profile = make_host_profile(lab, name=name) + make_disk_profile(host_profile, 500, name=name) + make_cpu_profile(host_profile) + make_interface_profile(host_profile, name=name) + make_ram_profile(host_profile) + + return host_profile + + +def make_host_profile(lab, host_type=0, name="test hostprofile"): + host_profile = HostProfile.objects.create( + host_type=host_type, + name=name, + description='test hostprofile instance' + ) + host_profile.labs.add(lab) + + return host_profile + + +def make_ram_profile(host, channels=4, amount=256): + return RamProfile.objects.create( + host=host, + amount=amount, + channels=channels + ) + + +def make_disk_profile(hostprofile, size=0, media_type="SSD", + name="test diskprofile", rotation=0, + interface="sata"): + return DiskProfile.objects.create( + name=name, + size=size, + media_type=media_type, + host=hostprofile, + rotation=rotation, + interface=interface + ) + + +def make_cpu_profile(hostprofile, + cores=4, + architecture="x86_64", + cpus=4,): + return CpuProfile.objects.create( + cores=cores, + architecture=architecture, + cpus=cpus, + host=hostprofile, + cflags='' + ) + + +def make_interface_profile(hostprofile, + speed=1000, + name="test interface profile", + nic_type="pcie"): + return InterfaceProfile.objects.create( + host=hostprofile, + name=name, + speed=speed, + nic_type=nic_type + ) + + +def make_image(lab, lab_id, owner, os, host_profile, + public=True, name="default image", description="default image"): + return Image.objects.create( + from_lab=lab, + lab_id=lab_id, + os=os, + host_type=host_profile, + public=public, + name=name, + description=description + ) + + +def make_scenario(name="test scenario"): + return Scenario.objects.create(name=name) + + +def make_installer(scenarios, name="test installer"): + installer = Installer.objects.create(name=name) + for scenario in scenarios: installer.sup_scenarios.add(scenario) - installer.save() return installer -def instantiate_os(supported_installers, - name="test operating system", - ): - os = Opsys() - os.name = name - os.save() - for installer in supported_installers: +def make_os(installers, name="test OS"): + os = Opsys.objects.create(name=name) + for installer in installers: os.sup_installers.add(installer) - os.save() + return os -def instantiate_host(host_profile, - lab, - labid="test_host", - name="test_host", - booked=False, - working=True, - config=None, - template=None, - bundle=None, - model="Model 1", - vendor="ACME"): - host = Host() - host.lab = lab - host.profile = host_profile - host.name = name - host.booked = booked - host.working = working - host.config = config - host.template = template - host.bundle = bundle - host.model = model - host.vendor = vendor - - host.save() - - return host - - -def instantiate_opnfvrole(name="Jumphost", - description="test opnfvrole"): - role = OPNFVRole() - role.name = name - role.description = description - role.save() - - return role - - -def instantiate_publicnet(vlan, - lab, - in_use=False, - cidr="0.0.0.0/0", - gateway="0.0.0.0"): - pubnet = PublicNetwork() - pubnet.lab = lab - pubnet.vlan = vlan - pubnet.cidr = cidr - pubnet.gateway = gateway - pubnet.save() - - return pubnet +def make_host(host_profile, lab, labid="test_host", name="test_host", + booked=False, working=True, config=None, template=None, + bundle=None, model="Model 1", vendor="ACME"): + return Host.objects.create( + lab=lab, + profile=host_profile, + name=name, + booked=booked, + working=working, + config=config, + template=template, + bundle=bundle, + model=model, + vendor=vendor + ) + + +def make_opnfv_role(name="Jumphost", description="test opnfvrole"): + return OPNFVRole.objects.create( + name=name, + description=description + ) + + +def make_public_net(vlan, lab, in_use=False, + cidr="0.0.0.0/0", gateway="0.0.0.0"): + return PublicNetwork.objects.create( + lab=lab, + vlan=vlan, + cidr=cidr, + gateway=gateway + ) |