Parse optional 8X lane count in ZTE opticalinfo so QSFP-DD multi-lane powers stay aligned.

Co-authored-by: Cursor <cursoragent@cursor.com>

netx_api/biz_state/parsers/zte/optical_brief.py

@@ -3,6 +3,8 @@
 Power cells look like ``-6.0/[-20.0,-3.0]`` (value / [low,high] threshold).
 50G+ (QSFP) modules print one lane per line; continuation lines are merged
 into comma-separated ``rx_power`` / ``tx_power`` (and matching thresholds).
+400G QSFP-DD may insert a lane-count token (``8X``) between Type and Wavelength;
+that token is absorbed into ``wavelength`` so Rx/Tx columns stay aligned.
 """
 
 from __future__ import annotations
@@ -23,7 +25,7 @@ _CELL_RE = re.compile(
 )
 _HDR_RE = re.compile(
     r"^(?P<iface>\S+)\s+(?P<otype>offline)\s*$|"
-    r"^(?P<iface2>\S+)\s+(?P<otype2>\S+)\s+(?P<wave>\S+)\s+"
+    r"^(?P<iface2>\S+)\s+(?P<otype2>\S+)\s+(?:(?P<lanes>\d+X)\s+)?(?P<wave>\S+)\s+"
     r"(?P<rx>\S+)\s+(?P<tx>\S+)\s+(?P<status>\S+)(?:\s+(?P<intensity>\S+))?\s*$",
     re.I,
 )
@@ -62,7 +64,7 @@ def _empty_row(iface: str, optic_type: str = "", wavelength: str = "") -> dict[s
     return {
         "interface": iface[:128],
         "optic_type": optic_type[:64],
-        "wavelength": wavelength[:32],
+        "wavelength": re.sub(r"\s+", " ", str(wavelength or "").strip())[:32],
         "rx_power": "",
         "rx_threshold": "",
         "tx_power": "",
@@ -172,7 +174,11 @@ def _hand_parse(*, raw_text: str, **_kw: Any) -> list[dict[str, Any]]:
                 _flush()
                 continue
             iface = m.group("iface2") or ""
-            cur = _empty_row(iface, m.group("otype2") or "", m.group("wave") or "")
+            wave = m.group("wave") or ""
+            lanes = m.group("lanes") or ""
+            if lanes:
+                wave = f"{lanes} {wave}".strip()
+            cur = _empty_row(iface, m.group("otype2") or "", wave)
             _append_lane(cur, m.group("rx") or "", m.group("tx") or "", m.group("status") or "")
             continue
         if cur is None:

netx_api/cli_templates/zte/zte_zxros_show_opticalinfo_brief.textfsm

@@ -1,6 +1,6 @@
 Value Filldown INTERFACE (\S+)
 Value Filldown OPTIC_TYPE (\S+)
-Value Filldown WAVELENGTH (\S+)
+Value Filldown WAVELENGTH ((?:\d+X\s+)?\S+)
 Value RX_POWER (\S+)
 Value TX_POWER (\S+)
 Value STATUS (\S+)

tests/test_zte_extended_parsers.py

@@ -74,6 +74,30 @@ cgei-0/1/0/1      100G-10km-QSFP28  1310nm      1.5/[-10.6,4.5]       0.9/[-4.3,
 gei-0/0/0/11      offline
 """
 
