154 lines
5.2 KiB
Python
154 lines
5.2 KiB
Python
"""Command-line interface for the GovOPlaN Kubernetes VM lab."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
from typing import Sequence
|
|
|
|
from .config import LabConfigError, load_config
|
|
from .lifecycle import (
|
|
LabOperationError,
|
|
create,
|
|
deploy,
|
|
destroy,
|
|
doctor,
|
|
enroll_admin,
|
|
pause,
|
|
resume,
|
|
status,
|
|
update,
|
|
verify,
|
|
)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="govoplan-lab",
|
|
description="Create and operate a libvirt-backed GovOPlaN Kubernetes test lab.",
|
|
)
|
|
parser.add_argument(
|
|
"--config",
|
|
type=Path,
|
|
default=Path("govoplan-lab.toml"),
|
|
help="Strict TOML lab inventory (default: ./govoplan-lab.toml).",
|
|
)
|
|
parser.add_argument("--verbose", action="store_true")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
doctor_parser = subparsers.add_parser("doctor", help="Validate inventory and prerequisites.")
|
|
doctor_parser.add_argument(
|
|
"--online",
|
|
action="store_true",
|
|
help="Also connect to every hypervisor and verify its toolchain.",
|
|
)
|
|
|
|
subparsers.add_parser("status", help="Show VM, Kubernetes node, and pod state.")
|
|
_mutation_parser(subparsers, "create", "Create or reuse all declared VMs.")
|
|
_mutation_parser(subparsers, "deploy", "Deploy shared state, K3s, and GovOPlaN.")
|
|
_mutation_parser(subparsers, "update", "Reconcile pinned K3s and GovOPlaN inputs serially.")
|
|
_mutation_parser(subparsers, "pause", "Gracefully stop the lab while preserving disks.")
|
|
_mutation_parser(subparsers, "resume", "Start a paused lab in dependency order.")
|
|
|
|
destroy_parser = _mutation_parser(
|
|
subparsers,
|
|
"destroy",
|
|
"Destroy lab-owned VMs and disks with an explicit name confirmation.",
|
|
)
|
|
destroy_parser.add_argument("--confirm", default="")
|
|
destroy_parser.add_argument(
|
|
"--purge-local-state",
|
|
action="store_true",
|
|
help="Also delete local secrets, manifests, and evidence after VM teardown.",
|
|
)
|
|
|
|
verify_parser = subparsers.add_parser(
|
|
"verify",
|
|
help=(
|
|
"Collect sanitized live-cluster evidence using an "
|
|
"Ops-read-authorized GOVOPLAN_OPS_API_KEY."
|
|
),
|
|
)
|
|
verify_parser.add_argument(
|
|
"--exercise-api-pod-loss",
|
|
action="store_true",
|
|
help="Delete one ready API pod during the bounded availability drill.",
|
|
)
|
|
enroll_parser = _mutation_parser(
|
|
subparsers,
|
|
"enroll-admin",
|
|
"Securely consume the first-administrator enrollment artifact.",
|
|
)
|
|
enroll_parser.add_argument("--email", required=True)
|
|
enroll_parser.add_argument("--display-name", default=None)
|
|
enroll_parser.add_argument("--tenant-slug", default="default")
|
|
enroll_parser.add_argument("--tenant-name", default="Default Tenant")
|
|
return parser
|
|
|
|
|
|
def _mutation_parser(
|
|
subparsers: argparse._SubParsersAction[argparse.ArgumentParser],
|
|
name: str,
|
|
help_text: str,
|
|
) -> argparse.ArgumentParser:
|
|
parser = subparsers.add_parser(name, help=help_text)
|
|
parser.add_argument(
|
|
"--apply",
|
|
action="store_true",
|
|
help="Perform mutations; without this flag the command is a dry run.",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
try:
|
|
config = load_config(args.config)
|
|
if args.command == "doctor":
|
|
return doctor(config, online=args.online, verbose=args.verbose)
|
|
if args.command == "status":
|
|
return status(config, verbose=args.verbose)
|
|
if args.command == "create":
|
|
create(config, apply=args.apply, verbose=args.verbose)
|
|
elif args.command == "deploy":
|
|
deploy(config, apply=args.apply, verbose=args.verbose)
|
|
elif args.command == "update":
|
|
update(config, apply=args.apply, verbose=args.verbose)
|
|
elif args.command == "pause":
|
|
pause(config, apply=args.apply, verbose=args.verbose)
|
|
elif args.command == "resume":
|
|
resume(config, apply=args.apply, verbose=args.verbose)
|
|
elif args.command == "destroy":
|
|
destroy(
|
|
config,
|
|
apply=args.apply,
|
|
confirmation=args.confirm,
|
|
purge_local_state=args.purge_local_state,
|
|
verbose=args.verbose,
|
|
)
|
|
elif args.command == "verify":
|
|
verify(
|
|
config,
|
|
exercise_api_pod_loss=args.exercise_api_pod_loss,
|
|
verbose=args.verbose,
|
|
)
|
|
elif args.command == "enroll-admin":
|
|
enroll_admin(
|
|
config,
|
|
email=args.email,
|
|
display_name=args.display_name,
|
|
tenant_slug=args.tenant_slug,
|
|
tenant_name=args.tenant_name,
|
|
apply=args.apply,
|
|
)
|
|
else:
|
|
raise RuntimeError(f"unsupported command: {args.command}")
|
|
return 0
|
|
except (LabConfigError, LabOperationError, OSError, ValueError) as exc:
|
|
print(f"error: {exc}", file=__import__("sys").stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|