194 lines
8.6 KiB
Python
Executable File
194 lines
8.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ALLOWED_TYPES = {
|
|
"PERFORMANCE", "CODE_STRUCTURE", "MODULE_STRUCTURE", "ARCHITECTURE",
|
|
"DATA_ACCESS", "RELIABILITY", "CONCURRENCY", "TRANSACTION",
|
|
"SECURITY", "OPERABILITY", "CONFIGURATION", "DEPENDENCY",
|
|
"BUILD", "TESTABILITY", "CLEANUP",
|
|
}
|
|
ALLOWED_SCOPES = {"LOCAL", "MODULE", "CROSS_MODULE", "PROJECT"}
|
|
ALLOWED_STATUSES = {"CANDIDATE", "READY", "BASELINING", "IN_PROGRESS", "VERIFYING", "WAITING_APPROVAL", "APPROVED", "MERGED", "REJECTED", "BLOCKED", "COMPLETE"}
|
|
PERFORMANCE_BASELINE_STATUSES = {"IN_PROGRESS", "VERIFYING", "WAITING_APPROVAL", "APPROVED", "MERGED", "COMPLETE"}
|
|
PERFORMANCE_COMPLETE_EVIDENCE_STATUSES = {"WAITING_APPROVAL", "APPROVED", "MERGED", "COMPLETE"}
|
|
|
|
|
|
def _load(item_dir: Path) -> dict:
|
|
path = item_dir / "work-item.json"
|
|
if not path.exists():
|
|
raise FileNotFoundError(path)
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _resolve(item_dir: Path, rel: str | None) -> Path | None:
|
|
if not rel:
|
|
return None
|
|
return item_dir / rel
|
|
|
|
|
|
def _comparison_fields(text: str) -> dict[str, str]:
|
|
labels = (
|
|
"Same measurement command/procedure",
|
|
"Same metric definitions",
|
|
"Same dataset/load profile",
|
|
"Environment materially equivalent",
|
|
"Result",
|
|
"Acceptance criteria satisfied",
|
|
)
|
|
values: dict[str, str] = {}
|
|
for raw in text.splitlines():
|
|
stripped = raw.strip().lstrip("- ")
|
|
for label in labels:
|
|
prefix = label + ":"
|
|
if stripped.startswith(prefix):
|
|
values[label] = stripped[len(prefix):].strip().upper()
|
|
return values
|
|
|
|
|
|
def verify_work_item(item_dir: Path) -> list[str]:
|
|
item_dir = Path(item_dir)
|
|
errors: list[str] = []
|
|
try:
|
|
data = _load(item_dir)
|
|
except (FileNotFoundError, json.JSONDecodeError) as exc:
|
|
return [f"invalid work-item.json: {exc}"]
|
|
|
|
required = (
|
|
"schemaVersion", "id", "project", "analysisRevision", "type", "scope",
|
|
"target", "priority", "status", "problem", "goal", "acceptanceCriteria", "evidence",
|
|
)
|
|
for key in required:
|
|
if key not in data:
|
|
errors.append(f"missing work item field: {key}")
|
|
|
|
item_type = data.get("type")
|
|
if item_type not in ALLOWED_TYPES:
|
|
errors.append(f"invalid type: {item_type}")
|
|
scope = data.get("scope")
|
|
if scope not in ALLOWED_SCOPES:
|
|
errors.append(f"invalid scope: {scope}")
|
|
status = data.get("status")
|
|
if status not in ALLOWED_STATUSES:
|
|
errors.append(f"invalid status: {status}")
|
|
|
|
if item_type == "PERFORMANCE":
|
|
contract = data.get("measurementContract")
|
|
if not isinstance(contract, dict):
|
|
errors.append("performance baseline measurement contract is required before refactoring")
|
|
else:
|
|
for key in ("command", "cwd", "environment", "dataset", "metrics"):
|
|
value = contract.get(key)
|
|
if value in (None, "", []):
|
|
errors.append(f"performance baseline measurement contract missing: {key}")
|
|
environment = _resolve(item_dir, contract.get("environment"))
|
|
if environment is not None and not environment.exists():
|
|
errors.append(f"performance environment evidence missing: {contract.get('environment')}")
|
|
|
|
evidence = data.get("evidence") or {}
|
|
baseline = evidence.get("baseline") or []
|
|
after = evidence.get("after") or []
|
|
comparison = evidence.get("comparison")
|
|
|
|
def validate_metadata(phase: str, require: bool) -> dict | None:
|
|
meta_path = item_dir / f"evidence/{phase}/metadata.json"
|
|
if not meta_path.exists():
|
|
if require:
|
|
errors.append(f"performance {phase} metadata is required")
|
|
return None
|
|
try:
|
|
meta = json.loads(meta_path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError as exc:
|
|
errors.append(f"performance {phase} metadata invalid: {exc}")
|
|
return None
|
|
for key in ("sourceRevision", "command", "cwd", "exitCode", "dataset", "metrics", "rawFiles"):
|
|
if key not in meta or meta.get(key) in (None, "", []):
|
|
if key == "exitCode" and meta.get(key) == 0:
|
|
continue
|
|
errors.append(f"performance {phase} metadata missing: {key}")
|
|
if isinstance(contract, dict):
|
|
for key in ("command", "cwd", "dataset", "metrics"):
|
|
if meta.get(key) != contract.get(key):
|
|
errors.append(f"performance {phase} metadata {key} differs from measurement contract")
|
|
if meta.get("exitCode") not in (0,):
|
|
errors.append(f"performance {phase} measurement exitCode is not zero")
|
|
for rel in meta.get("rawFiles") or []:
|
|
if not (meta_path.parent / rel).exists():
|
|
errors.append(f"performance {phase} metadata raw file missing: {rel}")
|
|
return meta
|
|
|
|
if status in PERFORMANCE_BASELINE_STATUSES:
|
|
if not baseline:
|
|
errors.append("performance baseline evidence is required before source changes")
|
|
else:
|
|
for rel in baseline:
|
|
if not (item_dir / rel).exists():
|
|
errors.append(f"performance baseline evidence missing: {rel}")
|
|
baseline_meta = validate_metadata("baseline", True)
|
|
if baseline_meta is not None and baseline_meta.get("sourceRevision") != data.get("analysisRevision"):
|
|
errors.append("performance baseline metadata sourceRevision differs from analysisRevision")
|
|
|
|
if status in PERFORMANCE_COMPLETE_EVIDENCE_STATUSES:
|
|
if not after:
|
|
errors.append("performance after evidence is required")
|
|
else:
|
|
for rel in after:
|
|
if not (item_dir / rel).exists():
|
|
errors.append(f"performance after evidence missing: {rel}")
|
|
validate_metadata("after", True)
|
|
if not comparison:
|
|
errors.append("performance comparison evidence is required")
|
|
elif not (item_dir / comparison).exists():
|
|
errors.append(f"performance comparison evidence missing: {comparison}")
|
|
else:
|
|
comparison_text = (item_dir / comparison).read_text(encoding="utf-8", errors="replace")
|
|
fields = _comparison_fields(comparison_text)
|
|
required_comparison_fields = (
|
|
"Same measurement command/procedure",
|
|
"Same metric definitions",
|
|
"Same dataset/load profile",
|
|
"Environment materially equivalent",
|
|
"Result",
|
|
"Acceptance criteria satisfied",
|
|
)
|
|
for field in required_comparison_fields:
|
|
if not fields.get(field):
|
|
errors.append(f"performance comparison missing field: {field}")
|
|
for field in required_comparison_fields[:4]:
|
|
value = fields.get(field)
|
|
if value and value not in {"YES", "NO"}:
|
|
errors.append(f"performance comparison invalid equivalence value for {field}: {value}")
|
|
result = fields.get("Result")
|
|
if result and result not in {"IMPROVED", "NEUTRAL", "REGRESSED", "INCOMPARABLE"}:
|
|
errors.append(f"performance comparison invalid result: {result}")
|
|
acceptance = fields.get("Acceptance criteria satisfied")
|
|
if acceptance and acceptance not in {"YES", "NO"}:
|
|
errors.append(f"performance comparison invalid acceptance value: {acceptance}")
|
|
equivalent = all(fields.get(field) == "YES" for field in required_comparison_fields[:4])
|
|
if not equivalent and result and result != "INCOMPARABLE":
|
|
errors.append("performance incomparable conditions cannot claim a comparable result")
|
|
|
|
return errors
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
argv = sys.argv[1:] if argv is None else argv
|
|
if len(argv) != 1:
|
|
print("usage: verify_refactor_work_item.py <work-item-directory>")
|
|
return 2
|
|
errors = verify_work_item(Path(argv[0]))
|
|
if errors:
|
|
print("REFACTOR WORK ITEM VERIFICATION: FAIL")
|
|
for error in errors:
|
|
print(f"- {error}")
|
|
return 1
|
|
print("REFACTOR WORK ITEM VERIFICATION: PASS")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|