SEBA-113 s-tag support for "any" and ranges
Change-Id: Ifcec2c42b16a835338b2e7f703dad0bc31931cfe
diff --git a/xos/synchronizer/steps/sync_fabric_crossconnect_service_instance.py b/xos/synchronizer/steps/sync_fabric_crossconnect_service_instance.py
old mode 100644
new mode 100755
index 4a8a4cf..ee067de
--- a/xos/synchronizer/steps/sync_fabric_crossconnect_service_instance.py
+++ b/xos/synchronizer/steps/sync_fabric_crossconnect_service_instance.py
@@ -66,6 +66,39 @@
s_tag = int(s_tag)
return (s_tag, dpid)
+ def range_matches(self, value, pattern):
+ value=int(value)
+ for this_range in pattern.split(","):
+ this_range = this_range.strip()
+ if "-" in this_range:
+ (first, last) = this_range.split("-")
+ first = int(first.strip())
+ last = int(last.strip())
+ if (value>=first) and (value<=last):
+ return True
+ elif this_range.lower()=="any":
+ return True
+ else:
+ if (value==int(this_range)):
+ return True
+ return False
+
+ def find_bng(self, s_tag):
+ # See if there's a mapping for our s-tag directly
+ bng_mappings = BNGPortMapping.objects.filter(s_tag=str(s_tag))
+ if bng_mappings:
+ return bng_mappings[0]
+
+ # TODO(smbaker): Examine miss performance, and if necessary set a flag in the save method to allow filtering
+ # of mappings based on whether they are ranges or any.
+
+ # See if there are any ranges or "any" that match
+ for bng_mapping in BNGPortMapping.objects.all():
+ if self.range_matches(s_tag, bng_mapping.s_tag):
+ return bng_mapping
+
+ return None
+
def sync_record(self, o):
self.log.info("Sync'ing Fabric Crossconnect Service Instance", service_instance=o)
@@ -77,10 +110,10 @@
dpid = si.get_westbound_service_instance_properties("switch_datapath_id")
west_port = si.get_westbound_service_instance_properties("switch_port")
- bng_mappings = BNGPortMapping.objects.filter(s_tag = s_tag)
- if not bng_mappings:
+ bng_mapping = self.find_bng(s_tag = s_tag)
+ if not bng_mapping:
raise Exception("Unable to determine BNG port for s_tag %s" % s_tag)
- east_port = bng_mappings[0].switch_port
+ east_port = bng_mapping.switch_port
data = { "deviceId": dpid,
"vlanId": s_tag,
diff --git a/xos/synchronizer/steps/test_sync_fabric_crossconnect_service_instance.py b/xos/synchronizer/steps/test_sync_fabric_crossconnect_service_instance.py
old mode 100644
new mode 100755
index 4455497..a1b3e23
--- a/xos/synchronizer/steps/test_sync_fabric_crossconnect_service_instance.py
+++ b/xos/synchronizer/steps/test_sync_fabric_crossconnect_service_instance.py
@@ -125,6 +125,54 @@
self.assertEqual(d["user"], "onos")
self.assertEqual(d["pass"], "rocks")
+ def test_range_matches_single(self):
+ self.assertTrue(self.sync_step().range_matches(123, "123"))
+
+ def test_range_matches_single_incorrect(self):
+ self.assertFalse(self.sync_step().range_matches(123, "456"))
+
+ def test_range_matches_range(self):
+ self.assertTrue(self.sync_step().range_matches(123, "122-124"))
+
+ def test_range_matches_range_incorrect(self):
+ self.assertFalse(self.sync_step().range_matches(123, "110-113"))
+
+ def test_range_matches_any(self):
+ self.assertTrue(self.sync_step().range_matches(123, "ANY"))
+ self.assertTrue(self.sync_step().range_matches(123, "any"))
+
+ def test_find_bng_single(self):
+ with patch.object(BNGPortMapping.objects, "get_items") as bng_objects, \
+ patch.object(self.sync_step, "range_matches") as range_matches:
+ bngmapping = BNGPortMapping(s_tag="111", switch_port=4)
+ bng_objects.return_value = [bngmapping]
+
+ # this should not be called
+ range_matches.return_value = False
+
+ found_bng = self.sync_step().find_bng(111)
+ self.assertTrue(found_bng)
+ self.assertEqual(found_bng.switch_port, 4)
+
+ range_matches.assert_not_called()
+
+ def test_find_bng_any(self):
+ with patch.object(BNGPortMapping.objects, "get_items") as bng_objects:
+ bngmapping = BNGPortMapping(s_tag="ANY", switch_port=4)
+ bng_objects.return_value = [bngmapping]
+
+ found_bng = self.sync_step().find_bng(111)
+ self.assertTrue(found_bng)
+ self.assertEqual(found_bng.switch_port, 4)
+
+ def test_find_bng_range(self):
+ with patch.object(BNGPortMapping.objects, "get_items") as bng_objects:
+ bngmapping = BNGPortMapping(s_tag="100-200", switch_port=4)
+ bng_objects.return_value = [bngmapping]
+
+ found_bng = self.sync_step().find_bng(111)
+ self.assertTrue(found_bng)
+ self.assertEqual(found_bng.switch_port, 4)
@requests_mock.Mocker()
def test_sync(self, m):
@@ -137,7 +185,7 @@
si = self.mock_westbound(fsi, s_tag=111, switch_datapath_id = "of:0000000000000201", switch_port = 3)
serviceinstance_objects.return_value = [si]
- bngmapping = BNGPortMapping(s_tag=111, switch_port=4)
+ bngmapping = BNGPortMapping(s_tag="111", switch_port=4)
bng_objects.return_value = [bngmapping]
desired_data = {"deviceId": "of:0000000000000201",
@@ -160,7 +208,7 @@
fsi = FabricCrossconnectServiceInstance(id=7777, owner=self.service)
- si = self.mock_westbound(fsi, s_tag=111, switch_datapath_id = "of:0000000000000201", switch_port = 3)
+ si = self.mock_westbound(fsi, s_tag="111", switch_datapath_id = "of:0000000000000201", switch_port = 3)
serviceinstance_objects.return_value = [si]
with self.assertRaises(Exception) as e: