diff --git a/netx_api/biz_state/collect_runner.py b/netx_api/biz_state/collect_runner.py index f27e136..37a1b9d 100644 --- a/netx_api/biz_state/collect_runner.py +++ b/netx_api/biz_state/collect_runner.py @@ -28,8 +28,14 @@ from ..models import ( ) from ..ne_netmiko import disable_target_paging, send_show_command from ..ne_session_factory import close_netmiko_connection, open_netmiko_connection +from .collect_session import ( + CollectSession, + build_parse_bundle, + resolve_aux_command, + run_primary_with_bundle, +) from .command_match import expand_from_bindings, match_command, normalize_command -from .parsers import get_parser, get_parser_meta, run_parser +from .parsers import get_parser from .profiles import get_profile _log = logging.getLogger("netx.biz_state.runner") @@ -420,8 +426,13 @@ def _run_collect_session( pass sdb = SessionLocal() - # Same-batch cache: concrete CLI -> raw + parse artifacts (aux reuse). - cmd_cache: dict[str, dict[str, Any]] = {} + session = CollectSession( + conn, + vendor=vendor_eff, + device_type=device_type_eff, + vendor_key=vendor_key, + read_timeout=per_cmd, + ) try: batch_row = sdb.get(BizStateBatch, batch_id) if not batch_row: @@ -481,177 +492,119 @@ def _run_collect_session( sdb.commit() continue - primary_ck = normalize_command(concrete) - cmd_cache[primary_ck] = { - "raw": cmd_row.raw_text or "", - "fsm_tables": {}, - "records": None, - "ok": True, - "error": "", - "cmd_row_id": cmd_row.id, - } - - raws: dict[str, str] = {"primary": cmd_row.raw_text or ""} - aux_records: dict[str, list[dict[str, Any]]] = {} - fsm_extra: dict[str, list[dict[str, Any]]] = {} - command_rules: dict[str, list[str]] = { - "primary": list( - (get_parser_meta(hit.profile.parser_id) or {}).get("rule_keys") - or () - ) - } + session.remember( + concrete, + raw=cmd_row.raw_text or "", + ok=True, + cmd_row_id=cmd_row.id, + ) + resolved_aux = [] + aux_results: dict[str, Any] = {} for aux in list(hit.profile.aux_commands or []): - aux_concrete = normalize_command(aux.command_template) - if not aux_concrete: - continue - aux_rules = list(aux.rule_keys or ()) - if not aux_rules and aux.parser_id: - aux_rules = list( - (get_parser_meta(aux.parser_id) or {}).get("rule_keys") or () + try: + ra = resolve_aux_command(aux) + except ValueError as exc: + aux_row = BizStateBatchCommand( + id=uuid4().hex, + batch_id=batch_id, + task_item_id=item_id, + profile_id=str(aux.profile_id or "")[:128], + raw_command=str(aux.key or "")[:512], + params_json={}, + parse_status="aux_failed", + message=f"aux_for={cmd_row.id};resolve:{exc}"[:1020], + created_at=_utcnow(), ) - command_rules[aux.key] = aux_rules - - cached = cmd_cache.get(aux_concrete) + cmd_count += 1 + sdb.add(aux_row) + sdb.commit() + continue + resolved_aux.append(ra) aux_row = BizStateBatchCommand( id=uuid4().hex, batch_id=batch_id, task_item_id=item_id, - profile_id=aux.profile_id or hit.profile.profile_id, - parser_id=aux.parser_id or "", + profile_id=ra.profile_id, + parser_id=ra.parser_id, metric_id="", - raw_command=aux_concrete[:512], + raw_command=ra.command[:512], params_json={}, created_at=_utcnow(), ) cmd_count += 1 - if cached and cached.get("ok"): - raws[aux.key] = str(cached.get("raw") or "") - if cached.get("records") is not None: - aux_records[aux.key] = list(cached.get("records") or []) - fsm_extra.update(cached.get("fsm_tables") or {}) + entry, cache_hit = session.fetch_and_parse( + ra.command, + parser_id=ra.parser_id, + textfsm_command=ra.textfsm_command, + cmd_row_id=aux_row.id, + ) + aux_results[ra.key] = entry + if cache_hit: aux_row.parse_status = "aux_cached" aux_row.message = ( - f"aux_for={cmd_row.id};cache_hit;src={cached.get('cmd_row_id') or ''}" + f"aux_for={cmd_row.id};cache_hit;src={entry.cmd_row_id}" )[:1020] aux_row.raw_text = "" - aux_row.row_count = len(cached.get("records") or []) - sdb.add(aux_row) - sdb.commit() - continue - - try: - aux_raw = send_show_command( - conn, aux_concrete, read_timeout=per_cmd - ) - aux_row.raw_text = str(aux_raw or "") - except Exception as exc: + aux_row.row_count = len(entry.records or []) + elif not entry.ok: aux_row.parse_status = "aux_failed" aux_row.message = ( - f"aux_for={cmd_row.id};{_format_error(exc)}" + f"aux_for={cmd_row.id};{entry.error}" )[:1020] - sdb.add(aux_row) - sdb.commit() - cmd_cache[aux_concrete] = { - "raw": "", - "fsm_tables": {}, - "records": [], - "ok": False, - "error": str(exc), - "cmd_row_id": aux_row.id, - } - raws[aux.key] = "" - continue - - aux_recs: list[dict[str, Any]] = [] - aux_fsm: dict[str, list[dict[str, Any]]] = {} - if aux.parser_id and get_parser(aux.parser_id): - try: - aux_recs, aux_fsm, _aux_keys = run_parser( - aux.parser_id, - raw_text=aux_row.raw_text or "", - vendor=vendor_eff, - device_type=device_type_eff, - command=aux.textfsm_command or aux_concrete, - textfsm_command=aux.textfsm_command or "", - params={}, - ) - except Exception as exc: - aux_row.parse_status = "aux_failed" - aux_row.message = ( - f"aux_for={cmd_row.id};parse:{_format_error(exc)}" - )[:1020] - sdb.add(aux_row) - sdb.commit() - cmd_cache[aux_concrete] = { - "raw": aux_row.raw_text or "", - "fsm_tables": {}, - "records": [], - "ok": False, - "error": str(exc), - "cmd_row_id": aux_row.id, - } - raws[aux.key] = aux_row.raw_text or "" - continue - elif aux_rules: - from ..ntc_parse import apply_rules, resolve_cli_platform - - plat = resolve_cli_platform( - vendor=vendor_eff, - device_type=device_type_eff, - vendor_key=vendor_key, - ) - aux_fsm = apply_rules( - platform=plat, - text=aux_row.raw_text or "", - rule_keys=aux_rules, - command=aux.textfsm_command or aux_concrete, - ) - - raws[aux.key] = aux_row.raw_text or "" - if aux_recs: - aux_records[aux.key] = aux_recs - fsm_extra.update(aux_fsm) - aux_row.parse_status = "aux" - aux_row.message = f"aux_for={cmd_row.id}"[:1020] - aux_row.row_count = len(aux_recs) + aux_row.raw_text = entry.raw + else: + aux_row.parse_status = "aux" + aux_row.message = f"aux_for={cmd_row.id}"[:1020] + aux_row.raw_text = entry.raw + aux_row.row_count = len(entry.records or []) + # refresh cache row id to this aux row on first fetch + entry.cmd_row_id = aux_row.id sdb.add(aux_row) sdb.commit() - cmd_cache[aux_concrete] = { - "raw": aux_row.raw_text or "", - "fsm_tables": dict(aux_fsm), - "records": list(aux_recs), - "ok": True, - "error": "", - "cmd_row_id": aux_row.id, - } + bundle = build_parse_bundle( + primary_raw=cmd_row.raw_text or "", + primary_parser_id=hit.profile.parser_id, + aux_results=aux_results, + resolved_aux=resolved_aux, + ) try: - records, fsm_tables, rule_keys = run_parser( + records, fsm_tables, rule_keys = run_primary_with_bundle( hit.profile.parser_id, - raw_text=cmd_row.raw_text or "", + bundle=bundle, vendor=vendor_eff, device_type=device_type_eff, command=hit.profile.textfsm_command or concrete, - params=merged, textfsm_command=hit.profile.textfsm_command or "", - raws=raws, - command_rules=command_rules, - aux_records=aux_records, - fsm_tables_extra=fsm_extra, + params=merged, + enrich_joins=list(hit.profile.enrich_joins or []), + ) + session.remember( + concrete, + raw=cmd_row.raw_text or "", + fsm_tables=fsm_tables, + records=records, + ok=True, + cmd_row_id=cmd_row.id, ) - cmd_cache[primary_ck]["fsm_tables"] = dict(fsm_tables) - cmd_cache[primary_ck]["records"] = list(records) hints = [] if rule_keys: nonempty = [k for k in rule_keys if fsm_tables.get(k)] hints.append( f"fsm_keys={','.join(rule_keys)};hit={','.join(nonempty)}" ) - if aux_records: + if bundle.aux_records: hints.append( "aux=" - + ",".join(f"{k}:{len(v)}" for k, v in aux_records.items()) + + ",".join( + f"{k}:{len(v)}" for k, v in bundle.aux_records.items() + ) + ) + if hit.profile.enrich_joins: + hints.append( + "enrich=" + + ",".join(j.from_aux for j in hit.profile.enrich_joins) ) if hints: cmd_row.message = ";".join(hints)[:1020] diff --git a/netx_api/biz_state/collect_session.py b/netx_api/biz_state/collect_session.py new file mode 100644 index 0000000..b52dcca --- /dev/null +++ b/netx_api/biz_state/collect_session.py @@ -0,0 +1,241 @@ +"""Per-batch CLI session helpers: aux resolution, command cache, parse bundle.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable + +from .command_match import normalize_command +from .enrich import EnrichJoin, apply_enrich_joins +from .parsers import get_parser, get_parser_meta, run_parser +from .profiles import AuxCommand, ParseProfile, get_profile + +SendFn = Callable[..., str] + + +@dataclass +class CachedCommand: + raw: str = "" + fsm_tables: dict[str, list[dict[str, Any]]] = field(default_factory=dict) + records: list[dict[str, Any]] = field(default_factory=list) + ok: bool = True + error: str = "" + cmd_row_id: str = "" + + +@dataclass(frozen=True) +class ResolvedAux: + key: str + profile_id: str + command: str + textfsm_command: str + parser_id: str + rule_keys: tuple[str, ...] + profile: ParseProfile + + +def resolve_aux_command(aux: AuxCommand) -> ResolvedAux: + """Resolve aux from ``profile_id`` (single source of truth).""" + key = str(aux.key or "").strip() + pid = str(aux.profile_id or "").strip() + if not key or not pid: + raise ValueError("AuxCommand requires key and profile_id") + prof = get_profile(pid) + if not prof: + raise ValueError(f"aux profile not found: {pid}") + cmd = normalize_command(prof.command_template) + if not cmd: + raise ValueError(f"aux profile {pid} has empty command_template") + textfsm = str(prof.textfsm_command or cmd).strip() + parser_id = str(prof.parser_id or "").strip() + meta = get_parser_meta(parser_id) if parser_id else None + rule_keys = tuple((meta or {}).get("rule_keys") or ()) + return ResolvedAux( + key=key, + profile_id=pid, + command=cmd, + textfsm_command=textfsm, + parser_id=parser_id, + rule_keys=rule_keys, + profile=prof, + ) + + +@dataclass +class ParseBundle: + """Inputs for primary ``run_parser`` after primary + aux collection.""" + + raws: dict[str, str] = field(default_factory=dict) + command_rules: dict[str, list[str]] = field(default_factory=dict) + aux_records: dict[str, list[dict[str, Any]]] = field(default_factory=dict) + fsm_extra: dict[str, list[dict[str, Any]]] = field(default_factory=dict) + + +class CollectSession: + """SSH session-scoped command cache + aux fetch/parse.""" + + def __init__( + self, + conn: Any, + *, + vendor: str = "", + device_type: str = "", + vendor_key: str = "", + read_timeout: int = 120, + send_fn: SendFn | None = None, + ) -> None: + self.conn = conn + self.vendor = vendor + self.device_type = device_type + self.vendor_key = vendor_key + self.read_timeout = int(read_timeout or 120) + self._send = send_fn + self.cache: dict[str, CachedCommand] = {} + + def _send_show(self, command: str) -> str: + if self._send is None: + from ..ne_netmiko import send_show_command + + return str(send_show_command(self.conn, command, read_timeout=self.read_timeout) or "") + return str(self._send(self.conn, command, read_timeout=self.read_timeout) or "") + + def remember( + self, + command: str, + *, + raw: str = "", + fsm_tables: dict[str, list[dict[str, Any]]] | None = None, + records: list[dict[str, Any]] | None = None, + ok: bool = True, + error: str = "", + cmd_row_id: str = "", + ) -> CachedCommand: + ck = normalize_command(command) + entry = CachedCommand( + raw=str(raw or ""), + fsm_tables=dict(fsm_tables or {}), + records=list(records or []), + ok=bool(ok), + error=str(error or ""), + cmd_row_id=str(cmd_row_id or ""), + ) + self.cache[ck] = entry + return entry + + def get_cached(self, command: str) -> CachedCommand | None: + ck = normalize_command(command) + hit = self.cache.get(ck) + if hit and hit.ok: + return hit + return None + + def fetch_and_parse( + self, + command: str, + *, + parser_id: str = "", + textfsm_command: str = "", + params: dict[str, str] | None = None, + cmd_row_id: str = "", + ) -> tuple[CachedCommand, bool]: + """Return ``(entry, cache_hit)``. On miss: CLI + optional parser.""" + cached = self.get_cached(command) + if cached is not None: + return cached, True + try: + raw = self._send_show(command) + except Exception as exc: + entry = self.remember( + command, + raw="", + ok=False, + error=f"{type(exc).__name__}: {exc}", + cmd_row_id=cmd_row_id, + ) + return entry, False + records: list[dict[str, Any]] = [] + fsm_tables: dict[str, list[dict[str, Any]]] = {} + if parser_id and get_parser(parser_id): + try: + records, fsm_tables, _keys = run_parser( + parser_id, + raw_text=raw, + vendor=self.vendor, + device_type=self.device_type, + command=textfsm_command or command, + textfsm_command=textfsm_command or "", + params=params or {}, + ) + except Exception as exc: + entry = self.remember( + command, + raw=raw, + ok=False, + error=f"parse: {type(exc).__name__}: {exc}", + cmd_row_id=cmd_row_id, + ) + return entry, False + entry = self.remember( + command, + raw=raw, + fsm_tables=fsm_tables, + records=records, + ok=True, + cmd_row_id=cmd_row_id, + ) + return entry, False + + +def primary_rule_keys(parser_id: str) -> list[str]: + meta = get_parser_meta(parser_id) or {} + return list(meta.get("rule_keys") or ()) + + +def build_parse_bundle( + *, + primary_raw: str, + primary_parser_id: str, + aux_results: dict[str, CachedCommand], + resolved_aux: list[ResolvedAux], +) -> ParseBundle: + bundle = ParseBundle( + raws={"primary": str(primary_raw or "")}, + command_rules={"primary": primary_rule_keys(primary_parser_id)}, + ) + for ra in resolved_aux: + entry = aux_results.get(ra.key) or CachedCommand(ok=False) + bundle.raws[ra.key] = entry.raw + bundle.command_rules[ra.key] = list(ra.rule_keys) + if entry.records: + bundle.aux_records[ra.key] = list(entry.records) + bundle.fsm_extra.update(entry.fsm_tables or {}) + return bundle + + +def run_primary_with_bundle( + parser_id: str, + *, + bundle: ParseBundle, + vendor: str = "", + device_type: str = "", + command: str = "", + textfsm_command: str = "", + params: dict[str, str] | None = None, + enrich_joins: list[EnrichJoin] | None = None, +) -> tuple[list[dict[str, Any]], dict[str, list[dict[str, Any]]], list[str]]: + records, fsm_tables, keys = run_parser( + parser_id, + raw_text=bundle.raws.get("primary") or "", + vendor=vendor, + device_type=device_type, + command=command, + textfsm_command=textfsm_command, + params=params, + raws=bundle.raws, + command_rules=bundle.command_rules, + aux_records=bundle.aux_records, + fsm_tables_extra=bundle.fsm_extra, + ) + if enrich_joins: + apply_enrich_joins(records, bundle.aux_records, enrich_joins) + return records, fsm_tables, keys diff --git a/netx_api/biz_state/enrich.py b/netx_api/biz_state/enrich.py new file mode 100644 index 0000000..569a4ab --- /dev/null +++ b/netx_api/biz_state/enrich.py @@ -0,0 +1,64 @@ +"""Declarative cross-command enrich (equal join) for biz-state metrics.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping, Sequence + + +@dataclass(frozen=True) +class EnrichJoin: + """Copy fields from an aux metric table onto primary rows by equal key. + + Example (ARP ← if_intf):: + + EnrichJoin(from_aux="if_intf", on="interface", take=("vrf",)) + """ + + from_aux: str + take: tuple[str, ...] + on: str = "" + left_on: str = "" + right_on: str = "" + fill_missing: bool = True # on miss, set take fields to "" + + +def _sides(join: EnrichJoin) -> tuple[str, str]: + left = str(join.left_on or join.on or "").strip() + right = str(join.right_on or join.on or "").strip() + if not left or not right: + raise ValueError(f"EnrichJoin {join.from_aux!r} needs on= or left_on/right_on") + return left, right + + +def apply_enrich_joins( + records: list[dict[str, Any]], + aux_records: Mapping[str, list[dict[str, Any]]] | None, + joins: Sequence[EnrichJoin] | None, +) -> list[dict[str, Any]]: + """In-place enrich of ``records``; returns the same list.""" + if not records or not joins: + return records + aux_map = aux_records or {} + for join in joins: + left, right = _sides(join) + take = [str(t).strip() for t in (join.take or ()) if str(t).strip()] + if not take: + continue + index: dict[str, dict[str, Any]] = {} + for row in aux_map.get(str(join.from_aux or "").strip()) or []: + if not isinstance(row, dict): + continue + key = str(row.get(right) or "").strip() + if key and key not in index: + index[key] = row + for rec in records: + if not isinstance(rec, dict): + continue + hit = index.get(str(rec.get(left) or "").strip()) + for field in take: + if hit is not None: + rec[field] = str(hit.get(field) or "").strip() + elif join.fill_missing: + rec[field] = "" + return records diff --git a/netx_api/biz_state/parsers/__init__.py b/netx_api/biz_state/parsers/__init__.py index 5ea217c..ad10d1a 100644 --- a/netx_api/biz_state/parsers/__init__.py +++ b/netx_api/biz_state/parsers/__init__.py @@ -3,29 +3,32 @@ Layout:: parsers/ - common/ # cross-vendor (lldp, vrf_list, vrf_route_summary, pipeline) + common/ # cross-vendor + pipeline.prefer_fsm zte/ # implemented status tables - cisco/ # skeleton — add command modules here - huawei/ - h3c/ - juniper/ - nokia/ - ericsson/ + cisco/ huawei/ … # skeletons Each vendor package exports ``PARSERS: dict[parser_id, normalize_fn]``. Adding a status metric ---------------------- -1. (Recommended) ``cli_templates//.textfsm`` + ``index`` line -2. ``parsers//.py``: ``RULE_KEYS = ("",)`` + - ``normalize(..., fsm_tables=...)`` (use ``common.pipeline.prefer_fsm``) -3. Register in the vendor ``PARSERS`` dict -4. ``profiles.py``: add ``ParseProfile`` (command match + ``FieldDef`` schema); - optional ``aux_commands`` for multi-command collect items +1. ``cli_templates//.textfsm`` + ``index`` (FSM first) +2. ``parsers//.py``: ``RULE_KEYS`` + ``normalize`` + ``prefer_fsm`` +3. Register in vendor ``PARSERS`` +4. ``profiles.py``: ``ParseProfile`` (match + ``FieldDef``) -Collect runs TextFSM rules first → ``fsm_tables``, then calls ``normalize`` with -``raw_text`` / ``raws`` / ``fsm_tables`` / ``aux_records``. Prefer FSM; hand-parse -only when FSM has no rows. +Cross-command (multi aux) +------------------------- +1. Implement each aux as a normal status profile (steps 1–4) +2. On the primary profile:: + + aux_commands=[AuxCommand(key="if_intf", profile_id="zte.if_intf")] + enrich_joins=[EnrichJoin(from_aux="if_intf", on="interface", take=("vrf",))] + +3. CollectSession caches identical concrete CLI in one batch; enrich runs after + primary normalize (no join logic inside the parser). + +Complex joins that cannot be expressed as equal-field copy still go in +``normalize`` using ``raws`` / ``aux_records`` / ``fsm_tables``. """ from __future__ import annotations diff --git a/netx_api/biz_state/parsers/zte/arp.py b/netx_api/biz_state/parsers/zte/arp.py index 8132543..eb0e6e9 100644 --- a/netx_api/biz_state/parsers/zte/arp.py +++ b/netx_api/biz_state/parsers/zte/arp.py @@ -1,7 +1,7 @@ """ZTE: show arp [| one-line]. -Pipeline: TextFSM rule ``zte_zxros_show_arp`` → map rows → hand fallback. -VRF filled from aux ``if_intf`` (show running-config if-intf), FSM-first. +Pipeline: TextFSM ``zte_zxros_show_arp`` → map → hand fallback. +Cross-command VRF comes from profile ``enrich_joins`` (if_intf), not here. """ from __future__ import annotations @@ -12,7 +12,6 @@ from typing import Any, Mapping from ....lldp_shared import resolve_vendor_key from ....ntc_parse import apply_rules, resolve_cli_platform, row_get from ..common.pipeline import prefer_fsm -from .if_intf import parse_if_intf_vrf_map RULE_KEYS = ("zte_zxros_show_arp",) @@ -34,7 +33,6 @@ def _row_from_fields( exter: str = "", inter: str = "", sub: str = "", - vrf: str = "", ) -> dict[str, Any] | None: ip = str(ip or "").strip() iface = str(iface or "").strip() @@ -51,7 +49,7 @@ def _row_from_fields( "inter_vlan": str(inter or "").strip()[:32], "sub_interface": str(sub or "").strip()[:128], "entry_type": "dynamic" if dynamic else "static", - "vrf": str(vrf or "").strip()[:128], + "vrf": "", } @@ -114,40 +112,6 @@ def _hand_parse(*, raw_text: str, **_kw: Any) -> list[dict[str, Any]]: return out -def _vrf_map_from_aux( - *, - raws: Mapping[str, str] | None, - aux_records: Mapping[str, list[dict[str, Any]]] | None, - fsm_tables: Mapping[str, list[dict[str, Any]]] | None, -) -> dict[str, str]: - recs = list((aux_records or {}).get("if_intf") or []) - if recs: - return parse_if_intf_vrf_map(rows=recs) - fsm_rows = (fsm_tables or {}).get("zte_zxros_show_running_config_if_intf") or [] - if fsm_rows: - mapped = [ - { - "interface": row_get(r, "INTERFACE", "interface"), - "vrf": row_get(r, "VRF", "vrf"), - } - for r in fsm_rows - ] - return parse_if_intf_vrf_map(rows=mapped) - raw = str((raws or {}).get("if_intf") or "") - if raw.strip(): - return parse_if_intf_vrf_map(raw_text=raw) - return {} - - -def _enrich_vrf(rows: list[dict[str, Any]], vrf_map: dict[str, str]) -> list[dict[str, Any]]: - if not rows: - return rows - for r in rows: - iface = str(r.get("interface") or "").strip() - r["vrf"] = (vrf_map.get(iface) or "")[:128] - return rows - - def normalize_arp( *, raw_text: str, @@ -159,7 +123,7 @@ def normalize_arp( raws: Mapping[str, str] | None = None, aux_records: Mapping[str, list[dict[str, Any]]] | None = None, ) -> list[dict[str, Any]]: - _ = params + _ = (params, raws, aux_records) tables = dict(fsm_tables or {}) if not any(tables.get(k) for k in RULE_KEYS): platform = resolve_cli_platform( @@ -172,7 +136,7 @@ def normalize_arp( tables = apply_rules( platform=platform, text=raw_text, rule_keys=RULE_KEYS, command=cmd ) - rows = prefer_fsm( + return prefer_fsm( tables, RULE_KEYS, _map_fsm_rows, @@ -182,8 +146,6 @@ def normalize_arp( device_type=device_type, command=command, ) - vrf_map = _vrf_map_from_aux(raws=raws, aux_records=aux_records, fsm_tables=tables) - return _enrich_vrf(rows, vrf_map) normalize_arp.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/profiles.py b/netx_api/biz_state/profiles.py index 0ce2a96..ef8bb81 100644 --- a/netx_api/biz_state/profiles.py +++ b/netx_api/biz_state/profiles.py @@ -8,6 +8,8 @@ from __future__ import annotations from dataclasses import dataclass, field from typing import Any +from .enrich import EnrichJoin + @dataclass(frozen=True) class FieldDef: @@ -36,14 +38,10 @@ class PlaceholderDef: @dataclass(frozen=True) class AuxCommand: - """Secondary CLI bound to a collect profile (same session, optional cache).""" + """Secondary collect: only key + profile_id (rest from that profile).""" key: str - command_template: str - textfsm_command: str = "" - parser_id: str = "" - rule_keys: tuple[str, ...] = () - profile_id: str = "" + profile_id: str @dataclass @@ -65,6 +63,7 @@ class ParseProfile: enabled: bool = True kind: str = "collect" # collect | discover aux_commands: list[AuxCommand] = field(default_factory=list) + enrich_joins: list[EnrichJoin] = field(default_factory=list) _LLDP_FIELDS: list[FieldDef] = [ @@ -358,14 +357,10 @@ def _zte_status_profiles() -> list[ParseProfile]: enabled=True, kind="collect", aux_commands=[ - AuxCommand( - key="if_intf", - command_template="show running-config if-intf", - textfsm_command="show running-config if-intf", - parser_id="if_intf", - rule_keys=("zte_zxros_show_running_config_if_intf",), - profile_id="zte.if_intf", - ) + AuxCommand(key="if_intf", profile_id="zte.if_intf"), + ], + enrich_joins=[ + EnrichJoin(from_aux="if_intf", on="interface", take=("vrf",)), ], ), ParseProfile( @@ -546,12 +541,19 @@ def profile_to_public_dict(p: ParseProfile, *, overrides: dict[str, Any] | None "aux_commands": [ { "key": a.key, - "command_template": a.command_template, - "textfsm_command": a.textfsm_command or a.command_template, - "parser_id": a.parser_id, - "rule_keys": list(a.rule_keys), "profile_id": a.profile_id, } for a in (p.aux_commands or []) ], + "enrich_joins": [ + { + "from_aux": j.from_aux, + "on": j.on, + "left_on": j.left_on, + "right_on": j.right_on, + "take": list(j.take), + "fill_missing": j.fill_missing, + } + for j in (p.enrich_joins or []) + ], } diff --git a/tests/test_enrich_framework.py b/tests/test_enrich_framework.py new file mode 100644 index 0000000..03a52a5 --- /dev/null +++ b/tests/test_enrich_framework.py @@ -0,0 +1,135 @@ +"""Tests for declarative enrich + slim aux (profile_id only).""" + +from __future__ import annotations + +import unittest + +from netx_api.biz_state.collect_session import ( + CachedCommand, + CollectSession, + build_parse_bundle, + resolve_aux_command, + run_primary_with_bundle, +) +from netx_api.biz_state.enrich import EnrichJoin, apply_enrich_joins +from netx_api.biz_state.parsers import run_parser +from netx_api.biz_state.profiles import AuxCommand, get_profile, reload_profiles +from netx_api.ntc_parse import _all_index_entries + + +IF_INTF_SAMPLE = """\ +! +interface cdgei-0/1/0/1.1 + ip vrf forwarding IuB_UP-evpn +$ +interface cdgei-0/1/1/2.1 + ip vrf forwarding 400G-1 +$ +""" + +ARP_MATCHING = """\ +IP Hardware Exter Inter Sub +Address Age Address Interface VlanID VlanID Interface +-------------------------------------------------------------------------------- +131.1.1.2 03:22:07 0011.2233.4455 cdgei-0/1/0/1.1 N/A N/A cdgei-0/1/0/1.1 +11.1.1.2 H 00aa.bbcc.ddee cdgei-0/1/1/2.1 N/A N/A N/A +10.0.0.1 01:00:00 aabb.ccdd.eeff gei-0/0/0/1 N/A N/A N/A +""" + + +class EnrichJoinTests(unittest.TestCase): + def test_apply_enrich(self) -> None: + rows = [ + {"interface": "a", "ip": "1.1.1.1", "vrf": ""}, + {"interface": "b", "ip": "2.2.2.2", "vrf": ""}, + ] + aux = {"if_intf": [{"interface": "a", "vrf": "V1"}]} + apply_enrich_joins( + rows, + aux, + [EnrichJoin(from_aux="if_intf", on="interface", take=("vrf",))], + ) + self.assertEqual(rows[0]["vrf"], "V1") + self.assertEqual(rows[1]["vrf"], "") + + +class AuxResolveTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + reload_profiles() + + def test_resolve_from_profile_id(self) -> None: + ra = resolve_aux_command(AuxCommand(key="if_intf", profile_id="zte.if_intf")) + self.assertEqual(ra.command, "show running-config if-intf") + self.assertEqual(ra.parser_id, "if_intf") + self.assertIn("zte_zxros_show_running_config_if_intf", ra.rule_keys) + + def test_arp_profile_slim(self) -> None: + p = get_profile("zte.arp") + assert p is not None + self.assertEqual(p.aux_commands[0].key, "if_intf") + self.assertEqual(p.aux_commands[0].profile_id, "zte.if_intf") + self.assertEqual(len(p.enrich_joins), 1) + self.assertEqual(p.enrich_joins[0].take, ("vrf",)) + + +class CollectSessionCacheTests(unittest.TestCase): + def test_cache_hit(self) -> None: + calls: list[str] = [] + + def send(_conn, cmd, read_timeout=0): + calls.append(cmd) + return "RAW" + + sess = CollectSession(None, send_fn=send, read_timeout=1) + e1, hit1 = sess.fetch_and_parse("show running-config if-intf", parser_id="") + e2, hit2 = sess.fetch_and_parse("show running-config if-intf", parser_id="") + self.assertFalse(hit1) + self.assertTrue(hit2) + self.assertEqual(calls, ["show running-config if-intf"]) + self.assertEqual(e2.raw, "RAW") + + +class ArpEnrichPipelineTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + _all_index_entries.cache_clear() + reload_profiles() + + def test_run_primary_with_enrich(self) -> None: + if_recs, if_fsm, _ = run_parser( + "if_intf", + raw_text=IF_INTF_SAMPLE, + vendor="zte", + device_type="zte_zxros", + command="show running-config if-intf", + ) + ra = resolve_aux_command(AuxCommand(key="if_intf", profile_id="zte.if_intf")) + bundle = build_parse_bundle( + primary_raw=ARP_MATCHING, + primary_parser_id="arp", + aux_results={ + "if_intf": CachedCommand( + raw=IF_INTF_SAMPLE, fsm_tables=if_fsm, records=if_recs, ok=True + ) + }, + resolved_aux=[ra], + ) + p = get_profile("zte.arp") + assert p is not None + records, _tables, _keys = run_primary_with_bundle( + "arp", + bundle=bundle, + vendor="zte", + device_type="zte_zxros", + command="show arp", + enrich_joins=list(p.enrich_joins), + ) + by_ip = {r["ip"]: r for r in records} + self.assertEqual(by_ip["131.1.1.2"]["vrf"], "IuB_UP-evpn") + self.assertEqual(by_ip["11.1.1.2"]["vrf"], "400G-1") + self.assertEqual(by_ip["10.0.0.1"]["vrf"], "") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_multi_command_arp.py b/tests/test_multi_command_arp.py index e6a2031..0f2bc7d 100644 --- a/tests/test_multi_command_arp.py +++ b/tests/test_multi_command_arp.py @@ -97,11 +97,18 @@ class ArpMultiCommandTests(unittest.TestCase): assert p is not None self.assertEqual(len(p.aux_commands), 1) self.assertEqual(p.aux_commands[0].key, "if_intf") - self.assertEqual(p.aux_commands[0].parser_id, "if_intf") - self.assertIsNotNone(get_profile("zte.if_intf")) + self.assertEqual(p.aux_commands[0].profile_id, "zte.if_intf") self.assertEqual(get_parser_meta("if_intf")["rule_keys"], ("zte_zxros_show_running_config_if_intf",)) def test_arp_enriches_vrf_via_aux_records(self) -> None: + from netx_api.biz_state.collect_session import ( + CachedCommand, + build_parse_bundle, + resolve_aux_command, + run_primary_with_bundle, + ) + from netx_api.biz_state.profiles import AuxCommand + if_recs, if_fsm, _ = run_parser( "if_intf", raw_text=IF_INTF_SAMPLE, @@ -109,20 +116,26 @@ class ArpMultiCommandTests(unittest.TestCase): device_type="zte_zxros", command="show running-config if-intf", ) - records, tables, keys = run_parser( + ra = resolve_aux_command(AuxCommand(key="if_intf", profile_id="zte.if_intf")) + bundle = build_parse_bundle( + primary_raw=ARP_MATCHING, + primary_parser_id="arp", + aux_results={ + "if_intf": CachedCommand( + raw=IF_INTF_SAMPLE, fsm_tables=if_fsm, records=if_recs, ok=True + ) + }, + resolved_aux=[ra], + ) + p = get_profile("zte.arp") + assert p is not None + records, tables, keys = run_primary_with_bundle( "arp", - raw_text=ARP_MATCHING, + bundle=bundle, vendor="zte", device_type="zte_zxros", - command="show arp | one-line", - textfsm_command="show arp", - raws={"primary": ARP_MATCHING, "if_intf": IF_INTF_SAMPLE}, - command_rules={ - "primary": ["zte_zxros_show_arp"], - "if_intf": ["zte_zxros_show_running_config_if_intf"], - }, - aux_records={"if_intf": if_recs}, - fsm_tables_extra=if_fsm, + command="show arp", + enrich_joins=list(p.enrich_joins), ) self.assertIn("zte_zxros_show_arp", keys) by_ip = {r["ip"]: r for r in records}