Files
stack/packages/mosaic/framework/tools/quality/scripts/framework-drift-check.py
T
2026-08-13 01:42:32 -05:00

133 lines
4.5 KiB
Python

#!/usr/bin/env python3
"""Compare deployed Mosaic framework tools with the shipped framework source.
The framework ownership manifest declares tools/** framework-owned. Consequently every
regular file shipped below source tools/ is expected below MOSAIC_HOME/tools/, except
the explicit operator credential carve-out. Files that exist only in the deployed tree
are operator/unknown state and are reported but never treated as framework drift.
"""
from __future__ import annotations
import argparse
import hashlib
import os
from pathlib import Path
import sys
OPERATOR_CARVE_OUTS = {"_lib/credentials.json"}
def digest(path: Path) -> str:
value = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
value.update(chunk)
return value.hexdigest()
def default_source_tools() -> Path:
# .../tools/quality/scripts/framework-drift-check.py -> .../tools
return Path(__file__).resolve().parents[2]
def normalize_source(path: Path) -> Path:
candidate = path.resolve()
if (candidate / "tools").is_dir():
candidate = candidate / "tools"
return candidate
def files_below(root: Path) -> dict[str, Path]:
return {
path.relative_to(root).as_posix(): path
for path in root.rglob("*")
if path.is_file()
}
def main() -> int:
parser = argparse.ArgumentParser(description="Detect deployed Mosaic framework-tool drift")
parser.add_argument(
"--source-root",
type=Path,
default=Path(os.environ["MOSAIC_FRAMEWORK_SOURCE_ROOT"])
if os.environ.get("MOSAIC_FRAMEWORK_SOURCE_ROOT")
else default_source_tools(),
help="shipped framework root or tools root (default: this script's shipped tools tree)",
)
parser.add_argument(
"--installed-root",
type=Path,
default=Path(os.environ.get("MOSAIC_HOME", Path.home() / ".config/mosaic")) / "tools",
help="deployed tools root (default: $MOSAIC_HOME/tools)",
)
parser.add_argument("--verbose", action="store_true", help="list in-sync paths too")
args = parser.parse_args()
source = normalize_source(args.source_root)
installed = args.installed_root.resolve()
if not source.is_dir():
print(f"[framework-drift] CANNOT_ASSERT source tools missing: {source}", file=sys.stderr)
return 2
if not installed.is_dir():
print(f"[framework-drift] CANNOT_ASSERT installed tools missing: {installed}", file=sys.stderr)
return 2
if source == installed:
print(
"[framework-drift] CANNOT_ASSERT source and installed roots are identical; "
"run the checker from the bundled package or pass --source-root",
file=sys.stderr,
)
return 2
source_files = files_below(source)
installed_files = files_below(installed)
required = sorted(set(source_files) - OPERATOR_CARVE_OUTS)
in_sync: list[str] = []
stale: list[str] = []
not_installed: list[str] = []
for relative in required:
deployed = installed / relative
if not deployed.is_file():
not_installed.append(relative)
elif digest(source_files[relative]) == digest(deployed):
in_sync.append(relative)
else:
stale.append(relative)
installed_only = sorted(set(installed_files) - set(source_files) - OPERATOR_CARVE_OUTS)
if args.verbose:
for relative in in_sync:
print(f"[framework-drift] IN_SYNC {relative}")
for relative in stale:
print(f"[framework-drift] STALE {relative}")
for relative in not_installed:
print(f"[framework-drift] NOT_INSTALLED {relative}")
if args.verbose:
for relative in installed_only:
print(f"[framework-drift] INSTALLED_ONLY operator-or-unknown {relative}")
print(
"[framework-drift] summary "
f"in-sync={len(in_sync)} stale={len(stale)} not-installed={len(not_installed)} "
f"installed-only={len(installed_only)}"
)
print(
"[framework-drift] classification tools/**=framework-owned-required; "
"tools/_lib/credentials.json=operator-owned-excluded; "
"installed-only=operator-or-unknown-preserved"
)
if stale or not_installed:
print(
"[framework-drift] FAIL deployed framework tools do not match shipped source; "
"schedule a reviewed framework reseed",
file=sys.stderr,
)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())