Implement ipCidrRouteStatus for default route and loopback (sonic-net#56

)

Signed-off-by: Qi Luo <[email protected]>

src/sonic_ax_impl/mibs/ietf/rfc4292.py

@@ -15,6 +15,27 @@ def __init__(self):
  self.db_conn = mibs.init_db()
  self.route_dest_map = {}
  self.route_dest_list = []
+ ## loopback ip string -> ip address object
+ self.loips = {}
+
+ def reinit_data(self):
+ """
+ Subclass update loopback information
+ """
+ self.loips = {}
+
+ self.db_conn.connect(mibs.APPL_DB)
+ loopbacks = self.db_conn.keys(mibs.APPL_DB, "INTF_TABLE:lo:*")
+ if not loopbacks:
+ return
+
+ # collect only ipv4 interfaces
+ for loopback in loopbacks:
+ lostr = loopback.decode()
+ loip = lostr[len("INTF_TABLE:lo:"):]
+ ipa = ipaddress.ip_address(loip)
+ if isinstance(ipa, ipaddress.IPv4Address):
+ self.loips[loip] = ipa
 
  def update_data(self):
  """
@@ -45,12 +66,23 @@ def update_data(self):
  sub_id = ip2tuple_v4(ipn.network_address) + ip2tuple_v4(ipn.netmask) + (self.tos,) + ip2tuple_v4(nh)
  self.route_dest_list.append(sub_id)
  self.route_dest_map[sub_id] = ipn.network_address.packed
+ elif ipnstr in self.loips:
+ ## Note: ipv4 /32 or ipv6 /128 routes will has no prefix ending
+ sub_id = ip2tuple_v4(ipnstr) + (255, 255, 255, 255) + (self.tos,) + (0, 0, 0, 0)
+ self.route_dest_list.append(sub_id)
+ self.route_dest_map[sub_id] = self.loips[ipnstr].packed
 
  self.route_dest_list.sort()
 
  def route_dest(self, sub_id):
  return self.route_dest_map.get(sub_id, None)
 
+ def route_status(self, sub_id):
+ if sub_id in self.route_dest_map:
+ return 1 ## active
+ else:
+ return None
+
  def get_next(self, sub_id):
  right = bisect_right(self.route_dest_list, sub_id)
  if right >= len(self.route_dest_list):
@@ -67,3 +99,6 @@ class IpCidrRouteTable(metaclass=MIBMeta, prefix='.1.3.6.1.2.1.4.24.4'):
 
  ipCidrRouteDest = \
  SubtreeMIBEntry('1.1', route_updater, ValueType.IP_ADDRESS, route_updater.route_dest)
+
+ ipCidrRouteStatus = \
+ SubtreeMIBEntry('1.16', route_updater, ValueType.INTEGER, route_updater.route_status)

tests/mock_tables/appl_db.json

@@ -333,6 +333,14 @@
  "ifname": "Ethernet0,Ethernet4,Ethernet8,Ethernet12,Ethernet16,Ethernet20,Ethernet24,Ethernet28,Ethernet32,Ethernet36,Ethernet40,Ethernet44,Ethernet48,Ethernet52",
  "nexthop": "10.0.0.1,10.0.0.3,10.0.0.5,10.0.0.7,10.0.0.9,10.0.0.11,10.0.0.13,10.0.0.15,10.0.0.17,10.0.0.19,10.0.0.21,10.0.0.23,10.0.0.25,10.0.0.27"
  },
+ "ROUTE_TABLE:10.1.0.32": {
+ "nexthop": "",
+ "ifname": "lo"
+ },
+ "ROUTE_TABLE:fc00:1::32": {
+ "nexthop": "",
+ "ifname": "lo"
+ },
  "LAG_MEMBER_TABLE:PortChannel01:Ethernet112": {
  "status": "enabled"
  },
@@ -369,5 +377,13 @@
  "admin_status": "up",
  "oper_status": "up",
  "mtu": "9216"
+ },
+ "INTF_TABLE:lo:10.1.0.32": {
+ "scope": "global",
+ "family": "IPv4"
+ },
+ "INTF_TABLE:lo:fc00:1::32": {
+ "scope": "global",
+ "family": "IPv6"
  }
 }

tests/test_forward.py

@@ -35,7 +35,7 @@ def test_network_order(self):
  ips = ".".join(str(int(x)) for x in list(ipb))
  self.assertEqual(ips, "0.1.2.3")
 
- def test_getnextpdu_first(self):
+ def test_getnextpdu_first_default(self):
  # oid.include = 1
  oid = ObjectIdentifier(10, 0, 1, 0, (1, 3, 6, 1, 2, 1, 4, 24, 4, 1))
  get_pdu = GetNextPDU(
@@ -125,7 +125,7 @@ def test_getnextpdu_empty(self):
  get_pdu = GetNextPDU(
  header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0),
  oids=(
- ObjectIdentifier(12, 0, 0, 0, (1, 3, 6, 1, 2, 1, 4, 24, 4, 1, 1, 1)),
+ ObjectIdentifier(12, 0, 0, 0, (1, 3, 6, 1, 2, 1, 4, 24, 4, 1, 1, 255)),
  )
  )
 
@@ -137,3 +137,41 @@ def test_getnextpdu_empty(self):
  value0 = response.values[0]
  self.assertEqual(value0.type_, ValueType.END_OF_MIB_VIEW)
 
+ def test_getpdu_loopback_status(self):
+ loip_tuple = (10, 1, 0, 32) # ref: appl_db.json
+ lomask_tuple = (255, 255, 255, 255)
+ emptyip_tuple = (0, 0, 0, 0)
+
+ oid = ObjectIdentifier(24, 0, 1, 0
+ , (1, 3, 6, 1, 2, 1, 4, 24, 4, 1, 16) + loip_tuple + lomask_tuple + (0,) + emptyip_tuple
+ )
+ get_pdu = GetPDU(
+ header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0),
+ oids=[oid]
+ )
+
+ encoded = get_pdu.encode()
+ response = get_pdu.make_response(self.lut)
+
+ value0 = response.values[0]
+ self.assertEqual(value0.type_, ValueType.INTEGER)
+ self.assertEqual(str(value0.name), str(oid))
+ self.assertEqual(value0.data, 1)
+
+ def test_getnextpdu_first_default_status(self):
+ oid = ObjectIdentifier(10, 0, 1, 0, (1, 3, 6, 1, 2, 1, 4, 24, 4, 1, 16))
+ get_pdu = GetNextPDU(
+ header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0),
+ oids=[oid]
+ )
+
+ encoded = get_pdu.encode()
+ response = get_pdu.make_response(self.lut)
+
+ n = len(response.values)
+ # self.assertEqual(n, 7)
+ value0 = response.values[0]
+ self.assertEqual(value0.type_, ValueType.INTEGER)
+ self.assertEqual(str(value0.name), '.1.3.6.1.2.1.4.24.4.1.16.0.0.0.0.0.0.0.0.0.10.0.0.1')
+ self.assertEqual(value0.data, 1)
+