Update Arista Traffic policy to set DSCP bits in the IP header along …

…with a validation check to make sure it falls within the 6 bit range.

PiperOrigin-RevId: 721012361

capirca/lib/arista_tp.py

@@ -400,6 +400,12 @@ def __str__(self):
             [ACTION_INDENT,
              "count %s" % self.term.counter, False])
 
+      # set dscp bits if any.
+      if self.term.dscp_set and self._is_valid_dscp_value(self.term.dscp_set):
+        term_block.append(
+            [ACTION_INDENT, f"set dscp {self.term.dscp_set}", False]
+        )
+
       term_block.append([MATCH_INDENT, "!", False])  # end of actions
     term_block.append([TERM_INDENT, "!", False])  # end of match entry
 
@@ -408,6 +414,11 @@ def __str__(self):
 
     return str(config)
 
+  def _is_valid_dscp_value(self, dscp):
+    if dscp.isdigit() and 0 <= int(dscp) <= 63:
+      return True
+    raise ValueError(f"Invalid DSCP value: {dscp}. Valid range is 0-63.")
+
   def _reflowComments(self, comments, max_length):
     """reflows capirca comments to stay within max_length.
 

tests/lib/arista_tp_test.py

@@ -262,6 +262,14 @@
   action:: accept
 }
 """
+GOOD_TERM_39 = """
+term good_term_39 {
+  protocol:: tcp
+  destination-port:: http
+  action:: accept
+  dscp-set:: 32
+}
+"""
 GOOD_TERM_COMMENT = """
 term good-term-comment {
   protocol:: udp
@@ -313,6 +321,14 @@
   action:: accept
 }
 """
+BAD_DSCP_TERM_1 = """
+term bad_dscp_term_1 {
+  protocol:: tcp
+  destination-port:: http
+  action:: accept
+  dscp-set:: ef
+}
+"""
 DEFAULT_TERM_1 = """
 term default-term-1 {
   action:: deny
@@ -1290,6 +1306,41 @@ def testAnyMixed(self):
     self.assertIn("match ipv6-ANY_MIXED ipv6", output, output)
     self.assertIn("destination prefix 2001:4860:4860::8844/128", output, output)
 
+  def testDscpSet(self):
+    self.naming.GetNetAddr.side_effect = [[
+        nacaddr.IP("8.8.4.4"),
+        nacaddr.IP("8.8.8.8"),
+        nacaddr.IP("2001:4860:4860::8844"),
+        nacaddr.IP("2001:4860:4860::8888"),
+    ]]
+    self.naming.GetServiceByProto.return_value = ["80"]
+    atp = arista_tp.AristaTrafficPolicy(
+        policy.ParsePolicy(GOOD_FIELD_SET_HEADER + GOOD_TERM_39, self.naming),
+        EXP_INFO,
+    )
+    output = str(atp)
+    self.assertIn("set dscp 32", output)
+
+  def testBadDscpSet(self):
+    self.naming.GetNetAddr.side_effect = [[
+        nacaddr.IP("8.8.4.4"),
+        nacaddr.IP("8.8.8.8"),
+        nacaddr.IP("2001:4860:4860::8844"),
+        nacaddr.IP("2001:4860:4860::8888"),
+    ]]
+    self.naming.GetServiceByProto.return_value = ["80"]
+    atp = arista_tp.AristaTrafficPolicy(
+        policy.ParsePolicy(
+            GOOD_FIELD_SET_HEADER + BAD_DSCP_TERM_1, self.naming
+        ),
+        EXP_INFO,
+    )
+    with self.assertRaises(ValueError) as msg:
+      str(atp)
+    self.assertEqual(
+        str(msg.exception), "Invalid DSCP value: ef. Valid range is 0-63."
+    )
+
   def testInetInet(self):
     self.naming.GetNetAddr.side_effect = [
         [nacaddr.IP("8.8.4.4"), nacaddr.IP("8.8.8.8")],