Prevent synchronizer to push OLTs to VOLTHA if they have feedback state
Moved the s_tag in the PONPort Model
Change-Id: I8e6b07ff925f0df7772a66425e9ddaa944cfe9fb
diff --git a/xos/synchronizer/steps/test_sync_olt_device.py b/xos/synchronizer/steps/test_sync_olt_device.py
index cac697d..cd4906f 100644
--- a/xos/synchronizer/steps/test_sync_olt_device.py
+++ b/xos/synchronizer/steps/test_sync_olt_device.py
@@ -40,6 +40,17 @@
raise Exception("Unable to find service=%s xproto=%s" % (service_name, xproto_name))
# END generate model from xproto
+def match_onos_req(req):
+ r = req.json()['devices']
+ if not r['abc']:
+ return False
+ else:
+ if not r['abc']['basic']['driver'] == 'pmc-olt':
+ return False
+ if not r['abc']['accessDevice']['vlan'] == "s_tag" or not r['abc']['accessDevice']['uplink'] == "129":
+ return False
+ return True
+
class TestSyncOLTDevice(unittest.TestCase):
def setUp(self):
@@ -61,6 +72,10 @@
from sync_olt_device import SyncOLTDevice
self.sync_step = SyncOLTDevice
+ pon_port = Mock()
+ pon_port.port_id = "00ff00"
+ pon_port.s_tag = "s_tag"
+
# create a mock service instance
o = Mock()
o.volt_service.voltha_url = "voltha_url"
@@ -74,13 +89,20 @@
o.host = "172.17.0.1"
o.port = "50060"
o.uplink = "129"
- o.vlan = "3"
o.driver = "pmc-olt"
+ # feedback state
+ o.device_id = None
+ o.admin_state = None
+ o.oper_status = None
+ o.of_id = None
+
o.tologdict.return_value = {'name': "Mock VOLTServiceInstance"}
o.save.return_value = "Saved"
+ o.ports.all.return_value = [pon_port]
+
self.o = o
def tearDown(self):
@@ -175,17 +197,6 @@
}
m.get("http://voltha_url/api/v1/logical_devices", status_code=200, json=logical_devices)
- def match_onos_req(req):
- r = req.json()['devices']
- if not r['abc']:
- return False
- else:
- if not r['abc']['basic']['driver'] == 'pmc-olt':
- return False
- if not r['abc']['accessDevice']['vlan'] == "3" or not r['abc']['accessDevice']['uplink'] == "129":
- return False
- return True
-
m.post("http://p_onos_url/onos/v1/network/configuration/", status_code=200, additional_matcher=match_onos_req, json={})
self.sync_step().sync_record(self.o)
@@ -195,6 +206,21 @@
self.o.save.assert_called_once()
@requests_mock.Mocker()
+ def test_sync_record_already_existing_in_voltha(self, m):
+
+ # mock device feedback state
+ self.o.device_id = "123"
+ self.o.admin_state = "ACTIVE"
+ self.o.oper_status = "ENABLED"
+ self.o.of_id = "abc"
+
+ m.post("http://p_onos_url/onos/v1/network/configuration/", status_code=200, additional_matcher=match_onos_req, json={})
+
+ self.sync_step().sync_record(self.o)
+ self.o.save.assert_not_called()
+
+
+ @requests_mock.Mocker()
def test_delete_record(self, m):
self.o.of_id = "abc"
self.o.device_id = "123"