Improve partial-batch diagnostics and stop treating vpnv6 prefixes as clock noise.

List failing profile/command/reason in batch.message (including unmatched and expand_all), and require whitespace/EOL after HH:MM:SS so prefixes like 56:16:10::/64 are parsed.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-23 16:43:28 +08:00
parent 9b6452971f
commit 7b5f03ad54
4 changed files with 259 additions and 42 deletions

View file

@ -127,7 +127,9 @@ class BizStateCollectFinalizeTests(unittest.TestCase):
assert batch is not None
self.assertEqual(batch.status, "partial")
self.assertTrue(str(batch.message or "").strip())
self.assertIn("aux", (batch.message or "").lower())
self.assertIn("aux_failed", (batch.message or "").lower())
self.assertIn("show interface brief", batch.message or "")
self.assertIn("parse boom", batch.message or "")
def test_finalize_partial_when_skipped_bindings_with_success(self) -> None:
from netx_api.models import BizStateBatchCommand
@ -172,6 +174,71 @@ class BizStateCollectFinalizeTests(unittest.TestCase):
batch = self.db.get(BizStateBatch, "b-finalize")
assert batch is not None
self.assertIn("skipped", (batch.message or "").lower())
self.assertIn("zte.bgp_vpnv4_neighbor_in", batch.message or "")
self.assertIn("bindings", (batch.message or "").lower())
def test_finalize_partial_lists_unmatched_and_failed_commands(self) -> None:
from netx_api.models import BizStateBatchCommand
self.db.add(
BizStateBatchCommand(
id="c-ok3",
batch_id="b-finalize",
profile_id="zte.arp",
metric_id="arp",
raw_command="show arp",
parse_status="ok",
row_count=1,
)
)
self.db.add(
BizStateBatchCommand(
id="c-unmatched",
batch_id="b-finalize",
profile_id="zte.legacy_x",
raw_command="show weird-legacy",
parse_status="unmatched",
message="no profile matched concrete command",
row_count=0,
)
)
self.db.add(
BizStateBatchCommand(
id="c-fail",
batch_id="b-finalize",
profile_id="zte.bgp_route",
metric_id="bgp_route",
raw_command="show bgp vpnv4 unicast neighbor in 1.1.1.1",
parse_status="failed",
message="parse: ValueError: boom",
row_count=0,
)
)
self.db.commit()
with patch.object(runner, "SessionLocal", self.Session):
with patch(
"netx_api.biz_state.compare_service.schedule_auto_compare_for_task",
):
status = runner._finalize_batch_status(
batch_id="b-finalize",
task_id="t-finalize",
cmd_count=3,
total_rows=1,
any_fail=True,
any_ok=True,
lane_errors=[],
)
self.assertEqual(status, "partial")
self.db.expire_all()
batch = self.db.get(BizStateBatch, "b-finalize")
assert batch is not None
msg = batch.message or ""
self.assertIn("unmatched=1", msg)
self.assertIn("failed=1", msg)
self.assertIn("show weird-legacy", msg)
self.assertIn("show bgp vpnv4", msg)
self.assertIn("zte.bgp_route", msg)
self.assertNotIn("partial success (some steps failed", msg)
def test_finalize_retries_once_on_operational_error(self) -> None:
calls = {"n": 0}

View file

@ -1242,6 +1242,43 @@ Route Distinguisher:10.0.0.1:100
self.assertTrue(all(r.get("status_codes") == "*i" for r in routes))
self.assertTrue(all(r.get("rd") == "10.0.0.1:100" for r in routes))
def test_bgp_route_vpnv6_prefix_not_skipped_as_clock(self) -> None:
"""Prefixes like 56:16:10::/64 must not match HH:MM:SS noise filter."""
from netx_api.biz_state.parsers.zte.bgp_route import _skip_noise_line
self.assertTrue(_skip_noise_line("09:30:01"))
self.assertTrue(_skip_noise_line("9:05:00 system ready"))
self.assertFalse(_skip_noise_line("56:16:10::/64"))
self.assertFalse(_skip_noise_line("* i 56:16:10::/64"))
raw = """
Routes Advertised to This Neighbor:
Status codes: * valid, i - internal
Total number of routes: 2
Dest Next Hop Metric LocPrf InTag Path
Route Distinguisher:65525:30001 (default for vrf CUST_V6)
* i 56:16:10::/64 ::FFFF:10.1.1.1 100 0 ?
* i 56:16:32::/64 ::FFFF:10.1.1.1 100 0 ?
"""
routes = normalize_bgp_route(
raw_text=raw,
command="show bgp vpnv6 unicast neighbor out 24.11.0.8 as 24208 | one-line",
vendor="ZTE",
device_type="zte_zxros",
params={
"neighbor": "24.11.0.8",
"direction": "out",
"afi": "vpnv6",
"local_as": "24208",
},
)
self.assertEqual(len(routes), 2)
self.assertEqual(
{r["network"] for r in routes},
{"56:16:10::/64", "56:16:32::/64"},
)
self.assertTrue(all(r.get("rd") == "65525:30001" for r in routes))
def test_bgp_route_prod_snippets_rd_vrf_and_wraps(self) -> None:
"""Live RR snippets: OUT without status, RD+VRF, vpnv6 *i + NH wrap."""
from pathlib import Path