diff --git a/docs/reviews/assets/biz-compare-desktop.png b/docs/reviews/assets/biz-compare-desktop.png new file mode 100644 index 0000000..17926f3 Binary files /dev/null and b/docs/reviews/assets/biz-compare-desktop.png differ diff --git a/docs/reviews/assets/biz-compare-error.png b/docs/reviews/assets/biz-compare-error.png new file mode 100644 index 0000000..e032b23 Binary files /dev/null and b/docs/reviews/assets/biz-compare-error.png differ diff --git a/docs/reviews/assets/biz-compare-legacy.png b/docs/reviews/assets/biz-compare-legacy.png new file mode 100644 index 0000000..a374dee Binary files /dev/null and b/docs/reviews/assets/biz-compare-legacy.png differ diff --git a/docs/reviews/assets/biz-compare-memory-benchmark.json b/docs/reviews/assets/biz-compare-memory-benchmark.json new file mode 100644 index 0000000..70290b7 --- /dev/null +++ b/docs/reviews/assets/biz-compare-memory-benchmark.json @@ -0,0 +1,29 @@ +{ + "date": "2026-10-10", + "baseline_ref": "afa8e06b2de64100f963ae30b3babfcb5d7009f4", + "python": "3.14.3", + "dataset": { + "rows_per_side": 1000000, + "fields": ["id", "neighbor", "state", "afi"], + "strata": 50, + "unchanged_limit": 5000, + "compact_unchanged": true, + "interface_transforms": false, + "all_rows_unchanged": true + }, + "measurement": "tracemalloc peak allocated during compare_rows; excludes input allocations, DB loading and process overhead", + "results": [ + { + "engine": "baseline", + "peak_engine_mib": 1144.43, + "seconds_with_tracemalloc": 92.4 + }, + { + "engine": "optimized", + "peak_engine_mib": 420.49, + "seconds_with_tracemalloc": 63.681 + } + ], + "results_and_sample_order_identical": true, + "test_result": "1 passed in 157.88s" +} diff --git a/docs/reviews/assets/biz-compare-narrow.png b/docs/reviews/assets/biz-compare-narrow.png new file mode 100644 index 0000000..f6f316a Binary files /dev/null and b/docs/reviews/assets/biz-compare-narrow.png differ diff --git a/docs/reviews/biz-compare-review-2026-10-10.md b/docs/reviews/biz-compare-review-2026-10-10.md new file mode 100644 index 0000000..f3150e1 --- /dev/null +++ b/docs/reviews/biz-compare-review-2026-10-10.md @@ -0,0 +1,120 @@ +# 业务对比检查与优化记录 + +检查日期:2026-10-10。范围:网络管理 → 割接对比 → 业务对比,包括前端任务/批次/结果界面、Python 与 PostgreSQL 对比路径、结果分页与搜索、短期列表缓存。 + +本次直接修改项目代码并添加回归测试,没有连接设备执行采集,也没有修改生产数据库。浏览器截图与交互测试使用固定 API 响应。 + +## 已修复的问题 + +### 1. 搜索会改变业务判定(P1) + +原实现分别按搜索条件过滤割接前、后的原表,再重新对比。例如 `state=up → down`,搜索 `up` 后只剩前侧,原本的“变化”就可能变成“删除”。重复身份键的业务还可能在筛选后错位配对。 + +现在:已存储的判定直接在结果表分页筛选;需要回查未保存全量明细的成功行时,先找到候选身份键,再补齐两侧的完整分组,按原有配对规则对比,最后筛选。达到读取上限的分组整体跳过,并返回截断标记,避免产生虚假的新增或删除。 + +### 2. PostgreSQL 搜索漏用部分模板过滤条件(P1) + +原搜索只下推 SQL 支持的过滤规则,对 `age_timer` 等不支持的规则直接忽略,可能将静态 ARP 等不属于模板范围的数据纳入结果。现在不支持下推的规则走流式读取并在 Python 中完整过滤。 + +### 3. 快速切换时旧请求覆盖当前任务(P1) + +前后采集任务的批次列表、打开任务、加载历史结果以及轮询缺少完整的过期保护。现在对批次请求使用清理标记,对任务与结果使用请求序号;关闭弹窗、切换路由或打开其他任务后,旧响应不能覆盖当前界面。 + +### 4. 结果加载错误容易被误读(P2) + +原查询失败时保留上一组明细,界面已经显示新筛选条件;切到未完成的表后,加载状态也可能无法结束。现在加载和失败都有独立状态,错误时清除旧明细并提供重试入口;切换表/页签后清理加载状态。 + +### 5. LLDP 搜索漏查(P2) + +LLDP 数据存于 `biz_state_lldp_neighbor`,原搜索只读 `biz_state_metric_row`。现在原表回查与精简结果的字段搜索都支持 LLDP 表。 + +### 6. 其他已修复边界 + +- 恰好命中读取上限时,原非 PostgreSQL 路径错误地报告“已截断”;现在多读一条判断。 +- 无效或有歧义的表标识可能落到其他表;现在返回 `sheet_not_found`。 +- `%`、`_` 等搜索字符按字面值匹配,避免不同查询路径含义不一致。 +- 关闭最后一项检查后,空数组被解释为“启用全部”;现在阻止关闭最后一项并提示至少保留一项。 +- 两个 HeroUI Input 使用了 `isDisabled`,实际需要 `disabled`;过滤值与容差输入的禁用状态已修正。 +- 列表缓存缺少并发合并,慢请求可以覆盖强制刷新结果或在失效后重新填充缓存;现在共享请求、保护失效状态并抑制过期的后台通知。 +- 清理中英文文案中的重复键,解决相关类型检查错误。 + +## 性能改动与实际收益范围 + +- 合并两套轮询,下一轮在本轮请求完成后启动。模拟浏览器测试中,运行任务在 5.2 秒内请求历史列表和详情各两次,没有重复轮询,也不会因慢请求持续叠加。 +- 已保存的失败、新增、变化等结果在数据库中筛选并分页,避免每次翻页重新加载和对比原始批次。回归测试明确验证这一查询路径不调用 `compare_rows`。 +- 同一前后采集任务的批次请求共用短期缓存,列表后台刷新合并同一请求;6 项缓存测试覆盖并发、强制刷新、失败重试与失效竞态。 +- 模板查找使用 Map,避免列表搜索与绘制逐项线性查找模板。 +- 修改筛选后先回到第一页,避免先请求旧页码再重复请求第一页。 + +这些改动减少了重复工作。以下后续基准使用合成数据,没有测量生产环境延迟。 + +## 后续优化(2026-10-10) + +### 历史执行配置与精简行还原 + +新运行在创建时将模板、启用的检查项、字段规则/过滤、接口归一化、端口映射、成功行保存策略和名称存入 `summary_json.config_snapshot`。后台执行、成功行回查、精简行还原以及导出共用该配置;排队之后修改配置,甚至删除原模板/映射,也不会改变这一运行使用的配置。无需数据库结构迁移。 + +修复了精简存储明细还原时将 `mapped_before` 直接设为原始 `before` 的问题。现在前侧依次归一化和映射,后侧归一化,保持与完整结果的显示及 CSV 一致。ZIP 新增 `config_snapshot.json`,便于核对当时配置;详情接口只返回快照模板与版本,不将完整端口映射加入轮询响应。 + +页面显示“执行配置已保存”;旧记录显示“旧记录:未保存执行配置”。旧运行无法可靠补造当时的配置,仍保留兼容查询,成功行回查及精简行还原会使用当前配置。快照也不保存原始采集行的副本,若原始批次数据被清理,依赖行 ID 的还原仍会缺少内容。 + +### Python 大表内存 + +成功样本按“层内序号、层首次出现顺序”排序,只保留样本上限内的候选和分层状态,避免先积累全部成功行。保留原分层轮询顺序;未截断的小表仍保持采集顺序。不需要接口变换时复用输入行;有变换时创建新行,不修改输入数据。稳定身份键顺序直接使用分组字典,省去再次计算全表键、额外键列表与集合。 + +每侧 1,000,000 行、50 个邻居分层、全部成功、保存 5,000 条精简样本的测试:优化前引擎新增分配峰值 **1144.43 MiB**,优化后 **420.49 MiB**,下降约 **63.3%**。完整结果与样本顺序的 SHA-256 一致。开启 `tracemalloc` 时耗时为 92.400 秒和 63.681 秒;该计时含内存追踪开销,不代表生产耗时。输入行在追踪开始前创建,峰值不包含输入行、数据库加载或 Python 进程的总内存。 + +基线为 `afa8e06b2de64100f963ae30b3babfcb5d7009f4` 中的对比引擎。原始指标见 [百万行测试结果](assets/biz-compare-memory-benchmark.json)。身份键分组仍需要随原表行数增长的内存;这次不是完全流式对比。 + +### 大表导出 + +导出改为按 `(seq, id)` 游标分批查询、分批还原精简行、逐行将 UTF-8 CSV 写入 ZIP,避免将全部明细与完整 CSV 同时保存在内存。相同 `seq` 的多条结果按 `id` 连续读取,不漏掉跨批次的同行序号记录。保留 BOM、字段顺序及逗号/引号/换行转义。最终压缩包仍由现有接口在内存中返回,压缩文件本身的大小仍计入内存。 + +## UI 改动 + +结果区分为当前检查项/全表通过率、结果类型、搜索与辅助操作三层。增加明确的分页范围、清空筛选、加载状态、错误重试和搜索无结果提示。筛选按钮增加按下状态,字段筛选增加展开状态与键盘焦点提示。后续增加了执行配置保存状态及旧记录限制提示。 + +窄屏检查项导航改为紧凑的横向排列,避免一个检查项占据整行的大块区域。浏览器检查覆盖 1440px、768px、390px;筛选栏没有横向溢出。结果表保留必要的横向滚动。 + +- [桌面结果区](assets/biz-compare-desktop.png) +- [768px 结果区](assets/biz-compare-narrow.png) +- [加载失败与重试](assets/biz-compare-error.png) +- [旧记录执行配置提示](assets/biz-compare-legacy.png) + +## 验证 + +- Python:87 项通过,1 项百万行基准默认跳过;单独启用百万行基准后通过。新增测试覆盖搜索、历史配置快照、有界抽样与游标导出。 +- 缓存:6 项 Node 测试通过。 +- 浏览器:10 个场景通过,包括配置保存状态、旧记录提示、结果布局、错误清除旧明细、重试、清空筛选、窄屏布局、旧任务响应保护、单套轮询、无运行时异常。 +- Vite 生产打包成功。 +- 本次业务对比页面的类型错误已经修复。全项目 `tsc -b` 仍被 `BizStatePage.tsx`、`BizMigrationPage.tsx` 中原有的类型/未使用变量错误阻塞。 +- 页面 ESLint 原有 34 个错误,主要是 `any` 与现有 React Hook 写法;本次未增加错误。结果请求依赖警告已修正。缓存模块没有 lint 错误。 +- PostgreSQL 路径验证了 SQL 编译与规则降级分支,没有在真实 PostgreSQL 数据库上执行查询或测量执行计划。 + +可重复执行的检查: + +```powershell +# 项目根目录 +.venv\Scripts\python.exe -m pytest tests/test_biz_state_compare_export.py tests/test_biz_state_compare_snapshot.py tests/test_biz_state_compare_memory.py tests/test_biz_state_compare_search.py tests/test_biz_state_compare.py tests/test_biz_state_compare_sql.py tests/test_biz_state_compare_recovery.py -q + +# 可选:百万行基准,耗时约数分钟,内存占用较大 +$env:NETX_BENCH_ROWS = '1000000' +$env:NETX_BENCH_REF = 'afa8e06b2de64100f963ae30b3babfcb5d7009f4' +.venv\Scripts\python.exe -m pytest tests/test_biz_state_compare_memory.py::test_memory_benchmark_against_git_baseline -q -s +Remove-Item Env:NETX_BENCH_ROWS, Env:NETX_BENCH_REF + +# web 目录 +node --experimental-strip-types --test tests/cutoverDataCache.test.mjs +node node_modules/vite/bin/vite.js build +node node_modules/typescript/bin/tsc -b +``` + +浏览器脚本为 `web/tests/bizCompare.browser.mjs`。先启动本地 Vite 服务,再设置 `NETX_PLAYWRIGHT_MODULE` 为可用的 Playwright 包路径;`NETX_TEST_URL` 默认是 `http://127.0.0.1:5179`,浏览器默认使用已安装的 Chrome。所有 `/v1` 请求由脚本提供固定响应,不会访问实际设备或数据库。 + +## 后续值得优化的部分 + +1. **Python 大表分组。** 有界抽样和冗余复制优化已完成,百万行基准下降约 63% 的引擎新增分配峰值。分组与输入仍随原表规模增长;进一步降低内存可研究排序流式配对或分区执行,并保持重复键配对语义。 +2. **PostgreSQL 搜索与深分页。** JSON 字段包含搜索和精确总数统计仍可能扫描大量结果。需要用真实数据查看 `EXPLAIN (ANALYZE, BUFFERS)`,再决定常用字段索引、搜索文本索引或游标分页;本次没有盲目添加生产索引。 +3. **历史配置与原始数据保留。** 新运行的执行配置快照已完成。旧运行仍无法还原未记录的当时配置;精简结果还原依赖原批次行,后续可为批次清理增加引用保护或保留策略。 +4. **重复键的业务语义。** 引擎按采集顺序配对同键行。如果业务应按集合判断,CLI 行顺序改变可能产生多条“不一致”。当前保留原有语义;可优先完善身份键,或另行实现并测试 Python/SQL 一致的集合匹配。 +5. **原表回查的数量上限。** 每侧候选命中最多 2000 条,完整分组补齐每侧预算 10000 条,最终展示最多 200 对;这不是全表匹配总数。界面会提示截断,大范围检索需要收窄条件。 diff --git a/netx_api/biz_state/compare_engine.py b/netx_api/biz_state/compare_engine.py index 3871cc4..5957d79 100644 --- a/netx_api/biz_state/compare_engine.py +++ b/netx_api/biz_state/compare_engine.py @@ -2,8 +2,10 @@ from __future__ import annotations -from collections import defaultdict, deque -from typing import Any, Mapping, Sequence +from collections import defaultdict +from heapq import heappush, heapreplace +from itertools import chain +from typing import Any, Iterable, Mapping, Sequence from .compare_rules import field_rule_map, values_equal, explain_diff from .iface_normalize import ( @@ -28,36 +30,52 @@ def stratum_key(row: Mapping[str, Any] | None) -> str: return "|".join(parts) if parts else "_" -def stratify_take(items: list[Any], limit: int, *, key_fn) -> list[Any]: - """Round-robin across strata so one neighbor/direction cannot consume the whole sample.""" - lim = max(0, int(limit)) - if lim <= 0 or not items: +class _StratifiedSample: + """Keep the first N round-robin ranks in O(N) space, even with many strata.""" + + def __init__(self, limit: int) -> None: + self.limit = max(0, int(limit)) + self.count = 0 + self.strata: dict[str, tuple[int, int]] = {} + self.heap: list[tuple[int, int, int, Any]] = [] + + def add(self, item: Any, key: str) -> None: + serial = self.count + self.count += 1 + if not self.limit: + return + state = self.strata.get(key) + if state is None: + # Later strata cannot beat the first item of N earlier strata. + if len(self.strata) >= self.limit: + return + index, round_n = len(self.strata), 0 + else: + index, round_n = state + self.strata[key] = (index, round_n + 1) + entry = (-round_n, -index, serial, item) + if len(self.heap) < self.limit: + heappush(self.heap, entry) + elif entry[:2] > self.heap[0][:2]: + heapreplace(self.heap, entry) + + def picked(self) -> list[Any]: + # The legacy helper preserves encounter order when no truncation occurs. + if self.count <= self.limit: + entries = sorted(self.heap, key=lambda e: e[2]) + else: + entries = sorted(self.heap, key=lambda e: (-e[0], -e[1])) + return [entry[3] for entry in entries] + + +def stratify_take(items: Iterable[Any], limit: int, *, key_fn) -> list[Any]: + """Round-robin across strata without retaining every candidate.""" + sample = _StratifiedSample(limit) + if not sample.limit: return [] - if len(items) <= lim: - return list(items) - buckets: dict[str, deque[Any]] = defaultdict(deque) - order: list[str] = [] - for it in items: - sk = str(key_fn(it) or "_") - if sk not in buckets: - order.append(sk) - buckets[sk].append(it) - out: list[Any] = [] - while len(out) < lim and buckets: - drained: list[str] = [] - for sk in order: - q = buckets.get(sk) - if not q: - drained.append(sk) - continue - out.append(q.popleft()) - if len(out) >= lim: - break - for sk in drained: - buckets.pop(sk, None) - if sk in order: - order = [x for x in order if x != sk] - return out + for item in items: + sample.add(item, str(key_fn(item) or "_")) + return sample.picked() def apply_port_map( @@ -198,10 +216,10 @@ def compare_rows( before_norm = apply_iface_normalize_rows( before_rows, iface_fields=iface_list, rules=norm_rules - ) + ) if norm_rules and iface_list else before_rows after_norm = apply_iface_normalize_rows( after_rows, iface_fields=iface_list, rules=norm_rules - ) + ) if norm_rules and iface_list else after_rows ignore_ports = False # No map → optionally ignore port renames by dropping iface from match key. @@ -216,9 +234,16 @@ def compare_rows( match_keys = list(key_fields) else: # Auto heuristic (legacy default) - before_c = [row_key(r, candidate) for r in before_norm] - after_c = [row_key(r, candidate) for r in after_norm] - if len(before_c) == len(set(before_c)) and len(after_c) == len(set(after_c)): + def unique_keys(rows: list[dict[str, Any]]) -> bool: + seen: set[tuple[str, ...]] = set() + for row in rows: + key = row_key(row, candidate) + if key in seen: + return False + seen.add(key) + return True + + if unique_keys(before_norm) and unique_keys(after_norm): match_keys = candidate ignore_ports = True else: @@ -226,15 +251,12 @@ def compare_rows( else: match_keys = list(key_fields) - before_mapped: list[dict[str, Any]] = [ - apply_port_map(r, iface_fields=iface_list, port_map=pmap) for r in before_norm - ] - before_groups: dict[tuple[str, ...], list[tuple[dict[str, Any], dict[str, Any]]]] = ( defaultdict(list) ) after_groups: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list) - for orig, mapped in zip(before_rows, before_mapped): + for orig, norm in zip(before_rows, before_norm): + mapped = apply_port_map(norm, iface_fields=iface_list, port_map=pmap) if iface_list else norm before_groups[row_key(mapped, match_keys)].append((orig, mapped)) for r in after_norm: after_groups[row_key(r, match_keys)].append(r) @@ -246,6 +268,7 @@ def compare_rows( multi_before_keys: list[tuple[str, ...]] = [] multi_after_keys: list[tuple[str, ...]] = [] unchanged_candidates: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]] = [] + sample = _StratifiedSample(limit_n) if include_unchanged and limit_n is not None else None def _key_obj(row: dict[str, Any]) -> dict[str, Any]: return {f: row.get(f, "") for f in key_fields} @@ -294,18 +317,7 @@ def compare_rows( ) # Stable key order: before encounter order, then after-only keys - seen_keys: set[tuple[str, ...]] = set() - ordered_keys: list[tuple[str, ...]] = [] - for orig, mapped in zip(before_rows, before_mapped): - k = row_key(mapped, match_keys) - if k not in seen_keys: - seen_keys.add(k) - ordered_keys.append(k) - for r in after_norm: - k = row_key(r, match_keys) - if k not in seen_keys: - seen_keys.add(k) - ordered_keys.append(k) + ordered_keys = chain(before_groups, (k for k in after_groups if k not in before_groups)) for k in ordered_keys: b_list = before_groups.get(k) or [] a_list = after_groups.get(k) or [] @@ -377,17 +389,14 @@ def compare_rows( else: unchanged += 1 if include_unchanged: - unchanged_candidates.append((orig, mapped, after)) + if sample is not None: + if sample.limit: + sample.add((orig, mapped, after), stratum_key(mapped)) + else: + unchanged_candidates.append((orig, mapped, after)) - if include_unchanged and unchanged_candidates: - if limit_n is None: - picked = unchanged_candidates - else: - picked = stratify_take( - unchanged_candidates, - limit_n, - key_fn=lambda t: stratum_key(t[1]), - ) + if include_unchanged: + picked = sample.picked() if sample is not None else unchanged_candidates for orig, mapped, after_row in picked: _append_unchanged_diff(orig, mapped, after_row) diff --git a/netx_api/biz_state/compare_service.py b/netx_api/biz_state/compare_service.py index e11b551..43e097b 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -8,13 +8,14 @@ import logging import threading import time import zipfile +from copy import deepcopy from datetime import datetime -from typing import Any, Callable +from typing import Any, Callable, Iterable, Iterator from uuid import uuid4 from fastapi import HTTPException -from sqlalchemy import and_, or_ -from sqlalchemy.orm import Session +from sqlalchemy import String, and_, cast, func, or_, tuple_ +from sqlalchemy.orm import Session, aliased from ..models import ( BizCompareDiff, @@ -29,7 +30,7 @@ from ..models import ( BizStateTask, ) from ..timeutil import utcnow_naive -from .compare_engine import compare_rows, mapping_stats +from .compare_engine import apply_port_map, compare_rows, mapping_stats, row_key from .compare_rules import ( ROW_FILTER_PRESETS, apply_row_filters, @@ -193,6 +194,7 @@ _FAIL_COMPACT_MIN = 50_000 # Live search (kw): load at most this many matching rows per side, return ≤ this many pairs. _LIVE_SEARCH_LOAD_CAP = 2_000 _LIVE_SEARCH_RESULT_CAP = 200 +_LIVE_SEARCH_GROUP_LOAD_CAP = 10_000 # Success-row persist policy (see resolve_unchanged_policy) _STORE_UNCHANGED_MODES = frozenset({"auto", "always", "never", "sample", "keys"}) _UNCHANGED_FULL_MAX = 20_000 @@ -269,7 +271,7 @@ def _strip_netx(row: Any) -> dict[str, Any]: return {k: v for k, v in row.items() if k != "_netx"} -def _diff_search_text(d: dict[str, Any]) -> str: +def _diff_search_text(d: dict[str, Any], *, limit: int | None = _SEARCH_TEXT_MAX) -> str: parts = [str(d.get("kind") or "")] for key in ("key", "before", "after", "mapped_before", "changes"): val = d.get(key) @@ -278,7 +280,8 @@ def _diff_search_text(d: dict[str, Any]) -> str: parts.append(json.dumps(val, ensure_ascii=False, default=str, separators=(",", ":"))) except Exception: parts.append(str(val)) - return " ".join(parts)[:_SEARCH_TEXT_MAX] + value = " ".join(parts) + return value if limit is None else value[:limit] def _top_changed_fields(diffs: list[dict[str, Any]], *, limit: int = 8) -> list[dict[str, Any]]: @@ -422,7 +425,10 @@ def _metric_rows_by_ids(db: Session, ids: list[str]) -> dict[str, dict[str, Any] return out -def _hydrate_diff_rows(db: Session, items: list[dict[str, Any]]) -> list[dict[str, Any]]: +def _hydrate_diff_rows( + db: Session, items: list[dict[str, Any]], *, + run: BizCompareRun | None = None, sheet: dict[str, Any] | None = None, +) -> list[dict[str, Any]]: """Fill empty before/after from metric tables when row_ids are present.""" need: list[str] = [] for it in items: @@ -433,16 +439,24 @@ def _hydrate_diff_rows(db: Session, items: list[dict[str, Any]]) -> list[dict[st if not need: return items by_id = _metric_rows_by_ids(db, need) + pmap, norm_rules = _run_transform_config(db, run) if run else ({}, []) + iface_fields = list((sheet or {}).get("iface_fields") or []) for it in items: brid = str(it.get("before_row_id") or "") arid = str(it.get("after_row_id") or "") if not it.get("before") and brid and brid in by_id: it["before"] = by_id[brid] if not it.get("after") and arid and arid in by_id: - it["after"] = by_id[arid] - # Success compact: no mapped_before stored — UI falls back to before + it["after"] = apply_iface_normalize_rows( + [by_id[arid]], iface_fields=iface_fields, rules=norm_rules, + )[0] if not it.get("mapped_before") and it.get("before"): - it["mapped_before"] = dict(it["before"]) + normalized = apply_iface_normalize_rows( + [it["before"]], iface_fields=iface_fields, rules=norm_rules, + )[0] + it["mapped_before"] = apply_port_map( + normalized, iface_fields=iface_fields, port_map=pmap, + ) return items @@ -471,6 +485,7 @@ def _filter_inline_diffs( *, kind: str, kw: str, + field_q: dict[str, str] | None = None, ) -> list[dict[str, Any]]: kind_n = (kind or "diff").strip().lower() kw_n = (kw or "").strip().lower() @@ -484,13 +499,62 @@ def _filter_inline_diffs( elif kind_n != "all" and dk != kind_n: continue if kw_n: - blob = _diff_search_text(d).lower() + blob = _diff_search_text(d, limit=None).lower() if kw_n not in blob: continue + if field_q and not _diff_matches_search(d, kw="", field_q=field_q): + continue out.append(d) return out +def _diff_matches_search(d: dict[str, Any], *, kw: str, field_q: dict[str, str]) -> bool: + """Search a completed pair, never filter its two sides before pairing.""" + if kw and kw.lower() not in _diff_search_text(d, limit=None).lower(): + return False + return all( + any( + val.lower() in str((d.get(side) or {}).get(name, "")).lower() + for side in ("key", "before", "mapped_before", "after") + ) + for name, val in field_q.items() + ) + + +def _search_stored_diffs(q: Any, *, metric_id: str, kw: str, field_q: dict[str, str]) -> Any: + """Filter persisted verdicts in SQL; join source rows for compact payloads.""" + from ..models import BizStateMetricRow + + sides = [BizCompareDiff.key_json, BizCompareDiff.before_json, + BizCompareDiff.mapped_before_json, BizCompareDiff.after_json] + sources = [] + for row_id in (BizCompareDiff.before_row_id, BizCompareDiff.after_row_id): + model = BizStateLldpNeighbor if metric_id == "lldp_neighbor" else BizStateMetricRow + src = aliased(model) + q = q.outerjoin(src, src.id == row_id) + sources.append(src) + if kw: + expressions = [BizCompareDiff.search_text, *[cast(s, String) for s in sides]] + for src in sources: + if metric_id == "lldp_neighbor": + expressions.extend(getattr(src, f) for f in + ("local_if", "remote_sys", "remote_if", "remote_ip", "protocol")) + else: + expressions.append(cast(src.data_json, String)) + q = q.filter(or_(*[func.lower(e).contains(kw.lower(), autoescape=True) for e in expressions])) + for name, val in field_q.items(): + expressions = [s[name].as_string() for s in sides] + for src in sources: + if metric_id == "lldp_neighbor": + if name in ("local_if", "remote_sys", "remote_if", "remote_ip", "protocol"): + expressions.append(getattr(src, name)) + else: + expressions.append(src.data_json[name].as_string()) + q = q.filter(or_(*[func.lower(cast(e, String)).contains(val.lower(), autoescape=True) + for e in expressions])) + return q + + def _sheet_meta_from_summary(summary: dict[str, Any], run: BizCompareRun, tpl: Any) -> list[dict[str, Any]]: sheets = list(summary.get("sheets") or []) if sheets: @@ -2193,6 +2257,18 @@ def _create_running_run( ) -> BizCompareRun: first_metric = str(sheets_cfg[0].get("metric_id") or "") pending_sheets = [_pending_sheet_meta(s) for s in sheets_cfg] + mapping = db.get(BizPortMapping, job.mapping_id) if job.mapping_id else None + config_snapshot = deepcopy({ + "version": 1, + "template": _template_out(tpl), + "sheets": sheets_cfg, + "port_map": _port_map_dict(db, str(job.mapping_id or "")), + "store_unchanged": normalize_store_unchanged(getattr(job, "store_unchanged", None)), + "job_name": job.name or "", + "before_task_id": job.before_task_id or "", + "after_task_id": job.after_task_id or "", + "mapping_name": (mapping.name if mapping else "") or "", + }) run = BizCompareRun( id=uuid4().hex, job_id=job.id, @@ -2203,6 +2279,7 @@ def _create_running_run( metric_id=first_metric, status="running", summary_json={ + "config_snapshot": config_snapshot, "progress": { "phase": "queued", "sheet_index": 0, @@ -2232,6 +2309,26 @@ def _create_running_run( return run +def _run_config_snapshot(run: BizCompareRun) -> dict[str, Any] | None: + snapshot = (run.summary_json or {}).get("config_snapshot") + if isinstance(snapshot, dict) and snapshot.get("version") == 1: + return snapshot + return None + + +def _run_transform_config( + db: Session, run: BizCompareRun, +) -> tuple[dict[str, str], list[dict[str, str]]]: + snapshot = _run_config_snapshot(run) + if snapshot is not None: + return dict(snapshot.get("port_map") or {}), normalize_iface_rules( + (snapshot.get("template") or {}).get("iface_normalize_rules") or [], + ) + # Legacy runs did not record these settings; retain their fallback behavior. + tpl = db.get(BizCompareTemplate, run.template_id) if run.template_id else None + return _port_map_dict(db, str(run.mapping_id or "")), template_iface_normalize(tpl) + + def _set_run_progress( db: Session, run: BizCompareRun, @@ -2309,14 +2406,15 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: run = db.get(BizCompareRun, run_id) if not run: raise HTTPException(status_code=404, detail="run_not_found") + snapshot = _run_config_snapshot(run) j = db.get(BizCompareJob, run.job_id) if not j: run.status = "failed" run.message = "job_not_found" db.commit() raise HTTPException(status_code=404, detail="job_not_found") - tpl = db.get(BizCompareTemplate, run.template_id) - if not tpl: + tpl = db.get(BizCompareTemplate, run.template_id) if snapshot is None else None + if not tpl and snapshot is None: run.status = "failed" run.message = "template_not_found" db.commit() @@ -2324,25 +2422,25 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: before_batch_id = str(run.before_batch_id or "") after_batch_id = str(run.after_batch_id or "") - sheets_cfg = _filter_enabled_sheets( - template_metrics(tpl), getattr(j, "enabled_sheet_ids", None) - ) + if snapshot is not None: + sheets_cfg = deepcopy(snapshot.get("sheets") or []) + else: + sheets_cfg = _filter_enabled_sheets(template_metrics(tpl), getattr(j, "enabled_sheet_ids", None)) if not sheets_cfg: run.status = "failed" run.message = "no_enabled_sheets" db.commit() raise HTTPException(status_code=400, detail="no_enabled_sheets") - sheets_cfg = _order_sheets_small_first( - db, - sheets_cfg, - before_batch_id=before_batch_id, - after_batch_id=after_batch_id, - ) + if snapshot is None: + sheets_cfg = _order_sheets_small_first( + db, sheets_cfg, before_batch_id=before_batch_id, after_batch_id=after_batch_id, + ) started_mono = time.monotonic() - store_mode = normalize_store_unchanged(getattr(j, "store_unchanged", None)) - pmap = _port_map_dict(db, run.mapping_id) - norm_rules = template_iface_normalize(tpl) + store_mode = normalize_store_unchanged( + snapshot.get("store_unchanged") if snapshot is not None else getattr(j, "store_unchanged", None), + ) + pmap, norm_rules = _run_transform_config(db, run) unchanged_listed_total = 0 unchanged_truncated_any = False unchanged_compact_any = False @@ -2636,6 +2734,8 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: }, "sheets": sheet_metas, } + if snapshot is not None: + summary_payload["config_snapshot"] = snapshot run = db.get(BizCompareRun, run_id) or run if str(run.status or "") == "cancelled": return get_run(db, run.id) @@ -2842,7 +2942,7 @@ def _csv_cell(v: Any) -> str: return s -def _sheet_csv(sheet: dict[str, Any]) -> str: +def _sheet_csv_lines(sheet: dict[str, Any], diffs: Iterable[dict[str, Any]]) -> Iterator[str]: keys = list(sheet.get("key_fields") or []) key_set = set(keys) compare = [f for f in list(sheet.get("compare_fields") or []) if f not in key_set] @@ -2861,8 +2961,8 @@ def _sheet_csv(sheet: dict[str, Any]) -> str: headers.append(f"{f}__post") else: headers.append(f) - lines = [",".join(_csv_cell(h) for h in headers)] - for d in list(sheet.get("diffs") or []): + yield ",".join(_csv_cell(h) for h in headers) + for d in diffs: kind = str(d.get("kind") or "") pre = dict(d.get("mapped_before") or d.get("before") or {}) post = dict(d.get("after") or {}) @@ -2887,8 +2987,11 @@ def _sheet_csv(sheet: dict[str, Any]) -> str: row.append(pre.get(f, "")) else: row.append(post.get(f, pre.get(f, ""))) - lines.append(",".join(_csv_cell(x) for x in row)) - return "\ufeff" + "\n".join(lines) + "\n" + yield ",".join(_csv_cell(x) for x in row) + + +def _sheet_csv(sheet: dict[str, Any]) -> str: + return "\ufeff" + "\n".join(_sheet_csv_lines(sheet, sheet.get("diffs") or [])) + "\n" def _enrich_summary(summary: dict[str, Any], sheets: list[dict[str, Any]]) -> dict[str, Any]: @@ -3001,8 +3104,13 @@ def get_run(db: Session, run_id: str) -> dict[str, Any]: r = db.get(BizCompareRun, run_id) if not r: raise HTTPException(status_code=404, detail="run_not_found") - tpl = db.get(BizCompareTemplate, r.template_id) if r.template_id else None summary = dict(r.summary_json or {}) + snapshot = _run_config_snapshot(r) + tpl = db.get(BizCompareTemplate, r.template_id) if r.template_id and snapshot is None else None + template_out = ( + deepcopy(snapshot.get("template")) if snapshot is not None + else (_template_out(tpl) if tpl else None) + ) raw_sheets = _sheet_meta_from_summary(summary, r, tpl) # Never return full diffs in run detail (million-row safe) sheets = [ @@ -3023,22 +3131,31 @@ def get_run(db: Session, run_id: str) -> dict[str, Any]: ] enriched = _enrich_summary(summary, raw_sheets) stored = "rows" if _run_has_diff_rows(db, run_id) else "inline" - job = db.get(BizCompareJob, r.job_id) if r.job_id else None - mapping = db.get(BizPortMapping, r.mapping_id) if r.mapping_id else None + job = db.get(BizCompareJob, r.job_id) if r.job_id and snapshot is None else None + mapping = db.get(BizPortMapping, r.mapping_id) if r.mapping_id and snapshot is None else None before_side = _compare_side( - db, r.before_batch_id, fallback_task_id=(job.before_task_id if job else "") + db, r.before_batch_id, fallback_task_id=( + snapshot.get("before_task_id", "") if snapshot is not None else (job.before_task_id if job else "") + ), ) after_side = _compare_side( - db, r.after_batch_id, fallback_task_id=(job.after_task_id if job else "") + db, r.after_batch_id, fallback_task_id=( + snapshot.get("after_task_id", "") if snapshot is not None else (job.after_task_id if job else "") + ), ) return { "id": r.id, "job_id": r.job_id, - "job_name": (job.name if job else "") or "", + "job_name": ( + snapshot.get("job_name", "") if snapshot is not None else ((job.name if job else "") or "") + ), "template_id": r.template_id, - "template_name": (tpl.name if tpl else "") or "", + "template_name": (template_out or {}).get("name") or "", "mapping_id": r.mapping_id, - "mapping_name": (mapping.name if mapping else "") or "", + "mapping_name": ( + snapshot.get("mapping_name", "") if snapshot is not None else ((mapping.name if mapping else "") or "") + ), + "config_snapshot_version": snapshot["version"] if snapshot is not None else None, "before_batch_id": r.before_batch_id, "after_batch_id": r.after_batch_id, "before": before_side, @@ -3052,7 +3169,7 @@ def get_run(db: Session, run_id: str) -> dict[str, Any]: "mapping_stats": r.mapping_stats_json or {}, "message": r.message, "created_at": r.created_at.isoformat() + "Z" if r.created_at else None, - "template": _template_out(tpl) if tpl else None, + "template": template_out, } @@ -3152,13 +3269,17 @@ def _field_qf_sql(field_q: dict[str, str], *, prefix: str = "qf") -> tuple[str, for i, (name, val) in enumerate(field_q.items()): sf = _safe_field(name) key = f"{prefix}_{i}" - params[key] = f"%{val.lower()}%" + params[key] = _literal_search_pattern(val) parts.append( f"lower(trim(both from coalesce(data_json->>'{sf}', ''))) LIKE :{key}" ) return "(" + " AND ".join(parts) + ")", params +def _literal_search_pattern(value: str) -> str: + return "%" + value.lower().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%" + + def _load_metric_rows_for_search( db: Session, *, @@ -3188,7 +3309,9 @@ def _load_metric_rows_for_search( lim = max(1, min(int(cap), _LIVE_SEARCH_LOAD_CAP)) filters = [f for f in (row_filters or []) if isinstance(f, dict)] - if _dialect_is_postgres(db): + if mid != "lldp_neighbor" and _dialect_is_postgres(db) and ( + not filters or _filters_sql_compatible(filters) + ): filter_sql, filter_params = ("TRUE", {}) if filters and _filters_sql_compatible(filters): filter_sql, filter_params = compile_row_filters_sql(filters) @@ -3202,7 +3325,7 @@ def _load_metric_rows_for_search( **qf_params, } if needle: - params["kw"] = f"%{needle.lower()}%" + params["kw"] = _literal_search_pattern(needle) search_parts.append(_kw_match_sql(key_fields)) search_sql = " AND ".join(f"({p})" for p in search_parts if p and p != "TRUE") if not search_sql: @@ -3246,20 +3369,21 @@ def _load_metric_rows_for_search( return out, truncated # Non-PG / fallback: scan with early stop (OK for tests / small sheets) - q = ( - db.query(BizStateMetricRow) - .filter( - BizStateMetricRow.batch_id == bid, - BizStateMetricRow.metric_id == mid, - ) - .order_by(BizStateMetricRow.seq.asc(), BizStateMetricRow.id.asc()) - ) + model = BizStateLldpNeighbor if mid == "lldp_neighbor" else BizStateMetricRow + q = db.query(model).filter(model.batch_id == bid) + if mid == "lldp_neighbor": + q = q.order_by(model.id.asc()) + else: + q = q.filter(model.metric_id == mid).order_by(model.seq.asc(), model.id.asc()) out = [] truncated = False needle_l = needle.lower() key_set = [str(k).strip() for k in key_fields if str(k).strip()] for r in q.yield_per(500): - data = dict(r.data_json or {}) + data = ( + {f: getattr(r, f) for f in ("local_if", "remote_sys", "remote_if", "remote_ip", "protocol")} + if mid == "lldp_neighbor" else dict(r.data_json or {}) + ) row = { **data, "_netx": { @@ -3297,14 +3421,75 @@ def _load_metric_rows_for_search( if not hit: db.expunge(r) continue - out.append(row) db.expunge(r) if len(out) >= lim: truncated = True break + out.append(row) return out, truncated +def _load_search_key_groups( + db: Session, *, batch_id: str, metric_id: str, keys: set[tuple[str, ...]], + match_keys: list[str], iface_fields: list[str], row_filters: list[dict[str, Any]], + norm_rules: list[dict[str, str]], port_map: dict[str, str], +) -> tuple[list[dict[str, Any]], set[tuple[str, ...]]]: + """Complete candidate key groups so a one-sided search cannot change verdicts. + + Interface transforms run in Python. Other identity fields narrow the SQL + query. If the load budget is exceeded, omit entire groups rather than compare + partial groups and manufacture additions/deletions. + """ + from ..models import BizStateMetricRow + + model = BizStateLldpNeighbor if metric_id == "lldp_neighbor" else BizStateMetricRow + invariant = [f for f in match_keys if f not in iface_fields] + positions = [match_keys.index(f) for f in invariant] + coarse = sorted({tuple(k[i] for i in positions) for k in keys}) + out: list[dict[str, Any]] = [] + complete: set[tuple[str, ...]] = set() + # Bound memory even when a candidate identifies a very large duplicate group. + budget = _LIVE_SEARCH_GROUP_LOAD_CAP + for start in range(0, len(coarse), 100): + group = coarse[start:start + 100] + group_set = set(group) + target = {k for k in keys if tuple(k[i] for i in positions) in group_set} + q = db.query(model).filter(model.batch_id == batch_id) + if metric_id == "lldp_neighbor": + q = q.order_by(model.id.asc()) + else: + q = q.filter(model.metric_id == metric_id).order_by(model.seq.asc(), model.id.asc()) + if invariant: + exprs = [func.trim(func.coalesce( + cast(getattr(model, f) if metric_id == "lldp_neighbor" + else model.data_json[f].as_string(), String), "")) for f in invariant] + q = q.filter(tuple_(*exprs).in_(group)) + rows = q.limit(budget + 1).all() + if len(rows) > budget: + for r in rows: + db.expunge(r) + # No partial verdicts. Later groups may still fit the remaining budget. + continue + budget -= len(rows) + complete.update(target) + for r in rows: + data = ( + {f: getattr(r, f) for f in ("local_if", "remote_sys", "remote_if", "remote_ip", "protocol")} + if metric_id == "lldp_neighbor" else dict(r.data_json or {}) + ) + row = {**data, "_netx": {"row_id": r.id}} + db.expunge(r) + if row_filters and not all(row_matches_filter(row, f) for f in row_filters): + continue + normalized = apply_iface_normalize_rows( + [row], iface_fields=iface_fields, rules=norm_rules, + )[0] + mapped = apply_port_map(normalized, iface_fields=iface_fields, port_map=port_map) + if row_key(mapped, match_keys) in target: + out.append(row) + return out, complete + + def _live_search_sheet_diffs( db: Session, run: BizCompareRun, @@ -3316,7 +3501,7 @@ def _live_search_sheet_diffs( page: int, page_size: int, ) -> dict[str, Any]: - """Search before/after metric tables, zip-compare, filter by kind tab.""" + """Find candidate keys, complete both sides, then compare and filter pairs.""" mid_src = str(sheet.get("metric_id") or "").strip() sid = sheet_key(sheet) key_fields = list(sheet.get("key_fields") or []) @@ -3328,7 +3513,7 @@ def _live_search_sheet_diffs( ) row_filters = list(sheet.get("row_filters") or []) # Older runs may lack row_filters on sheet meta — fall back to template - if not row_filters and run.template_id: + if "row_filters" not in sheet and _run_config_snapshot(run) is None and run.template_id: tpl = db.get(BizCompareTemplate, run.template_id) if tpl: for s in template_metrics(tpl): @@ -3363,9 +3548,7 @@ def _live_search_sheet_diffs( ignore_ports = sheet.get("ignore_port_changes") if ignore_ports is not None: ignore_ports = bool(ignore_ports) - pmap = _port_map_dict(db, str(run.mapping_id or "")) - tpl = db.get(BizCompareTemplate, run.template_id) if run.template_id else None - norm_rules = template_iface_normalize(tpl) + pmap, norm_rules = _run_transform_config(db, run) before_rows, trunc_b = _load_metric_rows_for_search( db, @@ -3385,6 +3568,31 @@ def _live_search_sheet_diffs( kw=free_kw, field_q=merged_q, ) + match_keys = list((sheet.get("summary") or {}).get("match_key_fields") or key_fields) + if ignore_ports is True and not pmap: + match_keys = [f for f in key_fields if f not in iface_fields] or key_fields + keys: set[tuple[str, ...]] = set() + for rows, mapping in ((before_rows, pmap), (after_rows, {})): + for row in apply_iface_normalize_rows(rows, iface_fields=iface_fields, rules=norm_rules): + keys.add(row_key(apply_port_map(row, iface_fields=iface_fields, port_map=mapping), match_keys)) + before_rows, complete_b = _load_search_key_groups( + db, batch_id=str(run.before_batch_id or ""), metric_id=mid_src, keys=keys, + match_keys=match_keys, iface_fields=iface_fields, row_filters=row_filters, + norm_rules=norm_rules, port_map=pmap, + ) + after_rows, complete_a = _load_search_key_groups( + db, batch_id=str(run.after_batch_id or ""), metric_id=mid_src, keys=keys, + match_keys=match_keys, iface_fields=iface_fields, row_filters=row_filters, + norm_rules=norm_rules, port_map={}, + ) + complete = complete_b & complete_a + def keep_complete(rows: list[dict[str, Any]], mapping: dict[str, str]) -> list[dict[str, Any]]: + normalized = apply_iface_normalize_rows(rows, iface_fields=iface_fields, rules=norm_rules) + return [orig for orig, norm in zip(rows, normalized) if row_key( + apply_port_map(norm, iface_fields=iface_fields, port_map=mapping), match_keys, + ) in complete] + before_rows = keep_complete(before_rows, pmap) + after_rows = keep_complete(after_rows, {}) result = compare_rows( before_rows=before_rows, after_rows=after_rows, @@ -3394,17 +3602,19 @@ def _live_search_sheet_diffs( port_map=pmap, field_rules=field_rules, iface_normalize_rules=norm_rules, - ignore_port_changes=ignore_ports, + ignore_port_changes=(match_keys != key_fields) if not pmap else False, include_unchanged=True, unchanged_limit=None, compact_unchanged=False, ) kind_n = (kind or "diff").strip().lower() filtered = [ - d for d in list(result.get("diffs") or []) if _kind_allows(kind_n, str(d.get("kind") or "")) + d for d in list(result.get("diffs") or []) + if _kind_allows(kind_n, str(d.get("kind") or "")) + and _diff_matches_search(d, kw=free_kw, field_q=merged_q) ] # Cap pairs returned to keep UI snappy - truncated = bool(trunc_b or trunc_a or len(filtered) > _LIVE_SEARCH_RESULT_CAP) + truncated = bool(trunc_b or trunc_a or complete != keys or len(filtered) > _LIVE_SEARCH_RESULT_CAP) filtered = filtered[:_LIVE_SEARCH_RESULT_CAP] total = len(filtered) start = (page - 1) * page_size @@ -3452,10 +3662,17 @@ def list_run_diffs( sheets = _sheet_meta_from_summary(summary, r, tpl) asked = (metric_id or "").strip() sheet = _lookup_sheet(sheets, asked) if asked else (sheets[0] if sheets else None) + if asked and sheet is None: + raise HTTPException(status_code=404, detail="sheet_not_found") mid = sheet_key(sheet) if sheet else (asked or str(r.metric_id or "")) - # Unified search: kw and/or field filters → live source lookup; kind tab only filters. - if (kw_n or field_q) and sheet: + # Only reconstruct missing success rows. Persisted verdicts must stay stable + # while searching and avoid loading/recomparing source tables on every page. + sheet_summary = (sheet or {}).get("summary") or {} + success_missing = int(sheet_summary.get("unchanged") or 0) > int( + sheet_summary.get("unchanged_listed") or 0 + ) + if (kw_n or field_q) and sheet and kind_n in ("all", "unchanged") and success_missing: return _live_search_sheet_diffs( db, r, @@ -3480,6 +3697,9 @@ def list_run_diffs( q = q.filter(BizCompareDiff.kind == "changed") elif kind_n != "all": q = q.filter(BizCompareDiff.kind == kind_n) + if kw_n or field_q: + q = _search_stored_diffs(q, metric_id=str((sheet or {}).get("metric_id") or mid), + kw=kw_n, field_q=field_q) total = q.count() rows = ( q.order_by(BizCompareDiff.seq.asc(), BizCompareDiff.id.asc()) @@ -3487,7 +3707,7 @@ def list_run_diffs( .limit(size_n) .all() ) - items = _hydrate_diff_rows(db, [_diff_row_out(x) for x in rows]) + items = _hydrate_diff_rows(db, [_diff_row_out(x) for x in rows], run=r, sheet=sheet) return { "total": total, "page": page_n, @@ -3505,7 +3725,7 @@ def list_run_diffs( inline = list((sheet or {}).get("diffs") or []) if not inline and mid == r.metric_id: inline = list(r.diffs_json or []) - filtered = _filter_inline_diffs(inline, kind=kind_n, kw="") + filtered = _filter_inline_diffs(inline, kind=kind_n, kw=kw_n, field_q=field_q) total = len(filtered) start = (page_n - 1) * size_n page_items = filtered[start : start + size_n] @@ -3520,44 +3740,53 @@ def list_run_diffs( } -def _iter_sheet_diffs(db: Session, run_id: str, metric_id: str) -> list[dict[str, Any]]: - """Load all diffs for one sheet (export). Prefer row table; fall back to inline. +def _iter_sheet_diffs(db: Session, run_id: str, metric_id: str) -> Iterator[dict[str, Any]]: + """Stream one sheet using a stable (seq, id) cursor; fall back to inline. Compact success rows are hydrated in chunks from metric tables. """ + r = db.get(BizCompareRun, run_id) + if not r: + return + tpl = db.get(BizCompareTemplate, r.template_id) if r.template_id else None + sheets = _sheet_meta_from_summary(dict(r.summary_json or {}), r, tpl) + sheet = next((s for s in sheets if sheet_key(s) == metric_id), None) if _run_has_diff_rows(db, run_id): - out: list[dict[str, Any]] = [] - offset = 0 + cursor: tuple[int, str] | None = None while True: - rows = ( - db.query(BizCompareDiff) - .filter(BizCompareDiff.run_id == run_id, BizCompareDiff.metric_id == metric_id) + q = db.query(BizCompareDiff).filter( + BizCompareDiff.run_id == run_id, BizCompareDiff.metric_id == metric_id, + ) + if cursor is not None: + seq, diff_id = cursor + q = q.filter(or_(BizCompareDiff.seq > seq, and_( + BizCompareDiff.seq == seq, BizCompareDiff.id > diff_id, + ))) + rows = (q .order_by(BizCompareDiff.seq.asc(), BizCompareDiff.id.asc()) - .offset(offset) .limit(_DIFF_CHUNK) .all() ) if not rows: break - chunk = _hydrate_diff_rows(db, [_diff_row_out(x) for x in rows]) - out.extend(chunk) - offset += len(rows) + chunk = _hydrate_diff_rows(db, [_diff_row_out(x) for x in rows], run=r, sheet=sheet) + cursor = (rows[-1].seq, rows[-1].id) + yield from chunk if len(rows) < _DIFF_CHUNK: break - return out - r = db.get(BizCompareRun, run_id) - if not r: - return [] + # Release the previous chunk before loading the next one. + del rows, chunk + return summary = dict(r.summary_json or {}) sheets = list(summary.get("sheets") or []) for sh in sheets: if sheet_key(sh) == metric_id or ( str(sh.get("metric_id") or "") == metric_id and sheet_key(sh) == metric_id ): - return list(sh.get("diffs") or []) + yield from sh.get("diffs") or [] + return if metric_id == r.metric_id: - return list(r.diffs_json or []) - return [] + yield from r.diffs_json or [] def export_run_zip(db: Session, run_id: str) -> bytes: @@ -3586,14 +3815,17 @@ def export_run_zip(db: Session, run_id: str) -> bytes: f"+{card.get('added')}/-{card.get('removed')}/~{card.get('changed')}/= {card.get('unchanged')}" ) zf.writestr("manifest.txt", "\n".join(manifest) + "\n") + run = db.get(BizCompareRun, run_id) + snapshot = _run_config_snapshot(run) if run else None + if snapshot is not None: + zf.writestr("config_snapshot.json", json.dumps(snapshot, ensure_ascii=False, indent=2)) for sheet in list(detail.get("sheets") or []): sid = sheet_key(sheet) or "sheet" safe = "".join(ch if ch.isalnum() or ch in "-_." else "_" for ch in sid)[:80] or "sheet" - sheet_full = { - **sheet, - "diffs": _iter_sheet_diffs(db, run_id, sid), - } - zf.writestr(f"tables/{safe}.csv", _sheet_csv(sheet_full)) + with zf.open(f"tables/{safe}.csv", "w", force_zip64=True) as csv_file: + csv_file.write(b"\xef\xbb\xbf") + for line in _sheet_csv_lines(sheet, _iter_sheet_diffs(db, run_id, sid)): + csv_file.write((line + "\n").encode("utf-8")) sum_lines = ["metric_id,mode,before,after,added,removed,changed,unchanged,diff_count,pass_rate"] for card in list(s.get("sheet_cards") or []): sum_lines.append( diff --git a/tests/test_biz_state_compare_export.py b/tests/test_biz_state_compare_export.py new file mode 100644 index 0000000..d176c40 --- /dev/null +++ b/tests/test_biz_state_compare_export.py @@ -0,0 +1,87 @@ +"""Export must page lazily, keep tied sequence rows, and produce valid UTF-8 CSV.""" + +from __future__ import annotations + +import csv +import io +import unittest +import zipfile +from unittest.mock import patch + +from sqlalchemy import create_engine, event +from sqlalchemy.orm import Session + +from netx_api.biz_state.compare_service import _iter_sheet_diffs, export_run_zip +from netx_api.db import Base +from netx_api.models import BizCompareDiff, BizCompareRun + + +class CompareExportTests(unittest.TestCase): + def setUp(self): + self.engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(self.engine) + self.db = Session(self.engine) + self.sheet = {"sheet_id": "sheet", "metric_id": "test", "key_fields": ["id"], + "compare_fields": ["note"], "display_fields": ["id", "note"], + "summary": {"unchanged": 7}, "status": "done"} + self.run = BizCompareRun(id="run", metric_id="test", status="success", + summary_json={"sheets": [self.sheet]}) + self.db.add(self.run) + for seq, ids in ((0, "cba"), (1, "ed"), (2, "gf")): + for rid in ids: + note = '中文, "quoted"\nsecond line' + self.db.add(BizCompareDiff(id=rid, run_id="run", metric_id="sheet", seq=seq, + kind="unchanged", key_json={"id": rid}, + before_json={"id": rid, "note": note}, + after_json={"id": rid, "note": note})) + self.db.commit() + + def tearDown(self): + self.db.close() + self.engine.dispose() + + def test_cursor_handles_tied_sequence_numbers_across_chunks(self): + statements = [] + def record(_conn, _cursor, statement, _parameters, _context, _executemany): + if "ORDER BY biz_compare_diff.seq" in statement: + statements.append(statement) + event.listen(self.engine, "before_cursor_execute", record) + with patch("netx_api.biz_state.compare_service._DIFF_CHUNK", 2): + iterator = _iter_sheet_diffs(self.db, "run", "sheet") + self.assertEqual(statements, []) + self.assertEqual(next(iterator)["key"]["id"], "a") + self.assertEqual(next(iterator)["key"]["id"], "b") + self.assertEqual(len(statements), 1, "do not load the next chunk early") + self.assertEqual([d["key"]["id"] for d in iterator], list("cdefg")) + self.assertEqual(len(statements), 4) + self.assertTrue(all("biz_compare_diff.seq >" in q and "biz_compare_diff.id >" in q + for q in statements[1:])) + + def test_export_csv_escapes_unicode_quotes_commas_and_newlines(self): + with patch("netx_api.biz_state.compare_service._DIFF_CHUNK", 2): + archive_bytes = export_run_zip(self.db, "run") + with zipfile.ZipFile(io.BytesIO(archive_bytes)) as archive: + payload = archive.read("tables/sheet.csv") + self.assertTrue(payload.startswith(b"\xef\xbb\xbf")) + rows = list(csv.DictReader(io.StringIO(payload.decode("utf-8-sig")))) + self.assertEqual([r["id"] for r in rows], list("abcdefg")) + self.assertTrue(all(r["note__pre"] == '中文, "quoted"\nsecond line' for r in rows)) + self.assertTrue(all(r["note__pre"] == r["note__post"] for r in rows)) + + def test_other_sheets_do_not_leak_into_export(self): + self.db.add(BizCompareDiff(id="other", run_id="run", metric_id="another", seq=0, + kind="added", key_json={"id": "other"}, after_json={"note": "other"})) + self.db.commit() + with patch("netx_api.biz_state.compare_service._DIFF_CHUNK", 2): + self.assertEqual([d["key"]["id"] for d in _iter_sheet_diffs(self.db, "run", "sheet")], list("abcdefg")) + + def test_legacy_inline_and_empty_exports(self): + self.db.query(BizCompareDiff).delete() + inline = {"kind": "added", "key": {"id": "legacy"}, "after": {"note": "value"}} + self.run.summary_json = {"sheets": [{**self.sheet, "diffs": [inline]}]} + self.db.commit() + self.assertEqual(list(_iter_sheet_diffs(self.db, "run", "sheet")), [inline]) + self.run.summary_json = {"sheets": [self.sheet]} + self.db.commit() + with zipfile.ZipFile(io.BytesIO(export_run_zip(self.db, "run"))) as archive: + self.assertEqual(archive.read("tables/sheet.csv").decode("utf-8-sig"), "kind,id,note__pre,note__post\n") diff --git a/tests/test_biz_state_compare_memory.py b/tests/test_biz_state_compare_memory.py new file mode 100644 index 0000000..7850798 --- /dev/null +++ b/tests/test_biz_state_compare_memory.py @@ -0,0 +1,103 @@ +"""Bounded sampling regressions and an opt-in comparison memory benchmark.""" + +from __future__ import annotations + +import gc +import hashlib +import json +import os +import random +import subprocess +import time +import tracemalloc +import types +from collections import defaultdict, deque +from copy import deepcopy + +import pytest + +from netx_api.biz_state.compare_engine import _StratifiedSample, compare_rows, stratify_take + + +def reference_sample(items, limit): + if len(items) <= limit: + return list(items) + buckets = defaultdict(deque) + for item in items: + buckets[item[0]].append(item) + out = [] + while buckets and len(out) < limit: + for key in list(buckets): + out.append(buckets[key].popleft()) + if not buckets[key]: + del buckets[key] + if len(out) >= limit: + break + return out + + +def test_streaming_sample_preserves_round_robin_and_small_input_order(): + rng = random.Random(42) + for _ in range(100): + items = [(str(rng.randrange(20)), i) for i in range(rng.randrange(100))] + for limit in (0, 1, 5, 20, 100): + expected = reference_sample(items, limit) + assert stratify_take(iter(items), limit, key_fn=lambda item: item[0]) == expected + + +def test_sample_storage_is_bounded_for_both_many_rows_and_many_strata(): + sample = _StratifiedSample(50) + for i in range(20_000): + sample.add(i, str(i) if i >= 10_000 else "one-large-peer") + assert len(sample.heap) == 50 + assert len(sample.strata) == 50 + assert sample.picked() == [0, *range(10_000, 10_049)] + + +@pytest.mark.parametrize("iface_fields,rules,pmap", [ + ([], [], {}), + (["interface"], [], {}), + (["interface"], [{"from": "GE", "to": "gei"}], {"gei-1": "gei-2"}), +]) +def test_compare_does_not_mutate_inputs_when_reusing_rows(iface_fields, rules, pmap): + before = [{"id": "1", "interface": "GE-1", "state": "up"}] + after = [{"id": "1", "interface": "GE-2", "state": "up"}] + original = deepcopy((before, after)) + compare_rows(before_rows=before, after_rows=after, key_fields=["id"], + iface_fields=iface_fields, compare_fields=["state"], port_map=pmap, + iface_normalize_rules=rules, include_unchanged=True, unchanged_limit=1) + assert (before, after) == original + + +def test_memory_benchmark_against_git_baseline(): + """Run explicitly with NETX_BENCH_ROWS=1000000; not part of routine CI.""" + n = int(os.environ.get("NETX_BENCH_ROWS", "0")) + if not n: + pytest.skip("opt-in large synthetic benchmark") + baseline_ref = os.environ.get("NETX_BENCH_REF", "HEAD") + source = subprocess.run(["git", "show", f"{baseline_ref}:netx_api/biz_state/compare_engine.py"], + check=True, capture_output=True, text=True, encoding="utf-8").stdout + baseline = types.ModuleType("netx_api.biz_state._benchmark_baseline") + baseline.__package__ = "netx_api.biz_state" + exec(compile(source, "baseline_compare_engine.py", "exec"), baseline.__dict__) + before = [{"id": str(i), "neighbor": str(i % 50), "state": "up", "afi": "vpnv4"} for i in range(n)] + after = [dict(row) for row in before] + records = [] + digests = [] + for name, compare in (("baseline", baseline.compare_rows), ("optimized", compare_rows)): + gc.collect() + tracemalloc.start() + started = time.perf_counter() + result = compare(before_rows=before, after_rows=after, key_fields=["id"], iface_fields=[], + compare_fields=["state"], include_unchanged=True, unchanged_limit=5000, + compact_unchanged=True) + elapsed = time.perf_counter() - started + _, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + digests.append(hashlib.sha256(json.dumps(result, sort_keys=True).encode()).hexdigest()) + records.append({"engine": name, "rows_per_side": n, "peak_engine_mib": round(peak / 2**20, 2), + "seconds_with_tracemalloc": round(elapsed, 3)}) + del result + assert digests[0] == digests[1], "large-sheet results and sample order must match" + print("\n" + json.dumps({"baseline_ref": baseline_ref, "input_allocations_included": False, + "results": records}, indent=2)) diff --git a/tests/test_biz_state_compare_search.py b/tests/test_biz_state_compare_search.py new file mode 100644 index 0000000..0b30c8f --- /dev/null +++ b/tests/test_biz_state_compare_search.py @@ -0,0 +1,186 @@ +"""Regression tests: searching a pair must not change its comparison verdict.""" + +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from fastapi import HTTPException +from sqlalchemy import create_engine +from sqlalchemy.dialects import postgresql +from sqlalchemy.orm import Session + +from netx_api.biz_state.compare_engine import compare_rows +from netx_api.biz_state.compare_service import ( + _load_metric_rows_for_search, + _literal_search_pattern, + _persist_sheet_diffs, + _search_stored_diffs, + list_run_diffs, +) +from netx_api.db import Base +from netx_api.models import ( + BizCompareDiff, BizCompareRun, BizCompareTemplate, BizPortMappingRow, + BizStateLldpNeighbor, BizStateMetricRow, +) + + +class CompareSearchTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(self.engine, tables=[m.__table__ for m in ( + BizCompareDiff, BizCompareRun, BizCompareTemplate, BizPortMappingRow, + BizStateLldpNeighbor, BizStateMetricRow, + )]) + self.db = Session(self.engine) + + def tearDown(self) -> None: + self.db.close() + self.engine.dispose() + + def seed(self, before, after, *, metric="test", keys=None, fields=None, + iface=None, mapping=None, sample=False, compact=False, filters=None): + keys = keys or ["id"] + iface = iface or [] + fields = fields or ["state"] + for side, rows in (("before", before), ("after", after)): + for seq, row in enumerate(rows): + rid = f"{side}-{seq}" + if metric == "lldp_neighbor": + self.db.add(BizStateLldpNeighbor(id=rid, batch_id=side, **row)) + else: + self.db.add(BizStateMetricRow(id=rid, batch_id=side, metric_id=metric, + seq=seq, data_json=dict(row))) + row["_netx"] = {"row_id": rid} + if mapping: + for b, a in mapping.items(): + self.db.add(BizPortMappingRow(mapping_id="map", before_if=b, after_if=a)) + result = compare_rows(before_rows=before, after_rows=after, key_fields=keys, + compare_fields=fields, iface_fields=iface, port_map=mapping or {}, + include_unchanged=True, compact_unchanged=compact) + sheet = {"sheet_id": "sheet", "metric_id": metric, "key_fields": keys, + "compare_fields": fields, "iface_fields": iface, + "summary": {**result["summary"], "unchanged_listed": 0 if sample + else result["summary"]["unchanged"]}, + "row_filters": filters or []} + self.db.add(BizCompareRun(id="run", status="success", before_batch_id="before", + after_batch_id="after", mapping_id="map" if mapping else "", + summary_json={"sheets": [sheet]})) + self.db.commit() + _persist_sheet_diffs(self.db, run_id="run", metric_id="sheet", + diffs=[d for d in result["diffs"] + if not sample or d["kind"] != "unchanged"]) + return result + + def search(self, **kwargs): + return list_run_diffs(self.db, "run", metric_id="sheet", **kwargs) + + def test_search_before_value_preserves_changed_verdict(self): + self.seed([{"id": "1", "state": "up"}], [{"id": "1", "state": "down"}]) + with patch("netx_api.biz_state.compare_service.compare_rows") as compare: + result = self.search(kw="up", kind="changed") + compare.assert_not_called() + self.assertEqual(result["total"], 1) + self.assertEqual(result["items"][0]["after"]["state"], "down") + self.assertEqual(self.search(kw="up", kind="removed")["total"], 0) + + def test_field_filter_matches_after_value_on_compact_diff(self): + self.seed([{"id": "1", "state": "up"}], [{"id": "1", "state": "down"}], compact=True) + result = self.search(qf={"state": "down"}, kind="changed") + self.assertEqual(result["total"], 1) + self.assertEqual(result["items"][0]["before"]["state"], "up") + + def test_sample_lookup_completes_pair_matching_only_one_side(self): + self.seed([{"id": "1", "state": "up", "age": "old"}], + [{"id": "1", "state": "up", "age": "new"}], sample=True) + result = self.search(kw="old", kind="unchanged") + self.assertEqual(result["source"], "live") + self.assertEqual(result["total"], 1) + self.assertEqual(result["items"][0]["after"]["age"], "new") + + def test_sample_lookup_does_not_turn_changed_into_removed(self): + self.seed([{"id": "1", "state": "up"}, {"id": "2", "state": "up"}], + [{"id": "1", "state": "down"}, {"id": "2", "state": "up"}], sample=True) + result = self.search(qf={"id": "1", "state": "up"}, kind="all") + self.assertEqual([d["kind"] for d in result["items"]], ["changed"]) + + def test_sample_lookup_keeps_duplicate_group_pairing(self): + self.seed([{"id": "1", "state": "up", "note": "first"}, + {"id": "1", "state": "down", "note": "second"}], + [{"id": "1", "state": "up", "note": "other"}, + {"id": "1", "state": "down", "note": "other"}], sample=True) + result = self.search(kw="second", kind="all") + self.assertEqual([d["kind"] for d in result["items"]], ["unchanged"]) + self.assertEqual(result["items"][0]["after"]["state"], "down") + + def test_sample_search_is_not_limited_by_persisted_search_text_size(self): + self.seed([{"id": "1", "state": "up", "note": "x" * 5000 + "needle"}], + [{"id": "1", "state": "up", "note": "other"}], sample=True) + self.assertEqual(self.search(kw="needle", kind="unchanged")["total"], 1) + + def test_sample_lookup_applies_port_mapping_before_matching(self): + self.seed([{"id": "1", "interface": "old", "state": "up"}], + [{"id": "1", "interface": "new", "state": "up"}], + keys=["id", "interface"], iface=["interface"], mapping={"old": "new"}, sample=True) + result = self.search(qf={"interface": "old"}, kind="unchanged") + self.assertEqual(result["total"], 1) + self.assertEqual(result["items"][0]["after"]["interface"], "new") + + def test_lldp_source_search(self): + self.seed([{"local_if": "gei-1", "remote_sys": "peer", "remote_if": "port", "remote_ip": "1"}], + [{"local_if": "gei-1", "remote_sys": "peer", "remote_if": "port", "remote_ip": "1"}], + metric="lldp_neighbor", keys=["local_if", "remote_sys"], fields=["remote_ip"], + iface=["local_if"], sample=True) + self.assertEqual(self.search(kw="peer", kind="unchanged")["total"], 1) + + def test_lldp_compact_diff_field_search(self): + self.seed([{"local_if": "gei-1", "remote_sys": "peer", "remote_ip": "1"}], + [{"local_if": "gei-1", "remote_sys": "peer", "remote_ip": "2"}], + metric="lldp_neighbor", keys=["local_if", "remote_sys"], fields=["remote_ip"], + iface=["local_if"], compact=True) + self.assertEqual(self.search(qf={"remote_ip": "2"}, kind="changed")["total"], 1) + + def test_exact_load_cap_is_not_truncated(self): + self.seed([{"id": "1", "state": "up"}], [{"id": "1", "state": "up"}]) + rows, truncated = _load_metric_rows_for_search(self.db, batch_id="before", metric_id="test", + row_filters=[], key_fields=["id"], kw="up", cap=1) + self.assertEqual(len(rows), 1) + self.assertFalse(truncated) + + def test_incomplete_duplicate_groups_are_omitted(self): + self.seed([{"id": "1", "state": "up"}, {"id": "1", "state": "down"}], + [{"id": "1", "state": "up"}, {"id": "1", "state": "down"}], sample=True) + with patch("netx_api.biz_state.compare_service._LIVE_SEARCH_GROUP_LOAD_CAP", 1): + result = self.search(kw="up", kind="all") + self.assertEqual(result["items"], []) + self.assertTrue(result["truncated"]) + + def test_postgres_search_pattern_treats_wildcards_literally(self): + self.assertEqual(_literal_search_pattern("a_10%"), "%a\\_10\\%%") + + def test_postgres_unsupported_filter_uses_python(self): + self.seed([{"id": "1", "state": "up", "age": "H"}], + [{"id": "1", "state": "up", "age": "H"}]) + with patch("netx_api.biz_state.compare_sql._dialect_is_postgres", return_value=True): + rows, _ = _load_metric_rows_for_search(self.db, batch_id="before", metric_id="test", + row_filters=[{"field": "age", "op": "age_timer"}], key_fields=["id"], kw="up") + self.assertEqual(rows, []) + + def test_unknown_sheet_is_rejected(self): + self.seed([{"id": "1", "state": "up"}], [{"id": "1", "state": "up"}]) + with self.assertRaises(HTTPException) as error: + list_run_diffs(self.db, "run", metric_id="missing") + self.assertEqual(error.exception.detail, "sheet_not_found") + + def test_stored_search_escapes_like_wildcards(self): + self.seed([{"id": "x_1", "state": "up"}, {"id": "xy1", "state": "up"}], + [{"id": "x_1", "state": "down"}, {"id": "xy1", "state": "down"}]) + self.assertEqual(self.search(qf={"id": "x_1"}, kind="changed")["total"], 1) + query = _search_stored_diffs(self.db.query(BizCompareDiff), metric_id="test", + kw="up", field_q={"id": "x_1"}) + compiled = query.statement.compile(dialect=postgresql.dialect()) + self.assertIn("LEFT OUTER JOIN", str(compiled)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_biz_state_compare_snapshot.py b/tests/test_biz_state_compare_snapshot.py new file mode 100644 index 0000000..3736fa8 --- /dev/null +++ b/tests/test_biz_state_compare_snapshot.py @@ -0,0 +1,170 @@ +"""Historical comparisons must use the settings captured when they were queued.""" + +from __future__ import annotations + +import csv +import io +import json +import unittest +import zipfile +from copy import deepcopy +from unittest.mock import patch + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from netx_api.biz_state.compare_service import ( + _create_running_run, _execute_compare_into_run, cancel_compare_run, + export_run_zip, get_run, list_run_diffs, template_metrics, +) +from netx_api.db import Base +from netx_api.models import ( + BizCompareDiff, BizCompareJob, BizCompareTemplate, BizPortMapping, + BizPortMappingRow, BizStateMetricRow, +) + + +class CompareSnapshotTests(unittest.TestCase): + def setUp(self): + self.engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(self.engine) + self.factory = sessionmaker(bind=self.engine, expire_on_commit=False) + self.db = self.factory() + # Progress uses a separate session in production; keep every test write local. + self.session_patch = patch("netx_api.db.SessionLocal", self.factory) + self.session_patch.start() + self.sheet = { + "sheet_id": "sheet", "metric_id": "test", "title": "Original sheet", + "key_fields": ["id", "interface"], "iface_fields": ["interface"], + "compare_fields": ["state"], "display_fields": ["id", "interface", "state"], + "row_filters": [], "ignore_port_changes": False, + } + self.tpl = BizCompareTemplate( + id="tpl", name="Original template", metrics_json=[deepcopy(self.sheet)], + iface_normalize_json=[{"from": "GE", "to": "gei"}], + ) + self.job = BizCompareJob(id="job", name="Original job", template_id="tpl", + mapping_id="map", store_unchanged="sample") + self.mapping = BizPortMapping(id="map", name="Original mapping") + self.pair = BizPortMappingRow(mapping_id="map", before_if="gei-1", after_if="gei-2") + self.db.add_all([self.tpl, self.job, self.mapping, self.pair]) + for side, iface, note in (("before", "GE-1.100", "old"), ("after", "GE-2.100", "new")): + self.db.add(BizStateMetricRow(id=side, batch_id=side, metric_id="test", seq=0, + data_json={"id": "1", "interface": iface, + "state": "up", "note": note})) + self.db.commit() + self.run = _create_running_run( + self.db, job=self.job, tpl=self.tpl, before_batch_id="before", + after_batch_id="after", sheets_cfg=template_metrics(self.tpl), + ) + + def tearDown(self): + self.session_patch.stop() + self.db.close() + self.engine.dispose() + + def edit_current_settings(self): + changed = {**self.sheet, "title": "Edited sheet", "compare_fields": ["note"], + "row_filters": [{"field": "state", "op": "eq", "value": "down"}]} + self.tpl.name = "Edited template" + self.tpl.metrics_json = [changed] + self.tpl.iface_normalize_json = [{"from": "GE", "to": "other"}] + self.pair.after_if = "gei-99" + self.job.name = "Edited job" + self.job.store_unchanged = "never" + self.job.enabled_sheet_ids = ["different-sheet"] + self.mapping.name = "Edited mapping" + self.db.commit() + + def execute(self): + return _execute_compare_into_run(self.db, self.run.id) + + def test_queued_run_uses_original_sheet_rules_mapping_and_storage(self): + original = deepcopy(self.run.summary_json["config_snapshot"]) + self.edit_current_settings() + detail = self.execute() + self.assertEqual(detail["status"], "success") + self.assertEqual(detail["summary"]["unchanged"], 1) + self.assertEqual(detail["summary"]["unchanged_listed"], 1) + self.assertEqual(detail["summary"]["store_unchanged"], "sample") + self.assertEqual(detail["sheets"][0]["compare_fields"], ["state"]) + self.assertEqual(self.run.summary_json["config_snapshot"], original) + + def test_compact_hydration_restores_original_normalized_values(self): + self.execute() + self.edit_current_settings() + diff = list_run_diffs(self.db, self.run.id, metric_id="sheet", kind="unchanged")["items"][0] + self.assertEqual(diff["before"]["interface"], "GE-1.100") + self.assertEqual(diff["mapped_before"]["interface"], "gei-2.100") + self.assertEqual(diff["after"]["interface"], "gei-2.100") + + def test_live_search_does_not_inherit_new_filters_or_transforms(self): + with patch("netx_api.biz_state.compare_service._UNCHANGED_SAMPLE_MAX", 0): + self.execute() + self.edit_current_settings() + result = list_run_diffs(self.db, self.run.id, metric_id="sheet", kind="unchanged", kw="old") + self.assertEqual(result["source"], "live") + self.assertEqual(result["total"], 1) + self.assertEqual(result["items"][0]["mapped_before"]["interface"], "gei-2.100") + + def test_snapshot_can_be_read_after_mapping_and_template_are_deleted(self): + self.execute() + self.db.query(BizPortMappingRow).delete() + self.db.delete(self.mapping) + self.db.delete(self.tpl) + self.db.commit() + detail = get_run(self.db, self.run.id) + self.assertEqual(detail["template_name"], "Original template") + self.assertEqual(detail["mapping_name"], "Original mapping") + diff = list_run_diffs(self.db, self.run.id, metric_id="sheet", kind="unchanged")["items"][0] + self.assertEqual(diff["mapped_before"]["interface"], "gei-2.100") + + def test_queued_run_can_execute_after_its_template_and_mapping_are_deleted(self): + self.db.query(BizPortMappingRow).delete() + self.db.delete(self.mapping) + self.db.delete(self.tpl) + self.db.commit() + detail = self.execute() + self.assertEqual(detail["status"], "success") + self.assertEqual(detail["summary"]["unchanged"], 1) + + def test_detail_returns_original_names_and_independent_template_copy(self): + self.edit_current_settings() + detail = get_run(self.db, self.run.id) + self.assertEqual(detail["job_name"], "Original job") + self.assertEqual(detail["config_snapshot_version"], 1) + self.assertEqual(detail["template"]["iface_normalize_rules"], [{"from": "GE", "to": "gei"}]) + self.assertNotIn("config_snapshot", detail["summary"]) + detail["template"]["iface_normalize_rules"].clear() + self.assertTrue(self.run.summary_json["config_snapshot"]["template"]["iface_normalize_rules"]) + + def test_export_contains_original_config_and_hydrated_csv(self): + self.execute() + self.edit_current_settings() + with zipfile.ZipFile(io.BytesIO(export_run_zip(self.db, self.run.id))) as archive: + config = json.loads(archive.read("config_snapshot.json")) + self.assertEqual(config["port_map"], {"gei-1": "gei-2"}) + self.assertEqual(config["template"]["name"], "Original template") + rows = list(csv.DictReader(io.StringIO(archive.read("tables/sheet.csv").decode("utf-8-sig")))) + self.assertEqual(rows[0]["interface"], "gei-2.100") + + def test_failed_and_cancelled_runs_keep_snapshot(self): + original = deepcopy(self.run.summary_json["config_snapshot"]) + with patch("netx_api.biz_state.compare_service._run_sheet", side_effect=RuntimeError("test-failure")): + with self.assertRaisesRegex(RuntimeError, "test-failure"): + self.execute() + self.assertEqual(self.run.summary_json["config_snapshot"], original) + self.run.status = "running" + self.db.commit() + cancel_compare_run(self.db, self.run.id) + self.assertEqual(self.run.summary_json["config_snapshot"], original) + + def test_legacy_runs_without_snapshot_remain_readable(self): + self.execute() + self.run.summary_json = {k: v for k, v in self.run.summary_json.items() if k != "config_snapshot"} + self.db.commit() + detail = get_run(self.db, self.run.id) + self.assertEqual(detail["status"], "success") + self.assertIsNone(detail["config_snapshot_version"]) + self.assertEqual(list_run_diffs(self.db, self.run.id, metric_id="sheet", kind="unchanged")["total"], 1) + self.assertEqual(self.db.query(BizCompareDiff).count(), 1) diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 1061040..7c247dd 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -474,6 +474,10 @@ const en = { sideAfter: "After", sidesVs: "vs", sidesTitle: "Compare sides", + configSnapshot: "Execution settings saved", + configSnapshotHint: "Template rules and port mapping are fixed for this run. Later edits do not affect historical lookup. The downloaded report includes the execution settings.", + configLegacy: "Legacy run: settings not saved", + configLegacyHint: "This run has no configuration snapshot. Success lookup and compact row reconstruction use the current settings, which may differ from the original execution.", pick: "Select…", sameAsBefore: "Same as before / select…", autoHint: "auto: new after-task batches trigger compare", @@ -532,8 +536,14 @@ const en = { ranWithDuration: "Compare finished ({{s}}s)", unchangedNotStored: "Success rows were counted but not stored. Set “Store success rows” to sample and re-run for spot-check.", unchangedSampleHint: "{{total}} success rows total; browsing a stratified sample of {{listed}}. Search any route for live source lookup. Pass rate uses all {{total}}.", - liveSearchHint: "Field filters look up source batches live; tabs only filter kind. E.g. direction=out, network=1.1.1.1.", - liveSearchTruncatedHint: "Search results truncated — narrow the field filters.", + liveSearchHint: "Pairs are completed from source batches before filtering. Pass rate still uses the full sheet.", + liveSearchTruncatedHint: "Search limit reached; some entries are omitted. Narrow the filters. Incomplete key groups were skipped.", + clearSearch: "Clear filters", + resultLoading: "Loading comparison details…", + resultRange: "{{from}}–{{to}} of {{total}}", + resultLoadFailed: "Could not load comparison details", + retry: "Retry", + searchEmpty: "No results match these filters. Clear filters or select another result type.", clearKeyFilters: "Clear fields", storeUnchanged: "Store success rows", storeUnchangedAuto: "Auto (full if small / sample 5k + hydrate if large)", @@ -567,8 +577,6 @@ const en = { pickBatchRun: "Select compare run", pickRun: "Select run…", runCount: "{{n}} runs", - resultEmpty: "No matching diff rows", - resultFilterPh: "Free text, or direction:out network:1.1.1.1", kindAll: "All", kindDiff: "Fail", kindAdded: "Added", @@ -712,7 +720,6 @@ const en = { newBaseline: "New baseline batch", needProjectFields: "Enter a name and pick old/new portrait tasks", needExpectPorts: "Select at least one expected object for this batch", - needBaselineFirst: "Save an old baseline batch first to list expect objects", needBothBaselines: "Save both old and new baseline batches before evaluating", errOldBaselineRequired: "Old baseline batch is required", errNewBaselineRequired: "New baseline batch is required", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index 1c9b0fc..2eb360f 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -473,6 +473,10 @@ const zh = { sideAfter: "操作后", sidesVs: "比对", sidesTitle: "比对双方", + configSnapshot: "执行配置已保存", + configSnapshotHint: "此记录的模板规则和端口映射已固定;后续编辑不影响历史回查,下载报告包含执行配置。", + configLegacy: "旧记录:未保存执行配置", + configLegacyHint: "此记录创建时未保存配置快照;成功行回查和精简明细还原使用当前配置,修改配置后可能与当时结果不同。", pick: "选择…", sameAsBefore: "同操作前 / 选择…", autoHint: "auto:操作后任务新批次到达会自动跑比对", @@ -531,8 +535,14 @@ const zh = { ranWithDuration: "比对完成(耗时 {{s}} 秒)", unchangedNotStored: "成功行仅统计数量未落库。可在任务配置将「成功行保存」改为抽样后重新比对(抽查用)。", unchangedSampleHint: "成功共 {{total}} 条,明细抽样 {{listed}} 条(分层抽查)。要查任意路由请输入筛选条件——将按原表即时判定。通过率按全部 {{total}} 计。", - liveSearchHint: "按字段回查原表即时判定;页签只过滤种类。例:direction=out,network=1.1.1.1。", - liveSearchTruncatedHint: "搜索结果已截断,请再收窄字段条件。", + liveSearchHint: "从原始批次补齐同一业务的前后数据后筛选;通过率仍按全表统计。", + liveSearchTruncatedHint: "搜索已达上限,部分业务未展示。请收窄条件;不完整的业务分组已跳过。", + clearSearch: "清空筛选", + resultLoading: "正在加载对比明细…", + resultRange: "{{from}}–{{to}} / {{total}} 条", + resultLoadFailed: "对比明细加载失败", + retry: "重试", + searchEmpty: "当前筛选条件下没有匹配结果,可清空筛选或切换结果类型。", clearKeyFilters: "清空字段", storeUnchanged: "成功行保存", storeUnchangedAuto: "自动(小表全量 / 大表抽样 5000+原表补全)", @@ -778,7 +788,6 @@ const zh = { metaBaseline: "基线", baselineOk: "已设", baselineMissing: "缺", - needBaselineFirst: "先设基线", enableHighfreq: "开启高频", highfreqReady: "高频已就绪", collectNow: "立即采集", @@ -830,10 +839,6 @@ const zh = { currentMissingBanner: "缺当前批次:{{metrics}}", highfreqReadyInactive: "高频已绑定(窗口未开,未立即采集)", collectPartial: "部分失败(成功 {{ok}} / 失败 {{fail}})", - confirmProjectDone: "确认完成项目并暂停全部高频?", - projectMarkedDone: "已完成,高频已停", - confirmDeleteProject: "确认删除项目?将停高频并清批次记录。", - projectDeleted: "已删除", errMetricNeedsBindings: "监控项「{{metric}}」需要参数绑定({{params}}),割接高频暂不支持,请去掉该项", errNoProfileForMetric: "监控项「{{metric}}」在当前设备厂商下无可用采集 profile", acceptTitle: "本批验收", diff --git a/web/src/index.css b/web/src/index.css index 2046460..96e00ff 100644 --- a/web/src/index.css +++ b/web/src/index.css @@ -11931,6 +11931,22 @@ body:has(.bs-cmp-board:fullscreen) { opacity: 0.75; } +.bs-cmp-board__config { + flex: 0 1 auto; + min-width: 0; + padding: 4px 7px; + border: 1px solid var(--bs-cmp-border, rgba(148, 163, 184, 0.22)); + border-radius: 6px; + color: var(--nm-muted, #94a3b8); + font-size: 11px; + line-height: 1.4; + overflow-wrap: anywhere; +} + +.bs-cmp-board__config.is-legacy { + color: #fbbf24; +} + .bs-cmp-board__actions { margin-left: auto; flex: 0 0 auto; @@ -12497,7 +12513,7 @@ body:has(.bs-cmp-board:fullscreen) { flex-wrap: wrap; align-items: center; gap: 4px; - flex: 1 1 280px; + flex: 1 1 100%; min-width: 0; padding: 2px; border-radius: 8px; @@ -12615,13 +12631,13 @@ body:has(.bs-cmp-board:fullscreen) { .bs-cmp-filter-bar { display: flex; - flex-wrap: nowrap; + flex-wrap: wrap; gap: 8px; align-items: center; - flex: 1 1 200px; - min-width: 160px; - max-width: 320px; - margin-left: auto; + flex: 1 1 100%; + min-width: 0; + max-width: none; + margin-left: 0; min-height: 32px; } @@ -12630,6 +12646,8 @@ body:has(.bs-cmp-board:fullscreen) { .bs-cmp-filter-bar input { min-width: 0; width: 100%; + flex: 1 1 220px; + max-width: 440px; } .bs-cmp-filter-bar .bs-sheet-count { @@ -12717,11 +12735,57 @@ body:has(.bs-cmp-board:fullscreen) { } .bs-cmp-result-table.is-loading { - opacity: 0.55; + opacity: 1; /* Keep header/filters clickable; only mute row hit-testing via opacity cue */ pointer-events: auto; } +.bs-cmp-result-state { + min-height: 180px; + display: flex; + flex-direction: column; + align-items: center; + justify-content: center; + gap: 12px; + padding: 24px; + text-align: center; + white-space: normal; + overflow-wrap: anywhere; + color: var(--bs-cmp-muted); +} + +.bs-cmp-result-state strong { + color: var(--bs-cmp-ink); + font-size: 14px; +} + +.bs-cmp-result-state.is-error strong { + color: var(--danger, #fca5a5); +} + +.bs-cmp-strip__kind:focus-visible, +.bs-cmp-strip__toggle:focus-visible, +.bs-cmp-nav__item:focus-visible { + outline: 2px solid var(--accent, #60a5fa); + outline-offset: 2px; +} + +@media (max-width: 600px) { + .bs-cmp-strip__sheet { + flex-wrap: wrap; + } + + .bs-cmp-filter-bar input { + flex-basis: 100%; + max-width: none; + } + + .bs-cmp-key-filter { + flex: 1 1 120px; + max-width: none; + } +} + .bs-cmp-board.is-fullscreen .bs-cmp-result-table { max-height: none; min-height: 0; @@ -12752,6 +12816,7 @@ body:has(.bs-cmp-board:fullscreen) { @media (max-width: 960px) { .bs-cmp-board__body { grid-template-columns: 1fr; + grid-template-rows: auto minmax(0, 1fr); } .bs-cmp-board__body.is-nav-collapsed { @@ -12760,6 +12825,7 @@ body:has(.bs-cmp-board:fullscreen) { .bs-cmp-nav { max-height: 180px; + align-self: start; } .bs-cmp-nav.is-collapsed { @@ -12769,6 +12835,8 @@ body:has(.bs-cmp-board:fullscreen) { .bs-cmp-nav__list { flex-direction: row; flex-wrap: nowrap; + flex: 0 0 auto; + align-items: flex-start; overflow-x: auto; overflow-y: hidden; } @@ -12779,8 +12847,9 @@ body:has(.bs-cmp-board:fullscreen) { .bs-cmp-nav__item { flex: 0 0 auto; - min-width: 180px; - grid-template-columns: 8px minmax(0, 1fr); + width: auto; + min-width: 240px; + grid-template-columns: 8px minmax(0, 1fr) auto; } } diff --git a/web/src/pages/network/BizComparePage.tsx b/web/src/pages/network/BizComparePage.tsx index 22f2213..57c5079 100644 --- a/web/src/pages/network/BizComparePage.tsx +++ b/web/src/pages/network/BizComparePage.tsx @@ -845,7 +845,7 @@ function TplRowFiltersEditor({ Array.isArray(filt.value) ? filt.value.join(",") : String(filt.value ?? "") } placeholder={t("bizCompare.filterValue")} - isDisabled={["empty", "not_empty", "age_timer"].includes(filt.op || "")} + disabled={["empty", "not_empty", "age_timer"].includes(filt.op || "")} onChange={(e) => { const next = groups.map((g) => g.map((x) => ({ ...x }))); const op = next[gi][fi]?.op || "eq"; @@ -893,10 +893,6 @@ function TplRowFiltersEditor({ ); } -function metricLabel(id: string) { - return id; -} - export type BizComparePageMode = "jobs" | "templates" | "all"; export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePageMode }) { @@ -988,6 +984,12 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage const [diffsLoading, setDiffsLoading] = useState(false); const [diffsSource, setDiffsSource] = useState(""); const [diffsTruncated, setDiffsTruncated] = useState(false); + const [diffsError, setDiffsError] = useState(""); + const [diffsRetry, setDiffsRetry] = useState(0); + const jobRequestRef = useRef(0); + const runRequestRef = useRef(0); + const listsRequestRef = useRef(0); + const diffFiltersRef = useRef(""); const boardRef = useRef(null); const tableScrollRef = useRef(null); const tableScrollPosRef = useRef({ top: 0, left: 0 }); @@ -996,6 +998,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage const tplImportRef = useRef(null); const refresh = useCallback(async (opts?: { force?: boolean }) => { + const request = ++listsRequestRef.current; type Bundle = { taskRes: Awaited>; tpl: Awaited>; @@ -1014,6 +1017,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage return { taskRes, tpl, maps, j, met }; }; const apply = (b: Bundle) => { + if (request !== listsRequestRef.current) return; setTasks((b.taskRes.items || []) as TaskOpt[]); setTemplates((b.tpl.items || []) as Template[]); setMappings((b.maps.items || []) as Mapping[]); @@ -1038,8 +1042,12 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage // eslint-disable-next-line react-hooks/exhaustive-deps }, [refresh]); + useEffect(() => () => { listsRequestRef.current += 1; }, []); + // Keep local tab + dismiss overlays when route mode flips (component may be reused). useEffect(() => { + jobRequestRef.current += 1; + runRequestRef.current += 1; if (pageMode === "templates") setPageTab("templates"); else if (pageMode === "jobs") setPageTab("jobs"); setTplOpen(false); @@ -1053,39 +1061,55 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage void document.exitFullscreen().catch(() => undefined); } setBoardFs(false); + return () => { + jobRequestRef.current += 1; + runRequestRef.current += 1; + }; }, [pageMode]); useEffect(() => { + let cancelled = false; + setBeforeBatches([]); void (async () => { - if (!beforeTaskId) { - setBeforeBatches([]); - return; + if (!beforeTaskId) return; + try { + const b = await cutoverCachedGet(`bizCompare:batches:${beforeTaskId}`, + () => bizStateListBatches(beforeTaskId), { ttlMs: 5000 }); + if (!cancelled) setBeforeBatches((b.items || []) as BatchOpt[]); + } catch (e) { + if (!cancelled) showError(formatErr(e)); } - const b = await bizStateListBatches(beforeTaskId); - setBeforeBatches((b.items || []) as BatchOpt[]); })(); - }, [beforeTaskId]); + return () => { cancelled = true; }; + }, [beforeTaskId, showError]); useEffect(() => { + let cancelled = false; + setAfterBatches([]); void (async () => { - if (!afterTaskId) { - setAfterBatches([]); - return; + if (!afterTaskId) return; + try { + const b = await cutoverCachedGet(`bizCompare:batches:${afterTaskId}`, + () => bizStateListBatches(afterTaskId), { ttlMs: 5000 }); + if (!cancelled) setAfterBatches((b.items || []) as BatchOpt[]); + } catch (e) { + if (!cancelled) showError(formatErr(e)); } - const b = await bizStateListBatches(afterTaskId); - setAfterBatches((b.items || []) as BatchOpt[]); })(); - }, [afterTaskId]); + return () => { cancelled = true; }; + }, [afterTaskId, showError]); + + const templatesById = useMemo(() => new Map(templates.map((x) => [x.id, x])), [templates]); const filteredJobs = useMemo(() => { const kw = debouncedListKw.trim().toLowerCase(); if (!kw) return jobs; return jobs.filter((j) => { - const tpl = templates.find((x) => x.id === j.template_id); + const tpl = templatesById.get(j.template_id); const mids = (tpl?.metric_ids || templateSheets(tpl).map((s) => s.metric_id)).join(" "); return `${j.name} ${j.mode} ${j.status} ${tpl?.name || ""} ${mids}`.toLowerCase().includes(kw); }); - }, [jobs, templates, debouncedListKw]); + }, [jobs, templatesById, debouncedListKw]); const filteredTemplates = useMemo(() => { const kw = debouncedListKw.trim().toLowerCase(); @@ -1135,10 +1159,14 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage const next = currentlyOn.includes(sid) ? currentlyOn.filter((id) => id !== sid) : [...currentlyOn, sid]; + if (!next.length) { + showError(t("bizCompare.needSheets")); + return; + } // Empty list means "all on" (new template sheets auto-included) setEnabledSheetIds(next.length === all.length ? [] : next); }, - [jobSheetAllIds, enabledSheetIds], + [jobSheetAllIds, enabledSheetIds, showError, t], ); const setJobTemplateAndSheets = useCallback( @@ -1195,28 +1223,18 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage () => runSheets.find((s) => sheetIdentity(s) === resultSheetId) || runSheets[0] || null, [runSheets, resultSheetId], ); + const activeRunSheetId = sheetIdentity(activeRunSheet); + const activeRunSheetStatus = String(activeRunSheet?.status || ""); // Clear field filters when switching sheet useEffect(() => { setResultKeyFilters({}); }, [resultSheetId]); - // Reset page when sheet / filter / page size changes - useEffect(() => { - setResultPage(1); - }, [ - resultSheetId, - kindFilter, - debouncedResultKw, - debouncedKeyFiltersJson, - resultPageSize, - runDetail?.id, - ]); - useEffect(() => { const runId = String(runDetail?.id || ""); - const mid = resultSheetId || sheetIdentity(activeRunSheet) || ""; - const sheetSt = String(activeRunSheet?.status || ""); + const mid = resultSheetId || activeRunSheetId || ""; + const sheetSt = activeRunSheetStatus; // Block only while *this* sheet is still in flight — done sheets are readable mid-run const sheetStillRunning = ["pending", "running", "queued"].includes(sheetSt); if (!runId || !mid || jobDetailTab !== "result" || sheetStillRunning) { @@ -1224,11 +1242,30 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage setResultTotal(0); setDiffsSource(""); setDiffsTruncated(false); + setDiffsLoading(false); + setDiffsError(""); return; } + const filterKey = JSON.stringify([runId, mid, kindFilter, debouncedResultKw, + debouncedKeyFiltersJson, resultPageSize]); + if (diffFiltersRef.current !== filterKey) { + diffFiltersRef.current = filterKey; + if (resultPage !== 1) { + setPagedDiffs([]); + setResultTotal(0); + setDiffsLoading(true); + setResultPage(1); + return; + } + } let cancelled = false; void (async () => { setDiffsLoading(true); + setDiffsError(""); + setPagedDiffs([]); + setResultTotal(0); + setDiffsSource(""); + setDiffsTruncated(false); try { const res = await bizCompareListRunDiffs({ runId, @@ -1253,8 +1290,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage if (resultPage > pages) setResultPage(pages); } catch (e) { if (!cancelled) { - // Keep previous rows to avoid strip/table jump; only clear on hard empty run - showError(formatErr(e)); + setDiffsError(formatErr(e)); } } finally { if (!cancelled) setDiffsLoading(false); @@ -1267,15 +1303,16 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage runDetail?.id, runDetail?.status, resultSheetId, - activeRunSheet?.metric_id, - activeRunSheet?.status, + activeRunSheetId, + activeRunSheetStatus, kindFilter, debouncedResultKw, debouncedKeyFiltersJson, + debouncedKeyFilters, resultPage, resultPageSize, jobDetailTab, - showError, + diffsRetry, ]); const resultColumns = useMemo(() => { @@ -1991,9 +2028,12 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage }; const openJob = async (id: string) => { + const request = ++jobRequestRef.current; + runRequestRef.current += 1; setJobId(id); setJobDetailTab("config"); setRunDetail(null); + setRuns([]); setResultSheetId(""); setKindFilter("diff"); setResultKw(""); @@ -2001,20 +2041,24 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage if (job) resetJobForm(job); try { const r = await bizCompareListRuns(id, 50); + if (request !== jobRequestRef.current) return; const items = r.items || []; setRuns(items); if (items.length) { const latest = await bizCompareGetRun(String((items as any[])[0].id)); + if (request !== jobRequestRef.current) return; setRunDetail(latest); // Land on batch list so stuck/cancelled runs are visible and actionable setJobDetailTab("runs"); } } catch (e) { - showError(formatErr(e)); + if (request === jobRequestRef.current) showError(formatErr(e)); } }; const closeJob = () => { + jobRequestRef.current += 1; + runRequestRef.current += 1; if (document.fullscreenElement === boardRef.current) { void document.exitFullscreen().catch(() => undefined); } @@ -2061,20 +2105,26 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage setJobDetailTab("runs"); return; } + const request = jobRequestRef.current; setBusy(true); try { await bizCompareUpdateJob(jobId, jobConfigBody()); // Async enqueue — returns immediately with status=running; poll below. const run = await bizCompareRunJob(jobId); + if (request !== jobRequestRef.current) return; + runRequestRef.current += 1; setRunDetail(run); setJobDetailTab("runs"); showOk(t("bizCompare.runStarted")); const r = await bizCompareListRuns(jobId, 50); + if (request !== jobRequestRef.current) return; setRuns(r.items || []); await refresh({ force: true }); } catch (e) { - showError(formatErr(e)); - setJobDetailTab("runs"); + if (request === jobRequestRef.current) { + showError(formatErr(e)); + setJobDetailTab("runs"); + } } finally { setBusy(false); } @@ -2082,20 +2132,25 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage const runProgress = (runDetail?.summary?.progress || {}) as RunProgressInfo; const runEngine = String(runProgress.engine || "").toLowerCase(); - // Poll active compare runs so the modal can be closed and reopened safely. + // One poller; schedule the next tick after requests settle to prevent overlap. useEffect(() => { - if (!runIsActive || !runDetail?.id) return; + if (!jobId || (!runIsActive && !(jobHasActiveRun && jobDetailTab === "runs"))) return; let cancelled = false; let notified = false; + let timer: number | undefined; + const jobRequest = jobRequestRef.current; + const runId = String(runDetail?.id || ""); const tick = async () => { + const runRequest = runRequestRef.current; try { - const d = await bizCompareGetRun(String(runDetail.id)); - if (cancelled) return; - setRunDetail(d); - if (jobId) { - const r = await bizCompareListRuns(jobId); - if (!cancelled) setRuns(r.items || []); - } + const [r, d] = await Promise.all([ + bizCompareListRuns(jobId, 50), + runIsActive && runId ? bizCompareGetRun(runId) : Promise.resolve(null), + ]); + if (cancelled || jobRequest !== jobRequestRef.current) return; + setRuns(r.items || []); + if (!d || runRequest !== runRequestRef.current) return; + setRunDetail((prev: typeof runDetail) => String(prev?.id || "") === runId ? d : prev); const st = String(d.status || ""); if (!notified && st === "success") { notified = true; @@ -2111,62 +2166,45 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage } } catch (e) { if (!cancelled) showError(formatErr(e)); + } finally { + if (!cancelled) timer = window.setTimeout(() => void tick(), 2000); } }; void tick(); - const id = window.setInterval(() => void tick(), 2000); return () => { cancelled = true; - window.clearInterval(id); + window.clearTimeout(timer); }; - }, [runDetail?.id, runIsActive, jobId, showOk, showError, t]); - - // Keep the runs list fresh while any batch on this job is active - useEffect(() => { - if (!jobId || !jobHasActiveRun || jobDetailTab !== "runs") return; - let cancelled = false; - const tick = async () => { - try { - const r = await bizCompareListRuns(jobId, 50); - if (!cancelled) setRuns(r.items || []); - const curId = String(runDetail?.id || ""); - if (curId) { - const d = await bizCompareGetRun(curId); - if (!cancelled) setRunDetail(d); - } - } catch { - /* ignore list poll errors */ - } - }; - void tick(); - const id = window.setInterval(() => void tick(), 2500); - return () => { - cancelled = true; - window.clearInterval(id); - }; - }, [jobId, jobHasActiveRun, jobDetailTab, runDetail?.id]); + }, [runDetail?.id, runIsActive, jobId, jobHasActiveRun, jobDetailTab, showOk, showError, t]); const loadRun = async (runId: string) => { + const request = ++runRequestRef.current; + const jobRequest = jobRequestRef.current; try { const d = await bizCompareGetRun(runId); + if (request !== runRequestRef.current || jobRequest !== jobRequestRef.current) return; setRunDetail(d); setKindFilter("diff"); setResultKw(""); + setResultKeyFilters({}); setJobDetailTab("result"); } catch (e) { - showError(formatErr(e)); + if (request === runRequestRef.current && jobRequest === jobRequestRef.current) showError(formatErr(e)); } }; const cancelRun = async (runId: string) => { if (!runId) return; if (!window.confirm(t("bizCompare.confirmCancelRun"))) return; + const request = jobRequestRef.current; setBusy(true); try { const d = await bizCompareCancelRun(runId); - if (String(runDetail?.id || "") === runId) setRunDetail(d); + if (request !== jobRequestRef.current) return; + setRunDetail((prev: typeof runDetail) => String(prev?.id || "") === runId ? d : prev); if (jobId) { const r = await bizCompareListRuns(jobId, 50); + if (request !== jobRequestRef.current) return; setRuns(r.items || []); } showOk(t("bizCompare.runCancelled")); @@ -2180,19 +2218,24 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage const removeRun = async (runId: string) => { if (!runId) return; if (!window.confirm(t("bizCompare.confirmDeleteRun"))) return; + const request = jobRequestRef.current; + const runRequest = runRequestRef.current; setBusy(true); try { const wasCurrent = String(runDetail?.id || "") === runId; await bizCompareDeleteRun(runId); + if (request !== jobRequestRef.current) return; let nextRuns: typeof runs = []; if (jobId) { const r = await bizCompareListRuns(jobId, 50); + if (request !== jobRequestRef.current) return; nextRuns = r.items || []; } else { nextRuns = (runs || []).filter((r) => String(r.id) !== runId); } setRuns(nextRuns); - if (wasCurrent) { + if (wasCurrent && runRequest === runRequestRef.current) { + runRequestRef.current += 1; setRunDetail(null); setResultSheetId(""); setJobDetailTab("runs"); @@ -2542,7 +2585,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage {filteredJobs.map((j) => { - const tpl = templates.find((x) => x.id === j.template_id); + const tpl = templatesById.get(j.template_id); const n = templateSheets(tpl).length; return ( @@ -2882,7 +2925,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage : "" } placeholder={cmpMode === "percent" ? "%" : ""} - isDisabled={!isCompare || !needsTol} + disabled={!isCompare || !needsTol} onChange={(e) => { const raw = e.target.value.trim(); const tol = raw === "" ? undefined : Number(raw); @@ -3113,7 +3156,12 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage className={jobDetailTab === "runs" ? "is-active" : undefined} onPress={() => { setJobDetailTab("runs"); - if (jobId) void bizCompareListRuns(jobId, 50).then((r) => setRuns(r.items || [])); + const request = jobRequestRef.current; + if (jobId) void bizCompareListRuns(jobId, 50).then((r) => { + if (request === jobRequestRef.current) setRuns(r.items || []); + }).catch((e) => { + if (request === jobRequestRef.current) showError(formatErr(e)); + }); }} > {t("bizCompare.tabRuns")} @@ -3354,6 +3402,18 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage })()} ) : null} + {runDetail ? ( + + {t(runDetail.config_snapshot_version + ? "bizCompare.configSnapshot" + : "bizCompare.configLegacy")} + + ) : null} {!boardFs ? ( <> setResultKw(e.target.value)} /> - - {diffsLoading ? "…" : `${pagedDiffs.length}/${resultTotal}`} + + {diffsLoading ? "…" : t("bizCompare.resultRange", { + from: resultTotal ? (resultPage - 1) * resultPageSize + 1 : 0, + to: Math.min(resultPage * resultPageSize, resultTotal), + total: resultTotal, + })} {resultSearchKeyFields.length ? ( + {(resultKw.trim() || activeKeyFilterCount > 0) ? ( + + ) : null} {resultSearchKeyFields.length && keyFiltersVisible ? (
@@ -3856,10 +3931,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage ) : null} {!isLiveSearch && kindFilter === "unchanged" && - (runDetail?.summary?.unchanged_truncated || + (activeRunSheet?.summary?.unchanged_truncated || (Number( - (activeRunSheet?.summary as any)?.unchanged || - runDetail?.summary?.unchanged || + (activeRunSheet?.summary as any)?.unchanged ?? + runDetail?.summary?.unchanged ?? 0, ) > Number( @@ -3899,6 +3974,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage className={`pt-list-table-wrap bs-sheet-table bs-cmp-result-table${ diffsLoading ? " is-loading" : "" }`} + aria-busy={diffsLoading} > @@ -4004,6 +4080,22 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage })()} + {diffsLoading || diffsError ? ( + + + + ) : null} {pagedDiffs.map((d, i) => { const pre = pickSideRow( d.mapped_before as Record | null | undefined, @@ -4081,13 +4173,13 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage ); })} - {runDetail && !diffsLoading && !pagedDiffs.length ? ( + {runDetail && !diffsLoading && !diffsError && !pagedDiffs.length ? (
+
+ {t(diffsError ? "bizCompare.resultLoadFailed" : "bizCompare.resultLoading")} + {diffsError ? <> + {diffsError} + + : null} +
+
- {kindFilter === "unchanged" - ? Number(runDetail?.summary?.unchanged || 0) > 0 && - !Number(runDetail?.summary?.unchanged_listed || 0) + {hasResultSearch ? t("bizCompare.searchEmpty") : kindFilter === "unchanged" + ? Number(activeRunSheet?.summary?.unchanged || 0) > 0 && + !Number(activeRunSheet?.summary?.unchanged_listed || 0) ? t("bizCompare.unchangedNotStored") : t("bizCompare.resultEmpty") : activeSheetPending diff --git a/web/src/pages/network/cutoverDataCache.ts b/web/src/pages/network/cutoverDataCache.ts index cc0748b..ff5e03b 100644 --- a/web/src/pages/network/cutoverDataCache.ts +++ b/web/src/pages/network/cutoverDataCache.ts @@ -7,6 +7,7 @@ type Entry = { at: number; data: unknown }; const store = new Map(); +const inFlight = new Map>(); const DEFAULT_TTL_MS = 45_000; export async function cutoverCachedGet( @@ -21,9 +22,17 @@ export async function cutoverCachedGet( return hit.data as T; } } - const data = await fetcher(); - store.set(key, { at: Date.now(), data }); - return data; + const existing = inFlight.get(key); + if (!opts?.force && existing) return existing as Promise; + const request = Promise.resolve().then(fetcher).then((data) => { + // An invalidation or forced refresh may supersede this request. + if (inFlight.get(key) === request) store.set(key, { at: Date.now(), data }); + return data; + }).finally(() => { + if (inFlight.get(key) === request) inFlight.delete(key); + }); + inFlight.set(key, request); + return request; } /** Return stale immediately (if any), refresh in background and notify. */ @@ -34,27 +43,28 @@ export async function cutoverCachedGetSWR( ): Promise { const hit = store.get(key); if (hit) { - void fetcher() + void cutoverCachedGet(key, fetcher, { ttlMs: 0 }) .then((data) => { - store.set(key, { at: Date.now(), data }); - onFresh?.(data); + if (store.get(key)?.data === data) onFresh?.(data); }) .catch(() => { /* keep stale */ }); return hit.data as T; } - const data = await fetcher(); - store.set(key, { at: Date.now(), data }); - return data; + return cutoverCachedGet(key, fetcher); } export function invalidateCutoverCache(prefix = ""): void { if (!prefix) { store.clear(); + inFlight.clear(); return; } for (const k of [...store.keys()]) { if (k.startsWith(prefix)) store.delete(k); } + for (const k of [...inFlight.keys()]) { + if (k.startsWith(prefix)) inFlight.delete(k); + } } diff --git a/web/tests/bizCompare.browser.mjs b/web/tests/bizCompare.browser.mjs new file mode 100644 index 0000000..6d3f62d --- /dev/null +++ b/web/tests/bizCompare.browser.mjs @@ -0,0 +1,133 @@ +// Run against a local Vite server. All API responses are fixtures; no device or DB is touched. +// NETX_PLAYWRIGHT_MODULE can point at a bundled Playwright package. +import assert from "node:assert/strict"; +import { createRequire } from "node:module"; +import { mkdir } from "node:fs/promises"; +import path from "node:path"; + +const require = createRequire(import.meta.url); +const { chromium } = require(process.env.NETX_PLAYWRIGHT_MODULE || "playwright"); +const base = process.env.NETX_TEST_URL || "http://127.0.0.1:5179"; +const output = path.resolve(process.env.NETX_TEST_OUTPUT || "../docs/reviews/assets"); +await mkdir(output, { recursive: true }); +const browser = await chromium.launch({ headless: true, channel: process.env.NETX_TEST_BROWSER || "chrome" }); +const page = await browser.newPage({ viewport: { width: 1440, height: 1000 } }); +const errors = []; +page.on("pageerror", (error) => errors.push(error.message)); +const fields = ["neighbor", "vrf", "remote_as", "state"]; +const sheet = { + sheet_id: "peer", title: "BGP 邻居", metric_id: "bgp_peer", + key_fields: ["neighbor", "vrf"], iface_fields: [], compare_fields: ["remote_as", "state"], + status: "done", summary: { changed: 1, removed: 1, added: 1, unchanged: 2, unchanged_listed: 2 }, +}; +const side = (label, ne_name, time) => ({ label, ne_name, started_at: time, ne_ip: "10.0.0.1" }); +const run = { + id: "run-1", status: "success", job_id: "job-1", config_snapshot_version: 1, created_at: "2026-10-10T10:08:00Z", + before: side("before", "PE-BEFORE", "2026-10-10T09:00:00Z"), + after: side("after", "PE-AFTER", "2026-10-10T10:00:00Z"), + sheets: [sheet], + summary: { ...sheet.summary, before_count: 4, after_count: 4, pass_rate: 50, duration_ms: 1250, + sheet_cards: [{ ...sheet, ...sheet.summary, before_count: 4, after_count: 4, pass_rate: 50 }] }, +}; +const job = { id: "job-1", name: "割接前后 · 业务验收", template_id: "tpl", mode: "manual", status: "ready", + before_task_id: "t1", after_task_id: "t2", before_batch_id: "b", after_batch_id: "a" }; +const row = (neighbor, state, remote_as = "64512") => ({ neighbor, vrf: "core", state, remote_as }); +const diffs = [ + { kind: "changed", key: { neighbor: "10.2.1.1", vrf: "core" }, + before: row("10.2.1.1", "Established"), after: row("10.2.1.1", "Idle"), + changes: { state: { before: "Established", after: "Idle" } } }, + { kind: "removed", key: { neighbor: "10.2.1.2", vrf: "core" }, before: row("10.2.1.2", "Established"), after: null, changes: {} }, + { kind: "added", key: { neighbor: "10.2.1.3", vrf: "core" }, before: null, after: row("10.2.1.3", "Established"), changes: {} }, + { kind: "unchanged", key: { neighbor: "10.2.1.4", vrf: "core" }, before: row("10.2.1.4", "Established"), after: row("10.2.1.4", "Established"), changes: {} }, +]; +let failDiffs = false; +let active = false; +let legacy = false; +let pollCounts = { runs: 0, detail: 0 }; +await page.route("**/v1/**", async (route) => { + const url = new URL(route.request().url()); + const p = url.pathname; + let data = {}; + if (p === "/v1/auth/me") data = { user: { id: "test", username: "test", role: "admin", is_active: true } }; + else if (p.endsWith("/compare/jobs")) data = { items: [job, { ...job, id: "job-slow", name: "慢任务" }] }; + else if (p.endsWith("/compare/templates")) data = { items: [{ id: "tpl", name: "割接验收模板", metrics: [sheet] }] }; + else if (p.endsWith("/compare/metrics")) data = { items: [{ metric_id: "bgp_peer", fields: fields.map((name) => ({ name, display_name: name, dtype: "string", is_key: sheet.key_fields.includes(name) })) }] }; + else if (p.endsWith("/tasks")) data = { items: [{ id: "t1", ne_name: "PE-BEFORE", ne_ip: "10.0.0.1" }, { id: "t2", ne_name: "PE-AFTER", ne_ip: "10.0.0.2" }] }; + else if (p.endsWith("/batches")) data = { items: [{ id: p.includes("t1") ? "b" : "a", status: "success", row_count: 4 }] }; + else if (p.endsWith("/runs") && p.includes("/jobs/")) { + pollCounts.runs++; + if (p.includes("job-slow")) await new Promise((resolve) => setTimeout(resolve, 900)); + data = { items: [{ ...run, status: active ? "running" : "success" }] }; + } else if (p.endsWith("/diffs")) { + if (failDiffs) { await route.fulfill({ status: 503, contentType: "application/json", body: JSON.stringify({ detail: "测试:连接暂时中断" }) }); return; } + const kind = url.searchParams.get("kind") || "diff"; + const kw = url.searchParams.get("kw") || ""; + const items = diffs.filter((d) => (kind === "all" || (kind === "diff" ? ["removed", "changed"].includes(d.kind) : d.kind === kind)) && JSON.stringify(d).includes(kw)); + data = { items, total: items.length, page: 1, page_size: 100, source: "stored", truncated: false }; + } else if (p.endsWith("/runs/run-1")) { + pollCounts.detail++; + data = { ...run, config_snapshot_version: legacy ? null : 1, status: active ? "running" : "success" }; + } else if (p.endsWith("/mappings")) data = { items: [] }; + else data = { items: [] }; + await route.fulfill({ status: 200, contentType: "application/json", body: JSON.stringify(data) }); +}); + +try { + await page.goto(`${base}/network/cutover/biz-compare`); + await page.getByRole("row").filter({ hasText: "割接前后 · 业务验收" }).getByRole("button", { name: "详情", exact: true }).click(); + await page.getByRole("button", { name: "比对结果", exact: true }).click(); + await page.locator(".bs-cmp-diff-table tbody tr.bs-cmp-row").first().waitFor(); + assert.equal(await page.locator(".bs-cmp-diff-table tbody tr.bs-cmp-row").count(), 2); + assert.equal(await page.locator(".bs-cmp-board__config").textContent(), "执行配置已保存"); + await page.screenshot({ path: path.join(output, "biz-compare-desktop.png"), fullPage: true }); + + const search = page.locator(".bs-cmp-filter-bar input"); + failDiffs = true; + await search.fill("故障"); + await page.getByText("对比明细加载失败", { exact: true }).waitFor(); + assert.equal(await page.locator(".bs-cmp-row").count(), 0); + await page.screenshot({ path: path.join(output, "biz-compare-error.png"), fullPage: true }); + failDiffs = false; + await page.getByRole("button", { name: "重试", exact: true }).click(); + await page.getByText("当前筛选条件下没有匹配结果,可清空筛选或切换结果类型。", { exact: true }).waitFor(); + await page.getByRole("button", { name: "清空筛选", exact: true }).click(); + await page.locator(".bs-cmp-row").first().waitFor(); + + await page.setViewportSize({ width: 768, height: 1000 }); + await page.screenshot({ path: path.join(output, "biz-compare-narrow.png"), fullPage: true }); + const fits = await page.locator(".bs-cmp-strip").evaluate((element) => element.scrollWidth <= element.clientWidth + 1); + assert.ok(fits, "filter toolbar must fit its available width"); + await page.setViewportSize({ width: 390, height: 1000 }); + assert.ok(await page.locator(".bs-cmp-strip").evaluate((element) => element.scrollWidth <= element.clientWidth + 1)); + assert.ok(await page.locator(".bs-cmp-board__toolbar").evaluate((element) => element.scrollWidth <= element.clientWidth + 1)); + await page.setViewportSize({ width: 768, height: 1000 }); + + await page.getByRole("button", { name: "关闭", exact: true }).click(); + await page.getByRole("row").filter({ hasText: "慢任务" }).getByRole("button", { name: "详情", exact: true }).click(); + await page.getByRole("button", { name: "关闭", exact: true }).click(); + await page.getByRole("row").filter({ hasText: "割接前后 · 业务验收" }).getByRole("button", { name: "详情", exact: true }).click(); + await page.getByRole("button", { name: "比对结果", exact: true }).click(); + await page.locator(".bs-cmp-row").first().waitFor(); + await page.waitForTimeout(1100); + assert.equal(await page.locator(".bs-cmp-row").count(), 2, "slow old job must not change the current tab"); + + legacy = true; + await page.reload(); + await page.getByRole("row").filter({ hasText: "割接前后 · 业务验收" }).getByRole("button", { name: "详情", exact: true }).click(); + await page.getByRole("button", { name: "比对结果", exact: true }).click(); + await page.getByText("旧记录:未保存执行配置", { exact: true }).waitFor(); + await page.screenshot({ path: path.join(output, "biz-compare-legacy.png"), fullPage: true }); + + active = true; + await page.reload(); + await page.getByRole("row").filter({ hasText: "割接前后 · 业务验收" }).getByRole("button", { name: "详情", exact: true }).click(); + await page.locator(".bs-cmp-runs").waitFor(); + pollCounts = { runs: 0, detail: 0 }; + await page.waitForTimeout(5200); + assert.ok(pollCounts.runs <= 3 && pollCounts.detail <= 3, JSON.stringify(pollCounts)); + assert.ok(pollCounts.runs >= 2 && pollCounts.detail >= 2, "active run should keep polling"); + assert.deepEqual(errors, []); + console.log(JSON.stringify({ passed: ["result layout", "saved execution settings", "legacy settings warning", "error clears stale rows", "retry", "clear filters", "768px toolbar", "old job response ignored", "single poller", "no runtime errors"], pollCounts, output })); +} finally { + await browser.close(); +} diff --git a/web/tests/cutoverDataCache.test.mjs b/web/tests/cutoverDataCache.test.mjs new file mode 100644 index 0000000..8c12694 --- /dev/null +++ b/web/tests/cutoverDataCache.test.mjs @@ -0,0 +1,72 @@ +import assert from "node:assert/strict"; +import { beforeEach, test } from "node:test"; +import { cutoverCachedGet, cutoverCachedGetSWR, invalidateCutoverCache } from "../src/pages/network/cutoverDataCache.ts"; + +const deferred = () => { + let resolve, reject; + const promise = new Promise((ok, fail) => { resolve = ok; reject = fail; }); + return { promise, resolve, reject }; +}; + +beforeEach(() => invalidateCutoverCache()); + +test("concurrent readers share one fetch", async () => { + const pending = deferred(); + let calls = 0; + const fetcher = () => { calls++; return pending.promise; }; + const first = cutoverCachedGet("batches", fetcher); + const second = cutoverCachedGet("batches", fetcher); + pending.resolve("fresh"); + assert.deepEqual(await Promise.all([first, second]), ["fresh", "fresh"]); + assert.equal(calls, 1); +}); + +test("invalidated pending requests cannot repopulate the cache", async () => { + const pending = deferred(); + const old = cutoverCachedGet("batches", () => pending.promise); + invalidateCutoverCache("batches"); + pending.resolve("stale"); + await old; + assert.equal(await cutoverCachedGet("batches", async () => "fresh"), "fresh"); +}); + +test("forced refresh supersedes a slow older request", async () => { + const pending = deferred(); + const old = cutoverCachedGet("lists", () => pending.promise); + assert.equal(await cutoverCachedGet("lists", async () => "fresh", { force: true }), "fresh"); + pending.resolve("old"); + await old; + assert.equal(await cutoverCachedGet("lists", async () => "unexpected"), "fresh"); +}); + +test("a failed fetch can be retried", async () => { + await assert.rejects(cutoverCachedGet("lists", async () => { throw new Error("offline"); })); + assert.equal(await cutoverCachedGet("lists", async () => "recovered"), "recovered"); +}); + +test("SWR readers share background refresh and each receives the fresh value", async () => { + await cutoverCachedGet("lists", async () => "cached"); + const pending = deferred(); + let calls = 0; + const notifications = []; + const fetcher = () => { calls++; return pending.promise; }; + const first = await cutoverCachedGetSWR("lists", fetcher, (x) => notifications.push(x)); + const second = await cutoverCachedGetSWR("lists", fetcher, (x) => notifications.push(x)); + assert.equal(first, "cached"); + assert.equal(second, "cached"); + pending.resolve("fresh"); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(calls, 1); + assert.deepEqual(notifications, ["fresh", "fresh"]); +}); + +test("invalidated SWR requests do not notify with stale data", async () => { + await cutoverCachedGet("lists", async () => "cached"); + const pending = deferred(); + const notifications = []; + await cutoverCachedGetSWR("lists", () => pending.promise, (x) => notifications.push(x)); + invalidateCutoverCache("lists"); + pending.resolve("stale"); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(notifications, []); +});