# 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 api.models import ( Job, JobStatus, JobFactory, HostNetworkRelation, HostHardwareRelation, SoftwareRelation, AccessConfig, SnapshotRelation ) from resource_inventory.models import ( OPNFVRole, HostProfile, ConfigState, ) from django.test import TestCase, Client from dashboard.testing_utils import ( 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.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() def setUp(self): self.booking, self.compute_hostnames, self.jump_hostname = self.create_multinode_generic_booking() 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}] compute_networks = [{"name": "private", "tagged": True, "public": False}] 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"] = "net" + str(i) universal_networks.append(net) 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") return booking, compute_hostnames, "jump" def make_networks(self, hostprofile, nets): """ Distribute 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 ################################################################# # Complete Job Tests ################################################################# def test_complete_job_makes_access_configs(self): JobFactory.makeCompleteJob(self.booking) job = Job.objects.get(booking=self.booking) self.assertIsNotNone(job) access_configs = AccessConfig.objects.filter(accessrelation__job=job) vpn_configs = access_configs.filter(access_type="vpn") ssh_configs = access_configs.filter(access_type="ssh") self.assertFalse(AccessConfig.objects.exclude(access_type__in=["vpn", "ssh"]).exists()) all_users = list(self.booking.collaborators.all()) all_users.append(self.booking.owner) for user in all_users: self.assertTrue(vpn_configs.filter(user=user).exists()) self.assertTrue(ssh_configs.filter(user=user).exists()) def test_complete_job_makes_network_configs(self): JobFactory.makeCompleteJob(self.booking) job = Job.objects.get(booking=self.booking) self.assertIsNotNone(job) booking_hosts = self.booking.resource.hosts.all() netrelations = HostNetworkRelation.objects.filter(job=job) netconfigs = [r.config for r in netrelations] netrelation_hosts = [r.host for r in netrelations] for config in netconfigs: 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_complete_job_makes_hardware_configs(self): JobFactory.makeCompleteJob(self.booking) job = Job.objects.get(booking=self.booking) self.assertIsNotNone(job) hardware_relations = HostHardwareRelation.objects.filter(job=job) 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 hardware_relations: self.assertTrue(relation.host in booking_hosts) self.assertEqual(relation.status, JobStatus.NEW) config = relation.config host = relation.host self.assertEqual(config.get_delta()["hostname"], host.template.resource.name) def test_complete_job_makes_software_configs(self): JobFactory.makeCompleteJob(self.booking) 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.host_opnfv_config.first().role.name if str(role_name).lower() == "jumphost": self.assertEqual(host.template.resource.name, self.jump_hostname) elif str(role_name).lower() == "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 test_make_snapshot_task(self): host = self.booking.resource.hosts.first() image = make_image(self.lab, -1, None, None, host.profile) Job.objects.create(booking=self.booking) JobFactory.makeSnapshotTask(image, self.booking, host) snap_relation = SnapshotRelation.objects.get(job=self.booking.job) config = snap_relation.config self.assertEqual(host.id, config.host.id) self.assertEqual(config.dashboard_id, image.id) self.assertEqual(snap_relation.snapshot.id, image.id) def test_make_hardware_configs(self): hosts = self.booking.resource.hosts.all() job = Job.objects.create(booking=self.booking) JobFactory.makeHardwareConfigs(hosts=hosts, job=job) hardware_relations = HostHardwareRelation.objects.filter(job=job) self.assertEqual(hardware_relations.count(), hosts.count()) host_set = set([h.id for h in hosts]) for relation in hardware_relations: try: host_set.remove(relation.host.id) except KeyError: self.fail("Hardware Relation/Config not created for host " + str(relation.host)) # TODO: ConfigState needs to be fixed in factory methods relation.config.state = ConfigState.NEW self.assertEqual(relation.config.get_delta()["power"], "on") self.assertTrue(relation.config.get_delta()["ipmi_create"]) # TODO: the rest of hwconf attrs self.assertEqual(len(host_set), 0)