##############################################################################
# Copyright (c) 2019 Sawyer Bergeron, Parker Berberian, and others.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
# 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,
)
from resource_inventory.models import (
OPNFVRole,
HostProfile,
)
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,
)
class ValidBookingCreatesValidJob(TestCase):
@classmethod
def setUpTestData(cls):
cls.loginuser = instantiate_user(False, username="newtestuser", password="testpassword")
cls.userprofile = instantiate_userprofile(cls.loginuser)
lab_user = instantiate_user(True)
cls.lab = instantiate_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)
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)
cls.lab_selected = 'lab_' + str(cls.lab.lab_user.id) + '_selected'
cls.host_selected = 'host_' + str(cls.host_profile.id) + '_selected'
cls.post_data = cls.build_post_data()
cls.client = Client()
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()
@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()
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()]
vpnconfigs = []
sshconfigs = []
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")
user_set = []
user_set.append(self.booking.owner)
user_set += self.booking.collaborators.all()
for configs in [vpnconfigs, sshconfigs]:
for user in user_set:
configusers = [c.user for c in configs]
self.assertTrue(user in configusers)
def test_valid_network_configs(self):
job = Job.objects.get(booking=self.booking)
self.assertIsNotNone(job)
booking_hosts = self.booking.resource.hosts.all()
netrelation_set = HostNetworkRelation.objects.filter(job=job)
netconfig_set = [r.config for r in netrelation_set]
netrelation_hosts = [r.host for r in netrelation_set]
for config in netconfig_set:
for interface in config.interfaces.all():
self.assertTrue(interface.host in booking_hosts)
# if no interfaces are referenced that shouldn't have vlans,
# and no vlans exist outside those accounted for in netconfigs,
# then the api is faithfully representing networks
# as netconfigs reference resource_inventory models directly
# this test relies on the assumption that
# every interface is configured, whether it does or does not have vlans
# if this is not true, the test fails
for host in booking_hosts:
self.assertTrue(host in netrelation_hosts)
relation = HostNetworkRelation.objects.filter(job=job).get(host=host)
# do 2 direction matching that interfaces are one to one
config = relation.config
for interface in config.interfaces.all():
self.assertTrue(interface in host.interfaces)
for interface in host.interfaces.all():
self.assertTrue(interface in config.interfaces)
for host in netrelation_hosts:
self.assertTrue(host in booking_hosts)
def test_valid_hardware_configs(self):
job = Job.objects.get(booking=self.booking)
self.assertIsNotNone(job)
hrelations = HostHardwareRelation.objects.filter(job=job).all()
job_hosts = [r.host for r in hrelations]
booking_hosts = self.booking.resource.hosts.all()
self.assertEqual(len(booking_hosts), len(job_hosts))
for relation in hrelations:
self.assertTrue(relation.host in booking_hosts)
self.assertEqual(relation.status, JobStatus.NEW)
config = relation.config
host = relation.host
self.assertEqual(config.hostname, host.template.resource.name)
def test_valid_software_configs(self):
job = Job.objects.get(booking=self.booking)
self.assertIsNotNone(job)
srelation = SoftwareRelation.objects.filter(job=job).first()
self.assertIsNotNone(srelation)
sconfig = srelation.config
self.assertIsNotNone(sconfig)
oconfig = sconfig.opnfv
self.assertIsNotNone(oconfig)
# not onetoone in models, but first() is safe here based on how ConfigBundle and a matching OPNFVConfig are created
# this should, however, be made explicit
self.assertEqual(oconfig.installer, self.booking.config_bundle.opnfv_config.first().installer.name)
self.assertEqual(oconfig.scenario, self.booking.config_bundle.opnfv_config.first().scenario.name)
for host in oconfig.roles.all():
role_name = host.config.opnfvRole.name
if str(role_name) == "Jumphost":
self.assertEqual(host.template.resource.name, self.jump_hostname)
elif str(role_name) == "Compute":
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