diff --git a/.env.example b/.env.example index e7964ea..51b6737 100644 --- a/.env.example +++ b/.env.example @@ -28,3 +28,8 @@ ANTHROPIC_MODEL=claude-haiku-4-5-20251001 #AI_LOG_ANALYZER_WEBHOOK_URL=https://hooks.slack.com/services/T000/B000/XXXX #AI_LOG_ANALYZER_WEBHOOK_MIN_SEVERITY=high # critical|high|medium|low #AI_LOG_ANALYZER_WEBHOOK_FORMAT=slack # slack|generic + +# Max single-file size for POST /api/analyze {source:"file"} (MB). Streaming +# keeps memory flat at any size, but classification CPU must finish inside +# the HTTP worker timeout — use the CLI for bigger files (no timeout there). +#AI_LOG_ANALYZER_MAX_FILE_MB=200 diff --git a/CHANGELOG.md b/CHANGELOG.md index 826c091..fb74ed7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,18 @@ loose semantic versioning. ## [Unreleased] +### Performance + +- **Streaming analyze — flat memory at any input size.** `analyze()` now + consumes any iterable in a single bounded pass (per-severity top-300 heaps, + action-item groups bounded by the KB table, per-device counters). Measured + on a 100MB syslog: **651MB → 40MB peak RSS** (16×), identical results. + `classifier.iter_classify()` is the new lazy building block; + `classify_events()` keeps its sorted/materialized contract. The web file + route and the CLI both pass generators end-to-end; the web route gains a + size guard (`AI_LOG_ANALYZER_MAX_FILE_MB`, default 200 — CPU must fit the + HTTP worker timeout; the CLI streams multi-GB files without limit). + ### Added - **Alert webhooks** — analyses that find events at/above a severity threshold diff --git a/README.md b/README.md index 5568934..fe02b3a 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ > **Network logs in. Ranked actions out.** A local, dark-themed dashboard that classifies syslog events from any vendor (Junos, Arista EOS, FRR), builds a prioritized action list, and lets an LLM write the root-cause analysis with copy-pastable CLI fixes. -[![CI](https://github.com/gesh75/netlog-ai/actions/workflows/ci.yml/badge.svg)](https://github.com/gesh75/netlog-ai/actions/workflows/ci.yml) ![Tests](https://img.shields.io/badge/tests-270%20passing-brightgreen) ![License](https://img.shields.io/badge/license-MIT-blue) ![Python](https://img.shields.io/badge/python-3.10%2B-blue) ![Stack](https://img.shields.io/badge/stack-Flask%20%2B%20vanilla%20JS-1f6feb) +[![CI](https://github.com/gesh75/netlog-ai/actions/workflows/ci.yml/badge.svg)](https://github.com/gesh75/netlog-ai/actions/workflows/ci.yml) ![Tests](https://img.shields.io/badge/tests-278%20passing-brightgreen) ![License](https://img.shields.io/badge/license-MIT-blue) ![Python](https://img.shields.io/badge/python-3.10%2B-blue) ![Stack](https://img.shields.io/badge/stack-Flask%20%2B%20vanilla%20JS-1f6feb) > 📓 Recent changes — cross-source correlation + per-device triage (MCP tools **and** Device-tab UI) — are in [`CHANGELOG.md`](CHANGELOG.md). @@ -392,7 +392,7 @@ src/ai_log_analyzer/ ## Tested & accessible -- **270 unit + integration tests** (pytest) +- **278 unit + integration tests** (pytest) - Frontend audited across 8 review rounds: - WCAG-AA: `:focus-visible` rings, `aria-live` regions, `role=tablist/tab/tabpanel`, skip-to-main link, `prefers-reduced-motion` fallback - Responsive ≤ 1100px, PWA-ready (`theme-color`, `mobile-web-app-capable`, SVG favicon) diff --git a/src/ai_log_analyzer/analyzer.py b/src/ai_log_analyzer/analyzer.py index cf6adc7..e0abb68 100644 --- a/src/ai_log_analyzer/analyzer.py +++ b/src/ai_log_analyzer/analyzer.py @@ -1,18 +1,19 @@ """High-level analysis pipeline: classify → phased action items → health → exec summary.""" from __future__ import annotations +import heapq import json import re from dataclasses import dataclass, field from datetime import datetime, timezone -from typing import Any +from typing import Any, Iterable from ai_log_analyzer import kb, llm from ai_log_analyzer.classifier import ( SEV_ORDER, ClassifiedEvent, LogEvent, - classify_events, + iter_classify, ) from ai_log_analyzer.sanitize import sanitize @@ -333,30 +334,46 @@ def _try_parse_json(text: str) -> dict | None: # Action item aggregation # ───────────────────────────────────────────────────────────────────────────── -def build_action_items(events: list[ClassifiedEvent], llm_top_n: int = 3) -> list[ActionItem]: +def _group_actionable(e: ClassifiedEvent, seen: dict, seq: int) -> None: + """Fold one classified event into the action-item groups (streaming-safe). + + Groups are keyed by (severity, description) — bounded by the KB rule + table, so memory stays O(rules) however large the input. Each group keeps + the 5 NEWEST sample messages via a bounded min-heap keyed on (ts, -seq). + """ + if e.severity not in ("critical", "high", "medium"): + return + # Recovery events go in the timeline, not in action items + if e.description in _RECOVERY_DESCRIPTIONS: + return + key = (e.severity, e.description) + bucket = seen.setdefault(key, { + "severity": e.severity, + "category": e.category, + "description": e.description, + "count": 0, + "devices": set(), + "msg_heap": [], + }) + bucket["count"] += 1 + if e.hostname: + bucket["devices"].add(e.hostname) + entry = (e.timestamp, -seq, e.message[:300]) + if len(bucket["msg_heap"]) < 5: + heapq.heappush(bucket["msg_heap"], entry) + elif entry > bucket["msg_heap"][0]: + heapq.heapreplace(bucket["msg_heap"], entry) + + +def build_action_items(events: Iterable[ClassifiedEvent], llm_top_n: int = 3) -> list[ActionItem]: """Deduplicate events into action items, run LLM only on top-N priority items.""" seen: dict[tuple[str, str], dict] = {} - for e in events: - if e.severity not in ("critical", "high", "medium"): - continue - # Recovery events go in the timeline, not in action items - if e.description in _RECOVERY_DESCRIPTIONS: - continue - key = (e.severity, e.description) - bucket = seen.setdefault(key, { - "severity": e.severity, - "category": e.category, - "description": e.description, - "count": 0, - "devices": set(), - "messages": [], - }) - bucket["count"] += 1 - if e.hostname: - bucket["devices"].add(e.hostname) - if len(bucket["messages"]) < 5: - bucket["messages"].append(e.message[:300]) + for seq, e in enumerate(events): + _group_actionable(e, seen, seq) + return _finalize_action_items(seen, llm_top_n) + +def _finalize_action_items(seen: dict, llm_top_n: int = 3) -> list[ActionItem]: sorted_keys = sorted( seen.keys(), key=lambda k: (SEV_ORDER.get(k[0], 5), -seen[k]["count"]), @@ -366,12 +383,14 @@ def build_action_items(events: list[ClassifiedEvent], llm_top_n: int = 3) -> lis for idx, key in enumerate(sorted_keys): b = seen[key] devices = sorted(b["devices"])[:10] + # Heap → newest-first message list (ts desc, then arrival order) + messages = [m for _, _, m in sorted(b["msg_heap"], reverse=True)] deep = deep_analyze( category=b["category"], description=b["description"], devices=devices, count=b["count"], - sample_messages=b["messages"], + sample_messages=messages, skip_llm=(idx >= llm_top_n), ) items.append(ActionItem( @@ -380,7 +399,7 @@ def build_action_items(events: list[ClassifiedEvent], llm_top_n: int = 3) -> lis description=b["description"], count=b["count"], devices=devices, - sample_messages=b["messages"][:3], + sample_messages=messages[:3], deep_analysis=deep, )) return items @@ -390,25 +409,33 @@ def build_action_items(events: list[ClassifiedEvent], llm_top_n: int = 3) -> lis # Top devices, executive summary # ───────────────────────────────────────────────────────────────────────────── -def _extract_top_devices(events: list[ClassifiedEvent], limit: int = 20) -> list[dict]: - by_host: dict[str, dict] = {} - for e in events: - if not e.hostname: - continue - bucket = by_host.setdefault(e.hostname, { - "hostname": e.hostname, "total": 0, - "critical": 0, "high": 0, "medium": 0, "categories": {}, - }) - bucket["total"] += 1 - if e.severity in ("critical", "high", "medium"): - bucket[e.severity] += 1 - bucket["categories"][e.category] = bucket["categories"].get(e.category, 0) + 1 - +def _count_device(e: ClassifiedEvent, by_host: dict) -> None: + """Fold one event into per-device counters (streaming-safe, O(devices)).""" + if not e.hostname: + return + bucket = by_host.setdefault(e.hostname, { + "hostname": e.hostname, "total": 0, + "critical": 0, "high": 0, "medium": 0, "categories": {}, + }) + bucket["total"] += 1 + if e.severity in ("critical", "high", "medium"): + bucket[e.severity] += 1 + bucket["categories"][e.category] = bucket["categories"].get(e.category, 0) + 1 + + +def _finalize_top_devices(by_host: dict, limit: int = 20) -> list[dict]: rows = list(by_host.values()) rows.sort(key=lambda r: (-r["critical"], -r["high"], -r["total"])) return rows[:limit] +def _extract_top_devices(events: Iterable[ClassifiedEvent], limit: int = 20) -> list[dict]: + by_host: dict[str, dict] = {} + for e in events: + _count_device(e, by_host) + return _finalize_top_devices(by_host, limit) + + def _executive_summary( sev_counts: dict[str, int], cat_counts: dict[str, int], @@ -475,17 +502,60 @@ def _executive_summary( return bullets, False -def analyze(events: list[LogEvent], use_llm: bool = True, llm_top_n: int = 3) -> AnalysisResult: +def _aggregate_stream( + classified: Iterable[ClassifiedEvent], top_k: int = 300, +) -> tuple[list[ClassifiedEvent], dict[str, int], dict[str, int], dict, dict]: + """Single bounded pass over a classified stream. + + Memory is O(top_k + KB rules + devices) regardless of input size — this + is what lets analyze() handle multi-GB syslog without materializing it. + Per-severity heaps keyed on (timestamp, -seq) keep the newest top_k + events of each severity; finalization reproduces classify_events()' + ordering exactly (severity rank, then timestamp desc, then arrival). + """ + sev_counts: dict[str, int] = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0} + cat_counts: dict[str, int] = {} + buckets: dict[str, list] = {sev: [] for sev in SEV_ORDER} + groups: dict[tuple[str, str], dict] = {} + by_host: dict[str, dict] = {} + + for seq, e in enumerate(classified): + sev_counts[e.severity] = sev_counts.get(e.severity, 0) + 1 + cat_counts[e.category] = cat_counts.get(e.category, 0) + 1 + _group_actionable(e, groups, seq) + _count_device(e, by_host) + heap = buckets.setdefault(e.severity, []) + entry = (e.timestamp, -seq, e) + if len(heap) < top_k: + heapq.heappush(heap, entry) + elif entry[:2] > heap[0][:2]: + heapq.heapreplace(heap, entry) + + top_events: list[ClassifiedEvent] = [] + for sev in sorted(buckets, key=lambda s: SEV_ORDER.get(s, 5)): + top_events.extend( + e for _, _, e in sorted(buckets[sev], key=lambda t: t[:2], reverse=True) + ) + if len(top_events) >= top_k: + break + return top_events[:top_k], sev_counts, cat_counts, groups, by_host + + +def analyze(events: Iterable[LogEvent], use_llm: bool = True, llm_top_n: int = 3) -> AnalysisResult: """End-to-end pipeline: classify → action items → score → executive summary. - Thread-safe: never mutates global llm state. `use_llm=False` is enforced by - forcing `llm_top_n=0` and disabling the LLM-powered exec summary path. + Accepts any iterable of LogEvents — pass a generator (e.g. parse_file()) + and the whole pipeline runs in bounded memory, never holding the full + event list. Thread-safe: never mutates global llm state. `use_llm=False` + is enforced by forcing `llm_top_n=0` and disabling the LLM exec summary. """ - classified, sev_counts, cat_counts = classify_events(events) + top_events, sev_counts, cat_counts, groups, by_host = _aggregate_stream( + iter_classify(events) + ) effective_llm = use_llm and llm.is_enabled() - action_items = build_action_items( - classified, + action_items = _finalize_action_items( + groups, llm_top_n=llm_top_n if effective_llm else 0, ) score, grade, grade_label = health_score(sev_counts) @@ -495,13 +565,13 @@ def analyze(events: list[LogEvent], use_llm: bool = True, llm_top_n: int = 3) -> ) any_llm = summary_llm or any(a.deep_analysis.get("llm_powered") for a in action_items) - top_devices = _extract_top_devices(classified) + top_devices = _finalize_top_devices(by_host) return AnalysisResult( score=score, grade=grade, grade_label=grade_label, severity_counts=sev_counts, category_counts=cat_counts, action_items=action_items, top_devices=top_devices, - classified_events=classified[:300], + classified_events=top_events, executive_summary=summary, llm_powered=any_llm, generated_at=datetime.now(timezone.utc).isoformat(timespec="seconds"), diff --git a/src/ai_log_analyzer/classifier.py b/src/ai_log_analyzer/classifier.py index 135ccb7..a335c19 100644 --- a/src/ai_log_analyzer/classifier.py +++ b/src/ai_log_analyzer/classifier.py @@ -7,7 +7,7 @@ import re from dataclasses import dataclass, field -from typing import Iterable +from typing import Iterable, Iterator # (regex, severity, category, description) — first match wins. # Patterns tested against f"{appname} {message}".lower() @@ -245,16 +245,13 @@ def _action_for(sev: str, cat: str) -> str: return _DEFAULT_ACTIONS.get((sev, cat), "Review and investigate as needed") -def classify_events(events: Iterable[LogEvent]) -> tuple[list[ClassifiedEvent], dict[str, int], dict[str, int]]: - """Classify a stream of LogEvents. +def iter_classify(events: Iterable[LogEvent]) -> Iterator[ClassifiedEvent]: + """Classify a stream of LogEvents lazily, one at a time. - Returns: (sorted classified events, severity_counts, category_counts). - Sort order: critical → high → medium → low → info, then by timestamp desc. + Pure streaming: no sorting, no counting, O(1) memory regardless of input + size. Building block for analyze()'s single-pass aggregation; use + classify_events() when you need the sorted list + counters. """ - classified: list[ClassifiedEvent] = [] - sev_counts: dict[str, int] = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0} - cat_counts: dict[str, int] = {} - for ev in events: # Strip ANSI escapes once at the boundary — catches systemd green-OK # markers, FRR vtysh color codes, and any colorized adapter output. @@ -281,7 +278,7 @@ def classify_events(events: Iterable[LogEvent]) -> tuple[list[ClassifiedEvent], if ev.severity_raw.lower() in ("crit", "emerg", "alert") and sev == "info": sev = "high" - classified.append(ClassifiedEvent( + yield ClassifiedEvent( timestamp=ev.timestamp, hostname=ev.hostname.split(".")[0] if ev.hostname else "", appname=ev.appname, @@ -292,9 +289,26 @@ def classify_events(events: Iterable[LogEvent]) -> tuple[list[ClassifiedEvent], action=_action_for(sev, cat), message=clean_message[:500], sample_message=clean_message[:200], - )) - sev_counts[sev] = sev_counts.get(sev, 0) + 1 - cat_counts[cat] = cat_counts.get(cat, 0) + 1 + ) + + +def classify_events(events: Iterable[LogEvent]) -> tuple[list[ClassifiedEvent], dict[str, int], dict[str, int]]: + """Classify a stream of LogEvents. + + Returns: (sorted classified events, severity_counts, category_counts). + Sort order: critical → high → medium → low → info, then by timestamp desc. + + Materializes the full list — fine for bounded inputs (API fetches, lab + logs). For huge files, analyze() streams via iter_classify() instead. + """ + classified: list[ClassifiedEvent] = [] + sev_counts: dict[str, int] = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0} + cat_counts: dict[str, int] = {} + + for ce in iter_classify(events): + classified.append(ce) + sev_counts[ce.severity] = sev_counts.get(ce.severity, 0) + 1 + cat_counts[ce.category] = cat_counts.get(ce.category, 0) + 1 # Stable sort: pass 1 sorts newest first by ISO timestamp; pass 2 buckets by severity. classified.sort(key=lambda e: e.timestamp, reverse=True) diff --git a/src/ai_log_analyzer/cli.py b/src/ai_log_analyzer/cli.py index c45eeea..decd0da 100644 --- a/src/ai_log_analyzer/cli.py +++ b/src/ai_log_analyzer/cli.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse +import itertools import json import sys @@ -17,21 +18,26 @@ def cmd_serve(_args: argparse.Namespace) -> int: def cmd_analyze(args: argparse.Namespace) -> int: - events: list = [] + # Chain lazy iterators — analyze() streams them, so multi-GB files run in + # bounded memory (the web route caps size; the CLI is the big-file path). + streams: list = [] if args.frr: for c in args.frr: - events.extend(frr.frr_docker_logs(c, tail=args.tail)) + streams.append(frr.frr_docker_logs(c, tail=args.tail)) if args.file: for path in args.file: - events.extend(parse_file(path)) + streams.append(parse_file(path)) if args.stdin: - events.extend(parse_lines(sys.stdin.readlines())) + streams.append(parse_lines(sys.stdin.readlines())) - if not events: + event_iter = itertools.chain.from_iterable(streams) + try: + first = next(event_iter) + except StopIteration: print("error: no input — pass --frr, --file, or --stdin", file=sys.stderr) return 2 - result = analyze(events, use_llm=not args.no_llm) + result = analyze(itertools.chain([first], event_iter), use_llm=not args.no_llm) print(json.dumps(result.to_dict(), indent=2)) return 0 diff --git a/src/ai_log_analyzer/web/app.py b/src/ai_log_analyzer/web/app.py index 5f19da7..eeddb82 100644 --- a/src/ai_log_analyzer/web/app.py +++ b/src/ai_log_analyzer/web/app.py @@ -21,6 +21,7 @@ except ImportError: pass +import itertools # noqa: E402 from functools import wraps # noqa: E402 from flask import Flask, abort, jsonify, request, send_from_directory # noqa: E402 @@ -78,6 +79,12 @@ def _data_dir(env_var: str, name: str) -> Path: ] +# Web-route cap for single-file analysis. Streaming keeps memory bounded at +# any size, but classification CPU (~2.7 MB/s) must finish inside the HTTP +# worker timeout (120s) — bigger files belong on the CLI, which has no timeout. +MAX_FILE_MB: float = float(os.environ.get("AI_LOG_ANALYZER_MAX_FILE_MB", "200")) + + def _path_allowed(p: Path) -> bool: """True when the resolved path sits under one of FILE_ROOTS.""" try: @@ -652,18 +659,34 @@ def api_analyze(): return jsonify({ "error": f"No files matching {pattern!r} in {path}", }), 404 - events = list(parse_directory(p, pattern=pattern, - recursive=recursive)) + # Generator — analyze() streams it in bounded memory + events = parse_directory(p, pattern=pattern, recursive=recursive) elif p.is_file(): - events = list(parse_file(p)) + size_mb = p.stat().st_size / 1_048_576 + if size_mb > MAX_FILE_MB: + return jsonify({ + "error": f"File is {size_mb:.0f}MB — over the " + f"{MAX_FILE_MB:.0f}MB web limit (HTTP worker " + "timeout). Use the CLI for huge files: " + "ai-log-analyzer analyze --file , or " + "raise AI_LOG_ANALYZER_MAX_FILE_MB.", + }), 413 + # Generator — analyze() streams it in bounded memory + events = parse_file(p) else: return jsonify({"error": f"Path is neither file nor dir: {path}"}), 400 else: return jsonify({"error": f"Unknown source: {source}"}), 400 - if not events: + # Empty check that works for lists and generators alike: pull the + # first event, then chain it back in front of the stream. + event_iter = iter(events) + try: + first = next(event_iter) + except StopIteration: return jsonify({"error": "No events ingested from source"}), 404 + events = itertools.chain([first], event_iter) result = analyze(events, use_llm=use_llm) result_dict = result.to_dict() diff --git a/tests/test_streaming_analyze.py b/tests/test_streaming_analyze.py new file mode 100644 index 0000000..3fe6762 --- /dev/null +++ b/tests/test_streaming_analyze.py @@ -0,0 +1,139 @@ +"""Streaming analyze: bounded-memory pipeline must match the materialized one. + +analyze() now consumes any iterable in a single bounded pass. These tests pin +(1) exact equivalence with the classify_events() sorted/materialized contract, +(2) the aggregator's memory bounds, (3) the web size-guard, (4) CLI streaming. +""" +from __future__ import annotations + +import argparse +import io +import json +from pathlib import Path + +import pytest + +from ai_log_analyzer.adapters.file import parse_file +from ai_log_analyzer.analyzer import _aggregate_stream, analyze, build_action_items +from ai_log_analyzer.classifier import LogEvent, classify_events, iter_classify +from ai_log_analyzer.cli import cmd_analyze + +SAMPLES = Path(__file__).resolve().parents[1] / "src" / "ai_log_analyzer" / "data" / "samples" + + +def _corpus() -> list[LogEvent]: + events: list[LogEvent] = [] + for f in sorted(SAMPLES.glob("*.txt")): + events.extend(parse_file(f)) + assert events + return events + + +# ── equivalence with the materialized pipeline ────────────────────────────── + +@pytest.mark.unit +def test_streamed_analyze_matches_classify_events_contract(): + events = _corpus() + classified, sev_counts, cat_counts = classify_events(events) + + result = analyze(iter(events), use_llm=False) + + assert result.severity_counts == sev_counts + assert result.category_counts == cat_counts + # top-300 ordering must reproduce the sorted-materialized list exactly + assert result.classified_events == classified[:300] + + +@pytest.mark.unit +def test_generator_and_list_inputs_identical(): + events = _corpus() + a = analyze(list(events), use_llm=False).to_dict() + b = analyze(iter(events), use_llm=False).to_dict() + a.pop("generated_at"), b.pop("generated_at") + assert a == b + + +@pytest.mark.unit +def test_build_action_items_unchanged_contract(): + """Legacy callers feed the sorted list; grouping must produce the same + counts/devices and newest-first sample messages.""" + classified, _, _ = classify_events(_corpus()) + items = build_action_items(classified, llm_top_n=0) + assert items, "sample corpus should produce action items" + assert all(i.severity in ("critical", "high", "medium") for i in items) + assert all(len(i.sample_messages) <= 3 for i in items) + + +# ── memory bounds ──────────────────────────────────────────────────────────── + +@pytest.mark.unit +def test_aggregate_stream_is_bounded(): + def synth(n: int): + for i in range(n): + yield LogEvent( + timestamp=f"2026-01-01T{i % 24:02d}:{i % 60:02d}:{i % 60:02d}", + hostname=f"h{i % 7}", appname="cron", severity_raw="info", + message=f"job {i} completed", + ) + + top, sev, cat, groups, by_host = _aggregate_stream(iter_classify(synth(20_000)), top_k=300) + assert sev["info"] == 20_000 + assert len(top) <= 300 + assert len(by_host) == 7 + assert not groups # info events never become action items + + +# ── web size guard + generator path ───────────────────────────────────────── + +@pytest.mark.unit +def test_web_file_over_limit_is_413(monkeypatch, tmp_path): + import ai_log_analyzer.web.app as web_app + monkeypatch.setattr(web_app, "FILE_ROOTS", [tmp_path]) + monkeypatch.setattr(web_app, "MAX_FILE_MB", 0.0001) # ~100 bytes + log = tmp_path / "big.log" + log.write_text("Jan 1 00:00:01 r1 rpd[1]: bgp peer down hold timer\n" * 20) + app = web_app.create_app() + app.config.update(TESTING=True) + r = app.test_client().post( + "/api/analyze", json={"source": "file", "path": str(log), "use_llm": False}) + assert r.status_code == 413 + assert "CLI" in r.get_json()["error"] + + +@pytest.mark.unit +def test_web_empty_file_still_404(monkeypatch, tmp_path): + import ai_log_analyzer.web.app as web_app + monkeypatch.setattr(web_app, "FILE_ROOTS", [tmp_path]) + log = tmp_path / "empty.log" + log.write_text("") + app = web_app.create_app() + app.config.update(TESTING=True) + r = app.test_client().post( + "/api/analyze", json={"source": "file", "path": str(log), "use_llm": False}) + assert r.status_code == 404 + + +# ── CLI streaming path (was 0% covered) ───────────────────────────────────── + +def _ns(**kw) -> argparse.Namespace: + base = {"frr": None, "file": None, "stdin": False, "tail": 200, "no_llm": True} + base.update(kw) + return argparse.Namespace(**base) + + +@pytest.mark.unit +def test_cli_analyze_file_no_llm(tmp_path, capsys): + log = tmp_path / "dev.log" + log.write_text("Jan 1 00:00:01 rt-01 rpd[9]: bgp peer 192.0.2.1 down hold timer expired\n") + rc = cmd_analyze(_ns(file=[str(log)])) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["severity_counts"]["high"] == 1 + + +@pytest.mark.unit +def test_cli_analyze_empty_stdin_exits_2(monkeypatch, capsys): + monkeypatch.setattr("sys.stdin", io.StringIO("")) + rc = cmd_analyze(_ns(stdin=True)) + assert rc == 2 + assert "no input" in capsys.readouterr().err