diff options
Diffstat (limited to 'src/dashboard/testing_utils.py')
| -rw-r--r-- | src/dashboard/testing_utils.py | 647 |
1 files changed, 264 insertions, 383 deletions
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 + ) |