The flake8 command in test.sh finds no longer finds any errors.
This may form a basis of a jenkins verify job as a sort of 'weak compile-time checks'
The flake8 command will not complain about line length, and will not complain about
django's manage.py file
Change-Id: Ic47cb4fc7ada55e64485661ab6881aef475018ff
Signed-off-by: Parker Berberian <pberberian@iol.unh.edu>
# 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
-##############################################################################
\ No newline at end of file
+##############################################################################
# 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
-##############################################################################
\ No newline at end of file
+##############################################################################
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-
-
'key_cert': key_cert
}
- return JIRA(server=settings.JIRA_URL, oauth=oauth_dict)
\ No newline at end of file
+ return JIRA(server=settings.JIRA_URL, oauth=oauth_dict)
def upload_to(object, filename):
return object.user.username + '/' + filename
+
class UserProfile(models.Model):
user = models.OneToOneField(User, on_delete=models.CASCADE)
timezone = models.CharField(max_length=100, blank=False, default='UTC')
net = PublicNetwork.objects.get(lab=self.lab_set.first(), vlan=vlan)
return not net.in_use
-
def is_available(self, vlans):
"""
'vlans' is either a single vlan id integer or a list of integers
self.save()
-
class Lab(models.Model):
lab_user = models.OneToOneField(User, on_delete=models.CASCADE)
name = models.CharField(max_length=200, primary_key=True, unique=True, null=False, blank=False)
key += random.choice(alphabet)
return key
-
def __str__(self):
return self.name
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-
-
The timezone should be UTC for anonymous users, for authenticated users it should be set
to user.userprofile.timezone
"""
- #default
+ # default
self.assertEqual(timezone.get_current_timezone_name(), 'UTC')
url = reverse('account:settings')
self.client.login(username='user2', password='user2')
self.client.get(url)
self.assertTrue(user2.userprofile)
-
-
"""
from django.conf.urls import url
-from account.views import *
+from account.views import (
+ AccountSettingsView,
+ JiraAuthenticatedView,
+ JiraLoginView,
+ JiraLogoutView,
+ UserListView,
+ account_resource_view,
+ account_booking_view,
+ account_images_view,
+ account_configuration_view,
+ account_detail_view
+)
app_name = "account"
urlpatterns = [
# Step 1. Get a request token from Jira.
try:
resp, content = client.request(settings.OAUTH_REQUEST_TOKEN_URL, "POST")
- except Exception as e:
+ except Exception:
messages.add_message(self.request, messages.ERROR,
'Error: Connection to Jira failed. Please contact an Administrator')
return '/'
self.request.session['request_token'] = dict(urllib.parse.parse_qsl(content.decode()))
# Step 3. Redirect the user to the authentication URL.
url = settings.OAUTH_AUTHORIZE_URL + '?oauth_token=' + \
- self.request.session['request_token']['oauth_token'] + \
- '&oauth_callback=' + settings.OAUTH_CALLBACK_URL
+ self.request.session['request_token']['oauth_token'] + \
+ '&oauth_callback=' + settings.OAUTH_CALLBACK_URL
return url
# Step 2. Request the authorized access token from Jira.
try:
resp, content = client.request(settings.OAUTH_ACCESS_TOKEN_URL, "POST")
- except Exception as e:
+ except Exception:
messages.add_message(self.request, messages.ERROR,
'Error: Connection to Jira failed. Please contact an Administrator')
return '/'
template = "account/details.html"
return render(request, template)
+
def account_resource_view(request):
"""
gathers a users genericResoureBundles and
context = {"resources": resources, "title": "My Resources"}
return render(request, template, context=context)
+
def account_booking_view(request):
if not request.user.is_authenticated:
return render(request, "dashboard/login.html", {'title': 'Authentication Required'})
context = {"title": "My Bookings", "bookings": bookings, "collab_bookings": collab_bookings}
return render(request, template, context=context)
+
def account_configuration_view(request):
if not request.user.is_authenticated:
return render(request, "dashboard/login.html", {'title': 'Authentication Required'})
context = {"title": "Configuration List", "configurations": configs}
return render(request, template, context=context)
+
def account_images_view(request):
if not request.user.is_authenticated:
return render(request, "dashboard/login.html", {'title': 'Authentication Required'})
template = "account/image_list.html"
my_images = Image.objects.filter(owner=request.user)
public_images = Image.objects.filter(public=True)
- context = {"title": "Images", "images": my_images, "public_images": public_images }
+ context = {"title": "Images", "images": my_images, "public_images": public_images}
return render(request, template, context=context)
-
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-
-
from django.apps import AppConfig
from django.contrib import admin
-from api.models import *
+from api.models import (
+ Job,
+ OpnfvApiConfig,
+ HardwareConfig,
+ NetworkConfig,
+ SoftwareConfig,
+ AccessRelation,
+ SoftwareRelation,
+ HostHardwareRelation,
+ HostNetworkRelation,
+)
class ApiConfig(AppConfig):
name = 'apiJobs'
+
admin.site.register(Job)
admin.site.register(OpnfvApiConfig)
admin.site.register(HardwareConfig)
##############################################################################
+from django.contrib.auth.models import User
from django.db import models
from django.core.exceptions import PermissionDenied
import json
import uuid
-from resource_inventory.models import *
from booking.models import Booking
+from resource_inventory.models import (
+ Lab,
+ HostProfile,
+ Host,
+ Image,
+ Interface
+)
class JobStatus(object):
prof = {}
prof['name'] = self.lab.name
prof['contact'] = {
- "phone": self.lab.contact_phone,
- "email": self.lab.contact_email
- }
+ "phone": self.lab.contact_phone,
+ "email": self.lab.contact_email
+ }
prof['host_count'] = []
for host in HostProfile.objects.filter(labs=self.lab):
count = Host.objects.filter(profile=host, lab=self.lab).count()
- prof['host_count'].append({
- "type": host.name,
- "count": count
- })
+ prof['host_count'].append(
+ {
+ "type": host.name,
+ "count": count
+ }
+ )
return prof
def get_inventory(self):
def serialize_images(self, images):
images_ser = []
for image in images:
- images_ser.append({
- "name": image.name,
- "lab_id": image.lab_id,
- "dashboard_id": image.id
- })
+ images_ser.append(
+ {
+ "name": image.name,
+ "lab_id": image.lab_id,
+ "dashboard_id": image.id
+ }
+ )
return images_ser
def serialize_host_profiles(self, profiles):
for profile in profiles:
p = {}
p['cpu'] = {
- "cores": profile.cpuprofile.first().cores,
- "arch": profile.cpuprofile.first().architecture,
- "cpus": profile.cpuprofile.first().cpus,
- }
+ "cores": profile.cpuprofile.first().cores,
+ "arch": profile.cpuprofile.first().architecture,
+ "cpus": profile.cpuprofile.first().cpus,
+ }
p['disks'] = []
for disk in profile.storageprofile.all():
d = {
- "size": disk.size,
- "type": disk.media_type,
- "name": disk.name
- }
+ "size": disk.size,
+ "type": disk.media_type,
+ "name": disk.name
+ }
p['disks'].append(d)
p['description'] = profile.description
p['interfaces'] = []
for iface in profile.interfaceprofile.all():
- p['interfaces'].append({
- "speed": iface.speed,
- "name": iface.name
- })
+ p['interfaces'].append(
+ {
+ "speed": iface.speed,
+ "name": iface.name
+ }
+ )
p['ram'] = {"amount": profile.ramprofile.first().amount}
p['name'] = profile.name
return False
return True
-
def get_delta(self, status):
d = {}
j = {}
def clear_delta(self):
self.delta = '{}'
+
class OpnfvApiConfig(models.Model):
installer = models.CharField(max_length=100)
self.save()
return json.loads(self.delta)
+
class AccessConfig(TaskConfig):
access_type = models.CharField(max_length=50)
user = models.ForeignKey(User, on_delete=models.CASCADE)
d['context'] = context
self.delta = json.dumps(d)
+
class SoftwareConfig(TaskConfig):
"""
handled opnfv installations, etc
def to_json(self):
return json.dumps(self.to_dict())
+
class HardwareConfig(TaskConfig):
"""
handles imaging, user accounts, etc
except:
job = Job.objects.create(status=JobStatus.NEW, booking=booking)
cls.makeHardwareConfigs(
- hosts=hosts,
- job=job
- )
+ hosts=hosts,
+ job=job
+ )
cls.makeNetworkConfigs(
- hosts=hosts,
- job=job
- )
+ hosts=hosts,
+ job=job
+ )
cls.makeSoftware(
- hosts=hosts,
- job=job
- )
+ hosts=hosts,
+ job=job
+ )
all_users = list(booking.collaborators.all())
all_users.append(booking.owner)
cls.makeAccessConfig(
- users=all_users,
- access_type="vpn",
- revoke=False,
- job=job
- )
+ users=all_users,
+ access_type="vpn",
+ revoke=False,
+ job=job
+ )
for user in all_users:
try:
cls.makeAccessConfig(
- users=[user],
- access_type="ssh",
- revoke=False,
- job=job,
- context={
- "key": user.userprofile.ssh_public_key.read(),
- "hosts": [host.labid for host in hosts]
- }
- )
+ users=[user],
+ access_type="ssh",
+ revoke=False,
+ job=job,
+ context={
+ "key": user.userprofile.ssh_public_key.read(),
+ "hosts": [host.labid for host in hosts]
+ }
+ )
except Exception:
continue
from rest_framework import serializers
-from resource_inventory.models import *
+from resource_inventory.models import (
+ HostConfiguration,
+ CpuProfile,
+ DiskProfile,
+ InterfaceProfile,
+ RamProfile,
+ Image,
+ Interface
+)
+
class BookingField(serializers.Field):
host_configs[host.name] = HostConfiguration.objects.get(host=host.template)
if "jumphost" not in ser and host_configs[host.name].opnfvRole.name.lower() == "jumphost":
ser['jumphost'] = host.name
- #host is a Host model
+ # host is a Host model
for i in range(len(host.interfaces.all())):
interface = host.interfaces.all()[i]
- #interface is an Interface model
+ # interface is an Interface model
for vlan in interface.config.all():
- #vlan is Vlan model
+ # vlan is Vlan model
if vlan.id not in networks:
networks[vlan.id] = []
- net_host = {"hostname": host.name, "tagged": vlan.tagged, "interface":i}
+ net_host = {"hostname": host.name, "tagged": vlan.tagged, "interface": i}
networks[vlan.id].append(net_host)
- #creates networking object of proper form
+ # creates networking object of proper form
networking = []
for vlanid in networks:
network = {}
ser['networking'] = networking
- #creates hosts object of correct form
+ # creates hosts object of correct form
hosts = []
for hostname in host_configs:
host = {"hostname": hostname}
"""
return None
+
class BookingSerializer(serializers.Serializer):
booking = BookingField()
-#Host Type stuff, for inventory
+# Host Type stuff, for inventory
class CPUSerializer(serializers.ModelSerializer):
class Meta:
model = CpuProfile
fields = ('cores', 'architecture', 'cpus')
+
class DiskSerializer(serializers.ModelSerializer):
class Meta:
model = DiskProfile
fields = ('size', 'media_type', 'name')
+
class InterfaceProfileSerializer(serializers.ModelSerializer):
class Meta:
model = InterfaceProfile
fields = ('speed', 'name')
+
class RamSerializer(serializers.ModelSerializer):
class Meta:
model = RamProfile
fields = ('amount', 'channels')
+
class HostTypeSerializer(serializers.Serializer):
name = serializers.CharField(max_length=200)
ram = RamSerializer()
disks = DiskSerializer()
cpu = CPUSerializer()
-#the rest of the inventory stuff
+
+# the rest of the inventory stuff
class NetworkSerializer(serializers.Serializer):
cidr = serializers.CharField(max_length=200)
gateway = serializers.IPAddressField(max_length=200)
vlan = serializers.IntegerField()
+
class ImageSerializer(serializers.ModelSerializer):
lab_id = serializers.IntegerField()
id = serializers.IntegerField(source="dashboard_id")
name = serializers.CharField(max_length=50)
description = serializers.CharField(max_length=200)
+
class Meta:
model = Image
+
class InterfaceField(serializers.Field):
def to_representation(self, interface):
pass
port_name=port_name
)
+
class InventoryHostSerializer(serializers.Serializer):
hostname = serializers.CharField(max_length=100)
host_type = serializers.CharField(max_length=100)
# 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
-##############################################################################
\ No newline at end of file
+##############################################################################
##############################################################################
from django.test import TestCase
from booking.models import Booking
-from resource_inventory.models import *
from account.models import Lab
-from api.serializers.booking_serializer import *
+from api.serializers.booking_serializer import BookingField
from datetime import timedelta
from django.utils import timezone
from django.contrib.auth.models import Permission, User
+from resource_inventory.models import (
+ Image,
+ OPNFVRole,
+ HostConfiguration,
+ HostProfile,
+ InterfaceProfile,
+ DiskProfile,
+ CpuProfile,
+ RamProfile,
+ GenericResourceBundle,
+ GenericResource,
+ GenericHost,
+ Host,
+ Vlan,
+ Interface,
+ ConfigBundle,
+ ResourceBundle
+)
class BookingSerializerTestCase(TestCase):
lab_user = User.objects.create(username="asfasdfasdf")
owner = User.objects.create(username="asfasdfasdffffff")
lab = Lab.objects.create(
- lab_user=lab_user,
- name="TestLab123123",
- contact_email="mail@email.com",
- contact_phone=""
- )
- jumphost=True
+ lab_user=lab_user,
+ name="TestLab123123",
+ contact_email="mail@email.com",
+ contact_phone=""
+ )
+ jumphost = True
for host in hosts:
image = Image.objects.create(
- lab_id=12,
- from_lab=lab,
- name="this is a test image",
- owner=owner
- )
+ lab_id=12,
+ from_lab=lab,
+ name="this is a test image",
+ owner=owner
+ )
name = "jumphost"
if not jumphost:
name = "compute"
role = OPNFVRole.objects.create(
- name=name,
- description="stuff"
- )
+ name=name,
+ description="stuff"
+ )
HostConfiguration.objects.create(
- host=host,
- image=image,
- bundle=config,
- opnfvRole=role
- )
- jumphost=False
-
+ host=host,
+ image=image,
+ bundle=config,
+ opnfvRole=role
+ )
+ jumphost = False
def setUp(self):
self.serializer = BookingField()
name='Test profile',
description='a test profile'
)
- interfaceProfile = InterfaceProfile.objects.create(
+ InterfaceProfile.objects.create(
speed=1000,
name='eno3',
host=hostProfile
)
- diskProfile = DiskProfile.objects.create(
+ DiskProfile.objects.create(
size=1000,
media_type="SSD",
name='/dev/sda',
host=hostProfile
)
- cpuProfile = CpuProfile.objects.create(
+ CpuProfile.objects.create(
cores=96,
architecture="x86_64",
cpus=2,
host=hostProfile
)
- ramProfile = RamProfile.objects.create(
+ RamProfile.objects.create(
amount=256,
channels=4,
host=hostProfile
)
- #create GenericResourceBundle
+ # create GenericResourceBundle
genericBundle = GenericResourceBundle.objects.create()
gres1 = GenericResource.objects.create(
conf = ConfigBundle.objects.create(owner=user1, name="test conf")
self.makeHostConfigurations([gHost1, gHost2], conf)
- #actual resource bundle
+ # actual resource bundle
bundle = ResourceBundle.objects.create(
- template = genericBundle
+ template=genericBundle
)
host1 = Host.objects.create(
# finally, can create booking
self.booking = Booking.objects.create(
owner=user1,
- start = timezone.now(),
- end = timezone.now() + timedelta(weeks=1),
+ start=timezone.now(),
+ end=timezone.now() + timedelta(weeks=1),
purpose='Testing',
resource=bundle,
config_bundle=conf
from django.urls import path
from rest_framework import routers
-from api.views import *
+from api.views import (
+ BookingViewSet,
+ UserViewSet,
+ lab_profile,
+ lab_status,
+ lab_inventory,
+ specific_job,
+ specific_task,
+ new_jobs,
+ current_jobs,
+ done_jobs,
+ GenerateTokenView
+)
router = routers.DefaultRouter()
router.register(r'bookings', BookingViewSet)
from rest_framework.authtoken.models import Token
from django.views.decorators.csrf import csrf_exempt
-import json
-
-from api.serializers.booking_serializer import *
+from api.serializers.booking_serializer import BookingSerializer
from api.serializers.old_serializers import UserSerializer
from account.models import UserProfile
from booking.models import Booking
-from api.models import *
+from api.models import LabManagerTracker, get_task
from notifier.manager import NotificationHandler
from django.contrib import admin
-from booking.models import *
+from booking.models import Booking
admin.site.register(Booking)
from resource_inventory.models import ResourceBundle, ConfigBundle
from account.models import Lab
-from django.conf import settings
from django.contrib.auth.models import User
from django.db import models
-from jira import JIRA
-from jira import JIRAError
import resource_inventory.resource_manager
class Booking(models.Model):
id = models.AutoField(primary_key=True)
- owner = models.ForeignKey(User, models.CASCADE, related_name='owner') # delete if user is deleted
+ owner = models.ForeignKey(User, models.CASCADE, related_name='owner')
collaborators = models.ManyToManyField(User, related_name='collaborators')
start = models.DateTimeField()
end = models.DateTimeField()
jira_issue_status = models.CharField(max_length=50, blank=True)
purpose = models.CharField(max_length=300, blank=False)
ext_count = models.IntegerField(default=2)
- resource = models.ForeignKey(ResourceBundle, on_delete=models.SET_NULL, null=True) #need to decide behavior here on delete
+ resource = models.ForeignKey(ResourceBundle, on_delete=models.SET_NULL, null=True)
config_bundle = models.ForeignKey(ConfigBundle, on_delete=models.SET_NULL, null=True)
project = models.CharField(max_length=100, default="", blank=True, null=True)
lab = models.ForeignKey(Lab, null=True, on_delete=models.SET_NULL)
Will return a dictionary of names and 2-D array of x and y data points.
e.g. {"plot1": [["x1", "x2", "x3"],["y1", "y2", "y3]]}
x values will be dates in string
- every change (booking start / end) will be reflected, instead of one data point per day
- y values are the integer number of bookings/users active at some point in the given date
- span is the number of days to plot. The last x value will always be the current time
+ every change (booking start / end) will be reflected,
+ instead of one data point per day
+ y values are the integer number of bookings/users active at
+ some point in the given date span is the number of days to plot.
+ The last x value will always be the current time
"""
x_set = set()
x = []
users = []
now = datetime.datetime.now(pytz.utc)
delta = datetime.timedelta(days=span)
- end = now-delta
+ end = now - delta
bookings = Booking.objects.filter(start__lte=now, end__gte=end)
for booking in bookings:
x_set.add(booking.start)
from datetime import timedelta
-from django.contrib.auth.models import Permission
+from django.contrib.auth.models import Permission, User
from django.test import TestCase
from django.utils import timezone
-from booking.models import *
+# from booking.models import *
+from booking.models import Booking
from resource_inventory.models import ResourceBundle, GenericResourceBundle, ConfigBundle
+
class BookingModelTestCase(TestCase):
+
count = 0
+
def setUp(self):
self.owner = User.objects.create(username='owner')
self.res1 = ResourceBundle.objects.create(
- template=GenericResourceBundle.objects.create(name="gbundle" + str(self.count))
- )
+ template=GenericResourceBundle.objects.create(
+ name="gbundle" + str(self.count)
+ )
+ )
self.count += 1
self.res2 = ResourceBundle.objects.create(
- template=GenericResourceBundle.objects.create(name="gbundle2" + str(self.count))
- )
+ template=GenericResourceBundle.objects.create(
+ name="gbundle2" + str(self.count)
+ )
+ )
self.count += 1
self.user1 = User.objects.create(username='user1')
self.user1.user_permissions.add(self.add_booking_perm)
self.user1 = User.objects.get(pk=self.user1.id)
- self.config_bundle = ConfigBundle.objects.create(owner=self.user1, name="test config")
+ self.config_bundle = ConfigBundle.objects.create(
+ owner=self.user1,
+ name="test config"
+ )
def test_start_end(self):
"""
- if the start of a booking is greater or equal then the end, saving should raise a
- ValueException
+ if the start of a booking is greater or equal then the end,
+ saving should raise a ValueException
"""
start = timezone.now()
end = start - timedelta(weeks=1)
resource=self.res1,
owner=self.user1,
config_bundle=self.config_bundle
- )
+ )
end = start
self.assertRaises(
ValueError,
resource=self.res1,
owner=self.user1,
config_bundle=self.config_bundle
- )
+ )
def test_conflicts(self):
"""
- saving an overlapping booking on the same resource should raise a ValueException
+ saving an overlapping booking on the same resource
+ should raise a ValueException
saving for different resources should succeed
"""
start = timezone.now()
owner=self.user1,
resource=self.res1,
config_bundle=self.config_bundle
- )
)
+ )
self.assertRaises(
ValueError,
resource=self.res1,
owner=self.user1,
config_bundle=self.config_bundle
- )
+ )
+
self.assertRaises(
ValueError,
Booking.objects.create,
resource=self.res1,
owner=self.user1,
config_bundle=self.config_bundle
- )
+ )
self.assertRaises(
ValueError,
resource=self.res1,
owner=self.user1,
config_bundle=self.config_bundle
- )
+ )
self.assertRaises(
ValueError,
resource=self.res1,
owner=self.user1,
config_bundle=self.config_bundle
- )
+ )
self.assertRaises(
ValueError,
resource=self.res1,
owner=self.user1,
config_bundle=self.config_bundle
- )
+ )
self.assertRaises(
ValueError,
resource=self.res1,
owner=self.user1,
config_bundle=self.config_bundle
- )
+ )
self.assertTrue(
Booking.objects.create(
owner=self.user1,
resource=self.res1,
config_bundle=self.config_bundle
- )
)
+ )
+
self.assertTrue(
Booking.objects.create(
start=end,
owner=self.user1,
resource=self.res1,
config_bundle=self.config_bundle
- )
)
+ )
self.assertTrue(
Booking.objects.create(
owner=self.user1,
resource=self.res1,
config_bundle=self.config_bundle
- )
)
+ )
self.assertTrue(
Booking.objects.create(
owner=self.user1,
resource=self.res1,
config_bundle=self.config_bundle
- )
)
+ )
self.assertTrue(
Booking.objects.create(
owner=self.user1,
resource=self.res2,
config_bundle=self.config_bundle
- )
)
+ )
def test_extensions(self):
"""
self.assertTrue(booking.save())
except Exception:
self.fail("save() threw an exception")
-
"""
from django.conf.urls import url
-from booking.views import *
+from booking.views import (
+ booking_detail_view,
+ BookingDeleteView,
+ bookingDelete,
+ BookingListView,
+ booking_stats_view,
+ booking_stats_json
+)
app_name = "booking"
urlpatterns = [
from django.contrib import messages
from django.shortcuts import get_object_or_404
from django.http import JsonResponse
-from django.urls import reverse
from django.utils import timezone
from django.views import View
-from django.views.generic import FormView
from django.views.generic import TemplateView
from django.shortcuts import redirect, render
import json
allowed_users.add(booking.owner)
if user not in allowed_users:
return render(request, "dashboard/login.html", {'title': 'This page is private'})
- return render(request, "booking/booking_detail.html", {
- 'title': 'Booking Details',
- 'booking': booking,
- 'pdf': ResourceManager().makePDF(booking.resource),
- 'user_id': user.id})
+
+ return render(
+ request,
+ "booking/booking_detail.html",
+ {
+ 'title': 'Booking Details',
+ 'booking': booking,
+ 'pdf': ResourceManager().makePDF(booking.resource),
+ 'user_id': user.id
+ })
def booking_stats_view(request):
return render(
- request,
- "booking/stats.html",
- context={
- "data": StatisticsManager.getContinuousBookingTimeSeries(),
- "title": "Booking Statistics"
- }
- )
+ request,
+ "booking/stats.html",
+ context={"data": StatisticsManager.getContinuousBookingTimeSeries(), "title": "Booking Statistics"}
+ )
def booking_stats_json(request):
##############################################################################
from django.conf import settings
+
def debug(context):
return {'DEBUG': settings.DEBUG}
"""
pass
+
class ModelValidationException(Exception):
"""
Validation before saving model returned issues
"""
pass
+
class ResourceAvailabilityException(ResourceProvisioningException):
"""
Requested resources are not *currently* available
"""
pass
+
class ResourceExistenceException(ResourceAvailabilityException):
"""
Requested resources do not exist or do not match any known resources
class NonUniqueHostnameException(Exception):
pass
+
class InvalidHostnameException(Exception):
pass
+
class InvalidVlanConfigurationException(Exception):
pass
+
class NetworkExistsException(Exception):
pass
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-
-from django.test import TestCase
-from booking.models import Booking
-from resource_inventory.models import *
-from account.models import *
-from api.serializers.booking_serializer import *
-from datetime import timedelta
-from django.utils import timezone
-from django.contrib.auth.models import Permission, User
import json
import yaml
+from account.models import Lab, UserProfile
+from django.contrib.auth.models import User
+from resource_inventory.models import (
+ HostProfile,
+ InterfaceProfile,
+ DiskProfile,
+ CpuProfile,
+ RamProfile,
+ VlanManager,
+ Scenario,
+ Installer,
+ Opsys,
+ OPNFVRole,
+ Image,
+ Interface,
+ Host
+)
+
class Populator:
self.generic_bundle_count = 0
self.booking_count = 0
-
def make_host_profile(self, lab, data):
hostProfile = HostProfile.objects.create(
host_type=data['host']['type'],
)
interfaceProfile.save()
-
for disk_data in data['disks']:
diskProfile = DiskProfile.objects.create(
user_sbergeron.save()
user_sbergeron_prof = UserProfile.objects.create(user=user_sbergeron)
user_sbergeron_prof.save()
- return [user_sbergeron, user_pberberian,]
-
+ return [user_sbergeron, user_pberberian]
def make_labs(self):
unh_iol = User.objects.create(username="unh_iol")
unh_iol.save()
vlans = []
reserved = []
- for i in range(1,4096):
+ for i in range(1, 4096):
vlans.append(1)
reserved.append(0)
- # TODO: put reserved vlans here
iol = Lab.objects.create(
- lab_user=unh_iol,
- name="UNH_IOL",
- vlan_manager=VlanManager.objects.create(
- vlans = json.dumps(vlans),
- reserved_vlans = json.dumps(reserved),
- allow_overlapping = False,
- block_size = 20,
- ),
- api_token = Lab.make_api_token(),
- contact_email = "nfv-lab@iol.unh.edu",
- location = "University of New Hampshire, Durham NH, 03824 USA"
- )
+ lab_user=unh_iol,
+ name="UNH_IOL",
+ vlan_manager=VlanManager.objects.create(
+ vlans=json.dumps(vlans),
+ reserved_vlans=json.dumps(reserved),
+ allow_overlapping=False,
+ block_size=20,
+ ),
+ api_token=Lab.make_api_token(),
+ contact_email="nfv-lab@iol.unh.edu",
+ location="University of New Hampshire, Durham NH, 03824 USA"
+ )
return [iol]
-
def make_configurations(self):
- #scenarios
+ # scenarios
scen1 = Scenario.objects.create(name="os-nosdn-nofeature-noha")
scen2 = Scenario.objects.create(name="os-odl-kvm-ha")
scen3 = Scenario.objects.create(name="os-nosdn-nofeature-ha")
- #installers
+ # installers
fuel = Installer.objects.create(name="Fuel")
fuel.sup_scenarios.add(scen1)
fuel.sup_scenarios.add(scen3)
compass.sup_scenarios.add(scen3)
compass.save()
- #operating systems
+ # operating systems
ubuntu = Opsys.objects.create(name="Ubuntu")
ubuntu.sup_installers.add(compass)
ubuntu.sup_installers.add(joid)
suse.sup_installers.add(fuel)
suse.save()
-
- #opnfv roles
- compute = OPNFVRole.objects.create(name="Compute", description="Does the heavy lifting")
- controller = OPNFVRole.objects.create(name="Controller", description="Controls everything")
- jumphost = OPNFVRole.objects.create(name="Jumphost", description="Entry Point")
+ # opnfv roles
+ OPNFVRole.objects.create(name="Compute", description="Does the heavy lifting")
+ OPNFVRole.objects.create(name="Controller", description="Controls everything")
+ OPNFVRole.objects.create(name="Jumphost", description="Entry Point")
lab = Lab.objects.first()
user = UserProfile.objects.first().user
- image = Image.objects.create(
- lab_id=23,
- name="hpe centos",
- from_lab=lab,
- owner=user,
- host_type=HostProfile.objects.get(name="hpe")
- )
- image = Image.objects.create(
- lab_id=25,
- name="hpe ubuntu",
- from_lab=lab,
- owner=user,
- host_type=HostProfile.objects.get(name="hpe")
- )
+ Image.objects.create(
+ lab_id=23,
+ name="hpe centos",
+ from_lab=lab,
+ owner=user,
+ host_type=HostProfile.objects.get(name="hpe")
+ )
+ Image.objects.create(
+ lab_id=25,
+ name="hpe ubuntu",
+ from_lab=lab,
+ owner=user,
+ host_type=HostProfile.objects.get(name="hpe")
+ )
- image = Image.objects.create(
- lab_id=26,
- name="hpe suse",
- from_lab=lab,
- owner=user,
- host_type=HostProfile.objects.get(name="hpe")
- )
- image = Image.objects.create(
- lab_id=27,
- name="arm ubuntu",
- from_lab=lab,
- owner=user,
- host_type=HostProfile.objects.get(name="arm")
- )
+ Image.objects.create(
+ lab_id=26,
+ name="hpe suse",
+ from_lab=lab,
+ owner=user,
+ host_type=HostProfile.objects.get(name="hpe")
+ )
+
+ Image.objects.create(
+ lab_id=27,
+ name="arm ubuntu",
+ from_lab=lab,
+ owner=user,
+ host_type=HostProfile.objects.get(name="arm")
+ )
def make_lab_hosts(self, hostcount, profile, lab, data, offset=1):
for i in range(hostcount):
- name="Host_" + lab.name + "_" + profile.name + "_" + str(i + offset)
+ name = "Host_" + lab.name + "_" + profile.name + "_" + str(i + offset)
host = Host.objects.create(
- name=name,
- lab=lab,
- profile=profile,
- labid=data[i]['labid']
- )
+ name=name,
+ lab=lab,
+ profile=profile,
+ labid=data[i]['labid']
+ )
for iface_profile in profile.interfaceprofile.all():
iface_data = data[i]['interfaces'][iface_profile.name]
Interface.objects.create(
- mac_address=iface_data['mac'],
- bus_address=iface_data['bus'],
- name=iface_profile.name,
- host=host
- )
+ mac_address=iface_data['mac'],
+ bus_address=iface_data['bus'],
+ name=iface_profile.name,
+ host=host
+ )
def make_profile_data(self):
"""
for prof in ["hpe", "arm"]: # TODO
profile_dict = {}
host = {
- "name": prof,
- "type": 0,
- "description": "some LaaS servers"
- }
+ "name": prof,
+ "type": 0,
+ "description": "some LaaS servers"
+ }
profile_dict['host'] = host
profile_dict['interfaces'] = []
for interface in [{"name": "eno1", "speed": 1000}, {"name": "eno2", "speed": 10000}]: # TODO
if len(host_data_dict) < 1:
continue
host_profile = HostProfile.objects.create(
- name=host_profile_name,
- description=""
- )
+ name=host_profile_name,
+ description=""
+ )
host_profile.labs.add(lab)
example_host_data = list(host_data_dict.values())[0]
cpu_data = example_host_data['cpu']
CpuProfile.objects.create(
- cores=cpu_data['cores'],
- architecture=cpu_data['arch'],
- cpus=cpu_data['cpus'],
- host=host_profile
- )
+ cores=cpu_data['cores'],
+ architecture=cpu_data['arch'],
+ cpus=cpu_data['cpus'],
+ host=host_profile
+ )
ram_data = example_host_data['memory']
RamProfile.objects.create(
- amount=int(ram_data[:-1]),
- channels=1,
- host=host_profile
- )
+ amount=int(ram_data[:-1]),
+ channels=1,
+ host=host_profile
+ )
disks_data = example_host_data['disk']
for disk_data in disks_data:
size = 0
try:
- size=int(disk_data['size'].split('.')[0])
+ size = int(disk_data['size'].split('.')[0])
except:
- size=int(disk_data['size'].split('.')[0][:-1])
+ size = int(disk_data['size'].split('.')[0][:-1])
DiskProfile.objects.create(
- size=size,
- media_type="SSD",
- name=disk_data['name'],
- host=host_profile
- )
+ size=size,
+ media_type="SSD",
+ name=disk_data['name'],
+ host=host_profile
+ )
ifaces_data = example_host_data['interface']
for iface_data in ifaces_data:
InterfaceProfile.objects.create(
- speed=iface_data['speed'],
- name=iface_data['name'],
- host=host_profile
- )
+ speed=iface_data['speed'],
+ name=iface_data['name'],
+ host=host_profile
+ )
# all profiles created
for hostname, host_data in host_data_dict.items():
host = Host.objects.create(
- name=hostname,
- labid=hostname,
- profile=host_profile,
- lab=lab
- )
+ name=hostname,
+ labid=hostname,
+ profile=host_profile,
+ lab=lab
+ )
for iface_data in host_data['interface']:
Interface.objects.create(
- mac_address=iface_data['mac'],
- bus_address=iface_data['busaddr'],
- name=iface_data['name'],
- host=host
- )
+ mac_address=iface_data['mac'],
+ bus_address=iface_data['busaddr'],
+ name=iface_data['name'],
+ host=host
+ )
def populate(self):
self.labs = self.make_labs()
from django.db.models import Q
from booking.models import Booking
from notifier.manager import NotificationHandler
-from api.models import *
+from api.models import JobStatus, SoftwareRelation, HostHardwareRelation, HostNetworkRelation, AccessRelation
from resource_inventory.resource_manager import ResourceManager
config.clear_delta()
config.set_power("off")
config.save()
- hostrelation.status=JobStatus.NEW
+ hostrelation.status = JobStatus.NEW
hostrelation.save()
def cleanup_network(qs):
interface.config.clear()
network.add_interface(interface)
network.save()
- hostrelation.status=JobStatus.NEW
+ hostrelation.status = JobStatus.NEW
hostrelation.save()
def cleanup_software(qs):
software = relation.config.opnfv
software.clear_delta()
software.save()
- relation.status=JobStatus.NEW
+ relation.status = JobStatus.NEW
relation.save()
def cleanup_access(qs):
hardware = ~Q(~Q(job__hosthardwarerelation__status=200))
bookings = Booking.objects.filter(
- networks,
- hardware,
- end__lt=timezone.now(),
- job__complete=True,
- resource__isnull=False
- )
+ networks,
+ hardware,
+ end__lt=timezone.now(),
+ job__complete=True,
+ resource__isnull=False
+ )
for booking in bookings:
ResourceManager.getInstance().deleteResourceBundle(booking.resource)
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-
-
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-
-
2. Add a URL to urlpatterns: url(r'^blog/', include('blog.urls'))
"""
from django.conf.urls import url
-from dashboard.views import *
+from dashboard.views import (
+ landing_view,
+ lab_list_view,
+ lab_detail_view,
+ host_profile_detail_view
+)
-app_name="dashboard"
+app_name = "dashboard"
urlpatterns = [
url(r'^$', landing_view, name='index'),
url(r'^lab/$', lab_list_view, name='all_labs'),
from django.shortcuts import render
from django.http import HttpResponseRedirect
-from booking.models import Booking
from account.models import Lab
-from resource_inventory.models import *
-from workflow.views import *
-from workflow.workflow_manager import *
+from resource_inventory.models import Image, HostProfile
+from workflow.views import create_session
+from workflow.workflow_manager import ManagerTracker
def lab_list_view(request):
if user:
images = images | Image.objects.filter(from_lab=lab).filter(owner=user)
- return render(request, "dashboard/lab_detail.html",
- {'title': "Lab Overview",
- 'lab': lab,
- 'hostprofiles': lab.hostprofiles.all(),
- 'images': images})
+ return render(
+ request,
+ "dashboard/lab_detail.html",
+ {
+ 'title': "Lab Overview",
+ 'lab': lab,
+ 'hostprofiles': lab.hostprofiles.all(),
+ 'images': images
+ }
+ )
def host_profile_detail_view(request):
- return render(request, "dashboard/host_profile_detail.html",
- {'title': "Host Types",
- })
+ return render(
+ request,
+ "dashboard/host_profile_detail.html",
+ {
+ 'title': "Host Types",
+ }
+ )
def landing_view(request):
try:
manager = ManagerTracker.managers[request.session['manager_session']]
-
- except KeyError as e:
+ except KeyError:
pass
if manager is not None:
- #no manager detected, don't display continue button
+ # no manager detected, don't display continue button
manager_detected = True
if request.method == 'GET':
request.session['manager_session'] = mgr_uuid
return HttpResponseRedirect('/wf/')
- except KeyError as e:
+ except KeyError:
pass
+#!/usr/bin/env python
+
##############################################################################
# Copyright (c) 2016 Max Breitenfeldt and others.
#
##############################################################################
-#!/usr/bin/env python
import os
import sys
the last is the title for the collaborators'
"""
owner_notif = Notification.objects.create(
- title=titles[0],
- content=render_to_string(template, context={
+ title=titles[0],
+ content=render_to_string(
+ template,
+ context={
"booking": booking,
"owner": True
- })
- )
+ }
+ )
+ )
owner_notif.recipients.add(booking.owner)
if not booking.collaborators.all().exists():
return # no collaborators - were done
collab_notif = Notification.objects.create(
- title=titles[-1],
- content=render_to_string(template, context={
+ title=titles[-1],
+ content=render_to_string(
+ template,
+ context={
"booking": booking,
"owner": False
- })
- )
+ }
+ )
+ )
for c in booking.collaborators.all():
collab_notif.recipients.add(c)
# gather up all the relevant messages from the lab
for task in all_tasks:
if (not hasattr(task, "user")) or task.user == user:
- user_tasklist.append({
- "title": task.type_str + " Message: ",
- "content": task.message
- })
+ user_tasklist.append(
+ {
+ "title": task.type_str + " Message: ",
+ "content": task.message
+ }
+ )
# gather up all the other needed info
context = {
- "user_name": user.userprofile.full_name,
- "messages": user_tasklist,
- "booking_url": os.environ.get("DASHBOARD_URL", "<Dashboard url>") + "/booking/detail/" + str(job.booking.id) + "/"
- }
+ "user_name": user.userprofile.full_name,
+ "messages": user_tasklist,
+ "booking_url": os.environ.get("DASHBOARD_URL", "<Dashboard url>") + "/booking/detail/" + str(job.booking.id) + "/"
+ }
# render email template
message = render_to_string(template_name, context)
# finally, send the email
send_mail(
- "Your Booking is Ready",
- message,
- os.environ.get("DEFAULT_FROM_EMAIL", "opnfv@pharos-dashboard"),
- user.userprofile.email_addr,
- fail_silently=False
- )
+ "Your Booking is Ready",
+ message,
+ os.environ.get("DEFAULT_FROM_EMAIL", "opnfv@pharos-dashboard"),
+ user.userprofile.email_addr,
+ fail_silently=False
+ )
@classmethod
def email_booking_over(cls, booking):
users.append(booking.owner)
for user in users:
context = {
- "user_name": user.userprofile.full_name,
- "booking": booking,
- "hosts": hostnames,
- "booking_url": os.environ.get("DASHBOARD_URL", "<Dashboard url>") + "/booking/detail/" + str(booking.id) + "/"
- }
+ "user_name": user.userprofile.full_name,
+ "booking": booking,
+ "hosts": hostnames,
+ "booking_url": os.environ.get("DASHBOARD_URL", "<Dashboard url>") + "/booking/detail/" + str(booking.id) + "/"
+ }
message = render_to_string(template_name, context)
send_mail(
- "Your Booking has Expired",
- message,
- os.environ.get("DEFAULT_FROM_EMAIL", "opnfv@pharos-dashboard"),
- user.userprofile.email_addr,
- fail_silently=False
- )
+ "Your Booking has Expired",
+ message,
+ os.environ.get("DEFAULT_FROM_EMAIL", "opnfv@pharos-dashboard"),
+ user.userprofile.email_addr,
+ fail_silently=False
+ )
@classmethod
def task_updated(cls, task):
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-
from django.test import TestCase
-from notifier.models import *
-from django.contrib.auth.models import User
+
class DispatchTestCase(TestCase):
# This is a stub, it will be filled out as this feature is remade with saner practices.
from django.test import TestCase
-from notifier.models import *
+from notifier.models import Notifier
from django.contrib.auth.models import User
+
class NotifierTestCase(TestCase):
def test_valid_notifier_saves(self):
from django.conf.urls import url
-from notifier.views import *
+from notifier.views import InboxView, NotificationView
app_name = "notifier"
urlpatterns = [
-
-
url(r'^$', InboxView, name='messages'),
- url(r'^notification/(?P<notification_id>[0-9]+)/$', NotificationView, name='notifier_single')
+ url(r'^notification/(?P<notification_id>[0-9]+)/$', NotificationView, name='notifier_single')
]
@app.task(bind=True)
def debug_task(self):
- print('Request: {0!r}'.format(self.request))
\ No newline at end of file
+ print('Request: {0!r}'.format(self.request))
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
- 'DIRS': [os.path.join(BASE_DIR, 'templates')]
- ,
+ 'DIRS': [os.path.join(BASE_DIR, 'templates')],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
EMAIL_PORT = os.environ['EMAIL_PORT']
EMAIL_HOST_USER = os.environ['EMAIL_HOST_USER']
EMAIL_HOST_PASSWORD = os.environ['EMAIL_HOST_PASSWORD']
-EMAIL_USE_TLS=True
+EMAIL_USE_TLS = True
DEFAULT_EMAIL_FROM = os.environ.get('DEFAULT_EMAIL_FROM', 'webmaster@localhost')
-SESSION_ENGINE="django.contrib.sessions.backends.signed_cookies"
+SESSION_ENGINE = "django.contrib.sessions.backends.signed_cookies"
from django.contrib import admin
-from resource_inventory.models import *
+from resource_inventory.models import (
+ HostProfile,
+ InterfaceProfile,
+ DiskProfile,
+ CpuProfile,
+ RamProfile,
+ GenericResourceBundle,
+ GenericResource,
+ GenericHost,
+ GenericInterface,
+ Host,
+ Interface,
+ Network,
+ Vlan,
+ ResourceBundle,
+ Scenario,
+ Installer,
+ Opsys,
+ ConfigBundle,
+ OPNFVConfig,
+ OPNFVRole,
+ Image,
+ HostConfiguration
+)
profiles = [HostProfile, InterfaceProfile, DiskProfile, CpuProfile, RamProfile]
speed = models.IntegerField()
name = models.CharField(max_length=100)
host = models.ForeignKey(HostProfile, on_delete=models.DO_NOTHING, related_name='interfaceprofile')
- nic_type = models.CharField(max_length=50, choices=[
- ("onboard", "onboard"),
- ("pcie", "pcie")
- ], default="onboard")
+ nic_type = models.CharField(
+ max_length=50,
+ choices=[
+ ("onboard", "onboard"),
+ ("pcie", "pcie")
+ ],
+ default="onboard"
+ )
def __str__(self):
return self.name + " for " + str(self.host)
name = models.CharField(max_length=50)
host = models.ForeignKey(HostProfile, on_delete=models.DO_NOTHING, related_name='storageprofile')
rotation = models.IntegerField(default=0)
- interface = models.CharField(max_length=50, choices=[
+ interface = models.CharField(
+ max_length=50,
+ choices=[
("sata", "sata"),
("sas", "sas"),
("ssd", "ssd"),
("nvme", "nvme"),
("scsi", "scsi"),
("iscsi", "iscsi"),
- ], default="sata")
+ ],
+ default="sata"
+ )
def __str__(self):
return self.name + " for " + str(self.host)
return str(self.amount) + "G for " + str(self.host)
-##Networking -- located here due to import order requirements
+# Networking -- located here due to import order requirements
class Network(models.Model):
id = models.AutoField(primary_key=True)
vlan_id = models.IntegerField()
def __str__(self):
return self.name
+
class Vlan(models.Model):
id = models.AutoField(primary_key=True)
vlan_id = models.IntegerField()
def __str__(self):
return self.name
+
class Installer(models.Model):
id = models.AutoField(primary_key=True)
name = models.CharField(max_length=200)
def __str__(self):
return self.name
+
class Opsys(models.Model):
id = models.AutoField(primary_key=True)
name = models.CharField(max_length=100)
def __str__(self):
return self.name
+
class ConfigBundle(models.Model):
id = models.AutoField(primary_key=True)
- owner = models.ForeignKey(User, on_delete=models.CASCADE) #consider setting to root user?
+ owner = models.ForeignKey(User, on_delete=models.CASCADE) # consider setting to root user?
name = models.CharField(max_length=200, unique=True)
description = models.CharField(max_length=1000, default="")
bundle = models.ForeignKey(GenericResourceBundle, null=True, on_delete=models.CASCADE)
def __str__(self):
return self.name
+
class OPNFVConfig(models.Model):
id = models.AutoField(primary_key=True)
installer = models.ForeignKey(Installer, on_delete=models.CASCADE)
def __str__(self):
return "OPNFV job with " + str(self.installer) + " and " + str(self.scenario)
+
class OPNFVRole(models.Model):
id = models.AutoField(primary_key=True)
name = models.CharField(max_length=200)
def __str__(self):
return self.name
+
class Image(models.Model):
"""
model for representing OS images / snapshots of hosts
name = models.CharField(max_length=200)
owner = models.ForeignKey(User, null=True, on_delete=models.SET_NULL)
public = models.BooleanField(default=True)
- host_type = models.ForeignKey(HostProfile, on_delete=models.CASCADE) #may need to change to models.SET() once images are transferrable between compatible host types
+ # may need to change host_type.on_delete to models.SET() once images are transferrable between compatible host types
+ host_type = models.ForeignKey(HostProfile, on_delete=models.CASCADE)
description = models.TextField()
def __str__(self):
return self.name
+
class HostConfiguration(models.Model):
"""
model to represent a complete configuration for a single
host = models.ForeignKey(GenericHost, related_name="configuration", on_delete=models.CASCADE)
image = models.ForeignKey(Image, on_delete=models.PROTECT)
bundle = models.ForeignKey(ConfigBundle, related_name="hostConfigurations", null=True, on_delete=models.CASCADE)
- opnfvRole = models.ForeignKey(OPNFVRole, on_delete=models.PROTECT) #need protocol for phasing out a role if we are going to allow that to happen
+ opnfvRole = models.ForeignKey(OPNFVRole, on_delete=models.PROTECT)
def __str__(self):
return "config with " + str(self.host) + " and image " + str(self.image)
##############################################################################
-from django.core.exceptions import *
from django.template.loader import render_to_string
import booking
-from dashboard.exceptions import *
-from resource_inventory.models import *
+from dashboard.exceptions import (
+ ResourceExistenceException,
+ ResourceAvailabilityException,
+ ResourceProvisioningException,
+ ModelValidationException
+)
+from resource_inventory.models import Host, HostConfiguration, ResourceBundle
+
class ResourceManager:
ResourceManager.instance = ResourceManager()
return ResourceManager.instance
- #public interface
+ # public interface
def deleteResourceBundle(self, resourceBundle):
for host in Host.objects.filter(bundle=resourceBundle):
self.releaseHost(host)
hosts = genericResourceBundle.getHosts()
- #current supported case: user creating new booking
- #currently unsupported: editing existing booking
+ # current supported case: user creating new booking
+ # currently unsupported: editing existing booking
physical_hosts = []
for host in hosts:
- host_config=None
+ host_config = None
if config:
host_config = HostConfiguration.objects.get(bundle=config, host=host)
try:
for vlan in generic_interface.vlans.all():
physical_interface.config.add(vlan)
- #private interface
+ # private interface
def acquireHost(self, genericHost, labName):
host_full_set = Host.objects.filter(lab__name__exact=labName, profile=genericHost.profile)
if not host_full_set.first():
booking_owner = booking.models.Booking.objects.get(resource=resource).owner
owner = booking_owner.username
email = booking_owner.userprofile.email_addr
- except Exception as e:
+ except Exception:
pass
details['owner'] = owner
return pdf_nodes
-
def get_pdf_host(self, host):
host_info = {}
host_info['name'] = host.template.resource.name
from resource.inventory_manager import InventoryManager
from resource.resource_manager import ResourceManager
-from resource.models import *
+from account.models import Lab
+from resource.models import (
+ Host,
+ Vlan,
+ Interface,
+ ResourceBundle,
+ GenericHost,
+ GenericResourceBundle,
+ CpuProfile,
+ RamProfile,
+ DiskProfile,
+ HostProfile,
+ InterfaceProfile
+)
class InventoryManagerTestCase(TestCase):
name='Test profile',
description='a test profile'
)
- interfaceProfile = InterfaceProfile.objects.create(
+ InterfaceProfile.objects.create(
speed=1000,
name='eno3',
host=hostProfile
)
- diskProfile = DiskProfile.objects.create(
+ DiskProfile.objects.create(
size=1000,
media_type="SSD",
name='/dev/sda',
host=hostProfile
)
- cpuProfile = CpuProfile.objects.create(
+ CpuProfile.objects.create(
cores=96,
architecture="x86_64",
cpus=2,
host=hostProfile
)
- ramProfile = RamProfile.objects.create(
+ RamProfile.objects.create(
amount=256,
channels=4,
host=hostProfile
)
- #create GenericResourceBundle
+ # create GenericResourceBundle
genericBundle = GenericResourceBundle.objects.create()
self.gHost1 = GenericHost.objects.create(
profile=hostProfile
)
- #actual resource bundle
+ # actual resource bundle
bundle = ResourceBundle.objects.create(template=genericBundle)
self.host1 = Host.objects.create(
vlan1 = Vlan.objects.create(vlan_id=300, tagged=False)
vlan2 = Vlan.objects.create(vlan_id=300, tagged=False)
- iface1 = Interface.objects.create(
+ Interface.objects.create(
mac_address='00:11:22:33:44:55',
bus_address='some bus address',
switch_name='switch1',
config=vlan1,
host=self.host1
)
- iface2 = Interface.objects.create(
+ Interface.objects.create(
mac_address='00:11:22:33:44:56',
bus_address='some bus address',
switch_name='switch1',
name='Test profile',
description='a test profile'
)
- interfaceProfile = InterfaceProfile.objects.create(
+ InterfaceProfile.objects.create(
speed=1000,
name='eno3',
host=hostProfile
)
- diskProfile = DiskProfile.objects.create(
+ DiskProfile.objects.create(
size=1000,
media_type="SSD",
name='/dev/sda',
host=hostProfile
)
- cpuProfile = CpuProfile.objects.create(
+ CpuProfile.objects.create(
cores=96,
architecture="x86_64",
cpus=2,
host=hostProfile
)
- ramProfile = RamProfile.objects.create(
+ RamProfile.objects.create(
amount=256,
channels=4,
host=hostProfile
)
- #create GenericResourceBundle
+ # create GenericResourceBundle
genericBundle = GenericResourceBundle.objects.create()
self.gHost1 = GenericHost.objects.create(
profile=hostProfile
)
- #actual resource bundle
+ # actual resource bundle
bundle = ResourceBundle.objects.create(template=genericBundle)
self.host1 = Host.objects.create(
vlan1 = Vlan.objects.create(vlan_id=300, tagged=False)
vlan2 = Vlan.objects.create(vlan_id=300, tagged=False)
- iface1 = Interface.objects.create(
+ Interface.objects.create(
mac_address='00:11:22:33:44:55',
bus_address='some bus address',
switch_name='switch1',
config=vlan1,
host=self.host1
)
- iface2 = Interface.objects.create(
+ Interface.objects.create(
mac_address='00:11:22:33:44:56',
bus_address='some bus address',
switch_name='switch1',
)
def test_convert_bundle(self):
- bundle = ResourceManager.getInstance().convertResoureBundle(self.genericBundle, self.lab.name)
+ ResourceManager.getInstance().convertResoureBundle(self.genericBundle, self.lab.name)
# verify bundle configuration
from django.test import TestCase
from django.contrib.auth.models import User
from account.models import Lab
-from resource_inventory.models import *
+from resource_inventory.models import (
+ Scenario,
+ Installer,
+ Opsys,
+ ConfigBundle,
+ OPNFVConfig,
+ OPNFVRole,
+ Image,
+ HostProfile,
+ GenericResourceBundle,
+ GenericResource,
+ GenericHost,
+ HostConfiguration
+)
class ConfigUtil():
- count=0
+ count = 0
@staticmethod
def makeScenario():
@staticmethod
def makeInstaller():
- inst = Installer.objects.create(
- name = "testInstaller"
- )
+ inst = Installer.objects.create(name="testInstaller")
inst.sup_scenarios = [ConfigUtil.makeScenario()]
return inst
@staticmethod
def makeOpsys():
- os = Opsys.objects.create(
- name = "test Operating System"
- )
+ os = Opsys.objects.create(name="test Operating System")
os.sup_installers = [ConfigUtil.makeInstaller()]
return os
def makeConfigBundle():
user = User.objects.create(username="test_user" + str(ConfigUtil.count))
ConfigUtil.count += 1
- return ConfigBundle.objects.create(
- owner = user
- )
+ return ConfigBundle.objects.create(owner=user)
@staticmethod
def makeOPNFVConfig():
scenario = ConfigUtil.makeScenario()
bundle = ConfigUtil.makeConfigBundle()
return OPNFVConfig.objects.create(
- installer=installer,
- scenario=scenario,
- bundle=bundle
- )
+ installer=installer,
+ scenario=scenario,
+ bundle=bundle
+ )
@staticmethod
def makeOPNFVRole():
return OPNFVRole.objects.create(
- name="Test role",
- description="This is a test role"
- )
+ name="Test role",
+ description="This is a test role"
+ )
@staticmethod
def makeImage():
owner = User.objects.create(username="another test user")
lab_user = User.objects.create(username="labUserForTests")
lab = Lab.objects.create(
- lab_user=lab_user,
- name="this is lab for testing",
- contact_email="email@mail.com",
- contact_phone="123-4567"
- )
+ lab_user=lab_user,
+ name="this is lab for testing",
+ contact_email="email@mail.com",
+ contact_phone="123-4567"
+ )
return Image.objects.create(
- lab_id=0,
- from_lab=lab,
- name="an image for testing",
- owner=owner
- )
-
+ lab_id=0,
+ from_lab=lab,
+ name="an image for testing",
+ owner=owner
+ )
@staticmethod
def makeGenericHost():
profile = HostProfile.objects.create(
- host_type=0,
- name="test lab for config bundle",
- description="this is a test profile"
- )
+ host_type=0,
+ name="test lab for config bundle",
+ description="this is a test profile"
+ )
user = User.objects.create(username="test sample user 12")
bundle = GenericResourceBundle.objects.create(
- name="Generic bundle for config tests",
- xml="",
- owner=user,
- description=""
- )
+ name="Generic bundle for config tests",
+ xml="",
+ owner=user,
+ description=""
+ )
resource = GenericResource.objects.create(
- bundle=bundle,
- name="a test generic resource"
- )
+ bundle=bundle,
+ name="a test generic resource"
+ )
return GenericHost.objects.create(
- profile=profile,
- resource=resource
- )
+ profile=profile,
+ resource=resource
+ )
@staticmethod
def makeHostConfiguration():
bundle = ConfigUtil.makeConfigBundle()
opnfvRole = ConfigUtil.makeOPNFVRole()
return HostConfiguration.objects.create(
- host=host,
- image=image,
- bundle=bundle,
- opnfvRole=opnfvRole
- )
+ host=host,
+ image=image,
+ bundle=bundle,
+ opnfvRole=opnfvRole
+ )
class ScenarioTestCase(TestCase):
def test_save(self):
self.assertTrue(ConfigUtil.makeScenario())
+
class InstallerTestCase(TestCase):
def test_save(self):
self.assertTrue(ConfigUtil.makeInstaller())
+
class OperatingSystemTestCase(TestCase):
def test_save(self):
self.assertTrue(ConfigUtil.makeOpsys())
+
class ConfigBundleTestCase(TestCase):
def test_save(self):
self.assertTrue(ConfigUtil.makeConfigBundle())
+
class OPNFVConfigTestCase(TestCase):
def test_save(self):
self.assertTrue(ConfigUtil.makeOPNFVConfig())
+
class OPNFVRoleTestCase(TestCase):
def test_save(self):
##############################################################################
-from django.shortcuts import render
-from django.views import View
from django.views.generic import TemplateView
from resource_inventory.models import Host
+
class HostView(TemplateView):
template_name = "resource/hosts.html"
def get_context_data(self, **kwargs):
context = super(HostView, self).get_context_data(**kwargs)
hosts = Host.objects.filter(working=True)
- context.update({'hosts':hosts, 'title':"Hardware Resources"})
+ context.update({'hosts': hosts, 'title': "Hardware Resources"})
return context
import json
from datetime import timedelta
-from account.models import UserProfile
from booking.models import Booking
from workflow.models import WorkflowStep
-from workflow.forms import ResourceSelectorForm, SWConfigSelectorForm, BookingMetaForm, ConfirmationForm
+from workflow.forms import ResourceSelectorForm, SWConfigSelectorForm, BookingMetaForm
from resource_inventory.models import GenericResourceBundle, ResourceBundle, ConfigBundle
+
class Resource_Select(WorkflowStep):
template = 'booking/steps/resource_select.html'
title = "Select Resource"
edit = self.repo_get(self.repo.EDIT, False)
user = self.repo_get(self.repo.SESSION_USER)
context['form'] = ResourceSelectorForm(
- data={"user": user},
- chosen_resource=default,
- bundle=bundle,
- edit=edit
- )
+ data={"user": user},
+ chosen_resource=default,
+ bundle=bundle,
+ edit=edit
+ )
return context
- def post_render(self, request):
+ def post_render(self, request):
form = ResourceSelectorForm(request.POST)
context = self.get_context()
if form.is_valid():
gresource_bundle = GenericResourceBundle.objects.get(id=selected_id)
except ValueError:
# we want the bundle in the repo
- gresource_bundle = self.repo_get(self.repo.GRESOURCE_BUNDLE_MODELS,{}).get("bundle", GenericResourceBundle())
+ gresource_bundle = self.repo_get(
+ self.repo.GRESOURCE_BUNDLE_MODELS,
+ {}
+ ).get("bundle", GenericResourceBundle())
self.repo_put(
- self.repo_key,
- gresource_bundle
- )
+ self.repo_key,
+ gresource_bundle
+ )
confirm = self.repo_get(self.repo.CONFIRMATION)
if self.confirm_key not in confirm:
confirm[self.confirm_key] = {}
self.metastep.set_invalid("Please complete the fields highlighted in red to continue")
return render(request, self.template, context)
+
class Booking_Resource_Select(Resource_Select):
def __init__(self, *args, **kwargs):
try:
config_bundle = self.repo_get(self.repo.BOOKING_MODELS)['booking'].config_bundle
if default:
- return default # select created grb, even if preselected config bundle
+ return default # select created grb, even if preselected config bundle
return config_bundle.bundle
except:
pass
self.repo_put(self.repo.BOOKING_MODELS, models)
return response
+
class SWConfig_Select(WorkflowStep):
template = 'booking/steps/swconfig_select.html'
title = "Select Software Configuration"
return self.render(request)
-
def get_context(self):
context = super(SWConfig_Select, self).get_context()
default = []
try:
chosen_bundle = booking.config_bundle
default.append(chosen_bundle.id)
- bundle=chosen_bundle
+ bundle = chosen_bundle
except:
if created_bundle:
default.append("repo bundle")
context['form'] = SWConfigSelectorForm(chosen_software=default, bundle=bundle, edit=edit, resource=grb)
return context
+
class Booking_Meta(WorkflowStep):
template = 'booking/steps/booking_meta.html'
title = "Extra Info"
users = models.get("collaborators", [])
for user in users:
default.append(user.id)
- except Exception as e:
+ except Exception:
pass
default_user = self.repo_get(self.repo.SESSION_USER)
user_data = form.cleaned_data['users']
confirm['booking']['collaborators'] = []
- user_data = user_data[2:-2] #fixes malformed string from querydict
+ user_data = user_data[2:-2] # fixes malformed string from querydict
if user_data:
form_users = json.loads(user_data)
for user_json in form_users:
import django.forms as forms
from django.forms import widgets
-from django.contrib.auth.models import User
from django.utils.safestring import mark_safe
from django.template.loader import render_to_string
-from django.core import serializers
from django.forms.widgets import NumberInput
-from django.db.models import F
-import json
-
-from resource_inventory.models import *
from account.models import Lab
from account.models import UserProfile
+from resource_inventory.models import (
+ GenericResourceBundle,
+ ConfigBundle,
+ OPNFVRole,
+ Image,
+ Installer,
+ Scenario
+)
class SearchableSelectMultipleWidget(widgets.SelectMultiple):
context = self.get_context(attrs)
return mark_safe(render_to_string(self.template_name, context))
- def get_context(self,attrs):
- return {'items':self.items,
- 'name':self.name,
- 'show_from_noentry':self.show_from_noentry,
- 'show_x_results':self.show_x_results,
- 'results_scrollable':self.results_scrollable,
- 'selectable_limit':self.selectable_limit,
- 'placeholder':self.placeholder,
- 'initial':self.initial,
- 'default_entry':self.default_entry,
- 'edit': self.edit,
- 'wf_type': self.wf_type
+ def get_context(self, attrs):
+ return {
+ 'items': self.items,
+ 'name': self.name,
+ 'show_from_noentry': self.show_from_noentry,
+ 'show_x_results': self.show_x_results,
+ 'results_scrollable': self.results_scrollable,
+ 'selectable_limit': self.selectable_limit,
+ 'placeholder': self.placeholder,
+ 'initial': self.initial,
+ 'default_entry': self.default_entry,
+ 'edit': self.edit,
+ 'wf_type': self.wf_type
}
+
class ResourceSelectorForm(forms.Form):
def __init__(self, data=None, **kwargs):
bundle = kwargs.pop("bundle")
if "edit" in kwargs:
edit = kwargs.pop("edit")
- super(ResourceSelectorForm, self).__init__(data=data,**kwargs)
+ super(ResourceSelectorForm, self).__init__(data=data, **kwargs)
queryset = GenericResourceBundle.objects.select_related("owner").all()
if data and 'user' in data:
queryset = queryset.filter(owner=data['user'])
attrs = self.build_search_widget_attrs(chosen_resource, bundle, edit, queryset)
self.fields['generic_resource_bundle'] = forms.CharField(
- widget=SearchableSelectMultipleWidget(attrs=attrs)
- )
+ widget=SearchableSelectMultipleWidget(attrs=attrs)
+ )
def build_search_widget_attrs(self, chosen_resource, bundle, edit, queryset):
resources = {}
displayable['string'] = bundle.description
displayable['id'] = "repo bundle"
resources["repo bundle"] = displayable
- attrs={
+ attrs = {
'set': resources,
'show_from_noentry': "true",
'show_x_results': -1,
}
return attrs
+
class SWConfigSelectorForm(forms.Form):
def __init__(self, *args, **kwargs):
edit = kwargs.pop("edit")
if "resource" in kwargs:
resource = kwargs.pop("resource")
- super(SWConfigSelectorForm, self).__init__(*args,**kwargs)
- attrs = self.build_search_widget_attrs(chosen_software,bundle, edit, resource)
+ super(SWConfigSelectorForm, self).__init__(*args, **kwargs)
+ attrs = self.build_search_widget_attrs(chosen_software, bundle, edit, resource)
self.fields['software_bundle'] = forms.CharField(
- widget=SearchableSelectMultipleWidget(attrs=attrs)
- )
+ widget=SearchableSelectMultipleWidget(attrs=attrs)
+ )
def build_search_widget_attrs(self, chosen, bundle, edit, resource):
configs = {}
displayable['id'] = "repo bundle"
configs['repo bundle'] = displayable
- attrs={
+ attrs = {
'set': configs,
'show_from_noentry': "true",
'show_x_results': -1,
}
return attrs
+
class BookingMetaForm(forms.Form):
- length = forms.IntegerField(widget=NumberInput(attrs={'type':'range', 'min':"0", "max":"21", "value":"0"}))
+ length = forms.IntegerField(
+ widget=NumberInput(
+ attrs={
+ "type": "range",
+ 'min': "0",
+ "max": "21",
+ "value": "0"
+ }
+ )
+ )
purpose = forms.CharField(max_length=1000)
project = forms.CharField(max_length=400)
info_file = forms.CharField(max_length=1000, required=False)
self.fields['users'] = forms.CharField(
widget=SearchableSelectMultipleWidget(
attrs=self.build_search_widget_attrs(chosen_users, default_user=default_user)
- ),
+ ),
required=False
- )
+ )
def build_user_list(self):
"""
"""
try:
users = {}
- d_qset = UserProfile.objects.select_related('user').all().exclude(user__username=self.default_user);
+ d_qset = UserProfile.objects.select_related('user').all().exclude(user__username=self.default_user)
for userprofile in d_qset:
user = {
- 'id':userprofile.user.id,
- 'expanded_name':userprofile.full_name,
- 'small_name':userprofile.user.username,
- 'string':userprofile.email_addr
- }
+ 'id': userprofile.user.id,
+ 'expanded_name': userprofile.full_name,
+ 'small_name': userprofile.user.username,
+ 'string': userprofile.email_addr
+ }
users[userprofile.user.id] = user
return users
- except Exception as e:
+ except Exception:
pass
def build_search_widget_attrs(self, chosen_users, default_user="you"):
- attrs={
+ attrs = {
'set': self.build_user_list(),
'show_from_noentry': "false",
'show_x_results': 10,
}
return attrs
+
class MultipleSelectFilterWidget(forms.Widget):
def __init__(self, attrs=None):
super(MultipleSelectFilterWidget, self).__init__(attrs)
self.attrs = attrs
- self.template_name="dashboard/multiple_select_filter_widget.html"
+ self.template_name = "dashboard/multiple_select_filter_widget.html"
def render(self, name, value, attrs=None, renderer=None):
attrs = self.attrs
def get_context(self, name, value, attrs):
return attrs
+
class MultipleSelectFilterField(forms.Field):
- def __init__( self, required=True, widget=None, label=None, initial=None,
- help_text='', error_messages=None, show_hidden_initial=False,
- validators=(), localize=False, disabled=False, label_suffix=None):
+
+ def __init__(self, required=True, widget=None, label=None, initial=None,
+ help_text='', error_messages=None, show_hidden_initial=False,
+ validators=(), localize=False, disabled=False, label_suffix=None):
"""from the documentation:
# required -- Boolean that specifies whether the field is required.
# True by default.
# label_suffix -- Suffix to be added to the label. Overrides
# form's label_suffix.
"""
- #this is bad, but django forms are annoying
- self.widget=widget
+ # this is bad, but django forms are annoying
+ self.widget = widget
if self.widget is None:
self.widget = MultipleSelectFilterWidget()
super(MultipleSelectFilterField, self).__init__(
"""
return data
+
class FormUtils:
@staticmethod
def getLabData():
filter_objects = [("labs", labs.values()), ("hosts", hosts.values())]
context = {
- 'filter_objects': filter_objects,
- 'mapping': mapping,
- 'items': items
- }
+ 'filter_objects': filter_objects,
+ 'mapping': mapping,
+ 'items': items
+ }
return context
+
class HardwareDefinitionForm(forms.Form):
def __init__(self, *args, **kwargs):
attrs = FormUtils.getLabData()
attrs['selection_data'] = selection_data
self.fields['filter_field'] = MultipleSelectFilterField(
- widget=MultipleSelectFilterWidget(
- attrs=attrs
- )
- )
+ widget=MultipleSelectFilterWidget(
+ attrs=attrs
+ )
+ )
+
class PodDefinitionForm(forms.Form):
- fields = [ "xml" ]
+ fields = ["xml"]
xml = forms.CharField()
+
class ResourceMetaForm(forms.Form):
bundle_name = forms.CharField(label="POD Name")
bundle_description = forms.CharField(label="POD Description", widget=forms.Textarea)
+
class GenericHostMetaForm(forms.Form):
host_profile = forms.CharField(label="Host Type", disabled=True, required=False)
host_name = forms.CharField(label="Host Name")
+
class NetworkDefinitionForm(forms.Form):
def __init__(self, *args, **kwargs):
- fields = []
-
super(NetworkDefinitionForm, self).__init__(**kwargs)
+
class NetworkConfigurationForm(forms.Form):
def __init__(self, *args, **kwargs):
- fields = []
-
super(NetworkConfigurationForm).__init__(**kwargs)
+
class HostSoftwareDefinitionForm(forms.Form):
fields = ["host_name", "role", "image"]
role = forms.ModelChoiceField(queryset=OPNFVRole.objects.all())
image = forms.ModelChoiceField(queryset=Image.objects.all())
+
class SoftwareConfigurationForm(forms.Form):
name = forms.CharField(max_length=200)
installer = forms.ModelChoiceField(queryset=Installer.objects.all(), disabled=True, required=False)
scenario = forms.ModelChoiceField(queryset=Scenario.objects.all(), disabled=True, required=False)
+
class WorkflowSelectionForm(forms.Form):
fields = ['workflow']
empty_permitted = False
- workflow = forms.ChoiceField( choices=(
- (0, 'Booking'),
- (1, 'Resource Bundle'),
- (2, 'Software Configuration')
+ workflow = forms.ChoiceField(
+ choices=(
+ (0, 'Booking'),
+ (1, 'Resource Bundle'),
+ (2, 'Software Configuration')
),
label="Choose Workflow",
initial='booking',
- required=True)
+ required=True
+ )
+
class SnapshotHostSelectForm(forms.Form):
host = forms.CharField()
+
class SnapshotMetaForm(forms.Form):
name = forms.CharField()
description = forms.CharField()
+
class ConfirmationForm(forms.Form):
fields = ['confirm']
- confirm = forms.ChoiceField( choices=(
- (True, "Confirm"),
- (False, "Cancel"))
+ confirm = forms.ChoiceField(
+ choices=(
+ (True, "Confirm"),
+ (False, "Cancel")
)
+ )
##############################################################################
-from django.contrib.auth.models import User
-from django.db import models
from django.shortcuts import render
from django.contrib import messages
import requests
from workflow.forms import ConfirmationForm
-from api.models import *
-from dashboard.exceptions import *
-from resource_inventory.models import *
+from api.models import JobFactory
+from dashboard.exceptions import ResourceAvailabilityException, ModelValidationException
+from resource_inventory.models import Image, GenericInterface
from resource_inventory.resource_manager import ResourceManager
from notifier.manager import NotificationHandler
+from booking.models import Booking
class BookingAuthManager():
return None
return ptl
- except Exception as e:
+ except Exception:
return None
-
def booking_allowed(self, booking, repo):
"""
This is the method that will have to change whenever the booking policy changes in the Infra
which is checked using the provided info file
"""
if len(booking.resource.template.getHosts()) < 2:
- return True #if they only have one server, we dont care
+ return True # if they only have one server, we dont care
if booking.owner.userprofile.booking_privledge:
return True # admin override for this user
if repo.BOOKING_INFO_FILE not in repo.el:
return False # INFO file not provided
ptl_info = self.parse_url(repo.BOOKING_INFO_FILE)
- return ptl_info and ptl_info == booking.owner.userprofile.email_addr
-
+ return ptl_info and ptl_info == booking.owner.userprofile.email_addr
class WorkflowStep(object):
def repo_put(self, key, value):
return self.repo.put(key, value, self.id)
+
class Confirmation_Step(WorkflowStep):
template = 'workflow/confirm.html'
title = "Confirm Changes"
return 1 # There is a problem with these vlans
return 0
-
def get_context(self):
context = super(Confirmation_Step, self).get_context()
context['form'] = ConfirmationForm()
context['confirmation_info'] = yaml.dump(
- self.repo_get(self.repo.CONFIRMATION),
- default_flow_style=False
- ).strip()
+ self.repo_get(self.repo.CONFIRMATION),
+ default_flow_style=False
+ ).strip()
context['vlan_warning'] = self.get_vlan_warning()
return context
steps = []
active_index = 0
+
class Repository():
EDIT = "editing"
SNAPSHOT_DESC = "description of the snapshot"
BOOKING_INFO_FILE = "the INFO.yaml file for this user's booking"
-
def get(self, key, default, id):
self.add_get_history(key, id)
return self.el.get(key, default)
- def put(self,key,val, id):
+ def put(self, key, val, id):
self.add_put_history(key, id)
self.el[key] = val
self.add_history(key, id, self.put_history)
def add_history(self, key, id, history):
- if not key in history:
+ if key not in history:
history[key] = [id]
else:
history[key].append(id)
errors = self.make_snapshot()
if errors:
return errors
- #if GRB WF, create it
+ # if GRB WF, create it
if self.GRESOURCE_BUNDLE_MODELS in self.el:
errors = self.make_generic_resource_bundle()
if errors:
booking = self.el[self.BOOKING_MODELS]['booking']
NotificationHandler.notify_new_booking(booking)
-
def make_snapshot(self):
owner = self.el[self.SESSION_USER]
models = self.el[self.SNAPSHOT_MODELS]
image.host_type = host.profile
image.save()
-
def make_generic_resource_bundle(self):
owner = self.el[self.SESSION_USER]
if self.GRESOURCE_BUNDLE_MODELS in self.el:
for resource_name, mapping in models['vlans'].items():
for profile_name, vlan_set in mapping.items():
interface = GenericInterface.objects.get(
- profile__name=profile_name,
- host__resource__name=resource_name,
- host__resource__bundle=models['bundle']
- )
+ profile__name=profile_name,
+ host__resource__name=resource_name,
+ host__resource__bundle=models['bundle']
+ )
for vlan in vlan_set:
try:
vlan.save()
else:
return "GRB, no vlan set provided. CODE:0x0018"
-
else:
return "GRB no models given. CODE:0x0001"
self.el[self.VALIDATED_MODEL_GRB] = bundle
return False
-
def make_software_config_bundle(self):
- owner = self.el[self.SESSION_USER]
models = self.el[self.CONFIG_MODELS]
if 'bundle' in models:
bundle = models['bundle']
if 'opnfv' in models:
opnfvconfig = models['opnfv']
opnfvconfig.bundle = opnfvconfig.bundle
- if not opnfvconfig.scenario in opnfvconfig.installer.sup_scenarios.all():
+ if opnfvconfig.scenario not in opnfvconfig.installer.sup_scenarios.all():
return "SWC, scenario not supported by installer. CODE:0x000d"
try:
opnfvconfig.save()
self.el[self.VALIDATED_MODEL_CONFIG] = bundle
return False
-
def make_booking(self):
models = self.el[self.BOOKING_MODELS]
owner = self.el[self.SESSION_USER]
vlan_manager.reserve_vlans(vlans)
vlan_manager.reserve_public_vlan(public_vlan.vlan_id)
return True
- except Exception as e:
+ except Exception:
return False
-
def __init__(self):
self.el = {}
self.el[self.CONFIRMATION] = {}
from xml.dom import minidom
from workflow.models import WorkflowStep
-from workflow.forms import *
-from resource_inventory.models import *
-from dashboard.exceptions import *
+from account.models import Lab
+from workflow.forms import (
+ HardwareDefinitionForm,
+ NetworkDefinitionForm,
+ ResourceMetaForm,
+ GenericHostMetaForm
+)
+from resource_inventory.models import (
+ GenericResourceBundle,
+ Vlan,
+ GenericInterface,
+ GenericHost,
+ GenericResource,
+ HostProfile
+)
+from dashboard.exceptions import (
+ InvalidVlanConfigurationException,
+ NetworkExistsException,
+ InvalidHostnameException,
+ NonUniqueHostnameException
+)
import logging
logger = logging.getLogger(__name__)
title = "Define Hardware"
description = "Choose the type and amount of machines you want"
short_title = "hosts"
+
def get_context(self):
context = super(Define_Hardware, self).get_context()
selection_data = {"hosts": {}, "labs": {}}
selection_data['labs'] = {"lab_" + str(models.get("bundle").lab.lab_user.id): "true"}
form = HardwareDefinitionForm(
- selection_data=selection_data
- )
+ selection_data=selection_data
+ )
context['form'] = form
return context
self.context = self.get_context()
return render(request, self.template, self.context)
-
def update_models(self, data):
data = json.loads(data['filter_field'])
models = self.repo_get(self.repo.GRESOURCE_BUNDLE_MODELS, {})
if list(lab_dict.values())[0]: # True for lab the user selected
lab_user_id = int(list(lab_dict.keys())[0].split("_")[-1])
models['bundle'].lab = Lab.objects.get(lab_user__id=lab_user_id)
- break # if somehow we get two 'true' labs, we only use one
+ break # if somehow we get two 'true' labs, we only use one
# return to repo
self.repo_put(self.repo.GRESOURCE_BUNDLE_MODELS, models)
-
def update_confirmation(self):
confirm = self.repo_get(self.repo.CONFIRMATION, {})
if "resource" not in confirm:
confirm['resource']['lab'] = models['lab'].lab_user.username
self.repo_put(self.repo.CONFIRMATION, confirm)
-
def post_render(self, request):
try:
self.form = HardwareDefinitionForm(request.POST)
self.context = self.get_context()
return render(request, self.template, self.context)
+
class Define_Nets(WorkflowStep):
template = 'resource/steps/pod_definition.html'
title = "Define Networks"
vlans = lab.vlan_manager.get_vlan(count=lab.vlan_manager.block_size)
self.repo_put(self.repo.VLANS, vlans)
return vlans
- except Exception as e:
+ except Exception:
return None
def get_context(self):
added_list = []
added_dict = {}
context['added_hosts'] = []
- if not hostlist is None:
+ if hostlist is not None:
new_hostlist = []
for host in models['hosts']:
intcount = host.profile.interfaceprofile.count()
host['id'] = generic_host.resource.name
host['interfaces'] = []
for iface in host_profile.interfaceprofile.all():
- host['interfaces'].append({
+ host['interfaces'].append(
+ {
"name": iface.name,
- "description": "speed: " + str(iface.speed) + "M\ntype: " + iface.nic_type})
+ "description": "speed: " + str(iface.speed) + "M\ntype: " + iface.nic_type
+ }
+ )
host['value'] = {"name": generic_host.resource.name}
host['value']['description'] = generic_host.profile.description
context['hosts'].append(json.dumps(host))
else:
context['xml'] = False
- except Exception as e:
+ except Exception:
pass
+
return context
def post_render(self, request):
self.updateModels(xmlData)
# update model with xml
self.metastep.set_valid("Networks applied successfully")
- except Exception as e:
+ except Exception:
self.metastep.set_invalid("An error occurred when applying networks")
return self.render(request)
given_hosts, interfaces = self.parseXml(xmlData)
vlan_manager = models['bundle'].lab.vlan_manager
existing_host_list = models.get("hosts", [])
- existing_hosts = {} # maps id to host
+ existing_hosts = {} # maps id to host
for host in existing_host_list:
existing_hosts[host.resource.name] = host
for ifaceId in given_host['interfaces']:
iface = interfaces[ifaceId]
- iface_profile = existing_host.profile.interfaceprofile.get(name=iface['profile_name'])
if existing_host.resource.name not in models['vlans']:
models['vlans'][existing_host.resource.name] = {}
models['vlans'][existing_host.resource.name][iface['profile_name']] = []
interfaces = {} # maps id -> interface
xmlDom = minidom.parseString(xmlString)
root = xmlDom.documentElement.firstChild
- connections = []
netids = {}
untagged_ints = {}
for cell in root.childNodes:
cellId = cell.getAttribute('id')
if cell.getAttribute("edge"):
- #cell is a network connection
+ # cell is a network connection
escaped_json_str = cell.getAttribute("value")
json_str = escaped_json_str.replace('"', '"')
attributes = json.loads(json_str)
src = cell.getAttribute("source")
tgt = cell.getAttribute("target")
if src in parent_nets:
- #src is a network port
+ # src is a network port
network = networks[parent_nets[src]]
- if tgt in untagged_ints and tagged==False:
+ if tgt in untagged_ints and not tagged:
raise InvalidVlanConfigurationException("More than one untagged vlan on an interface")
interface = interfaces[tgt]
untagged_ints[tgt] = True
else:
network = networks[parent_nets[tgt]]
- if src in untagged_ints and tagged==False:
+ if src in untagged_ints and not tagged:
raise InvalidVlanConfigurationException("More than one untagged vlan on an interface")
interface = interfaces[src]
untagged_ints[src] = True
int_netid = int(nid)
assert public or int_netid > 1, "Net id is 1 or lower"
assert int_netid < 4095, "Net id is 4095 or greater"
- except Exception as e:
+ except Exception:
raise InvalidVlanConfigurationException("VLAN ID is not an integer more than 1 and less than 4095")
if nid in netids:
raise NetworkExistsException("Non unique network id found")
networks[cellId] = network
elif "host" in cellId: # cell is a host/machine
- #TODO gather host info
+ # TODO gather host info
cell_json_str = cell.getAttribute("value")
cell_json = json.loads(cell_json_str)
host = {"interfaces": [], "name": cellId, "profile_name": cell_json['name']}
if "network" in parentId:
parent_nets[cellId] = parentId
elif "host" in parentId:
- #TODO gather iface info
+ # TODO gather iface info
cell_json_str = cell.getAttribute("value")
cell_json = json.loads(cell_json_str)
iface = {"name": cellId, "networks": [], "profile_name": cell_json['name']}
import datetime
import json
-from resource_inventory.models import *
-from workflow.models import *
-from workflow.forms import *
+from booking.models import Booking
+from resource_inventory.models import Host, Image
+from workflow.models import WorkflowStep
+from workflow.forms import SnapshotMetaForm, SnapshotHostSelectForm
+
class Select_Host_Step(WorkflowStep):
template = "snapshot_workflow/steps/select_host.html"
for genericHost in booking.resource.template.getHosts():
booking_hosts[booking.id]['hosts'].append({"name": genericHost.resource.name})
-
context['booking_hosts'] = booking_hosts
chosen_host = self.repo_get(self.repo.SNAPSHOT_MODELS, {}).get("host")
self.metastep.set_valid("Success")
return self.render(request)
+
class Image_Meta_Step(WorkflowStep):
template = "snapshot_workflow/steps/meta.html"
title = "Additional Information"
context['form'] = SnapshotMetaForm()
return context
-
def post_render(self, request):
form = SnapshotMetaForm(request.POST)
if form.is_valid():
##############################################################################
-from django.forms import formset_factory, modelformset_factory
+from django.forms import formset_factory
from workflow.models import WorkflowStep
from workflow.forms import SoftwareConfigurationForm, HostSoftwareDefinitionForm
from workflow.booking_workflow import Resource_Select
-from resource_inventory.models import *
+from resource_inventory.models import Image, GenericHost, ConfigBundle, HostConfiguration, Installer, OPNFVConfig
-#resource selection step is reused from Booking workflow
-
-#TODO: change this: too hacky, just for presentation
-
+# resource selection step is reused from Booking workflow
class SWConf_Resource_Select(Resource_Select):
def __init__(self, *args, **kwargs):
super(SWConf_Resource_Select, self).__init__(*args, **kwargs)
self.repo_put(self.repo.CONFIG_MODELS, models)
return response
+
class Define_Software(WorkflowStep):
template = 'config_bundle/steps/define_software.html'
title = "Pick Software"
filter_data = {}
user = self.repo_get(self.repo.SESSION_USER)
- i=0;
+ i = 0
for host_data in hosts_initial:
host = GenericHost.objects.get(pk=host_data['host_id'])
excluded_images = Image.objects.exclude(owner=user).exclude(public=True)
role = form.cleaned_data['role']
bundle = models['bundle']
hostConfig = HostConfiguration(
- host=host,
- image=image,
- bundle=bundle,
- opnfvRole=role
+ host=host,
+ image=image,
+ bundle=bundle,
+ opnfvRole=role
)
models['host_configs'].append(hostConfig)
confirm_host = {"name": host.resource.name, "image": image.name, "role": role.name}
return self.render(request)
+
class Config_Software(WorkflowStep):
template = 'config_bundle/steps/config_software.html'
form = SoftwareConfigurationForm
- context = {'workspace_form':form}
+ context = {'workspace_form': form}
title = "Other Info"
description = "Give your software config a name, description, and other stuff"
short_title = "config info"
if "bundle" not in models:
models['bundle'] = ConfigBundle(owner=self.repo_get(self.repo.SESSION_USER))
-
confirm = self.repo_get(self.repo.CONFIRMATION, {})
if "configuration" not in confirm:
confirm['configuration'] = {}
installer = form.cleaned_data['installer']
scenario = form.cleaned_data['scenario']
opnfv = OPNFVConfig(
- bundle=models['bundle'],
- installer=installer,
- scenario=scenario
- )
+ bundle=models['bundle'],
+ installer=installer,
+ scenario=scenario
+ )
models['opnfv'] = opnfv
confirm['configuration']['installer'] = form.cleaned_data['installer'].name
confirm['configuration']['scenario'] = form.cleaned_data['scenario'].name
self.repo_put(self.repo.CONFIG_MODELS, models)
self.repo_put(self.repo.CONFIRMATION, confirm)
- except Exception as e:
+ except Exception:
pass
return self.render(request)
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-from django.test import TestCase, client
+
+from django.test import TestCase
from dashboard.populate_db import Populator
from workflow.tests import constants
from workflow.workflow_factory import WorkflowFactory
from workflow.models import Repository
-from workflow.resource_bundle_workflow import *
-from workflow.sw_bundle_workflow import *
-from workflow.booking_workflow import *
+from workflow.resource_bundle_workflow import Define_Hardware, Define_Nets, Resource_Meta_Info, Host_Meta_Info
+from workflow.sw_bundle_workflow import SWConf_Resource_Select, Define_Software, Config_Software
+from workflow.booking_workflow import Booking_Resource_Select, SWConfig_Select, Booking_Meta
from django.http import QueryDict, HttpRequest
from django.contrib.auth.models import User
-from django.core.management import call_command
-from resource_inventory.models import *
+from resource_inventory.models import (
+ Scenario,
+ Installer,
+ OPNFVRole,
+ Image,
+ GenericResourceBundle,
+ GenericHost,
+ HostProfile,
+ GenericResource,
+ ConfigBundle
+)
class BaseStepTestCase(TestCase):
grb_model = GenericResourceBundle.objects.filter(owner__username="user 1").first()
grb = [{"small_name": grb_model.name, "expanded_name": "user 1", "id": grb_model.id, "string": ""}]
grb = str(grb).replace("'", '"')
- data = {"generic_resource_bundle": grb }
+ data = {"generic_resource_bundle": grb}
response, context = self.step_test(Booking_Resource_Select, data)
self.assertTrue(True)
data = {}
response, context = self.step_test(SWConfig_Select, data)
+
class SoftwareConfigSelectTestCase(BaseStepTestCase):
def test_step_with_good_data(self):
config_model = ConfigBundle.objects.filter(owner__username="user 1").first()
- config = [{"expanded_name":"user 1", "small_name":config_model.name, "id":config_model.id, "string":""}]
+ config = [{"expanded_name": "user 1", "small_name": config_model.name, "id": config_model.id, "string": ""}]
config = str(config).replace("'", '"')
data = {"software_bundle": config}
response, context = self.step_test(SWConfig_Select, data)
data = {}
response, context = self.step_test(SWConfig_Select, data)
+
class BookingMetaTestCase(BaseStepTestCase):
def test_step_with_good_data(self):
user2 = User.objects.get(username="user 2")
john = User.objects.get(username="johnsmith")
users = [
- {"expanded_name":"", "id":user2.id, "small_name":user2.username, "string":user2.email},
- {"expanded_name":"", "id":john.id, "small_name":john.username, "string":john.email}
- ]
+ {"expanded_name": "", "id": user2.id, "small_name": user2.username, "string": user2.email},
+ {"expanded_name": "", "id": john.id, "small_name": john.username, "string": john.email}
+ ]
users = str(users).replace("'", '"')
data['users'] = users
response, context = self.step_test(Booking_Meta, data)
def test_step_with_good_data(self):
hosts = {"host_4": 1, "host_1": 1}
- labs = {"lab_1":"true"}
+ labs = {"lab_1": "true"}
data = {"hosts": hosts, "labs": labs}
response, context = self.step_test(Define_Hardware, data)
grb_model = GenericResourceBundle.objects.filter(owner__username="user 1").first()
grb = [{"small_name": grb_model.name, "expanded_name": "user 1", "id": grb_model.id, "string": ""}]
grb = str(grb).replace("'", '"')
- data = {"generic_resource_bundle": grb }
+ data = {"generic_resource_bundle": grb}
response, context = self.step_test(SWConf_Resource_Select, data)
def test_step_with_bad_data(self): # TODO
repo.el[repo.SWCONF_SELECTED_GRB] = grb
return repo
-
def test_step_with_good_data(self):
data = {"form-INITIAL_FORMS": 3, "form-MAX_NUM_FORMS": 1000}
data["form-MIN_NUM_FORMS"] = 0
def test_step_with_empty_data(self):
data = {}
response, context = self.step_test(Config_Software, data)
-
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+
from django.test import TestCase, Client
+
class SuperViewTestCase(TestCase):
url = "/"
client = Client()
class DefineHardwareViewTestCase(SuperViewTestCase):
url = "/wf/workflow/step/define_hardware"
+
class DefineNetworkViewTestCase(SuperViewTestCase):
url = "/wf/workflow/step/define_net"
+
class ResourceMetaViewTestCase(SuperViewTestCase):
url = "/wf/workflow/step/resource_meta"
+
class BookingMetaViewTestCase(SuperViewTestCase):
url = "/wf/workflow/step/booking_meta"
+
class SoftwareSelectViewTestCase(SuperViewTestCase):
url = "/wf/workflow/step/software_select"
+
class ResourceSelectViewTestCase(SuperViewTestCase):
url = "/wf/workflow/step/resource_select"
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-from django.test import TestCase, client
+
+from django.test import TestCase
from workflow.workflow_factory import WorkflowFactory
from dashboard.populate_db import Populator
-from resource_inventory.models import *
"""
POST to /wf/workflow {"cancel": ""}
"""
+
class WorkflowTestCase(TestCase):
@classmethod
self.assertIsNone(exception)
+
class BookingWorkflowTestCase(WorkflowTestCase):
@classmethod
def test_steps_render(self):
super(BookingWorkflowTestCase, self).render_steps()
+
class ResourceWorkflowTestCase(WorkflowTestCase):
@classmethod
def test_steps_render(self):
super(ResourceWorkflowTestCase, self).render_steps()
+
class ConfigWorkflowTestCase(WorkflowTestCase):
@classmethod
##############################################################################
-from django.conf.urls import url, include
+from django.conf.urls import url
from django.conf import settings
-from workflow.views import *
-from workflow.models import *
-from workflow.resource_bundle_workflow import *
-from workflow.booking_workflow import *
+from workflow.views import step_view, delete_session, manager_view, viewport_view
+from workflow.models import Repository
+from workflow.resource_bundle_workflow import Define_Hardware, Define_Nets, Resource_Meta_Info
+from workflow.booking_workflow import SWConfig_Select, Resource_Select, Booking_Meta
app_name = 'workflow'
urlpatterns = [
##############################################################################
-from django.http import HttpResponse, HttpRequest, HttpResponseGone
-from django.urls import reverse
-from django.shortcuts import render, redirect
-from django import forms
+from django.http import HttpResponse, HttpResponseGone
+from django.shortcuts import render
import uuid
-from workflow.forms import *
-from workflow.workflow_manager import *
+from workflow.workflow_manager import ManagerTracker, SessionManager
import logging
logger = logging.getLogger(__name__)
except KeyError:
return None
+
def delete_session(request):
try:
- manager = ManagerTracker.managers[request.session['manager_session']]
del ManagerTracker.managers[request.session['manager_session']]
return HttpResponse('')
- except KeyError:
+ except Exception:
return None
+
def step_view(request):
manager = attempt_auth(request)
if not manager:
- #no manager found, redirect to "lost" page
+ # no manager found, redirect to "lost" page
return no_workflow(request)
if request.GET.get('step') is not None:
manager.goto(int(request.GET.get('step')))
return manager.render(request)
+
def manager_view(request):
manager = attempt_auth(request)
if not manager:
return HttpResponseGone("No session found that relates to current request")
- if request.method == 'GET': #no need for this statement if only intercepting post requests
+ if request.method == 'GET':
+ # no need for this statement if only intercepting post requests
- #return general context for viewport page
+ # return general context for viewport page
return manager.status(request)
if request.method == 'POST':
logger.debug("add found")
target_id = None
if 'target' in request.POST:
- target_id=int(request.POST.get('target'))
+ target_id = int(request.POST.get('target'))
manager.add_workflow(workflow_type=int(request.POST.get('add')), target_id=target_id)
elif request.POST.get('edit') is not None and request.POST.get('edit_id') is not None:
logger.debug("edit found")
manager.add_workflow(workflow_type=request.POST.get('edit'), edit_object=int(request.POST.get('edit_id')))
elif request.POST.get('cancel') is not None:
- mgr = ManagerTracker.managers[request.session['manager_session']]
del ManagerTracker.managers[request.session['manager_session']]
- del mgr
return manager.status(request)
+
def viewport_view(request):
if not request.user.is_authenticated:
return login(request)
if manager is None:
return no_workflow(request)
- if request.method == 'GET':
+ if request.method == 'GET':
return render(request, 'workflow/viewport-base.html')
else:
pass
+
def create_session(wf_type, request):
wf = int(wf_type)
smgr = SessionManager(request=request)
return manager_uuid
+
def no_workflow(request):
logger.debug("There is no active workflow")
return render(request, 'workflow/no_workflow.html', {'title': "Not Found"})
+
def login(request):
return render(request, "dashboard/login.html", {'title': 'Authentication Required'})
##############################################################################
-from workflow.booking_workflow import *
-from workflow.resource_bundle_workflow import *
-from workflow.sw_bundle_workflow import *
-from workflow.snapshot_workflow import *
-from workflow.models import Workflow, Repository
+from workflow.booking_workflow import Booking_Resource_Select, SWConfig_Select, Booking_Meta
+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
import uuid
import logging
logger = logging.getLogger(__name__)
+
class BookingMetaWorkflow(object):
workflow_type = 0
color = "#0099ff"
is_child = False
+
class ResourceMetaWorkflow(object):
workflow_type = 1
color = "#ff6600"
+
class ConfigMetaWorkflow(object):
workflow_type = 2
color = "#00ffcc"
+
class MetaRelation(object):
def __init__(self, *args, **kwargs):
self.color = "#cccccc"
'depth': self.depth,
}
+
class MetaStep(object):
- #valid = 0 #0 is not checked, 1 is invalid, 2 is valid
UNTOUCHED = 0
INVALID = 100
def __ne__(self, other):
return self.id.int != other.id.int
+
class WorkflowFactory():
- #def __init__(self, *args, **kwargs):
booking_steps = [
- Booking_Resource_Select,
- SWConfig_Select,
- Booking_Meta
- ]
+ Booking_Resource_Select,
+ SWConfig_Select,
+ Booking_Meta
+ ]
resource_steps = [
- Define_Hardware,
- Define_Nets,
- Resource_Meta_Info,
- ]
+ Define_Hardware,
+ Define_Nets,
+ Resource_Meta_Info,
+ ]
config_steps = [
- SWConf_Resource_Select,
- Define_Software,
- Config_Software,
- ]
+ SWConf_Resource_Select,
+ Define_Software,
+ Config_Software,
+ ]
snapshot_steps = [
- Select_Host_Step,
- Image_Meta_Step
- ]
+ Select_Host_Step,
+ Image_Meta_Step
+ ]
def conjure(self, workflow_type=None, repo=None):
workflow_types = [
- self.booking_steps,
- self.resource_steps,
- self.config_steps,
- self.snapshot_steps,
+ self.booking_steps,
+ self.resource_steps,
+ self.config_steps,
+ self.snapshot_steps,
]
steps = self.make_steps(workflow_types[workflow_type], repository=repo)
##############################################################################
-from django.db import models
-from django.contrib.auth.models import User
-from django.core import serializers
-from django.http import HttpResponse, JsonResponse
+from django.http import JsonResponse
-import json
-import uuid
import random
-from resource_inventory.models import *
from booking.models import Booking
from workflow.workflow_factory import WorkflowFactory, MetaStep, MetaRelation
from workflow.models import Repository, Confirmation_Step
+from resource_inventory.models import (
+ GenericResourceBundle,
+ ConfigBundle,
+ HostConfiguration,
+ OPNFVConfig
+)
import logging
logger = logging.getLogger(__name__)
+
class SessionManager():
def __init__(self, request=None):
metaconfirm = MetaStep()
metaconfirm.index = 0
metaconfirm.short_title = "confirm"
- self.repository.el['steps'] = 1;
+ self.repository.el['steps'] = 1
self.metaworkflow = None
self.metaworkflows = []
self.metarelations = []
self.metarelations.append(relation)
self.initialized = True
-
def status(self, request):
try:
- workflows = []
steps = []
for step in self.step_meta:
steps.append(step.to_json())
responsejson['parents'] = parents
responsejson['children'] = children
return JsonResponse(responsejson, safe=False)
- except Exception as e:
+ except Exception:
pass
def render(self, request, **kwargs):
- #filter out when a step needs to handle post/form data
+ # filter out when a step needs to handle post/form data
# if 'workflow' in post data, this post request was meant for me, not step
if request.method == 'POST' and request.POST.get('workflow', None) is None:
return self.steps[self.active_index].post_render(request)
def goto(self, num, **kwargs):
self.repository.el['active_step'] = int(num)
self.active_index = int(num)
- #TODO: change to include some checking
+ # TODO: change to include some checking
def prefill_repo(self, target_id, workflow_type):
self.repository.el[self.repository.EDIT] = True
edit_object = ConfigBundle.objects.get(pk=target_id)
self.prefill_config(edit_object)
-
def prefill_booking(self, booking):
models = self.make_booking_models(booking)
confirmation = self.make_booking_confirm(booking)
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# first, basic lint with flake8
+find . -type f -name "*.py" -not -name "manage.py" | xargs flake8 --count --ignore E501
+
+
# this file should be executed from the dir it is in
docker exec -it dg01 python manage.py test -t ../src/
Code Review / laas.git / commitdiff