+# 400G QSFP-DD inserts lane-count token "8X" between Type and Wavelength.
+_OPTICAL_QSFP_DD_SAMPLE = """
+Interface         Type              Wavelength  RxPower(dBm)          TxPower(dBm)          Status    Intensity(Rx)
+cdgei-0/1/0/1     400G-10km-QSFP-DD 8X  1310nm  N/A/[-9.0,6.1]        0.9/[-2.7,6.1]        Unknown   Unknown
+                                                N/A/[-9.0,6.1]        1.0/[-2.7,6.1]        Unknown   Unknown
+                                                N/A/[-9.0,6.1]        1.2/[-2.7,6.1]        Unknown   Unknown
+                                                N/A/[-9.0,6.1]        1.2/[-2.7,6.1]        Unknown   Unknown
+                                                                      7.1
+cdgei-0/1/0/3     400G-10km-QSFP-DD 8X  1310nm  1.3/[-9.0,5.1]        2.2/[-2.7,6.1]        Normal    Normal
+                                                1.0/[-9.0,5.1]        2.1/[-2.7,6.1]        Normal    Normal
+                                                1.0/[-9.0,5.1]        2.2/[-2.7,6.1]        Normal    Normal
+                                                0.3/[-9.0,5.1]        1.7/[-2.7,6.1]        Normal    Normal
+                                                6.9                   8.1
+cdgei-0/1/0/5     400G-80km-QSFP-DD 8X  1547.715nm  -8.8/[-20.0,3.0]  -8.3/[-11.0,2.0]      Normal    Normal
+                                                -8.8                  -8.3
+cdgei-0/1/0/6     offline
+cgei-0/1/0/2:1    400G-500m-QSFP-DD 8X  1310nm  N/A/[-5.9,5.0]        2.3/[-2.9,5.0]        Unknown   Unknown
+                                                N/A/[-5.9,5.0]        2.4/[-2.9,5.0]        Unknown   Unknown
+                                                N/A/[-5.9,5.0]        2.3/[-2.9,5.0]        Unknown   Unknown
+                                                N/A/[-5.9,5.0]        2.2/[-2.9,5.0]        Unknown   Unknown
+                                                                      8.3
+xgei-0/2/0/1      10G-10km-SFP+     1310nm      -2.4/[-14.4,0.5]      -1.8/[-8.2,1.5]       Normal    Normal
+"""
+
 _BGP_ROUTE_IN = """
 Routes Learned From This Neighbor:
      Network             Next Hop        Metric     LocPrf     RtPrf   Path
@@ -186,6 +210,28 @@ class ZteExtendedParserTests(unittest.TestCase):
         self.assertEqual(by["cgei-0/1/0/1"]["tx_threshold"], "[-4.3,5.5]")
         self.assertEqual(by["gei-0/0/0/11"]["optic_type"], "offline")
 
+    def test_optical_qsfp_dd_lane_count_8x(self) -> None:
+        """400G QSFP-DD prints '8X' between Type and Wavelength — must not shift Rx/Tx."""
+        opt = normalize_optical_brief(
+            raw_text=_OPTICAL_QSFP_DD_SAMPLE,
+            vendor="zte",
+            device_type="zte_zxros",
+            command="show opticalinfo brief | one-line",
+        )
+        by = {r["interface"]: r for r in opt}
+        self.assertIn("cdgei-0/1/0/1", by)
+        self.assertEqual(by["cdgei-0/1/0/1"]["rx_power"], "N/A,N/A,N/A,N/A")
+        self.assertEqual(by["cdgei-0/1/0/1"]["tx_power"], "0.9,1.0,1.2,1.2")
+        self.assertTrue(str(by["cdgei-0/1/0/1"]["wavelength"]).startswith("8X"))
+        self.assertEqual(by["cdgei-0/1/0/3"]["rx_power"], "1.3,1.0,1.0,0.3")
+        self.assertEqual(by["cdgei-0/1/0/3"]["tx_power"], "2.2,2.1,2.2,1.7")
+        # Summary-only continuation (bare numbers / tx-only) must not add lanes
+        self.assertEqual(by["cdgei-0/1/0/5"]["rx_power"], "-8.8")
+        self.assertEqual(by["cdgei-0/1/0/5"]["tx_power"], "-8.3")
+        self.assertEqual(by["cdgei-0/1/0/6"]["status"], "offline")
+        self.assertEqual(by["cgei-0/1/0/2:1"]["tx_power"], "2.3,2.4,2.3,2.2")
+        self.assertEqual(by["xgei-0/2/0/1"]["rx_power"], "-2.4")
+
     def test_bgp_route_and_aux_render(self) -> None:
         routes = normalize_bgp_route(
             raw_text=_BGP_ROUTE_IN,