diff --git a/netx_api/biz_state/parsers/zte/optical_brief.py b/netx_api/biz_state/parsers/zte/optical_brief.py index d49bef3..d228300 100644 --- a/netx_api/biz_state/parsers/zte/optical_brief.py +++ b/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\S+)\s+(?Poffline)\s*$|" - r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" + r"^(?P\S+)\s+(?P\S+)\s+(?:(?P\d+X)\s+)?(?P\S+)\s+" r"(?P\S+)\s+(?P\S+)\s+(?P\S+)(?:\s+(?P\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: diff --git a/netx_api/cli_templates/zte/zte_zxros_show_opticalinfo_brief.textfsm b/netx_api/cli_templates/zte/zte_zxros_show_opticalinfo_brief.textfsm index a7b88fa..c94d8f5 100644 --- a/netx_api/cli_templates/zte/zte_zxros_show_opticalinfo_brief.textfsm +++ b/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+) diff --git a/tests/test_zte_extended_parsers.py b/tests/test_zte_extended_parsers.py index cb1eb77..5aadc99 100644 --- a/tests/test_zte_extended_parsers.py +++ b/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,