strepsiades
212 lines
7.2 KiB
Python
212 lines
7.2 KiB
Python
"""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())
|