Added nono profiles & orca integration - Added some extensions to start working on
This commit is contained in:
Executable
+191
@@ -0,0 +1,191 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Manage trusted project-to-nono-profile mappings for Orca-launched Pi."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
MAP_PATH = Path("/srv/data/code/harness-tools/profiles/orca/project-nono-profiles.json")
|
||||
DEFAULT_PROFILE = "pi-orca-local"
|
||||
|
||||
|
||||
def git_output(path: Path, *args: str) -> str | None:
|
||||
result = subprocess.run(
|
||||
["git", "-C", str(path), *args],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
value = result.stdout.strip()
|
||||
return value or None
|
||||
|
||||
|
||||
def canonical_project_root(path: Path) -> Path:
|
||||
candidate = path.expanduser().resolve()
|
||||
common_dir = git_output(candidate, "rev-parse", "--path-format=absolute", "--git-common-dir")
|
||||
if common_dir:
|
||||
common_path = Path(common_dir).resolve()
|
||||
if common_path.name == ".git":
|
||||
return common_path.parent
|
||||
|
||||
top_level = git_output(candidate, "rev-parse", "--show-toplevel")
|
||||
if top_level:
|
||||
return Path(top_level).resolve()
|
||||
return candidate
|
||||
|
||||
|
||||
def load_map() -> dict[str, Any]:
|
||||
try:
|
||||
contents = MAP_PATH.read_text(encoding="utf-8")
|
||||
data = json.loads(contents)
|
||||
except (OSError, json.JSONDecodeError) as error:
|
||||
raise ValueError(f"cannot load profile map: {error}") from error
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("profile map must be a JSON object")
|
||||
projects = data.get("projects")
|
||||
if not isinstance(projects, dict):
|
||||
raise ValueError("profile map projects must be a JSON object")
|
||||
project_roots = data.get("projectRoots", {})
|
||||
if not isinstance(project_roots, dict):
|
||||
raise ValueError("profile map projectRoots must be a JSON object")
|
||||
for root, profile in project_roots.items():
|
||||
if not isinstance(root, str) or not Path(root).is_absolute():
|
||||
raise ValueError("projectRoots keys must be absolute paths")
|
||||
if not isinstance(profile, str) or not profile:
|
||||
raise ValueError("projectRoots values must be non-empty profile names")
|
||||
default = data.get("defaultProfile", DEFAULT_PROFILE)
|
||||
if not isinstance(default, str) or not default:
|
||||
raise ValueError("defaultProfile must be a non-empty string")
|
||||
return data
|
||||
|
||||
|
||||
def save_map(data: dict[str, Any]) -> None:
|
||||
MAP_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
descriptor, temporary_name = tempfile.mkstemp(prefix=".project-nono-profiles.", dir=MAP_PATH.parent)
|
||||
temporary = Path(temporary_name)
|
||||
try:
|
||||
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
|
||||
json.dump(data, handle, indent=2, sort_keys=True)
|
||||
handle.write("\n")
|
||||
temporary.chmod(0o600)
|
||||
os.replace(temporary, MAP_PATH)
|
||||
finally:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
|
||||
def profile_exists(profile: str) -> bool:
|
||||
result = subprocess.run(
|
||||
["nono", "profile", "show", profile],
|
||||
check=False,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
return result.returncode == 0
|
||||
|
||||
|
||||
def resolve_profile(
|
||||
data: dict[str, Any], project: Path, launch_path: Path | None = None
|
||||
) -> tuple[str, str]:
|
||||
projects = data["projects"]
|
||||
mapped = projects.get(str(project))
|
||||
if isinstance(mapped, str) and mapped:
|
||||
return mapped, "project"
|
||||
|
||||
route_candidates = {project}
|
||||
if launch_path is not None:
|
||||
route_candidates.add(launch_path)
|
||||
|
||||
matching_roots: list[tuple[Path, str]] = []
|
||||
for root_value, profile in data.get("projectRoots", {}).items():
|
||||
root = Path(root_value)
|
||||
if any(candidate == root or root in candidate.parents for candidate in route_candidates):
|
||||
matching_roots.append((root, profile))
|
||||
if matching_roots:
|
||||
_root, profile = max(matching_roots, key=lambda match: len(match[0].parts))
|
||||
return profile, "project-root"
|
||||
|
||||
return data.get("defaultProfile", DEFAULT_PROFILE), "default"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
for command in ("resolve", "show", "unset", "unset-root"):
|
||||
subparser = subparsers.add_parser(command)
|
||||
subparser.add_argument("path", nargs="?", default=os.getcwd())
|
||||
|
||||
for command in ("set", "set-root"):
|
||||
set_parser = subparsers.add_parser(command)
|
||||
set_parser.add_argument("profile")
|
||||
set_parser.add_argument("path", nargs="?", default=os.getcwd())
|
||||
subparsers.add_parser("list")
|
||||
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
data = load_map()
|
||||
except ValueError as error:
|
||||
parser.error(str(error))
|
||||
|
||||
if args.command == "list":
|
||||
print(json.dumps(data, indent=2, sort_keys=True))
|
||||
return 0
|
||||
|
||||
launch_path = Path(args.path).expanduser().resolve()
|
||||
project = canonical_project_root(launch_path)
|
||||
|
||||
if args.command == "resolve":
|
||||
profile, _source = resolve_profile(data, project, launch_path)
|
||||
print(profile)
|
||||
return 0
|
||||
|
||||
if args.command == "show":
|
||||
profile, source = resolve_profile(data, project, launch_path)
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"launchPath": str(launch_path),
|
||||
"projectRoot": str(project),
|
||||
"profile": profile,
|
||||
"source": source,
|
||||
},
|
||||
indent=2,
|
||||
sort_keys=True,
|
||||
)
|
||||
)
|
||||
return 0
|
||||
|
||||
if args.command in ("set", "set-root"):
|
||||
if not profile_exists(args.profile):
|
||||
parser.error(f"nono profile does not exist: {args.profile}")
|
||||
mapping_key = "projects" if args.command == "set" else "projectRoots"
|
||||
mapping_path = project if args.command == "set" else launch_path
|
||||
data.setdefault(mapping_key, {})[str(mapping_path)] = args.profile
|
||||
save_map(data)
|
||||
print(f"{mapping_path} -> {args.profile}")
|
||||
return 0
|
||||
|
||||
if args.command in ("unset", "unset-root"):
|
||||
mapping_key = "projects" if args.command == "unset" else "projectRoots"
|
||||
mapping_path = project if args.command == "unset" else launch_path
|
||||
removed = data.setdefault(mapping_key, {}).pop(str(mapping_path), None)
|
||||
save_map(data)
|
||||
if removed is None:
|
||||
print(f"no mapping for {mapping_path}")
|
||||
else:
|
||||
print(f"removed {mapping_path} -> {removed}")
|
||||
return 0
|
||||
|
||||
raise AssertionError(f"unhandled command: {args.command}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user