Fixing subscriber update
Change-Id: Ia04f21078853b9c9aae604eb27f6f16e43b6d5ae
diff --git a/xos/synchronizer/models/models.py b/xos/synchronizer/models/models.py
index 34aff50..6063282 100644
--- a/xos/synchronizer/models/models.py
+++ b/xos/synchronizer/models/models.py
@@ -82,7 +82,16 @@
# validate c_tag
if hasattr(self, 'c_tag') and self.c_tag is not None:
- if self.c_tag in self.get_used_c_tags():
+ is_update_with_same_tag = False
+
+ if not self.is_new:
+ # if it is an update, but the tag is the same, skip validation
+ existing = RCORDSubscriber.objects.filter(c_tag=self.c_tag)
+
+ if len(existing) > 0 and existing[0].c_tag == self.c_tag and existing[0].id == self.id:
+ is_update_with_same_tag = True
+
+ if self.c_tag in self.get_used_c_tags() and not is_update_with_same_tag:
raise XOSValidationError("The c_tag you specified (%s) has already been used on device %s" % (self.c_tag, self.onu_device))
if not hasattr(self, "c_tag") or self.c_tag is None:
diff --git a/xos/synchronizer/models/test_models.py b/xos/synchronizer/models/test_models.py
index bb2a469..1f7186a 100644
--- a/xos/synchronizer/models/test_models.py
+++ b/xos/synchronizer/models/test_models.py
@@ -50,8 +50,10 @@
self.rcord_subscriber_class = RCORDSubscriber
self.rcord_subscriber = RCORDSubscriber()
+ self.rcord_subscriber.id = None # this is a new model
+ self.rcord_subscriber.is_new = True
self.rcord_subscriber.onu_device = "BRCM1234"
- self.rcord_subscriber.c_tag = "111"
+ self.rcord_subscriber.c_tag = 111
self.rcord_subscriber.ip_address = "1.1.1.1"
self.rcord_subscriber.mac_address = "00:AA:00:00:00:01"
self.rcord_subscriber.owner.leaf_model.access = "voltha"
@@ -96,7 +98,7 @@
"""
s = Mock()
- s.c_tag = "111"
+ s.c_tag = 111
s.onu_device = "BRCM1234"
self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s]
@@ -107,6 +109,36 @@
self.assertEqual(e.exception.message, "The c_tag you specified (111) has already been used on device BRCM1234")
self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
+ def test_validate_c_tag_on_update(self):
+ s = Mock()
+ s.c_tag = 111
+ s.onu_device = "BRCM1234"
+ s.id = 1
+
+ self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s]
+
+ self.rcord_subscriber.is_new = False
+ self.rcord_subscriber.id = 1
+ self.rcord_subscriber.save()
+
+ self.models_decl.RCORDSubscriber_decl.save.assert_called()
+
+ def test_validate_c_tag_on_update_fail(self):
+ s = Mock()
+ s.c_tag = 222
+ s.onu_device = "BRCM1234"
+ s.id = 2
+
+ self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s]
+
+ self.rcord_subscriber.id = 1
+ self.rcord_subscriber.is_new = False
+ self.rcord_subscriber.c_tag = 222
+ with self.assertRaises(Exception) as e:
+ self.rcord_subscriber.save()
+
+ self.assertEqual(e.exception.message, "The c_tag you specified (222) has already been used on device BRCM1234")
+ self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
def test_generate_c_tag(self):
s = Mock()