first commit
This commit is contained in:
strepsiades
@@ -0,0 +1,211 @@
|
||||
"""Analyze whether state definitions are uniquely identifiable by field signatures.
|
||||
|
||||
This script is intentionally placed under scripts/ (but is NOT a pytest test module)
|
||||
so it can be used as a repeatable, versioned check.
|
||||
|
||||
It expands each state's possible value set for:
|
||||
(binding_status, action, status) and optionally data_id_exist.
|
||||
|
||||
Wildcard semantics:
|
||||
- '*' means "any enum value" for enum fields.
|
||||
- list means "any of these values".
|
||||
|
||||
For data_id_exist (optional):
|
||||
- If present: bool | '*' | [bool, ...]
|
||||
- If missing: treated as unknown by default, or as wildcard with --data-id-missing-wildcard.
|
||||
|
||||
Usage:
|
||||
python scripts/analyze_state_signatures.py
|
||||
python scripts/analyze_state_signatures.py --with-data-id
|
||||
python scripts/analyze_state_signatures.py --fail-on-overlap
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_CFG = REPO_ROOT / "sync_state_machine" / "config" / "node_state_machine.yaml"
|
||||
|
||||
# Ensure repository root is importable when running as a script.
|
||||
if str(REPO_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Collision:
|
||||
signature: Tuple[Any, ...]
|
||||
states: Tuple[str, ...]
|
||||
|
||||
|
||||
def _load_yaml(path: Path) -> Dict[str, Any]:
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f)
|
||||
if not isinstance(data, dict):
|
||||
raise TypeError(f"Top-level YAML must be a mapping: {path}")
|
||||
return data
|
||||
|
||||
|
||||
def _import_enum_names() -> Tuple[Set[str], Set[str], Set[str]]:
|
||||
# Validate enum NAMES; values are implementation details.
|
||||
from sync_state_machine.common.types import BindingStatus, SyncAction, SyncStatus
|
||||
|
||||
binding = {e.name for e in BindingStatus}
|
||||
action = {e.name for e in SyncAction}
|
||||
status = {e.name for e in SyncStatus}
|
||||
return binding, action, status
|
||||
|
||||
|
||||
def _expand_enum_field(value: Any, *, all_values: Set[str], field_path: str) -> Set[str]:
|
||||
if isinstance(value, str):
|
||||
if value == "*":
|
||||
return set(all_values)
|
||||
return {value}
|
||||
if isinstance(value, list):
|
||||
expanded: Set[str] = set()
|
||||
for i, item in enumerate(value):
|
||||
if not isinstance(item, str):
|
||||
raise TypeError(f"{field_path}[{i}] must be a string")
|
||||
if item == "*":
|
||||
expanded |= set(all_values)
|
||||
else:
|
||||
expanded.add(item)
|
||||
return expanded
|
||||
raise TypeError(f"{field_path} must be str | list[str]")
|
||||
|
||||
|
||||
def _expand_data_id_exist(
|
||||
value: Any,
|
||||
*,
|
||||
field_path: str,
|
||||
missing_as_wildcard: bool,
|
||||
) -> Set[Optional[bool]]:
|
||||
if value is None:
|
||||
return {True, False} if missing_as_wildcard else {None}
|
||||
if value == "*":
|
||||
return {True, False}
|
||||
if isinstance(value, bool):
|
||||
return {value}
|
||||
if isinstance(value, list):
|
||||
out: Set[Optional[bool]] = set()
|
||||
for i, item in enumerate(value):
|
||||
if item == "*":
|
||||
out |= {True, False}
|
||||
elif isinstance(item, bool):
|
||||
out.add(item)
|
||||
else:
|
||||
raise TypeError(f"{field_path}[{i}] must be bool or '*'")
|
||||
return out
|
||||
raise TypeError(f"{field_path} must be bool | '*' | list[bool]")
|
||||
|
||||
|
||||
def _cartesian_product(*sets: Sequence[Any]) -> Iterable[Tuple[Any, ...]]:
|
||||
if not sets:
|
||||
return []
|
||||
acc: List[Tuple[Any, ...]] = [()]
|
||||
for s in sets:
|
||||
nxt: List[Tuple[Any, ...]] = []
|
||||
for prefix in acc:
|
||||
for v in s:
|
||||
nxt.append(prefix + (v,))
|
||||
acc = nxt
|
||||
return acc
|
||||
|
||||
|
||||
def _collect_state_signature_sets(
|
||||
cfg: Mapping[str, Any],
|
||||
*,
|
||||
with_data_id: bool,
|
||||
data_id_missing_as_wildcard: bool,
|
||||
) -> Dict[str, Set[Tuple[Any, ...]]]:
|
||||
states = cfg.get("states")
|
||||
if not isinstance(states, dict):
|
||||
raise TypeError("cfg.states must be a mapping")
|
||||
|
||||
binding_all, action_all, status_all = _import_enum_names()
|
||||
|
||||
out: Dict[str, Set[Tuple[Any, ...]]] = {}
|
||||
for sid, sdef in states.items():
|
||||
if not isinstance(sdef, dict):
|
||||
continue
|
||||
|
||||
binding = _expand_enum_field(sdef.get("binding_status"), all_values=binding_all, field_path=f"states.{sid}.binding_status")
|
||||
action = _expand_enum_field(sdef.get("action"), all_values=action_all, field_path=f"states.{sid}.action")
|
||||
status = _expand_enum_field(sdef.get("status"), all_values=status_all, field_path=f"states.{sid}.status")
|
||||
|
||||
if with_data_id:
|
||||
data_id = _expand_data_id_exist(
|
||||
sdef.get("data_id_exist"),
|
||||
field_path=f"states.{sid}.data_id_exist",
|
||||
missing_as_wildcard=data_id_missing_as_wildcard,
|
||||
)
|
||||
signatures = set(_cartesian_product(sorted(binding), sorted(action), sorted(status), sorted(data_id, key=lambda x: str(x))))
|
||||
else:
|
||||
signatures = set(_cartesian_product(sorted(binding), sorted(action), sorted(status)))
|
||||
|
||||
out[str(sid)] = signatures
|
||||
|
||||
return out
|
||||
|
||||
|
||||
def _find_collisions(state_signatures: Mapping[str, Set[Tuple[Any, ...]]]) -> List[Collision]:
|
||||
index: Dict[Tuple[Any, ...], List[str]] = {}
|
||||
for sid, sigs in state_signatures.items():
|
||||
for sig in sigs:
|
||||
index.setdefault(sig, []).append(sid)
|
||||
|
||||
collisions: List[Collision] = []
|
||||
for sig, sids in index.items():
|
||||
if len(sids) > 1:
|
||||
collisions.append(Collision(signature=sig, states=tuple(sorted(sids))))
|
||||
|
||||
collisions.sort(key=lambda c: (-len(c.states), c.signature))
|
||||
return collisions
|
||||
|
||||
|
||||
def main(argv: Optional[Sequence[str]] = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="Analyze state signature overlaps")
|
||||
parser.add_argument("--config", default=str(DEFAULT_CFG), help="Path to node_state_machine.yaml")
|
||||
parser.add_argument("--with-data-id", action="store_true", help="Include data_id_exist in signature")
|
||||
parser.add_argument(
|
||||
"--data-id-missing-wildcard",
|
||||
action="store_true",
|
||||
help="If data_id_exist is missing, treat it as wildcard (True/False) instead of unknown",
|
||||
)
|
||||
parser.add_argument("--fail-on-overlap", action="store_true", help="Exit non-zero if collisions exist")
|
||||
parser.add_argument("--show", type=int, default=30, help="Max collisions to print")
|
||||
args = parser.parse_args(list(argv) if argv is not None else None)
|
||||
|
||||
cfg = _load_yaml(Path(args.config))
|
||||
state_sigs = _collect_state_signature_sets(
|
||||
cfg,
|
||||
with_data_id=bool(args.with_data_id),
|
||||
data_id_missing_as_wildcard=bool(args.data_id_missing_wildcard),
|
||||
)
|
||||
collisions = _find_collisions(state_sigs)
|
||||
|
||||
dim = 4 if args.with_data_id else 3
|
||||
print(f"Config: {args.config}")
|
||||
print(f"Signature dims: {dim} ({'binding,action,status,data_id' if args.with_data_id else 'binding,action,status'})")
|
||||
print(f"States analyzed: {len(state_sigs)}")
|
||||
print(f"Collisions: {len(collisions)}")
|
||||
|
||||
if collisions:
|
||||
print("\nTop collisions:")
|
||||
for c in collisions[: args.show]:
|
||||
print(f" {c.signature} -> {', '.join(c.states)}")
|
||||
|
||||
if args.fail_on_overlap and collisions:
|
||||
return 2
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user