[SEBA-742] Tech-profile-id as a mandatory field in RCORDSubscriber
Change-Id: I3f7f1e816d7cac457e1e286d809fd2728fb6f0ee
diff --git a/xos/synchronizer/models/models.py b/xos/synchronizer/models/models.py
index 5a42ce5..c80cb63 100755
--- a/xos/synchronizer/models/models.py
+++ b/xos/synchronizer/models/models.py
@@ -15,7 +15,7 @@
import re
import socket
import random
-
+from django.core.exceptions import ObjectDoesNotExist
from xos.exceptions import XOSValidationError, XOSProgrammingError, XOSPermissionDenied, XOSConfigurationError
from models_decl import RCORDService_decl, RCORDSubscriber_decl, RCORDIpAddress_decl, BandwidthProfile_decl
@@ -82,7 +82,6 @@
else:
return tag
-
def generate_c_tag(self):
# unused_c_tags_for_s_tag() this function will return a list of unused c_tags for the given s_tag
@@ -165,6 +164,26 @@
else:
return None
+ def validate_tech_profile_id(self):
+ if self.owner.leaf_model.access != "voltha":
+ # if we're not using VOLTHA we don't need to validate this
+ return True
+
+ volt = None
+ for ps in self.owner.provider_services:
+ provider_service = ps.leaf_model
+ if provider_service.name.lower() == "volt":
+ volt = provider_service
+
+ technology = volt.get_olt_technology_from_unu_sn(self.onu_device)
+
+ try:
+ tp = volt.get_tech_profile(technology, self.tech_profile_id)
+ return True
+ except ObjectDoesNotExist:
+ return False
+
+
def save(self, *args, **kwargs):
self.validate_unique_service_specific_id(none_okay=True)
@@ -232,6 +251,10 @@
if not volt_service.has_access_device(self.onu_device):
raise XOSValidationError("The onu_device you specified (%s) does not exists" % self.onu_device)
+ # validate that the tech_profile_id actually exists
+ if not self.validate_tech_profile_id():
+ raise XOSValidationError("The technology profile you specified [%s] does not exist" % self.tech_profile_id)
+
super(RCORDSubscriber, self).save(*args, **kwargs)
self.invalidate_related_objects()
return
diff --git a/xos/synchronizer/models/rcord.xproto b/xos/synchronizer/models/rcord.xproto
index e1ebb9a..e08cf34 100644
--- a/xos/synchronizer/models/rcord.xproto
+++ b/xos/synchronizer/models/rcord.xproto
@@ -62,6 +62,10 @@
optional string mac_address = 18 [
help_text = "Subscriber MAC Address",
max_length = 256];
+ required int32 tech_profile_id = 23 [
+ help_text = "Technology profile id to be used along with Technology type to retreive the profile",
+ min_value = 0,
+ max_value = 65535];
// operator specific fields
optional string nas_port_id = 20 [
@@ -73,10 +77,6 @@
optional string remote_id = 22 [
help_text = "Option 82 Remote ID for DHCP relay agent",
max_length = 256];
- optional int32 tech_profile_id = 23 [
- help_text = "Technology profile id to be used along with Technology type to retreive the profile",
- min_value = 0,
- max_value = 65535];
required manytoone upstream_bps->BandwidthProfile:us_subscriber = 31:1001 [
help_text = "The subscriber the IP address belongs to"];
diff --git a/xos/synchronizer/models/test_models.py b/xos/synchronizer/models/test_models.py
index a0e5a4d..16b0114 100644
--- a/xos/synchronizer/models/test_models.py
+++ b/xos/synchronizer/models/test_models.py
@@ -65,10 +65,13 @@
self.module_patcher = patch.dict('sys.modules', modules)
self.module_patcher.start()
- self.volt = Mock()
-
from models import RCORDSubscriber, RCORDIpAddress
+ self.volt = Mock(name="vOLT")
+ self.volt.leaf_model.name = "vOLT"
+ self.volt.get_olt_technology_from_unu_sn.return_value = "xgspon"
+
+
self.rcord_subscriber_class = RCORDSubscriber
self.rcord_subscriber = RCORDSubscriber()
@@ -85,9 +88,12 @@
self.rcord_subscriber.owner.provider_services = [self.volt]
self.rcord_subscriber.list_of_unused_c_tags_for_s_tag = []
+
self.rcord_ip = RCORDIpAddress()
self.rcord_ip.subscriber = 1
+ # TODO add a test for validate_tech_profile_id
+
def tearDown(self):
sys.path = self.sys_path_save