test

backend/src/api/nft_manager.py

@@ -175,39 +175,39 @@ mgr = NftManager()
 
 # ---------- Models ----------
 class RawCmdRequest(BaseModel):
-    cmd: str = Field(...)
+    cmd: str = Field(..., description="Textual nft command to execute")
 
 class ExecResult(BaseModel):
-    rc: int
+    rc: int = Field(..., description="Return code from nft execution")
-    stdout: Optional[str]
+    stdout: Optional[str] = Field(None)
-    stderr: Optional[str]
+    stderr: Optional[str] = Field(None)
 
 
 class RuleOut(BaseModel):
-    handle: Optional[int]
+    handle: Optional[int] = Field(None)
-    expr: Any
+    expr: Any = Field(..., description="Machine-readable nft expression (original nft JSON expr).")
-    text: str
+    text: str = Field(..., description="Deterministic short display string derived from expr")
-    position: Optional[Any]
+    position: Optional[Any] = Field(None)
-    comment: Optional[str]
+    comment: Optional[str] = Field(None)
 
 
 class ChainOut(BaseModel):
-    name: str
+    name: str = Field(...)
-    type: Optional[str]
+    type: Optional[str] = Field(None)
-    hook: Optional[str]
+    hook: Optional[str] = Field(None)
-    priority: Optional[int]
+    priority: Optional[int] = Field(None)
-    policy: Optional[str]
+    policy: Optional[str] = Field(None)
-    rules: List[RuleOut]
+    rules: List[RuleOut] = Field(...)
 
 
 class TableOut(BaseModel):
-    family: str
+    family: str = Field(...)
-    name: str
+    name: str = Field(...)
-    chains: List[ChainOut]
+    chains: List[ChainOut] = Field(...)
 
 
 class RulesetModel(BaseModel):
-    tables: List[TableOut]
+    tables: List[TableOut] = Field(...)
 
 
 class CreateRuleRequest(BaseModel):
@@ -229,6 +229,7 @@ class RulesetOut(BaseModel):
 # ---------- Helpers ----------
 _handle_re = re.compile(r"\s+#\s*handle\s+\d+\s*$")
 
+
 def parse_priority(val: Any) -> Optional[int]:
     if val is None:
         return None
@@ -253,6 +254,7 @@ def parse_priority(val: Any) -> Optional[int]:
                 return p
     return None
 
+
 def rule_text_from_expr(expr: Any) -> str:
     if expr is None:
         return ""
@@ -370,6 +372,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
                 tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}})
                 chains_map = tables[(fam, table_name)]["chains"]
                 chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []})
+                # we keep expr exactly as provided from JSON; do not replace it
                 rule_obj: Dict[str, Any] = {"handle": handle, "expr": expr, "text": rule_text_from_expr(expr)}
                 if "position" in r:
                     rule_obj["position"] = r["position"]
@@ -407,7 +410,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
     return result
 
 
-# ---------- Text parsing helpers ----------
+# ---------- Text parsing helpers (enrichment only) ----------
 def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]]:
     result: Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]] = {}
     if not nft_text:
@@ -478,8 +481,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
             key = (fam, tname, cname)
             textual_entries = parsed.get(key, [])
             if not textual_entries:
-                # no textual lines for this chain; skip
+                continue  # nothing to enrich for this chain
-                continue
 
             handle_map: Dict[int, str] = {}
             ordered_lines: List[str] = []
@@ -493,6 +495,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
             rules = chain.get("rules", [])
             for idx, rule in enumerate(rules):
                 replaced = False
+                # ensure we don't remove/overwrite expr
                 h = rule.get("handle")
                 if isinstance(h, int) and h in handle_map:
                     rule["text"] = handle_map[h]
@@ -516,7 +519,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
                                 rule["text"] = ln
                                 replaced = True
                                 break
-                # only update when a match was found; otherwise keep JSON-derived text
+                # only update 'text' when a match was found; otherwise leave existing rule['text']
 
 
 def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
@@ -533,19 +536,20 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
             try:
                 chain_text = mgr.list_chain_text(fam, tname, cname) or ""
                 lines = [ln.rstrip() for ln in chain_text.splitlines() if ln.strip() != ""]
-                handle_map: Dict[str, str] = {}
+                handle_map: Dict[int, str] = {}
                 for ln in lines:
                     m = re.search(r"\bhandle\s+(\d+)\b", ln)
                     if m:
