[SEBA-110] Validation location
Change-Id: Idd155c05fa4cac5eb0733576c5c0fae74925e1ef
diff --git a/xos/synchronizer/steps/sync_att_workflow_driver_service_instance.py b/xos/synchronizer/steps/sync_att_workflow_driver_service_instance.py
index d2fdeeb..1726e50 100644
--- a/xos/synchronizer/steps/sync_att_workflow_driver_service_instance.py
+++ b/xos/synchronizer/steps/sync_att_workflow_driver_service_instance.py
@@ -14,7 +14,7 @@
import json
from synchronizers.new_base.syncstep import SyncStep, model_accessor
-from synchronizers.new_base.modelaccessor import AttWorkflowDriverServiceInstance, AttWorkflowDriverWhiteListEntry
+from synchronizers.new_base.modelaccessor import AttWorkflowDriverServiceInstance, AttWorkflowDriverWhiteListEntry, ONUDevice
from xosconfig import Config
from multistructlog import create_logger
@@ -25,7 +25,7 @@
provides = [AttWorkflowDriverServiceInstance]
observes = AttWorkflowDriverServiceInstance
- def validate_in_external_oss(self, si):
+ def validate_onu(self, si):
# This is where you may want to call your OSS Database to verify if this ONU can be activated
oss_service = si.owner.leaf_model
@@ -34,30 +34,30 @@
matching_entries = AttWorkflowDriverWhiteListEntry.objects.filter(owner_id=oss_service.id,
serial_number=si.serial_number)
- return len(matching_entries)>0
+ # check that it's in the whitelist
+ if len(matching_entries) == 0:
+ log.warn("ONU disable as not in whitelist", object=str(si), serial_number=si.serial_number, **si.tologdict())
+ return False
- def get_suscriber_c_tag(self, serial_number):
- # If it's up to your OSS to generate c_tags, fetch them here
- # otherwise XOS will generate one for your subscriber
- return None
+ whitelisted = matching_entries[0]
+ pon_port = ONUDevice.objects.get().pon_port
+ if pon_port.port_no != whitelisted.pon_port_id or si.of_dpid != whitelisted.device_id:
+ log.warn("ONU disable as location don't match", object=str(si), serial_number=si.serial_number,
+ **si.tologdict())
+ return False
- def sync_record(self, o):
- log.info("synching AttWorkflowDriverServiceInstance", object=str(o), **o.tologdict())
+ return True
- if not self.validate_in_external_oss(o):
- log.error("ONU with serial number %s is not valid in the OSS Database" % o.serial_number)
- o.valid = "invalid"
+ def sync_record(self, si):
+ log.info("synching AttWorkflowDriverServiceInstance", object=str(si), **si.tologdict())
+
+ if not self.validate_onu(si):
+ log.error("ONU with serial number %s is not valid in the OSS Database" % si.serial_number)
+ si.valid = "invalid"
else:
- if self.get_suscriber_c_tag(o.serial_number):
- self.c_tag = self.get_suscriber_c_tag(o.serial_number)
+ si.valid = "valid"
- o.valid = "valid"
-
- # Set no_sync=True to prevent the syncstep from running again, and set alway_update_timestamp=True to cause
- # the model_policy to run again.
- # TODO(smbaker): Revisit this after fixing this issue in the core.
- o.no_sync = True
- o.save(update_fields=["valid", "no_sync", "updated"], always_update_timestamp=True)
+ si.save()
def delete_record(self, o):
pass
diff --git a/xos/synchronizer/steps/test_sync_att_workflow_driver_service_instance.py b/xos/synchronizer/steps/test_sync_att_workflow_driver_service_instance.py
index db168f7..2b82b1b 100644
--- a/xos/synchronizer/steps/test_sync_att_workflow_driver_service_instance.py
+++ b/xos/synchronizer/steps/test_sync_att_workflow_driver_service_instance.py
@@ -76,7 +76,7 @@
self.sync_step = SyncAttWorkflowDriverServiceInstance
- self.oss = Mock()
+ self.oss = AttWorkflowDriverService()
self.oss.name = "oss"
self.oss.id = 5367
@@ -84,30 +84,62 @@
self.o = Mock()
self.o.serial_number = "BRCM1234"
self.o.of_dpid = "of:109299321"
+ self.o.pon_port_id = 1
self.o.owner.leaf_model = self.oss
self.o.tologdict.return_value = {}
+ self.pon_port = PONPort(
+ port_no=1
+ )
+ self.onu = ONUDevice(
+ serial_number=self.o.serial_number,
+ pon_port=self.pon_port
+ )
+
def tearDown(self):
self.o = None
sys.path = self.sys_path_save
def test_sync_valid(self):
- with patch.object(AttWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_items:
+ with patch.object(AttWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_items, \
+ patch.object(ONUDevice.objects, "get_items") as onu_items:
# Create a whitelist entry for self.o's serial number
- whitelist_entry = AttWorkflowDriverWhiteListEntry(owner_id=self.oss.id, serial_number=self.o.serial_number)
+ whitelist_entry = AttWorkflowDriverWhiteListEntry(
+ owner_id=self.oss.id,
+ serial_number=self.o.serial_number,
+ device_id=self.o.of_dpid,
+ pon_port_id=1,
+ )
whitelist_items.return_value = [whitelist_entry]
+ onu_items.return_value = [self.onu]
self.sync_step().sync_record(self.o)
self.assertEqual(self.o.valid, "valid")
- self.assertTrue(self.o.no_sync)
self.o.save.assert_called()
- def test_sync_rejected(self):
+ def test_sync_bad_location(self):
+ with patch.object(AttWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_items, \
+ patch.object(ONUDevice.objects, "get_items") as onu_items:
+ # Create a whitelist entry for self.o's serial number
+ whitelist_entry = AttWorkflowDriverWhiteListEntry(
+ owner_id=self.oss.id,
+ serial_number=self.o.serial_number,
+ device_id="foo",
+ pon_port_id=666
+ )
+ whitelist_items.return_value = [whitelist_entry]
+ onu_items.return_value = [self.onu]
+
+ self.sync_step().sync_record(self.o)
+
+ self.assertEqual(self.o.valid, "invalid")
+ self.o.save.assert_called()
+
+ def test_sync_no_whitelist(self):
self.sync_step().sync_record(self.o)
self.assertEqual(self.o.valid, "invalid")
- self.assertTrue(self.o.no_sync)
self.o.save.assert_called()
if __name__ == '__main__':