From fe4c1db7639ca2b38e0a36761d7ba796c317be0b Mon Sep 17 00:00:00 2001 From: oliver Date: Sun, 11 Oct 2026 07:33:17 +0800 Subject: [PATCH] fix: enforce business continuity and live cutover monitoring --- netx_api/biz_migration/auto_monitor.py | 104 ++++++++ netx_api/biz_migration/evaluate.py | 132 +++++++--- netx_api/biz_migration/monitor_templates.py | 10 +- netx_api/biz_migration/service.py | 255 +++++++++++++++++--- netx_api/biz_migration/validation.py | 70 ++++++ netx_api/biz_migration_router.py | 4 + netx_api/biz_state/collect_runner.py | 8 + tests/test_biz_migration_evaluate.py | 17 +- tests/test_biz_migration_evidence.py | 4 +- tests/test_biz_migration_hf_decouple.py | 6 +- tests/test_biz_migration_integrity.py | 239 ++++++++++++++++++ 11 files changed, 769 insertions(+), 80 deletions(-) create mode 100644 netx_api/biz_migration/auto_monitor.py create mode 100644 netx_api/biz_migration/validation.py create mode 100644 tests/test_biz_migration_integrity.py diff --git a/netx_api/biz_migration/auto_monitor.py b/netx_api/biz_migration/auto_monitor.py new file mode 100644 index 0000000..e037472 --- /dev/null +++ b/netx_api/biz_migration/auto_monitor.py @@ -0,0 +1,104 @@ +"""Coalesced collect-completion events for live cutover evaluation. + +One worker per process bounds concurrency. Project row locks in run_evaluate +serialize persistence across processes. No device collection runs here. +""" +from __future__ import annotations + +import hashlib +import json +import logging +import threading + +from sqlalchemy.orm import Session + +from ..models import BizMigrationBatch, BizMigrationProject, BizMigrationRun, BizMonitorTemplate, BizStateBatch +from . import service + +_log = logging.getLogger(__name__) +_lock = threading.Lock() +_pending: set[str] = set() +_worker_running = False + + +def try_auto_monitor_for_task(db: Session, task_id: str) -> int: + count = 0 + # Bindings are JSON for multiple sampling intervals; match in Python after + # limiting to projects with an active operation batch. + projects = db.query(BizMigrationProject).filter( + BizMigrationProject.status != "done", + BizMigrationProject.id.in_(db.query(BizMigrationBatch.project_id).filter( + BizMigrationBatch.status == "active")), + ).all() + for candidate in projects: + tids = service._all_hf_task_ids(candidate, "old") + service._all_hf_task_ids(candidate, "new") + if task_id not in tids or not candidate.old_baseline_batch_id or not candidate.new_baseline_batch_id: + continue + # Lock before deduplication so two worker processes cannot save the same + # snapshot. run_evaluate holds the same lock through the transaction. + proj = db.query(BizMigrationProject).filter( + BizMigrationProject.id == candidate.id).with_for_update().populate_existing().first() + batch = db.query(BizMigrationBatch).filter( + BizMigrationBatch.project_id == proj.id, BizMigrationBatch.status == "active" + ).order_by(BizMigrationBatch.started_at.desc()).first() + if not batch or proj.status == "done": + db.rollback() + continue + mt = db.get(BizMonitorTemplate, proj.monitor_template_id) if proj.monitor_template_id else None + if not mt: + db.rollback() + _log.error("cutover monitor template missing project=%s", proj.id) + continue + sheets, overrides, defaults, norms = service.resolve_evaluate_sheets(db, mt) + samples = [] + for tid in tids: + sample = db.query(BizStateBatch).filter( + BizStateBatch.task_id == tid, BizStateBatch.status.notin_(("running", "queued")) + ).order_by(BizStateBatch.started_at.desc(), BizStateBatch.id.desc()).first() + samples.append([tid, sample.id if sample else "", sample.status if sample else ""]) + payload = [samples, proj.old_baseline_batch_id, proj.new_baseline_batch_id, + batch.expect_set_json, service._port_map_dict(db, proj.mapping_id), + sheets, overrides, defaults, norms, service.resolve_collect_metric_ids(db, proj)] + fingerprint = hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode()).hexdigest() + last = db.query(BizMigrationRun).filter(BizMigrationRun.batch_id == batch.id).order_by( + BizMigrationRun.created_at.desc()).first() + if last and (last.summary_json or {}).get("auto_fingerprint") == fingerprint: + db.rollback() + continue + try: + result = service.run_evaluate(db, batch_id=batch.id, purpose="auto", _commit=False) + run = db.get(BizMigrationRun, result["id"]) + run.summary_json = {**run.summary_json, "auto_fingerprint": fingerprint} + db.commit() + count += 1 + except Exception: + db.rollback() + _log.exception("cutover monitor evaluate failed project=%s", proj.id) + return count + + +def schedule_auto_monitor_for_task(task_id: str) -> None: + global _worker_running + with _lock: + _pending.add(str(task_id)) + if _worker_running: + return + _worker_running = True + + def work() -> None: + global _worker_running + from ..db import SessionLocal + + while True: + with _lock: + if not _pending: + _worker_running = False + return + tid = _pending.pop() + try: + with SessionLocal() as db: + try_auto_monitor_for_task(db, tid) + except Exception: + _log.exception("cutover monitor worker failed task=%s", tid) + + threading.Thread(target=work, name="biz-cutover-monitor", daemon=True).start() diff --git a/netx_api/biz_migration/evaluate.py b/netx_api/biz_migration/evaluate.py index eb85841..1391765 100644 --- a/netx_api/biz_migration/evaluate.py +++ b/netx_api/biz_migration/evaluate.py @@ -5,6 +5,9 @@ from __future__ import annotations from typing import Any from ..biz_state.compare_engine import apply_port_map, compare_rows +from ..biz_state.compare_rules import ( + effective_compare_fields, field_rule_map, scalar_text, values_equal, +) PORT_METRIC_ID = "interface_brief" PORT_STATUS_FIELDS = ("admin", "phy", "prot") @@ -22,7 +25,7 @@ def normalize_expect_key(raw: str) -> str: Accepts legacy ``a|b`` multi-field keys and already-normalized ``\\x1f`` keys. """ - s = str(raw or "").strip() + s = scalar_text(raw).strip() if not s: return "" if KEY_SEP in s: @@ -65,13 +68,13 @@ def classify_status(row: dict[str, Any] | None, sheet_override: dict[str, Any] | return "none" down_vals = {str(x).lower() for x in (ov.get("down_values") or ["down"])} up_vals = {str(x).lower() for x in (ov.get("up_values") or ["up"])} - vals = [str(row.get(f) or "").strip().lower() for f in fields] + vals = [scalar_text(row.get(f)).strip().lower() for f in fields] vals = [v for v in vals if v] if not vals: return "none" if any(v in down_vals for v in vals): return "down" - if vals and all(v in up_vals for v in vals): + if len(vals) == len(fields) and all(v in up_vals for v in vals): return "up" return "other" @@ -219,7 +222,7 @@ def _field_tokens_from_row(row: dict[str, Any] | None, fields: set[str]) -> set[ return set() out: set[str] = set() for f in fields: - v = str(row.get(f) or "").strip() + v = scalar_text(row.get(f)).strip() if v: out.add(field_token(f, v)) return out @@ -245,8 +248,8 @@ def _side_tokens( def _norm_list_values(raw: Any) -> list[str]: if isinstance(raw, list): - return [str(x).strip().lower() for x in raw if str(x).strip()] - s = str(raw or "").strip() + return [scalar_text(x).strip().lower() for x in raw if scalar_text(x).strip()] + s = scalar_text(raw).strip() if not s: return [] if "," in s: @@ -266,18 +269,18 @@ def _eval_field_op( f = str(field or "").strip() if not f: return False - raw = "" if not row else str(row.get(f) or "").strip() + raw = "" if not row else scalar_text(row.get(f)).strip() val = raw.lower() o = str(op or "eq").strip().lower() if o in ("changed", "diff"): - base = "" if not base_row else str(base_row.get(f) or "").strip().lower() + base = "" if not base_row else scalar_text(base_row.get(f)).strip().lower() if not row: return False return base != val if o in ("unchanged", "same"): if not row: return False - base = "" if not base_row else str(base_row.get(f) or "").strip().lower() + base = "" if not base_row else scalar_text(base_row.get(f)).strip().lower() return base == val if o == "empty": return not val @@ -568,7 +571,7 @@ def _match_field_token_rules( if not field or not allowed: continue row = old_row if side == "old" else new_row if side == "new" else None - val = str((row or {}).get(field) or "").strip().lower() + val = scalar_text((row or {}).get(field)).strip().lower() if val not in allowed: return False return True @@ -858,6 +861,14 @@ def _port_identity(key_fields: list[str], iface_fields: list[str]) -> bool: return all(str(f) in iface_set for f in key_fields) +def _iface_selected(value: str, ports: set[str]) -> bool: + """Set lookup plus parent interface prefixes; independent of scope size.""" + if value in ports: + return True + parts = value.split(".") + return any(".".join(parts[:i]) in ports for i in range(len(parts) - 1, 0, -1)) + + def _unmapped_same_iface_anomaly( *, old_kind: str, @@ -892,6 +903,8 @@ def _remap_key_str( return key_str if not port_map: return key_str + if key_fields and not set(key_fields).intersection(iface_fields): + return key_str if (not key_fields or len(key_fields) == 1) and ( key_str in port_map or any( @@ -928,6 +941,8 @@ def _reverse_remap_key_str( return key_str if not rev_map: return key_str + if key_fields and not set(key_fields).intersection(iface_fields): + return key_str if (not key_fields or len(key_fields) == 1) and ( key_str in rev_map or any( @@ -1009,7 +1024,7 @@ def row_match_key( data = strip_netx(row) if iface_fields and rules: data = apply_iface_normalize(dict(data), iface_fields=iface_fields, rules=rules) - return KEY_SEP.join(str(data.get(k) or "").strip() for k in key_fields) + return KEY_SEP.join(scalar_text(data.get(k)).strip() for k in key_fields) def index_raw_by_match_key( @@ -1033,7 +1048,7 @@ def display_key_from_raw(raw: dict[str, Any] | None, key_fields: list[str]) -> s if not raw or not key_fields: return "" data = strip_netx(raw) - return "|".join(str(data.get(k) or "").strip() for k in key_fields) + return "|".join(scalar_text(data.get(k)).strip() for k in key_fields) def iface_lineage( @@ -1053,7 +1068,7 @@ def iface_lineage( pmap = port_map or {} out: list[dict[str, Any]] = [] for f in iface_fields: - raw_v = str(data.get(f) or "").strip() + raw_v = scalar_text(data.get(f)).strip() norm_v = normalize_iface_name(raw_v, rules) if rules else raw_v mapped = False map_to = "" @@ -1092,6 +1107,8 @@ def evaluate_metric_dual( out_of_expect: str = "strict", sheet_id: str = "", iface_normalize_rules: list[dict[str, str]] | None = None, + previous_expect: dict[str, set[str]] | None = None, + allow_mapping_scope: bool = True, ) -> dict[str, Any]: """Run old vs old-baseline, new vs new-baseline (or mapped old baseline), dual merge. @@ -1108,6 +1125,9 @@ def evaluate_metric_dual( expect_keys = expect_keys_for_metric( expect, metric_id=metric_id, iface_fields=iface_fields, sheet_id=sid ) + previous_keys = expect_keys_for_metric( + previous_expect or {}, metric_id=metric_id, iface_fields=iface_fields, sheet_id=sid + ) ov = sheet_override or {} if ov.get("skip_dual"): return { @@ -1178,6 +1198,7 @@ def evaluate_metric_dual( port_map=None, field_rules=field_rules, iface_normalize_rules=None, # already applied + include_unchanged=True, ) old_idx = build_diff_index_from_compare(old_cmp) @@ -1186,18 +1207,8 @@ def evaluate_metric_dual( if new_baseline_rows is not None: new_before = list(new_baseline_rows) new_baseline_mode = "provided" - # Explicit empty batch is as unusable as "no baseline" for presence semantics. - if not new_before and (old_baseline_rows or new_current_rows): - if port_map and iface_fields: - new_before = [ - apply_port_map(r, iface_fields=iface_fields, port_map=port_map) - for r in old_baseline_rows - ] - new_baseline_mode = "port_mapped" - new_baseline_missing = False - else: - new_baseline_missing = True - new_baseline_mode = "empty" + # A successfully collected empty baseline is valid on a replacement NE. + # Collection completeness is checked by the service, not inferred from rows. elif port_map and iface_fields: new_before = [ apply_port_map(r, iface_fields=iface_fields, port_map=port_map) @@ -1218,12 +1229,22 @@ def evaluate_metric_dual( compare_fields=compare_fields, port_map=None, field_rules=field_rules, + include_unchanged=True, ) new_idx = build_diff_index_from_compare(new_cmp) rev_map = {v: k for k, v in port_map.items()} map_scoped = _map_defines_iface_expect(metric_id, iface_fields, port_map) - if map_scoped: + # Mapping describes identity; an explicit batch selection defines scope. + # Port selection also covers business rows attached to those interfaces. + for scope, keys in ((expect, expect_keys), (previous_expect or {}, previous_keys)): + ports = scope.get("_ports") or set() + if ports and not keys and iface_fields: + for ks, diff in old_idx.items(): + row = diff.get("before") or diff.get("after") or {} + if any(_iface_selected(scalar_text(row.get(f)), ports) for f in iface_fields): + keys.add(ks) + if allow_mapping_scope and map_scoped and not any(expect.values()): scoped: set[str] = set() for ks in old_idx: if _key_in_port_map( @@ -1253,6 +1274,8 @@ def evaluate_metric_dual( for ok in sorted(expect_keys): _add(ok) + for ok in sorted(previous_keys): + _add(ok) for ks in sorted(old_idx): _add(ks) for ks in sorted(new_idx): @@ -1270,6 +1293,10 @@ def evaluate_metric_dual( progress_total = 0 anomaly = 0 anomaly_in_expect = 0 + rules_by_field = field_rule_map(field_rules) + integrity_fields = effective_compare_fields(compare_fields, field_rules) + # Status transitions are interpreted by the monitor's correction rules. + integrity_fields = [f for f in integrity_fields if f not in (ov.get("status_fields") or [])] for old_ks in canon_keys: new_ks = _remap_key_str( @@ -1279,11 +1306,12 @@ def evaluate_metric_dual( port_map=port_map, ) in_exp = old_ks in expect_keys + previous = old_ks in previous_keys and not in_exp if in_exp: progress_total += 1 od = old_idx.get(old_ks) - nd = new_idx.get(new_ks) or new_idx.get(old_ks) + nd = new_idx.get(new_ks) old_kind = str((od or {}).get("kind") or "") new_kind = str((nd or {}).get("kind") or "") @@ -1301,11 +1329,7 @@ def evaluate_metric_dual( old_st = classify_status(old_cur, ov) new_st = classify_status(new_cur, ov) - # Mapped iface rows are the expect set. Unmapped rows stay out of the - # migration compare, except same-name ports where old went down and the - # new device's same port is up — that is anomaly, not a successful cutover. - if map_scoped and not in_exp: - if not ( + if map_scoped and not (in_exp or previous) and ( _port_identity(key_fields, iface_fields) and _unmapped_same_iface_anomaly( old_kind=old_kind, @@ -1314,26 +1338,62 @@ def evaluate_metric_dual( new_status=new_st, ) ): - continue verdict, color, rule_hit = "anomaly", "red", "unmapped_same_iface" else: verdict, color, rule_hit = dual_verdict_ex( old_kind=old_kind or "", new_kind=new_kind or "", - in_expect=in_exp, + in_expect=in_exp or previous, window_active=window_active, acceptance=acceptance, old_status=old_st, new_status=new_st, success_patterns=success_patterns or None, anomaly_patterns=anomaly_patterns or None, - out_of_expect=out_of_expect, + # Business drift outside this batch must remain visible. + out_of_expect="strict", old_row=old_cur or None, new_row=new_cur or None, old_base=old_base or None, new_base=new_base or None, sheet_override=ov, ) + integrity_changes: list[dict[str, Any]] = [] + if in_exp or previous: + old_departed = old_kind == "removed" or old_st == "down" + new_healthy = bool(new_cur) and ( + new_st == "up" if ov.get("status_fields") else True + ) + if old_departed and not new_healthy: + verdict, color, rule_hit = "lost", "red", "service_not_received" + elif verdict == "migrated": + if not new_healthy: + verdict, color, rule_hit = "anomaly", "red", "target_unhealthy" + elif not old_base: + verdict, color, rule_hit = "anomaly", "red", "old_service_not_in_baseline" + else: + reference = apply_port_map(old_base, iface_fields=iface_fields, port_map=port_map) + integrity_changes = [ + {"field": f, "before": reference.get(f), "after": new_cur.get(f)} + for f in integrity_fields + if not values_equal(reference.get(f), new_cur.get(f), rule=rules_by_field.get(f)) + ] + if integrity_changes: + verdict, color, rule_hit = "anomaly", "red", "service_content_changed" + elif not old_departed and not success_patterns: + verdict, color, rule_hit = ( + ("unfinished", "red", "old_service_still_present") if acceptance + else ("migrating", "yellow", "old_service_still_present") + ) + if previous: + if verdict == "migrated": + verdict, rule_hit = "migrated_previous", "previous_batch_received" + elif color != "red": + verdict, color, rule_hit = "unfinished_previous", "red", "previous_batch_not_received" + # Retain all drift outside scope, but avoid filling the board with stable + # unmapped inventory when a port map defines the legacy migration scope. + if map_scoped and not in_exp and verdict == "not_involved": + continue if in_exp and verdict == "migrated" and color == "green": progress_ok += 1 if color == "red": @@ -1411,7 +1471,9 @@ def evaluate_metric_dual( "verdict": verdict, "color": color, "rule_hit": rule_hit, + "integrity_changes": integrity_changes, "in_expect": in_exp, + "previous_batch": previous, "old_kind": old_kind, "new_kind": new_kind, "old": old_disp, diff --git a/netx_api/biz_migration/monitor_templates.py b/netx_api/biz_migration/monitor_templates.py index b55f487..cee7ccf 100644 --- a/netx_api/biz_migration/monitor_templates.py +++ b/netx_api/biz_migration/monitor_templates.py @@ -8,7 +8,8 @@ from uuid import uuid4 from fastapi import HTTPException from sqlalchemy.orm import Session -from ..models import BizCompareTemplate, BizMonitorTemplate +from ..models import BizCompareTemplate, BizMigrationProject, BizMonitorTemplate +from .validation import validate_monitor_rules from ..timeutil import utcnow_naive from ..biz_state import compare_service as cmp_svc from ..biz_state.iface_normalize import default_zte_iface_normalize_rules @@ -456,6 +457,7 @@ def create_monitor_template(db: Session, body: dict[str, Any]) -> dict[str, Any] overrides = body.get("sheet_overrides") if not isinstance(overrides, list): overrides = [] + validate_monitor_rules(defaults, overrides) row = BizMonitorTemplate( id=uuid4().hex, name=name[:256], @@ -479,6 +481,10 @@ def update_monitor_template(db: Session, template_id: str, body: dict[str, Any]) row = db.get(BizMonitorTemplate, template_id) if not row: raise HTTPException(status_code=404, detail="monitor_template_not_found") + validate_monitor_rules(body.get("defaults", row.defaults_json or {}), + body.get("sheet_overrides", row.sheet_overrides_json or [])) + if body.get("compare_template_id") and not db.get(BizCompareTemplate, body["compare_template_id"]): + raise HTTPException(status_code=404, detail="compare_template_not_found") if "name" in body and body["name"] is not None: name = str(body["name"] or "").strip() if not name: @@ -515,5 +521,7 @@ def delete_monitor_template(db: Session, template_id: str) -> None: row = db.get(BizMonitorTemplate, template_id) if not row: raise HTTPException(status_code=404, detail="monitor_template_not_found") + if db.query(BizMigrationProject.id).filter(BizMigrationProject.monitor_template_id == template_id).first(): + raise HTTPException(status_code=409, detail="monitor_template_in_use") db.delete(row) db.commit() diff --git a/netx_api/biz_migration/service.py b/netx_api/biz_migration/service.py index 760a6c0..c39183b 100644 --- a/netx_api/biz_migration/service.py +++ b/netx_api/biz_migration/service.py @@ -13,7 +13,6 @@ from ..biz_state.compare_rules import apply_row_filters from ..biz_state.compare_service import ( _load_metric_rows, _port_map_dict, - batch_metric_collect_ok, sheet_key, sheet_title, template_metrics, @@ -28,6 +27,7 @@ from ..models import ( BizMonitorTemplate, BizPortMapping, BizStateBatch, + BizStateBatchCommand, BizStateTask, ) from ..timeutil import utcnow_naive @@ -149,7 +149,7 @@ def hf_task_id(proj: BizMigrationProject, side: str) -> str: def current_task_id(proj: BizMigrationProject, side: str) -> str: """Evaluate current batches: prefer HF, fall back to portrait.""" - return hf_task_id(proj, side) or portrait_task_id(proj, side) + return hf_task_id(proj, side) def _metric_interval_map(proj: BizMigrationProject) -> dict[str, int]: @@ -268,15 +268,54 @@ def _current_batch_for_metric( # Only honor pin when it belongs to this metric's HF task (multi-interval safe) if pinned and (not tid or str(pinned.task_id or "") == tid): # Pinned partial / failed metric → treat as missing (avoid false red) - if not batch_metric_collect_ok(db, pinned.id, metric_id): + if not _metric_sample_ok(db, pinned, metric_id): return None return pinned - batch = _latest_success_batch(db, tid) if tid else None - if batch and not batch_metric_collect_ok(db, batch.id, metric_id): + batch = ( + db.query(BizStateBatch) + .filter(BizStateBatch.task_id == tid, BizStateBatch.status.notin_(("running", "queued"))) + .order_by(BizStateBatch.started_at.desc(), BizStateBatch.id.desc()).first() + ) if tid else None + if batch and not _metric_sample_ok(db, batch, metric_id): return None return batch +def _metric_sample_ok(db: Session, batch: BizStateBatch, metric_id: str) -> bool: + """An empty result requires a successful command receipt, just like a full one. + + Batch success alone cannot prove this metric was collected. Unmatched and + skipped parsers are uncertainty, never evidence that a business disappeared. + """ + if batch.status not in ("success", "partial"): + return False + statuses = [str(s or "").lower() for (s,) in db.query(BizStateBatchCommand.parse_status) + .filter(BizStateBatchCommand.batch_id == batch.id, + BizStateBatchCommand.metric_id == metric_id).all()] + return bool(statuses) and all(s in ("ok", "ok_aux") for s in statuses) + + +def _sample_timing(proj: BizMigrationProject, mid: str, old: BizStateBatch, + new: BizStateBatch, *, check_age: bool) -> dict[str, Any]: + interval = max(60, _metric_interval_map(proj).get(mid) or proj.hf_interval_sec or 60) + old_at, new_at = old.ended_at or old.started_at, new.ended_at or new.started_at + age_limit, skew_limit = interval * 3, interval * 2 + reasons: list[str] = [] + if not old_at or not new_at: + reasons.append("sample_time_missing") + skew = None + else: + skew = abs((old_at - new_at).total_seconds()) + if skew > skew_limit: + reasons.append("sample_skew") + if check_age and max((utcnow_naive() - old_at).total_seconds(), + (utcnow_naive() - new_at).total_seconds()) > age_limit: + reasons.append("sample_stale") + return {"old_collected_at": _dt_iso(old_at), "new_collected_at": _dt_iso(new_at), + "age_limit_sec": age_limit, "skew_limit_sec": skew_limit, + "skew_sec": skew, "sample_issues": reasons} + + def find_portrait_task_for_ne(db: Session, *, source: str, ne_id: str) -> BizStateTask | None: """Best portrait (non-HF) biz_state task for an NE — for baseline binding.""" src = str(source or "managed").strip().lower() or "managed" @@ -655,6 +694,14 @@ def patch_project(db: Session, project_id: str, body: dict[str, Any]) -> dict[st p = db.get(BizMigrationProject, project_id) if not p: raise HTTPException(status_code=404, detail="project_not_found") + # Validate incoming baseline ownership before mutating any project field. + for side in ("old", "new"): + bid = str(body.get(f"{side}_baseline_batch_id", getattr(p, f"{side}_baseline_batch_id")) or "").strip() + tid = str(body.get(f"{side}_task_id", getattr(p, f"{side}_task_id")) or "").strip() + if bid: + sample = db.get(BizStateBatch, bid) + if not sample or sample.task_id != tid or sample.status not in ("success", "partial"): + raise HTTPException(status_code=400, detail=f"{side}_baseline_invalid") if "name" in body and body["name"] is not None: p.name = str(body["name"]).strip() or p.name if "note" in body and body["note"] is not None: @@ -768,11 +815,13 @@ def create_batch(db: Session, project_id: str, body: dict[str, Any]) -> dict[str raise HTTPException(status_code=404, detail="project_not_found") label = str(body.get("batch_label") or "").strip() or "batch" expect = body.get("expect_set") if isinstance(body.get("expect_set"), dict) else {} + status = str(body.get("status") or "pending").strip() + _validate_batch_status(db, project_id, "", status) b = BizMigrationBatch( id=uuid4().hex, project_id=project_id, batch_label=label, - status=str(body.get("status") or "pending").strip() or "pending", + status=status, expect_set_json=dict(expect or {}), note=str(body.get("note") or "")[:500], ) @@ -819,17 +868,35 @@ def run_evaluate( new_batch_id: str = "", acceptance: bool = False, purpose: str = "manual", + _commit: bool = True, ) -> dict[str, Any]: mb = db.get(BizMigrationBatch, batch_id) if not mb: raise HTTPException(status_code=404, detail="batch_not_found") - proj = db.get(BizMigrationProject, mb.project_id) + # PostgreSQL serializes evaluate/finish for this project across API workers. + proj = db.query(BizMigrationProject).filter(BizMigrationProject.id == mb.project_id).with_for_update().populate_existing().first() if not proj: raise HTTPException(status_code=404, detail="project_not_found") + db.refresh(mb) + if acceptance and mb.status not in ("active", "review"): + raise HTTPException(status_code=409, detail="monitor_window_closed") + if purpose == "auto" and (mb.status != "active" or proj.status == "done"): + raise HTTPException(status_code=409, detail="monitor_window_closed") if not proj.old_baseline_batch_id: raise HTTPException(status_code=400, detail="old_baseline_required") if not proj.new_baseline_batch_id: raise HTTPException(status_code=400, detail="new_baseline_required") + for side in ("old", "new"): + baseline = db.get(BizStateBatch, getattr(proj, f"{side}_baseline_batch_id")) + if not baseline or baseline.task_id != portrait_task_id(proj, side): + raise HTTPException(status_code=400, detail=f"{side}_baseline_task_mismatch") + if baseline.status not in ("success", "partial"): + raise HTTPException(status_code=400, detail=f"{side}_baseline_not_complete") + pin = (old_batch_id if side == "old" else new_batch_id).strip() + if pin: + pinned = db.get(BizStateBatch, pin) + if not pinned or pinned.task_id not in _all_hf_task_ids(proj, side): + raise HTTPException(status_code=400, detail=f"{side}_current_batch_task_mismatch") old_batch = db.get(BizStateBatch, old_batch_id.strip()) if old_batch_id.strip() else None new_batch = db.get(BizStateBatch, new_batch_id.strip()) if new_batch_id.strip() else None @@ -854,11 +921,41 @@ def run_evaluate( pin_new = new_batch_id.strip() port_map = _port_map_dict(db, proj.mapping_id) + if proj.mapping_id and not db.get(BizPortMapping, proj.mapping_id): + raise HTTPException(status_code=400, detail="mapping_missing") + if len(set(port_map.values())) != len(port_map): + raise HTTPException(status_code=400, detail="mapping_target_ambiguous") expect = parse_expect_set(mb.expect_set_json if isinstance(mb.expect_set_json, dict) else {}) + previous_expect: dict[str, set[str]] = {} + for prior in db.query(BizMigrationBatch).filter( + BizMigrationBatch.project_id == proj.id, BizMigrationBatch.id != mb.id, + BizMigrationBatch.status.in_(("review", "done")), + BizMigrationBatch.created_at <= mb.created_at, + ).all(): + for mid, keys in parse_expect_set(prior.expect_set_json).items(): + previous_expect.setdefault(mid, set()).update(keys) # Final acceptance: window closed → unfinished expect = red window_active = (mb.status == "active") and (not acceptance) - mt = resolve_project_monitor_template(db, proj) + # Evaluations never seed or commit default configuration inside the lock. + mt = db.get(BizMonitorTemplate, proj.monitor_template_id) if proj.monitor_template_id else None + if not mt: + raise HTTPException(status_code=400, detail="monitor_template_missing") + if mt.compare_template_id and not db.get(BizCompareTemplate, mt.compare_template_id): + raise HTTPException(status_code=400, detail="compare_template_missing") sheets, sheet_overrides, defaults, iface_norm = resolve_evaluate_sheets(db, mt) + from .validation import validate_monitor_rules + + validate_monitor_rules(defaults, sheet_overrides) + from ..biz_state.compare_validation import validate_template_body + + validate_template_body({"metrics": sheets, "iface_normalize_rules": iface_norm}) + known_sheets = {sheet_key(s) for s in sheets} + known_metrics = {str(s.get("metric_id") or "") for s in sheets} + for override in sheet_overrides: + if (override.get("sheet_id") and override["sheet_id"] not in known_sheets) or ( + not override.get("sheet_id") and override.get("metric_id") not in known_metrics + ): + raise HTTPException(status_code=400, detail="monitor_override_sheet_missing") out_of_expect = str((defaults or {}).get("out_of_expect") or "strict").strip().lower() collect_ids = set(resolve_collect_metric_ids(db, proj)) @@ -867,6 +964,16 @@ def run_evaluate( metric_batches: dict[str, dict[str, str]] = {} seq = 0 verdict_counts: dict[str, int] = {} + rows_cache: dict[tuple[str, str], list[dict[str, Any]]] = {} + command_cache: dict[str, dict[str, Any]] = {} + scope_ids = {str(s.get("metric_id") or "") for s in sheets} | {sheet_key(s) for s in sheets} | {"_ports"} + uncovered_scope = sorted(k for k, values in expect.items() if values and k not in scope_ids) + + def metric_rows(bid: str, mid: str) -> list[dict[str, Any]]: + key = (bid, mid) + if key not in rows_cache: + rows_cache[key] = _load_metric_rows(db, batch_id=bid, metric_id=mid) + return rows_cache[key] for sheet in sheets: mid = str(sheet.get("metric_id") or "").strip() @@ -880,6 +987,9 @@ def run_evaluate( field_rules = list(sheet.get("field_rules") or []) sheet_ov = override_for_sheet(sheet_overrides, sheet_id=sid, metric_id=mid) if sheet_ov.get("skip_dual"): + sheet_cards.append({"metric_id": mid, "sheet_id": sid, "title": sheet_title(sheet), + "progress_ok": 0, "progress_total": 0, "anomaly": 0, + "collect_skipped": True}) continue # Metric not in HF collect set → skip dual eval (avoid false red) @@ -927,21 +1037,31 @@ def run_evaluate( continue old_cur_mid = old_cur_batch.id new_cur_mid = new_cur_batch.id + timing = _sample_timing(proj, mid, old_cur_batch, new_cur_batch, + check_age=acceptance or not (pin_old and pin_new)) + baseline_missing = [side for side in ("old", "new") if not _metric_sample_ok( + db, db.get(BizStateBatch, getattr(proj, f"{side}_baseline_batch_id")), mid)] + if timing["sample_issues"] or baseline_missing: + sheet_cards.append({"metric_id": mid, "sheet_id": sid, "title": sheet_title(sheet), + "progress_ok": 0, "progress_total": 0, "anomaly": 0, + "current_missing": True, "collect_incomplete": True, + "baseline_missing_sides": baseline_missing, **timing}) + continue old_base = apply_row_filters( - _load_metric_rows(db, batch_id=proj.old_baseline_batch_id, metric_id=mid), + metric_rows(proj.old_baseline_batch_id, mid), row_filters, ) old_now = apply_row_filters( - _load_metric_rows(db, batch_id=old_cur_mid, metric_id=mid), + metric_rows(old_cur_mid, mid), row_filters, ) new_base_rows = apply_row_filters( - _load_metric_rows(db, batch_id=proj.new_baseline_batch_id, metric_id=mid), + metric_rows(proj.new_baseline_batch_id, mid), row_filters, ) new_now = apply_row_filters( - _load_metric_rows(db, batch_id=new_cur_mid, metric_id=mid), + metric_rows(new_cur_mid, mid), row_filters, ) @@ -963,7 +1083,19 @@ def run_evaluate( sheet_override=sheet_ov, out_of_expect=out_of_expect, iface_normalize_rules=iface_norm, + previous_expect=previous_expect, + allow_mapping_scope=False, ) + # A target snapshot taken before the source's disappearance cannot prove + # loss. Wait for the next target receipt; final acceptance stays blocked. + awaiting = 0 + if (new_cur_batch.ended_at or new_cur_batch.started_at) < old_cur_batch.started_at: + for row in one["rows"]: + if row["in_expect"] and row["rule_hit"] == "service_not_received": + row.update(verdict="awaiting_peer", color="gray", rule_hit="target_sample_predates_source") + one["anomaly"] -= 1 + one["anomaly_in_expect"] -= 1 + awaiting += 1 old_tid = _hf_task_id_for_metric(proj, "old", mid) or current_task_id(proj, "old") new_tid = _hf_task_id_for_metric(proj, "new", mid) or current_task_id(proj, "new") old_task_brief = _task_brief(db, old_tid) @@ -973,6 +1105,7 @@ def run_evaluate( "new_batch_id": new_cur_mid, "old_task_id": old_tid, "new_task_id": new_tid, + **timing, } sheet_cards.append( { @@ -993,6 +1126,8 @@ def run_evaluate( "duplicate_key_list": list(one.get("duplicate_key_list") or []), "duplicate_keys_before": int(one.get("duplicate_keys_before") or 0), "duplicate_keys_after": int(one.get("duplicate_keys_after") or 0), + **timing, + "awaiting_peer": awaiting, } ) for r in one["rows"]: @@ -1003,6 +1138,7 @@ def run_evaluate( new_batch_id=new_cur_mid, old_task=old_task_brief, new_task=new_task_brief, + command_cache=command_cache, ) r["seq"] = seq r["sheet_id"] = sid @@ -1052,6 +1188,13 @@ def run_evaluate( if c.get("collect_skipped") ], "verdict_counts": verdict_counts, + "anomaly_outside_expect": sum(int(c.get("anomaly") or 0) - int(c.get("anomaly_in_expect") or 0) for c in active_cards), + "duplicate_keys": sum(int(c.get("duplicate_keys_before") or 0) + int(c.get("duplicate_keys_after") or 0) for c in active_cards), + "uncovered_scope": uncovered_scope, + "coverage_complete": bool(sheet_cards) and not uncovered_scope and not any(c.get("collect_skipped") or c.get("current_missing") or c.get("awaiting_peer") or c.get("duplicate_keys_before") or c.get("duplicate_keys_after") for c in sheet_cards), + "config_snapshot": {"sheets": sheets, "sheet_overrides": sheet_overrides, + "defaults": defaults, "iface_normalize": iface_norm, + "port_map": port_map, "expect_set": mb.expect_set_json}, "window_active": window_active, "expect_ports": sorted(expect.get("_ports") or ()), "old_current_task_id": current_task_id(proj, "old"), @@ -1098,6 +1241,8 @@ def run_evaluate( "old_status": r.get("old_status"), "new_status": r.get("new_status"), "rule_hit": r.get("rule_hit") or "", + "integrity_changes": r.get("integrity_changes") or [], + "previous_batch": bool(r.get("previous_batch")), "sheet_id": r.get("sheet_id") or "", "evidence": r.get("evidence") or {}, }, @@ -1109,8 +1254,10 @@ def run_evaluate( search_text=search[:2000], ) ) - db.commit() - db.refresh(run) + db.flush() + if _commit: + db.commit() + db.refresh(run) return run_to_dict(db, run) @@ -1147,7 +1294,12 @@ def _command_brief(db: Session, command_id: str) -> dict[str, Any]: cid = str(command_id or "").strip() if not cid: return {} - c = db.get(BizStateBatchCommand, cid) + # Select metadata only: raw CLI output can be several MB per command. + c = db.query(BizStateBatchCommand.id, BizStateBatchCommand.raw_command, + BizStateBatchCommand.parse_status, BizStateBatchCommand.row_count, + BizStateBatchCommand.profile_id, BizStateBatchCommand.parser_id, + BizStateBatchCommand.metric_id, BizStateBatchCommand.message).filter( + BizStateBatchCommand.id == cid).first() if not c: return {"command_id": cid} return { @@ -1170,6 +1322,7 @@ def _enrich_row_evidence( new_batch_id: str, old_task: dict[str, Any] | None, new_task: dict[str, Any] | None, + command_cache: dict[str, dict[str, Any]] | None = None, ) -> None: """Attach device / collect / show-command onto evaluate evidence (in-place).""" ev = dict(row.get("evidence") or {}) @@ -1179,7 +1332,11 @@ def _enrich_row_evidence( ): side_ev = dict(ev.get(side) or {}) netx = dict(side_ev.get("netx") or {}) - cmd = _command_brief(db, str(netx.get("batch_command_id") or "")) + cid = str(netx.get("batch_command_id") or "") + cache = command_cache if command_cache is not None else {} + if cid not in cache: + cache[cid] = _command_brief(db, cid) + cmd = cache[cid] side_ev["device"] = { "ne_id": str(brief.get("ne_id") or netx.get("ne_id") or ""), "ne_name": str(brief.get("ne_name") or ""), @@ -1221,6 +1378,8 @@ def diff_to_dict(d: BizMigrationDiff) -> dict[str, Any]: "old_status": kj.get("old_status") or "", "new_status": kj.get("new_status") or "", "rule_hit": kj.get("rule_hit") or "", + "integrity_changes": kj.get("integrity_changes") or [], + "previous_batch": bool(kj.get("previous_batch")), "evidence": dict(kj.get("evidence") or {}), "old_kind": d.old_kind, "new_kind": d.new_kind, @@ -1922,26 +2081,30 @@ def _persist_red_tickets_from_run( ) .all() ) - by_id: dict[tuple[str, str], BizMigrationRedTicket] = {} + by_id: dict[tuple[str, str, str], BizMigrationRedTicket] = {} for t in existing: + detail = t.detail_json if isinstance(t.detail_json, dict) else {} mk = str(getattr(t, "match_key_str", "") or "").strip() if not mk: detail = t.detail_json if isinstance(t.detail_json, dict) else {} mk = str(detail.get("match_old_key") or t.key_str or "").strip() if mk: - by_id[(str(t.metric_id or ""), mk)] = t + by_id[(str(t.metric_id or ""), str(detail.get("sheet_id") or t.metric_id or ""), + str(detail.get("match_old_key") or mk))] = t created_or_updated: list[BizMigrationRedTicket] = [] - seen: set[tuple[str, str]] = set() + seen: set[tuple[str, str, str]] = set() for d in diffs: kj = d.key_json if isinstance(d.key_json, dict) else {} mid = str(d.metric_id or "") match_key = str( kj.get("match_old_key") or kj.get("key_str") or kj.get("old_key") or "" - ).strip()[:256] + ).strip() if not match_key: match_key = str(kj.get("key_str") or "")[:256] - ident = (mid, match_key) + ident = (mid, str(kj.get("sheet_id") or mid), match_key) + from hashlib import sha256 + stored_key = match_key if len(match_key) <= 256 else "sha256:" + sha256(match_key.encode()).hexdigest() detail = { "old_kind": d.old_kind, "new_kind": d.new_kind, @@ -1952,6 +2115,8 @@ def _persist_red_tickets_from_run( "new_key": kj.get("new_key") or kj.get("new_key_str") or "", "match_old_key": kj.get("match_old_key") or match_key, "match_new_key": kj.get("match_new_key") or "", + "sheet_id": kj.get("sheet_id") or mid, + "integrity_changes": kj.get("integrity_changes") or [], "evidence": dict(kj.get("evidence") or {}), } prev = by_id.get(ident) @@ -1960,7 +2125,7 @@ def _persist_red_tickets_from_run( prev.run_id = run_id prev.key_str = str(kj.get("key_str") or "")[:256] prev.new_key_str = str(kj.get("new_key_str") or "")[:256] - prev.match_key_str = match_key + prev.match_key_str = stored_key prev.verdict = d.verdict prev.color = d.color or "red" prev.old_status = str(kj.get("old_status") or "")[:64] @@ -1980,7 +2145,7 @@ def _persist_red_tickets_from_run( metric_id=mid, key_str=str(kj.get("key_str") or "")[:256], new_key_str=str(kj.get("new_key_str") or "")[:256], - match_key_str=match_key, + match_key_str=stored_key, verdict=d.verdict, color=d.color or "red", old_status=str(kj.get("old_status") or "")[:64], @@ -2014,14 +2179,9 @@ def finish_batch( detail=f"batch_not_finishable:{mb.status}", ) - # Close window first so acceptance uses non-active rules - if mb.status == "active": - mb.ended_at = utcnow_naive() - mb.status = "done" if mark_done else "review" - mb.updated_at = utcnow_naive() - db.commit() - - run = run_evaluate(db, batch_id=batch_id, acceptance=True, purpose="acceptance") + # acceptance=True closes the logical window. Persist batch closure only after + # evaluation succeeds, so an API/collection failure keeps the batch operable. + run = run_evaluate(db, batch_id=batch_id, acceptance=True, purpose="acceptance", _commit=False) summary = dict(run.get("summary") or {}) progress = dict(summary.get("progress") or {}) anomaly = int(summary.get("anomaly") or 0) @@ -2036,6 +2196,10 @@ def finish_batch( fail_reasons.append("new_baseline_missing") if current_missing: fail_reasons.append("current_missing") + if not summary.get("coverage_complete", False): + fail_reasons.append("coverage_incomplete") + if summary.get("duplicate_keys"): + fail_reasons.append("duplicate_keys") if anomaly: fail_reasons.append("anomaly") if total > 0 and ok < total: @@ -2061,6 +2225,9 @@ def finish_batch( mb = db.get(BizMigrationBatch, batch_id) if not mb: raise HTTPException(status_code=404, detail="batch_not_found") + if mb.status == "active": + mb.ended_at = utcnow_naive() + mb.status = "done" if mark_done else "review" mb.accept_run_id = str(run.get("id") or "") mb.accept_status = "passed" if passed else "failed" mb.accept_summary_json = accept_summary @@ -2086,10 +2253,22 @@ def finish_batch( } +def _validate_batch_status(db: Session, project_id: str, batch_id: str, status: str) -> None: + if status not in ("pending", "active", "review", "done"): + raise HTTPException(status_code=400, detail="batch_status_invalid") + if status == "active": + db.query(BizMigrationProject).filter(BizMigrationProject.id == project_id).with_for_update().first() + if db.query(BizMigrationBatch.id).filter(BizMigrationBatch.project_id == project_id, + BizMigrationBatch.id != batch_id, BizMigrationBatch.status == "active").first(): + raise HTTPException(status_code=409, detail="another_batch_active") + + def patch_batch(db: Session, batch_id: str, body: dict[str, Any]) -> dict[str, Any]: b = db.get(BizMigrationBatch, batch_id) if not b: raise HTTPException(status_code=404, detail="batch_not_found") + if body.get("status") is not None: + _validate_batch_status(db, b.project_id, b.id, str(body["status"]).strip()) if "batch_label" in body and body["batch_label"] is not None: b.batch_label = str(body["batch_label"]).strip() or b.batch_label if "note" in body and body["note"] is not None: @@ -2135,6 +2314,8 @@ def list_run_diffs( run_id: str, *, metric_id: str = "", + sheet_id: str = "", + only_expect: bool = False, verdict: str = "", color: str = "", kw: str = "", @@ -2146,19 +2327,23 @@ def list_run_diffs( q = db.query(BizMigrationDiff).filter(BizMigrationDiff.run_id == run_id) if metric_id: q = q.filter(BizMigrationDiff.metric_id == metric_id) + if sheet_id: + q = q.filter(BizMigrationDiff.key_json["sheet_id"].as_string() == sheet_id) + if only_expect: + q = q.filter(BizMigrationDiff.in_expect.is_(True)) if verdict: q = q.filter(BizMigrationDiff.verdict == verdict) if color: q = q.filter(BizMigrationDiff.color == color) if kw.strip(): - like = f"%{kw.strip()}%" - q = q.filter(BizMigrationDiff.search_text.ilike(like)) + literal = kw.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + q = q.filter(BizMigrationDiff.search_text.ilike(f"%{literal}%", escape="\\")) total = q.count() rows = q.order_by(BizMigrationDiff.seq.asc()).offset(max(0, offset)).limit(min(500, max(1, limit))).all() return { "total": total, - "offset": offset, - "limit": limit, + "offset": max(0, offset), + "limit": min(500, max(1, limit)), "items": [diff_to_dict(d) for d in rows], } diff --git a/netx_api/biz_migration/validation.py b/netx_api/biz_migration/validation.py new file mode 100644 index 0000000..881a3b3 --- /dev/null +++ b/netx_api/biz_migration/validation.py @@ -0,0 +1,70 @@ +"""Validate monitor correction rules before they can affect live verdicts.""" +from typing import Any + +from fastapi import HTTPException + +from .evaluate import _groups_for_side, _normalize_cond_type + + +def validate_monitor_rules(defaults: Any, overrides: Any) -> None: + def fail(path: str) -> None: + raise HTTPException(status_code=400, detail=f"monitor_rule_invalid:{path}") + + if not isinstance(defaults, dict) or not isinstance(overrides, list): + fail("defaults/sheet_overrides") + if defaults.get("out_of_expect", "strict") not in ("strict", "warn", "ignore"): + fail("defaults.out_of_expect") + seen = set() + for i, ov in enumerate(overrides): + path = f"sheet_overrides[{i}]" + if not isinstance(ov, dict) or not (ov.get("sheet_id") or ov.get("metric_id")): + fail(path) + ident = ov.get("sheet_id") or ov.get("metric_id") + if ident in seen: + fail(path + ".duplicate") + seen.add(ident) + for key in ("status_fields", "up_values", "down_values"): + if key in ov and (not isinstance(ov[key], list) or any(not str(v).strip() for v in ov[key])): + fail(path + "." + key) + for key in ("success", "anomaly"): + patterns = ov.get(key, []) + if not isinstance(patterns, list): + fail(path + "." + key) + for j, pat in enumerate(patterns): + pp = f"{path}.{key}[{j}]" + if not isinstance(pat, dict): + fail(pp) + for side in ("old", "new"): + for attr in (side, side + "_conds", side + "_groups"): + if attr in pat and not isinstance(pat[attr], list): + fail(pp + "." + attr) + if any(isinstance(g, list) and not g for g in pat.get(side + "_groups", [])): + fail(pp + "." + side + "_groups.empty_group") + groups = _groups_for_side(pat, side) + if key == "success" and not groups: + fail(pp + "." + side + ".required") + for group in groups: + for cond in group: + cp = pp + "." + side + if isinstance(cond, str): + if cond not in ("added", "removed", "changed", "unchanged", "up", "down", "other", "none") and not (cond.startswith("field:") and cond.count(":") >= 2): + fail(cp + ".token") + elif isinstance(cond, dict): + typ = _normalize_cond_type(cond) + if typ not in ("presence", "status", "value"): + fail(cp + ".type") + val = cond.get("value", cond.get("kind", cond.get("status"))) + if typ == "presence" and val not in ("added", "removed", "changed", "unchanged"): + fail(cp + ".presence") + if typ == "status" and val not in ("up", "down", "other", "none"): + fail(cp + ".status") + if typ == "value": + op = cond.get("op", "eq") + if op not in ("eq", "ne", "in", "not_in", "empty", "not_empty", "changed", "unchanged", "same", "diff"): + fail(cp + ".op") + if op in ("eq", "ne", "in", "not_in") and (val is None or val == "" or val == []): + fail(cp + ".value") + else: + fail(cp + ".condition") + if key == "anomaly" and not any(_groups_for_side(pat, s) for s in ("old", "new")): + fail(pp + ".empty") diff --git a/netx_api/biz_migration_router.py b/netx_api/biz_migration_router.py index 29b4377..52508f8 100644 --- a/netx_api/biz_migration_router.py +++ b/netx_api/biz_migration_router.py @@ -285,6 +285,8 @@ def api_get_run(run_id: str, db: Session = Depends(get_db)): def api_list_diffs( run_id: str, metric_id: str = "", + sheet_id: str = "", + only_expect: bool = False, verdict: str = "", color: str = "", kw: str = "", @@ -296,6 +298,8 @@ def api_list_diffs( db, run_id, metric_id=metric_id, + sheet_id=sheet_id, + only_expect=only_expect, verdict=verdict, color=color, kw=kw, diff --git a/netx_api/biz_state/collect_runner.py b/netx_api/biz_state/collect_runner.py index ad04d77..cba6dd9 100644 --- a/netx_api/biz_state/collect_runner.py +++ b/netx_api/biz_state/collect_runner.py @@ -1720,6 +1720,14 @@ def _finalize_batch_status( batch.message = "" status = str(batch.status or "") db.commit() + # Failed/partial receipts also update cutover coverage; never keep an old + # success looking healthy after the latest collection has failed. + try: + from ..biz_migration.auto_monitor import schedule_auto_monitor_for_task + + schedule_auto_monitor_for_task(task_id) + except Exception: + _log.exception("cutover monitor schedule failed task=%s", task_id) # Only full success triggers auto compare; never block the collect thread. if status == "success" and not stopped: try: diff --git a/tests/test_biz_migration_evaluate.py b/tests/test_biz_migration_evaluate.py index 0ac413c..99d650f 100644 --- a/tests/test_biz_migration_evaluate.py +++ b/tests/test_biz_migration_evaluate.py @@ -733,7 +733,7 @@ class EvaluateMetricDualTests(unittest.TestCase): migrated = [r for r in out["rows"] if r["verdict"] == "migrated"] self.assertTrue(migrated, out["rows"]) - def test_out_of_expect_ignore_no_red(self): + def test_out_of_expect_ignore_cannot_hide_business_drift(self): old_base = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}] old_cur: list[dict] = [] new_cur = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}] @@ -753,9 +753,9 @@ class EvaluateMetricDualTests(unittest.TestCase): out_of_expect="ignore", ) reds = [r for r in out["rows"] if r["color"] == "red"] - self.assertFalse(reds, out["rows"]) + self.assertTrue(reds, out["rows"]) involved = [r for r in out["rows"] if r["verdict"] != "not_involved"] - self.assertFalse(involved, out["rows"]) + self.assertTrue(involved, out["rows"]) def test_unexpected_loss_is_anomaly(self): old_base = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}] @@ -823,7 +823,7 @@ class EvaluateMetricDualTests(unittest.TestCase): self.assertFalse(out["new_baseline_missing"]) self.assertEqual(out["new_baseline_mode"], "port_mapped") - def test_empty_provided_new_baseline_falls_back_or_missing(self): + def test_empty_collected_new_baseline_is_valid(self): old_base = [{"peer": "1.1.1.1", "state": "Established"}] new_cur = [{"peer": "1.1.1.1", "state": "Established"}] expect = parse_expect_set({"items": [{"metric_id": "bgp_peer", "key": "1.1.1.1"}]}) @@ -841,8 +841,8 @@ class EvaluateMetricDualTests(unittest.TestCase): window_active=True, sheet_override={}, ) - self.assertTrue(out["new_baseline_missing"]) - self.assertEqual(out["new_baseline_mode"], "empty") + self.assertFalse(out["new_baseline_missing"]) + self.assertEqual(out["new_baseline_mode"], "provided") # Still reports presence as added vs empty before — callers should heed the flag. kinds = {r["new_kind"] for r in out["rows"] if r.get("in_expect")} self.assertIn("added", kinds) @@ -1153,7 +1153,10 @@ class PortMapScopeTests(unittest.TestCase): ) keys = {r["key_str"] for r in out["rows"]} self.assertIn("10.0.0.1|gei-old", keys) - self.assertNotIn("10.0.0.2|gei-keep", keys) + self.assertIn("10.0.0.2|gei-keep", keys) + outside = next(r for r in out["rows"] if r["key_str"] == "10.0.0.2|gei-keep") + self.assertEqual(outside["color"], "red") + self.assertFalse(outside["in_expect"]) migrated = [r for r in out["rows"] if r["verdict"] == "migrated"] self.assertTrue(migrated, out["rows"]) self.assertEqual(migrated[0]["new_key_str"], "10.0.0.1|gei-new") diff --git a/tests/test_biz_migration_evidence.py b/tests/test_biz_migration_evidence.py index e22a259..89a4adb 100644 --- a/tests/test_biz_migration_evidence.py +++ b/tests/test_biz_migration_evidence.py @@ -149,6 +149,8 @@ class EvidencePersistTests(unittest.TestCase): command_count=1, ) self.db.add(b) + self.db.add(BizStateBatchCommand(id=f"receipt-{bid}", batch_id=bid, + metric_id="interface_brief", parse_status="ok", row_count=0)) self.db.commit() return b @@ -233,7 +235,7 @@ class EvidencePersistTests(unittest.TestCase): raw_command="show interface brief", interface="GE-old", ) - # New baseline empty → port_mapped mode uses mapped old + # Successfully collected empty new baseline is valid # Old current: gone (no rows) — migration off old # New current: xgei-new up self._seed_if_row( diff --git a/tests/test_biz_migration_hf_decouple.py b/tests/test_biz_migration_hf_decouple.py index 4fc3c53..ae63da9 100644 --- a/tests/test_biz_migration_hf_decouple.py +++ b/tests/test_biz_migration_hf_decouple.py @@ -14,6 +14,7 @@ from netx_api.models import ( BizMigrationProject, BizMonitorTemplate, BizStateBatch, + BizStateBatchCommand, BizStateTask, ) from netx_api.biz_migration import service as mig @@ -103,6 +104,9 @@ class HfDecoupleFlowTests(unittest.TestCase): row_count=1, ) self.db.add(b) + for mid in ("interface_brief", "arp"): + self.db.add(BizStateBatchCommand(id=f"receipt-{bid}-{mid}", batch_id=bid, + metric_id=mid, parse_status="ok", row_count=0)) self.db.commit() return b @@ -170,7 +174,7 @@ class HfDecoupleFlowTests(unittest.TestCase): batch = mig.create_batch( self.db, pid, {"batch_label": "n1", "expect_set": {"ports": ["gei-0/1"]}} ) - with mock.patch.object(mig, "resolve_evaluate_sheets", return_value=([], [], {}, [])): + with mock.patch.object(mig, "resolve_evaluate_sheets", return_value=([mig.port_sheet_def()], [], {}, [])): out = mig.run_evaluate(self.db, batch_id=batch["id"]) self.assertEqual(out["old_batch_id"], "cur_old") self.assertEqual(out["new_batch_id"], "cur_new") diff --git a/tests/test_biz_migration_integrity.py b/tests/test_biz_migration_integrity.py new file mode 100644 index 0000000..7a36f71 --- /dev/null +++ b/tests/test_biz_migration_integrity.py @@ -0,0 +1,239 @@ +"""Business continuity, wave scope, sample uncertainty and live-monitor regressions.""" +from datetime import timedelta + +import pytest +from fastapi import HTTPException +from sqlalchemy import create_engine, event +from sqlalchemy.orm import sessionmaker + +from netx_api.db import Base +from netx_api.models import (BizCompareTemplate, BizMonitorTemplate, BizMigrationProject, + BizMigrationBatch, BizMigrationRun, BizMigrationDiff, BizStateTask, BizStateBatch, + BizStateBatchCommand, BizStateMetricRow) +from netx_api.timeutil import utcnow_naive +from netx_api.biz_migration import service as svc +from netx_api.biz_migration.auto_monitor import try_auto_monitor_for_task +from netx_api.biz_migration.evaluate import evaluate_metric_dual, parse_expect_set, _eval_field_op +from netx_api.biz_migration.validation import validate_monitor_rules + + +def evaluate(**changes): + args = dict(metric_id="arp", sheet_id="arp", key_fields=["ip"], iface_fields=["port"], + compare_fields=["mac", "port"], old_baseline_rows=[{"ip": "a", "mac": "good", "port": "old"}], + old_current_rows=[], new_baseline_rows=[], + new_current_rows=[{"ip": "a", "mac": "good", "port": "new"}], + port_map={"old": "new"}, expect=parse_expect_set({"items": [{"metric_id": "arp", "key": "a"}]}), + window_active=True, allow_mapping_scope=False) + return evaluate_metric_dual(**{**args, **changes}) + + +def test_target_content_checked_against_old_business(): + out = evaluate(new_current_rows=[{"ip": "a", "mac": "wrong", "port": "new"}]) + assert out["progress_ok"] == 0 + assert out["rows"][0]["rule_hit"] == "service_content_changed" + assert out["rows"][0]["integrity_changes"][0]["field"] == "mac" + + +def test_content_normalization_and_ignore_follow_compare_template(): + out = evaluate(new_current_rows=[{"ip": "a", "mac": "GOOD", "port": "new"}], + field_rules=[{"field": "mac", "normalize": "upper"}]) + assert out["progress_ok"] == 1 + + +def test_down_target_cannot_pass_permissive_success_rule(): + out = evaluate(new_current_rows=[{"ip": "a", "state": "down"}], + sheet_override={"status_fields": ["state"], "up_values": ["up"], "down_values": ["down"], + "success": [{"old": ["removed"], "new": ["added"]}]}) + assert out["rows"][0]["verdict"] == "lost" + assert out["rows"][0]["color"] == "red" + + +def test_mapping_does_not_expand_wave_selection(): + out = evaluate(old_baseline_rows=[{"ip": "a", "port": "old"}, {"ip": "b", "port": "old2"}], + new_current_rows=[{"ip": "a", "port": "new"}], compare_fields=[], + port_map={"old": "new", "old2": "new2"}) + assert out["progress_total"] == 1 + assert next(r for r in out["rows"] if r["old_key"] == "b")["color"] == "red" + + +def test_port_selection_covers_business_with_nonkey_interface(): + out = evaluate(expect=parse_expect_set({"ports": ["old"]})) + assert out["progress_total"] == out["progress_ok"] == 1 + + +@pytest.mark.parametrize("healthy", [True, False]) +def test_previous_wave_remains_protected_without_changing_current_progress(healthy): + out = evaluate(expect={}, previous_expect=parse_expect_set({"items": [{"metric_id": "arp", "key": "a"}]}), + new_current_rows=[{"ip": "a", "mac": "good", "port": "new"}] if healthy else []) + assert out["progress_total"] == 0 + assert out["rows"][0]["verdict"] == ("migrated_previous" if healthy else "lost") + assert out["anomaly"] == (0 if healthy else 1) + + +@pytest.mark.parametrize("value", [0, False]) +def test_zero_false_are_not_blank_in_keys_or_conditions(value): + assert _eval_field_op({"id": value}, "id", "eq", value) + out = evaluate(old_baseline_rows=[{"ip": value}], new_current_rows=[{"ip": value}], + compare_fields=[], expect=parse_expect_set({"items": [{"metric_id": "arp", "key": [str(value).lower()]}]})) + assert out["rows"][0]["old_key"] == str(value).lower() + + +@pytest.fixture +def db(): + engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(engine) + with sessionmaker(bind=engine, expire_on_commit=False, autoflush=False)() as session: + now = utcnow_naive() - timedelta(seconds=1) + sheet = {"metric_id": "arp", "sheet_id": "s", "key_fields": ["ip"], "compare_fields": ["mac"]} + session.add(BizCompareTemplate(id="ct", name="cmp", metrics_json=[sheet])) + session.add(BizMonitorTemplate(id="mt", name="mon", compare_template_id="ct", + collect_metric_ids_json=["arp"], defaults_json={}, sheet_overrides_json=[])) + for tid in ("op", "np", "oh", "nh"): + session.add(BizStateTask(id=tid, ne_id=tid, ne_name=tid, purpose="cutover_hf" if tid.endswith("h") else "")) + session.add(BizMigrationProject(id="p", name="p", old_task_id="op", new_task_id="np", + old_hf_task_id="oh", new_hf_task_id="nh", old_baseline_batch_id="ob", new_baseline_batch_id="nb", + monitor_template_id="mt", hf_interval_sec=60, status="active")) + session.add(BizMigrationBatch(id="wave", project_id="p", batch_label="wave", status="active", + expect_set_json={"items": [{"metric_id": "arp", "sheet_id": "s", "key": "a"}]})) + for i, (bid, tid) in enumerate((("ob", "op"), ("nb", "np"), ("oc", "oh"), ("nc", "nh"))): + at = now - timedelta(milliseconds=100 * (4 - i)) + session.add(BizStateBatch(id=bid, task_id=tid, status="success", started_at=at, ended_at=at)) + session.add(BizStateBatchCommand(id="cmd-" + bid, batch_id=bid, metric_id="arp", parse_status="ok", raw_text="secret CLI payload")) + for bid in ("ob", "nc"): + session.add(BizStateMetricRow(id="row-" + bid, batch_id=bid, task_id="op" if bid == "ob" else "nh", + batch_command_id="cmd-" + bid, metric_id="arp", seq=0, data_json={"ip": "a", "mac": "good"})) + session.commit() + yield session + + +def test_collected_empty_baseline_can_pass_acceptance(db): + finished = svc.finish_batch(db, "wave") + assert finished["accept_summary"]["passed"] + assert finished["run"]["summary"]["config_snapshot"]["expect_set"] + + +@pytest.mark.parametrize("problem", ["missing_receipt", "failed_receipt", "stale", "skew", "duplicate", "skipped", "unknown_scope"]) +def test_uncertain_coverage_cannot_pass_acceptance(db, problem): + if problem == "missing_receipt": + db.delete(db.get(BizStateBatchCommand, "cmd-nb")) + elif problem == "failed_receipt": + db.get(BizStateBatchCommand, "cmd-nc").parse_status = "failed" + elif problem in ("stale", "skew"): + batch = db.get(BizStateBatch, "nc") + batch.started_at = batch.ended_at = utcnow_naive() - timedelta(minutes=10) + elif problem == "duplicate": + db.add(BizStateMetricRow(id="dup", batch_id="nc", metric_id="arp", data_json={"ip": "a", "mac": "good"})) + elif problem == "skipped": + db.get(BizMigrationProject, "p").collect_metric_ids_json = ["other"] + else: + db.get(BizMigrationBatch, "wave").expect_set_json = {"items": [{"sheet_id": "deleted", "metric_id": "arp", "key": "a"}]} + db.commit() + assert not svc.finish_batch(db, "wave")["accept_summary"]["passed"] + + +def test_invalid_baseline_and_pinned_batch_do_not_fallback(db): + with pytest.raises(HTTPException) as error: + svc.run_evaluate(db, batch_id="wave", old_batch_id="nb") + assert error.value.detail == "old_current_batch_task_mismatch" + db.get(BizMigrationProject, "p").old_baseline_batch_id = "nc" + db.commit() + with pytest.raises(HTTPException): + svc.finish_batch(db, "wave") + assert db.get(BizMigrationBatch, "wave").status == "active" + assert db.query(BizMigrationRun).count() == 0 + + +def test_latest_failed_collection_does_not_reuse_previous_success(db): + db.add(BizStateBatch(id="latest-failed", task_id="nh", status="failed", started_at=utcnow_naive(), ended_at=utcnow_naive())) + db.commit() + run = svc.run_evaluate(db, batch_id="wave") + assert run["summary"]["current_missing_metrics"] == ["s"] + assert not run["summary"]["coverage_complete"] + + +def test_target_older_than_source_is_uncertain_instead_of_false_loss(db): + db.delete(db.get(BizStateMetricRow, "row-nc")) + db.get(BizStateBatch, "oc").started_at = utcnow_naive() + db.get(BizStateBatch, "oc").ended_at = utcnow_naive() + db.commit() + run = svc.run_evaluate(db, batch_id="wave") + row = svc.list_run_diffs(db, run["id"])["items"][0] + assert row["verdict"] == "awaiting_peer" + assert run["summary"]["anomaly"] == 0 + assert not run["summary"]["coverage_complete"] + + +def test_auto_monitor_deduplicates_receipts_and_respects_closed_window(db): + assert try_auto_monitor_for_task(db, "oh") == 1 + assert try_auto_monitor_for_task(db, "nh") == 0 + assert db.query(BizMigrationRun).one().purpose == "auto" + db.get(BizMigrationBatch, "wave").status = "done" + db.commit() + assert try_auto_monitor_for_task(db, "oh") == 0 + + +def test_two_active_waves_are_rejected(db): + with pytest.raises(HTTPException) as error: + svc.create_batch(db, "p", {"status": "active", "batch_label": "overlap"}) + assert error.value.detail == "another_batch_active" + assert db.query(BizMigrationBatch).count() == 1 + + +def test_sheets_with_same_key_keep_separate_red_tickets(db): + run = svc.run_evaluate(db, batch_id="wave") + for sid in ("one", "two"): + db.add(BizMigrationDiff(id=sid, run_id=run["id"], metric_id="arp", color="red", + key_json={"sheet_id": sid, "match_old_key": "a"}, verdict="lost")) + db.flush() + tickets = svc._persist_red_tickets_from_run(db, project_id="p", batch_id="wave", run_id=run["id"]) + assert len(tickets) == 2 + assert {t.detail_json["sheet_id"] for t in tickets} == {"one", "two"} + + +def test_port_mapping_does_not_remap_business_key_with_same_text(): + out = evaluate(old_baseline_rows=[{"ip": "old", "port": "old"}], + new_current_rows=[{"ip": "old", "port": "new"}], compare_fields=["port"], + expect=parse_expect_set({"items": [{"metric_id": "arp", "key": "old"}]})) + assert out["progress_ok"] == 1 + + +def test_wave_acceptance_rolls_back_whole_transaction_on_ticket_failure(db): + from unittest.mock import patch + + with patch.object(svc, "_persist_red_tickets_from_run", side_effect=RuntimeError("write failed")): + with pytest.raises(RuntimeError): + svc.finish_batch(db, "wave") + db.rollback() + assert db.query(BizMigrationRun).count() == 0 + assert db.get(BizMigrationBatch, "wave").status == "active" + + +def test_command_metadata_is_cached_and_raw_cli_is_not_selected(db): + statements = [] + def capture(_conn, _cursor, statement, *_args): + statements.append(statement) + event.listen(db.bind, "before_cursor_execute", capture) + try: + run = svc.run_evaluate(db, batch_id="wave") + finally: + event.remove(db.bind, "before_cursor_execute", capture) + command_selects = [q for q in statements if q.startswith("SELECT") and "biz_state_batch_command" in q] + assert all("raw_text" not in q for q in command_selects) + assert run["summary"]["coverage_complete"] + + +def test_diff_filters_apply_before_pagination_and_escape_wildcards(db): + run = svc.run_evaluate(db, batch_id="wave") + db.add(BizMigrationDiff(id="outside", run_id=run["id"], metric_id="arp", seq=999, color="red", + key_json={"sheet_id": "outside"}, in_expect=False, search_text="a_100%")) + db.commit() + result = svc.list_run_diffs(db, run["id"], sheet_id="outside", color="red", limit=1) + assert result["total"] == 1 and result["items"][0]["id"] == "outside" + assert svc.list_run_diffs(db, run["id"], only_expect=True)["total"] == 1 + assert svc.list_run_diffs(db, run["id"], kw="_")["total"] == 1 + + +@pytest.mark.parametrize("condition", [{"type": "value", "field": "state", "op": "typo", "value": "up"}, {"type": "presence", "value": "typo"}]) +def test_invalid_correction_rules_fail_before_evaluation(condition): + with pytest.raises(HTTPException): + validate_monitor_rules({}, [{"metric_id": "arp", "success": [{"old_groups": [[condition]], "new": ["added"]}]}])