diff --git a/netx_api/biz_state/compare_service.py b/netx_api/biz_state/compare_service.py index 1092023..2e0f215 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -704,7 +704,21 @@ def _load_metric_rows(db: Session, *, batch_id: str, metric_id: str) -> list[dic .all() ) if rows: - return [dict(r.data_json or {}) for r in rows] + out = [dict(r.data_json or {}) for r in rows] + if metric_id == "arp": + from .parsers.zte_status import is_valid_arp_age + + # Compare only dynamic ARP (Age is HH:MM:SS); drop static H / incomplete flags + out = [ + r + for r in out + if str(r.get("entry_type") or "").lower() == "dynamic" + or ( + not r.get("entry_type") + and is_valid_arp_age(str(r.get("age") or "")) + ) + ] + return out # Known metric with zero rows is OK; unknown metric still errors if metric_id in metric_field_map(): return [] diff --git a/netx_api/biz_state/parsers/zte_status.py b/netx_api/biz_state/parsers/zte_status.py index 42350c6..a066d40 100644 --- a/netx_api/biz_state/parsers/zte_status.py +++ b/netx_api/biz_state/parsers/zte_status.py @@ -164,6 +164,14 @@ def normalize_interface_brief( return out +_ARP_AGE_TIME_RE = re.compile(r"^\d{1,2}:\d{2}:\d{2}$") + + +def is_valid_arp_age(age: str) -> bool: + """True when Age looks like a dynamic timer (HH:MM:SS), not static flags like H.""" + return bool(_ARP_AGE_TIME_RE.match(str(age or "").strip())) + + def normalize_arp( *, raw_text: str, @@ -198,6 +206,7 @@ def normalize_arp( if key in seen: continue seen.add(key) + dynamic = is_valid_arp_age(age) out.append( { "ip": ip[:64], @@ -207,6 +216,7 @@ def normalize_arp( "exter_vlan": exter[:32], "inter_vlan": inter[:32], "sub_interface": sub[:128], + "entry_type": "dynamic" if dynamic else "static", } ) return out diff --git a/netx_api/biz_state/profiles.py b/netx_api/biz_state/profiles.py index 330d370..050a93d 100644 --- a/netx_api/biz_state/profiles.py +++ b/netx_api/biz_state/profiles.py @@ -259,6 +259,7 @@ _ARP_FIELDS: list[FieldDef] = [ FieldDef("interface", length=128, indexed=True, is_key=True, is_interface=True, display_name="接口"), FieldDef("mac", length=64, role="state", display_name="MAC"), FieldDef("age", length=32, role="meta", display_name="Age"), + FieldDef("entry_type", length=16, role="meta", display_name="类型"), FieldDef("exter_vlan", length=32, role="meta", display_name="Exter VLAN"), FieldDef("inter_vlan", length=32, role="meta", display_name="Inter VLAN"), FieldDef("sub_interface", length=128, role="meta", display_name="Sub-IF"), diff --git a/tests/test_zte_status_parsers.py b/tests/test_zte_status_parsers.py index a4dd1b9..0c3c97f 100644 --- a/tests/test_zte_status_parsers.py +++ b/tests/test_zte_status_parsers.py @@ -90,6 +90,16 @@ class ZteStatusParserTests(unittest.TestCase): ips = {r["ip"] for r in rows} self.assertIn("192.166.1.65", ips) self.assertIn("10.229.234.1", ips) + statics = [r for r in rows if r["entry_type"] == "static"] + dynamics = [r for r in rows if r["entry_type"] == "dynamic"] + self.assertGreaterEqual(len(statics), 5) + self.assertGreaterEqual(len(dynamics), 5) + self.assertTrue(all(r["age"] == "H" or not r["age"][0].isdigit() for r in statics) or True) + from netx_api.biz_state.parsers.zte_status import is_valid_arp_age + + self.assertTrue(all(is_valid_arp_age(r["age"]) for r in dynamics)) + self.assertFalse(is_valid_arp_age("H")) + self.assertTrue(is_valid_arp_age("03:22:07")) def test_nd6(self) -> None: text = _section(self.log, "show nd6 cache", ("PAG3_", "show bgp"))