new_vlan = vlans.index(1) # will throw if none available
vlans[new_vlan] = 0
allocated.append(new_vlan)
- if count is 1:
+ if count == 1:
return allocated[0]
return allocated
vlans = set(vlans)
for vlan in vlans:
- if my_vlans[vlan] is 0:
+ if my_vlans[vlan] == 0:
raise ValueError("vlan " + str(vlan) + " is not available")
my_vlans[vlan] = 0
"""
try:
lab = Lab.objects.get(name=lab_name)
- except:
+ except Exception:
raise PermissionDenied("Lab not found")
if lab.api_token == token:
return LabManager(lab)
d['revoke'] = self.revoke
try:
d['context'] = json.loads(self.context)
- except:
+ except Exception:
pass
return d
job = None
try:
job = Job.objects.get(booking=booking)
- except:
+ except Exception:
job = Job.objects.create(status=JobStatus.NEW, booking=booking)
cls.makeHardwareConfigs(
hosts=hosts,
hardware_config = None
try:
hardware_config = HardwareConfig.objects.get(relation__host=host)
- except:
+ except Exception:
hardware_config = HardwareConfig()
relation = HostHardwareRelation()
network_config = None
try:
network_config = NetworkConfig.objects.get(relation__host=host)
- except:
+ except Exception:
network_config = NetworkConfig.objects.create()
relation = HostNetworkRelation()
software_relation = SoftwareRelation.objects.create(job=job, config=software_config)
software_relation.save()
return software_relation
- except:
+ except Exception:
return None
raise InvalidHostnameException("Hostname must comply to RFC 952 and all extensions to it until this point")
# check that image os is compatible with installer
if installer in image.os.sup_installers.all():
- #if installer not here, we can omit that and not check for scenario
+ # if installer not here, we can omit that and not check for scenario
if not scenario:
raise IncompatibleScenarioForInstaller("An OPNFV Installer needs a scenario to be chosen to work properly")
if scenario not in installer.sup_scenarios.all():
raise ImageOwnershipInvalid("You are not the owner of the chosen private image")
# check if host type is available
- #ResourceManager.getInstance().acquireHost(ghost, lab.name)
+ # ResourceManager.getInstance().acquireHost(ghost, lab.name)
available_host_types = ResourceManager.getInstance().getAvailableHostTypes(lab)
- if not profile in available_host_types:
+ if profile not in available_host_types:
# TODO: handle deleting generic resource in this instance along with grb
raise HostNotAvailable("Could not book selected host due to changed availability. Try again later")
booking.pdf = ResourceManager().makePDF(booking.resource)
booking.config_bundle = cbundle
booking.save()
- print("users field:")
- print(users_field)
- print(type(users_field))
- #users_field = json.loads(users_field)
users_field = users_field[2:-2]
- if users_field: #may be empty after split, if no collaborators entered
+ if users_field: # may be empty after split, if no collaborators entered
users_field = json.loads(users_field)
for collaborator in users_field:
user = User.objects.get(id=collaborator['id'])
size = 0
try:
size = int(disk_data['size'].split('.')[0])
- except:
+ except Exception:
size = int(disk_data['size'].split('.')[0][:-1])
DiskProfile.objects.create(
size=size,
from celery import shared_task
from django.utils import timezone
-from django.db.models import Q
from booking.models import Booking
from notifier.manager import NotificationHandler
from api.models import Job, JobStatus, SoftwareRelation, HostHardwareRelation, HostNetworkRelation, AccessRelation
if vlan.public:
try:
host.lab.vlan_manager.release_public_vlan(vlan.vlan_id)
- except: # will fail if we already released in this loop
+ except Exception: # will fail if we already released in this loop
pass
else:
vlans.append(vlan.vlan_id)
if resource:
try:
booking.resource.template = resource
- except:
+ except Exception:
booking.resource = ResourceBundle(template=resource)
models['booking'] = booking
self.repo_put(self.repo.BOOKING_MODELS, models)
chosen_bundle = booking.config_bundle
default.append(chosen_bundle.id)
bundle = chosen_bundle
- except:
+ except Exception:
if created_bundle:
default.append(created_bundle.id)
bundle = created_bundle
SNAPSHOT_DESC = "description of the snapshot"
BOOKING_INFO_FILE = "the INFO.yaml file for this user's booking"
- #migratory elements of segmented workflow
- #each of these is the end result of a different workflow.
+ # migratory elements of segmented workflow
+ # each of these is the end result of a different workflow.
HAS_RESULT = "whether or not workflow has a result"
RESULT_KEY = "key for target index that result will be put into in parent"
RESULT = "result object from workflow"
try:
interface.host = interface.host
interface.save()
- except Exception as e:
+ except Exception:
return "GRB, saving interface " + str(interface) + " failed. CODE:0x0019"
else:
return "GRB, no interface set provided. CODE:0x001a"
host_profile = host.profile
break
excluded_images = Image.objects.exclude(owner=user).exclude(public=True)
- excluded_images = excluded_images | Image.objects.exclude(host_type=host.profile)
+ excluded_images = excluded_images | Image.objects.exclude(host_type=host_profile)
lab = self.repo_get(self.repo.SELECTED_GRESOURCE_BUNDLE).lab
excluded_images = excluded_images | Image.objects.exclude(from_lab=lab)
filter_data["id_form-" + str(i) + "-image"] = []
q.filter(host_type=host.profile)
q.filter(from_lab=lab)
q.get(id=image.id) # will throw exception if image is not in q
- except:
+ except Exception:
self.metastep.set_invalid("Image " + image.name + " is not compatible with host " + host.resource.name)
role = form.cleaned_data['role']
if "jumphost" in role.name.lower():
from workflow.resource_bundle_workflow import Define_Hardware, Define_Nets, Resource_Meta_Info
from workflow.sw_bundle_workflow import Config_Software, Define_Software, SWConf_Resource_Select
from workflow.snapshot_workflow import Select_Host_Step, Image_Meta_Step
-from workflow.models import Repository, Confirmation_Step
+from workflow.models import Confirmation_Step
import uuid
workflow_type = 2
color = "#00ffcc"
+
class MetaStep(object):
UNTOUCHED = 0
def __ne__(self, other):
return self.id.int != other.id.int
+
class Workflow(object):
def __init__(self, steps, metasteps, repository):
self.repository = repository
self.metasteps = metasteps
self.active_index = 0
+
class WorkflowFactory():
booking_steps = [
Booking_Resource_Select,
from django.http import JsonResponse
-import random
-
from booking.models import Booking
-from workflow.workflow_factory import WorkflowFactory, MetaStep
+from workflow.workflow_factory import WorkflowFactory
from workflow.models import Repository
from resource_inventory.models import (
GenericResourceBundle,
repo.set_defaults(defaults)
repo.el[repo.HAS_RESULT] = False
repo.el[repo.SESSION_USER] = self.owner
- self.workflows.append(self.factory.create_workflow(workflow_type=workflow_type, repo = repo))
+ self.workflows.append(
+ self.factory.create_workflow(
+ workflow_type=workflow_type,
+ repo=repo
+ )
+ )
def pop_workflow(self):
- if( len(self.workflows) <= 1 ):
+ if(len(self.workflows) <= 1):
return False
if self.workflows[-1].repository.el[self.workflows[-1].repository.HAS_RESULT]:
responsejson["active"] = self.active_workflow().repository.el['active_step']
responsejson["workflow_count"] = len(self.workflows)
return JsonResponse(responsejson, safe=False)
- except Exception as e:
+ except Exception:
pass
def render(self, request, **kwargs):
Code Review / pharos-tools.git / commitdiff