-                        handle_map[m.group(1)] = ln.strip()
+                        try:
+                            handle_map[int(m.group(1))] = ln.strip()
+                        except Exception:
+                            continue
 
                 for rule in ch.get("rules", []):
                     replaced = False
                     h = rule.get("handle")
-                    if h is not None:
+                    if isinstance(h, int) and h in handle_map:
-                        key = str(h)
+                        rule["text"] = handle_map[h]
-                        if key in handle_map:
-                            rule["text"] = handle_map[key]
                         replaced = True
 
                     if not replaced:
@@ -568,6 +572,58 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
                 continue
 
 
+# ---------- Normalization helper to satisfy Pydantic model ----------
+def normalize_custom_for_model(custom: Dict[str, Any]) -> None:
+    """
+    Ensure every rule has required keys/types for RulesetModel parsing:
+      - 'expr' must exist (if missing/None -> set to [])
+      - 'text' must exist and be a str (if missing -> derive via rule_text_from_expr or "")
+      - 'handle' coerced to int or None
+    This mutates `custom` in-place.
+    """
+    tables = custom.get("tables") or []
+    for t in tables:
+        for ch in t.get("chains", []):
+            rules = ch.get("rules", []) or []
+            for r in rules:
+                # expr: required by model; prefer existing value, but don't leave None
+                if "expr" not in r or r.get("expr") is None:
+                    # default to empty list (acceptable Any value and consistent with prior shape)
+                    r["expr"] = []
+
+                # text: must be str
+                txt = r.get("text")
+                if txt is None:
+                    try:
+                        r["text"] = rule_text_from_expr(r.get("expr"))
+                    except Exception:
+                        r["text"] = ""
+                else:
+                    # coerce to str if not string
+                    if not isinstance(txt, str):
+                        try:
+                            r["text"] = str(txt)
+                        except Exception:
+                            r["text"] = ""
+
+                # handle: ensure int or None
+                h = r.get("handle")
+                if isinstance(h, str):
+                    try:
+                        r["handle"] = int(h)
+                    except Exception:
+                        r["handle"] = None
+                elif not isinstance(h, int):
+                    # if it's e.g. float, bool, etc. convert if reasonable
+                    try:
+                        if isinstance(h, (float,)):
+                            r["handle"] = int(h)
+                        else:
+                            r["handle"] = None
+                    except Exception:
+                        r["handle"] = None
+
+
 # ---------- Routes ----------
 @router.get("/rules", response_model=RulesetOut, summary="List ruleset")
 def list_rules():
@@ -578,27 +634,33 @@ def list_rules():
             logger.debug("could not obtain nft JSON ruleset: %s", e)
             raise HTTPException(status_code=500, detail=f"Failed to obtain nft JSON ruleset: {e}")
 
+        # get textual snapshot (best-effort), used only to enrich 'text' values
         nft_text = ""
         try:
             nft_text = mgr.list_rules_text()
         except Exception:
             logger.debug("could not obtain textual nft ruleset (list_rules_text failed)")
 
+        # build predictable JSON-derived structure (must be the source of truth)
         custom = build_predictable_ruleset(nft_json)
 
-        # Enrich rule['text'] using the full textual ruleset (handle-first, position, index)
+        # enrich existing JSON rules with textual lines (handle -> line; position/index fallback)
         try:
             if nft_text:
                 populate_text_from_ruleset_text(custom, nft_text)
         except Exception as e:
             logger.debug("list_rules: populate_text_from_ruleset_text failed: %s", e)
 
-        # per-chain fallback (will not fail overall listing)
+        # per-chain fallback that queries nft for chain-level textual lines
         try:
             populate_text_from_chain_text(custom)
         except Exception as e:
             logger.debug("list_rules: populate_text_from_chain_text failed: %s", e)
 
+        # Normalize to satisfy Pydantic model (do this last)
+        normalize_custom_for_model(custom)
+
+        # Debug: log counts
         num_tables = len(custom.get("tables", []))
         num_rules = sum(len(ch.get("rules", [])) for t in custom.get("tables", []) for ch in t.get("chains", []))
         logger.info("list_rules: returning tables=%d rules=%d", num_tables, num_rules)
@@ -627,7 +689,12 @@ def create_rule_json(req: CreateRuleRequest):
         if req.expr is None:
             raise NftError("field 'expr' is required for JSON rule creation")
 
-        rendered = expr_to_text(req.expr)
+        rendered = None
+        try:
+            rendered = rule_text_from_expr(req.expr)
+        except Exception:
+            rendered = None
+
         if rendered is None:
             raise NftError(
                 "cannot render provided 'expr' to textual nft syntax. "