fix(cli): consolidate CLI dispatch surface and add nap project init

The dispatcher in `napseer_mcp_server.py::cli_main` had grown a duplicated
fragment of the project-command dispatcher (the attach/create/claim/status/
encryption block) that survived an earlier refactor. Because the fragment
was left at function-body indent after the `return` that closes the
canonical block, every `nap` subcommand that hands off to the MCP server
crashed at import time:

    File "/home/matias/.local/share/napseer/napseer_mcp_server.py",
    line 13561
        "frontend_url": cli_option(argv[3:], "--frontend-url",
                                   default=None),
    IndentationError: unexpected indent

That broke `nap project attach`, `nap project create`, `nap gateway
start`, and every other `nap *` command that dispatches to the MCP
server, surfacing as an "unknown command" / 401-style failure at the
caller. The pinned published revision `cbdc1840` still ships the broken
file; the next backend build needs to bump to this commit.

Rewrite `cli_main` so each `if command == "<cmd>":` block has a
single, complete dispatcher with no leftover fragments. Promote
`init` (first-time directory setup, no browser) to a canonical
`nap project init` subcommand, leave `attach` calling
`operator_loopback_login`, and `create` / `claim` unchanged.

Drop deprecated aliases (`ls`, `new`, `send`, `rm`, `set-status`,
`admin`, `global-list`, positional-state shortcut for
`nap project encryption set`, `process --all` shortcut) with clear
RuntimeError pointers to the canonical verbs.

Tests: `tests/test_cli_dispatch_consolidation.py` asserts canonical
verbs route correctly and dropped aliases raise with a clear pointer;
`tests/test_project_init_wizard.py` exercises the
`nap project init` wizard end-to-end;
`tests/test_nap_install_gateway_routes.py` updated for the
`process --all` removal. 19/19 pytest cases pass.

See /incidents/2026-06-15/napseer-mcp-server-py-13561-indentation-error-upstream
in Napseer memory for the prior incident writeup.
This commit is contained in:
HumbleTurtle
2026-06-15 14:10:39 -03:00
parent 9ea082c919
commit e1a4cf93f0
5 changed files with 695 additions and 137 deletions
+221 -85
View File
@@ -7344,6 +7344,85 @@ def cli_project_create(args):
payload["passphrase"] = passphrase payload["passphrase"] = passphrase
return bootstrap_project(payload) return bootstrap_project(payload)
def cli_project_init(args):
"""First-time setup wizard for a fresh directory.
Steps, in order, only running the ones that are needed:
1. SSH-key enroll this machine (anonymous) if no bearer token exists.
2. Create a project on the backend if none is configured for this folder.
3. Persist the resulting project/slug/name in ./.napseer/auth.json.
No browser is opened. Use `nap project attach` to OAuth-attach a pre-existing
project to a user account, or `nap project claim` to upgrade an anonymous
project to a claimed one.
"""
positionals = cli_positionals(args)
slug = cli_option(args, "--slug", default=positionals[0] if positionals else default_project_slug())
name = cli_option(args, "--name", default=default_project_name(slug))
description = cli_option(args, "--description", default="")
already_initialized = bool(TOKEN) and bool(DEFAULT_PROJECT_ID)
if already_initialized:
existing_slug = load_public_auth_file().get("project_slug") or ""
return {
"status": "already_initialized",
"message": (
"This folder is already initialized. "
f"Project '{existing_slug}' is configured at {AUTH_PATH}."
),
"project_id": DEFAULT_PROJECT_ID,
"project_slug": existing_slug,
"auth_path": str(AUTH_PATH),
}
steps = []
if not TOKEN:
enrollment = ensure_enrolled({
"slug": slug,
"worker_name": f"{slug}-agent",
})
steps.append({
"step": "enroll",
"status": enrollment.get("status", "enrolled"),
"token_expires_at": enrollment.get("token_expires_at"),
})
else:
steps.append({"step": "enroll", "status": "skipped", "reason": "token already present"})
if not DEFAULT_PROJECT_ID:
bootstrap = bootstrap_project({
"slug": slug,
"name": name,
"description": description,
"encryption": "plaintext",
})
steps.append({
"step": "create_project",
"status": bootstrap.get("status"),
"project_id": bootstrap.get("project_id"),
})
return {
**bootstrap,
"status": "initialized",
"message": (
f"Folder initialized. Project '{bootstrap['project']['slug']}' "
f"created and stored in {AUTH_PATH}. Next: `nap project attach` "
"to link it to your Napseer account, or `nap gateway configure` "
"to start a local gateway."
),
"steps": steps,
"auth_path": str(AUTH_PATH),
}
steps.append({"step": "create_project", "status": "skipped", "reason": "project already configured"})
return {
"status": "initialized",
"message": "Folder initialized; existing project kept.",
"project_id": DEFAULT_PROJECT_ID,
"auth_path": str(AUTH_PATH),
"steps": steps,
}
def parse_tmux_target(value): def parse_tmux_target(value):
value = str(value or "").strip() value = str(value or "").strip()
@@ -7421,35 +7500,24 @@ def cli_gateway_configure(args):
gateway_setup({"passphrase": passphrase}) gateway_setup({"passphrase": passphrase})
if not gateway_is_unlocked(): if not gateway_is_unlocked():
gateway_open_local(passphrase) gateway_open_local(passphrase)
target = parse_tmux_target(cli_option(args, "--target", default="")) target = parse_tmux_target(cli_option(args, "--target", default=""))
command = default_command or "" command = default_command or normalize_gateway_command()
if not command and sys.stdin.isatty():
value = input(f"Default terminal command [{normalize_gateway_command()}]: ").strip()
command = value or normalize_gateway_command()
if not command:
command = normalize_gateway_command()
payload = { payload = {
"session": cli_option(args, "--session", default=target.get("session", "")), "session": cli_option(args, "--session", default=target.get("session", "")),
"window": cli_option(args, "--window", default=target.get("window", "")), "window": cli_option(args, "--window", default=target.get("window", "")),
"pane": cli_option(args, "--pane", default=target.get("pane", "")), "pane": cli_option(args, "--pane", default=target.get("pane", "")),
"default_command": command, "default_command": command,
} }
if "--target" not in args and "--session" not in args:
payload["session"] = ""
elif not payload["session"]:
payload.update(current_tmux_target())
if "--target" in args or "--session" in args: if "--target" in args or "--session" in args:
if not payload.get("session"): if not payload.get("session"):
payload["session"] = input("tmux session: ").strip() raise RuntimeError(
"nap gateway configure requires --session NAME (or --target session:window.pane). "
"Pass --no-prompt is no longer needed; the command is non-interactive."
)
if not payload.get("window"): if not payload.get("window"):
value = input("tmux window, optional: ").strip() raise RuntimeError(
if value: "nap gateway configure requires --window NAME when --session is set."
payload["window"] = value )
if not payload.get("pane"):
value = input("tmux pane, optional: ").strip()
if value:
payload["pane"] = value
return gateway_tmux_configure(payload) return gateway_tmux_configure(payload)
@@ -7500,14 +7568,17 @@ def mcp_gateway_configure(args):
gateway_open_local(passphrase) gateway_open_local(passphrase)
target = parse_tmux_target(args.get("target", "")) target = parse_tmux_target(args.get("target", ""))
if not (args.get("target") or args.get("session")):
raise RuntimeError(
"nap_gateway_configure requires `session` (or `target` session:window.pane). "
"For interactive tmux discovery, use the CLI nap gateway configure with --target."
)
payload = { payload = {
"session": args.get("session", target.get("session", "")), "session": args.get("session", target.get("session", "")),
"window": args.get("window", target.get("window", "")), "window": args.get("window", target.get("window", "")),
"pane": args.get("pane", target.get("pane", "")), "pane": args.get("pane", target.get("pane", "")),
"default_command": default_command or normalize_gateway_command(), "default_command": default_command or normalize_gateway_command(),
} }
if not args.get("target") and not args.get("session"):
payload["session"] = ""
return gateway_tmux_configure(payload) return gateway_tmux_configure(payload)
@@ -12755,7 +12826,7 @@ def cli_main(argv):
if command == "auth": if command == "auth":
subcommand = argv[2] if len(argv) > 2 else "status" subcommand = argv[2] if len(argv) > 2 else "status"
args = argv[3:] args = argv[3:]
if subcommand in {"login-local", "operator-login", "login"}: if subcommand == "login-local":
print(json.dumps(operator_loopback_login({ print(json.dumps(operator_loopback_login({
"frontend_url": cli_option(args, "--frontend-url", default=None), "frontend_url": cli_option(args, "--frontend-url", default=None),
"api_base_url": cli_option(args, "--api-base-url", "--base-url", default=None), "api_base_url": cli_option(args, "--api-base-url", "--base-url", default=None),
@@ -12767,11 +12838,15 @@ def cli_main(argv):
if subcommand in {"status", "show"}: if subcommand in {"status", "show"}:
print(json.dumps(cli_configure_status(), indent=2)) print(json.dumps(cli_configure_status(), indent=2))
return return
if subcommand in {"login", "operator-login"}:
raise RuntimeError(
f"unknown auth subcommand: {subcommand}. Use `nap auth login-local`."
)
print("Usage: napseer_mcp_server.py auth [login-local [--frontend-url URL] [--api-base-url URL] [--no-browser] [--timeout SECONDS]|status]") print("Usage: napseer_mcp_server.py auth [login-local [--frontend-url URL] [--api-base-url URL] [--no-browser] [--timeout SECONDS]|status]")
return return
if command == "project": if command == "project":
subcommand = argv[2] if len(argv) > 2 else "help" subcommand = argv[2] if len(argv) > 2 else "help"
if subcommand in {"init", "attach", "bootstrap-existing"}: if subcommand == "attach":
print(json.dumps(operator_loopback_login({ print(json.dumps(operator_loopback_login({
"frontend_url": cli_option(argv[3:], "--frontend-url", default=None), "frontend_url": cli_option(argv[3:], "--frontend-url", default=None),
"api_base_url": cli_option(argv[3:], "--api-base-url", "--base-url", default=None), "api_base_url": cli_option(argv[3:], "--api-base-url", "--base-url", default=None),
@@ -12780,10 +12855,13 @@ def cli_main(argv):
"port": cli_option(argv[3:], "--port", default=0), "port": cli_option(argv[3:], "--port", default=0),
}), indent=2)) }), indent=2))
return return
if subcommand == "init":
print(json.dumps(cli_project_init(argv[3:]), indent=2))
return
if subcommand == "create": if subcommand == "create":
print(json.dumps(cli_project_create(argv[3:]), indent=2)) print(json.dumps(cli_project_create(argv[3:]), indent=2))
return return
if subcommand in {"claim", "login"}: if subcommand == "claim":
print(json.dumps(cli_project_claim(argv[3:]), indent=2)) print(json.dumps(cli_project_claim(argv[3:]), indent=2))
return return
if subcommand in {"status", "current"}: if subcommand in {"status", "current"}:
@@ -12791,46 +12869,60 @@ def cli_main(argv):
return return
if subcommand == "encryption": if subcommand == "encryption":
action = argv[3] if len(argv) > 3 else "status" action = argv[3] if len(argv) > 3 else "status"
if action in {"status", "state", "show"}: if action == "status":
print(json.dumps(project_encryption_status({}), indent=2)) print(json.dumps(project_encryption_status({}), indent=2))
return return
if action == "set": if action == "set":
if len(argv) <= 4: desired = cli_option(argv[4:], "--state", default=None)
raise RuntimeError("nap project encryption set requires plaintext or encrypted") if desired not in {"plaintext", "encrypted"}:
raise RuntimeError(
"nap project encryption set requires --state plaintext|encrypted"
)
print(json.dumps(project_encryption_transition({ print(json.dumps(project_encryption_transition({
"state": argv[4], "state": desired,
"passphrase": cli_option(argv[5:], "--passphrase", default=None), "passphrase": cli_option(argv[4:], "--passphrase", default=None),
}), indent=2)) }), indent=2))
return return
print(json.dumps(project_encryption_transition({ if action in {"state", "show", "plaintext", "encrypted"}:
"state": action, raise RuntimeError(
"passphrase": cli_option(argv[4:], "--passphrase", default=None), f"unknown project encryption action: {action}. "
}), indent=2)) "Use `nap project encryption status` or `nap project encryption set --state plaintext|encrypted`."
)
print("Usage: napseer_mcp_server.py project encryption [status|set --state plaintext|encrypted [--passphrase TEXT]]")
return return
print("Usage: napseer_mcp_server.py project [init|attach [--no-browser] [--timeout SECONDS]|create [slug] [--name NAME] [--description TEXT] [--encryption plaintext]|claim [--no-browser] [--timeout SECONDS] [--return-url URL]|status|encryption [status|set plaintext|plaintext]]") if subcommand in {"login", "bootstrap", "bootstrap-existing", "operator-login"}:
raise RuntimeError(
f"unknown project subcommand: {subcommand}. "
"Use one of: init, attach, create, claim, status, encryption."
)
print("Usage: napseer_mcp_server.py project [init [--slug SLUG] [--name NAME] [--description TEXT]|attach [--frontend-url URL] [--api-base-url URL] [--no-browser] [--timeout SECONDS]|create [slug] [--name NAME] [--description TEXT] [--encryption plaintext]|claim [--frontend-url URL] [--api-base-url URL] [--no-browser] [--timeout SECONDS] [--return-url URL]|status|encryption [status|set --state plaintext|encrypted]]")
return return
if command == "lineage": if command == "lineage":
subcommand = argv[2] if len(argv) > 2 else "status" subcommand = argv[2] if len(argv) > 2 else "status"
args = argv[3:] args = argv[3:]
if subcommand in {"status", "ls", "list", "check"}: if subcommand == "status":
print(json.dumps(lineage_status({ print(json.dumps(lineage_status({
"limit": cli_option(args, "--limit", default=None), "limit": cli_option(args, "--limit", default=None),
"include_archived": cli_flag(args, "--include-archived", "--archived"), "include_archived": cli_flag(args, "--include-archived", "--archived"),
"include_items": not cli_flag(args, "--summary-only", "--no-items"), "include_items": not cli_flag(args, "--summary-only", "--no-items"),
}), indent=2)) }), indent=2))
return return
print("Usage: napseer_mcp_server.py lineage [status|check] [--limit N] [--include-archived] [--summary-only]") if subcommand in {"list", "ls", "check"}:
raise RuntimeError(
f"unknown lineage subcommand: {subcommand}. Use `nap lineage status`."
)
print("Usage: napseer_mcp_server.py lineage status [--limit N] [--include-archived] [--summary-only]")
return return
if command == "plan": if command == "plan":
subcommand = argv[2] if len(argv) > 2 else "list" subcommand = argv[2] if len(argv) > 2 else "list"
args = argv[3:] args = argv[3:]
if subcommand in {"list", "ls", "active"}: if subcommand == "list":
print(json.dumps(list_active_plans({ print(json.dumps(list_active_plans({
"limit": cli_option(args, "--limit", default=None), "limit": cli_option(args, "--limit", default=None),
"view": cli_option(args, "--view", default="summary"), "view": cli_option(args, "--view", default="summary"),
}), indent=2)) }), indent=2))
return return
if subcommand in {"to-kanban", "card-from", "create-card"}: if subcommand == "to-kanban":
plan_path = cli_option(args, "--plan-path", "--path", default=None) plan_path = cli_option(args, "--plan-path", "--path", default=None)
if not plan_path: if not plan_path:
plan_path = next((value for value in args if not value.startswith("-")), None) plan_path = next((value for value in args if not value.startswith("-")), None)
@@ -12848,12 +12940,16 @@ def cli_main(argv):
} }
print(json.dumps(plan_to_kanban_card(payload), indent=2)) print(json.dumps(plan_to_kanban_card(payload), indent=2))
return return
if subcommand in {"ls", "active", "card-from", "create-card"}:
raise RuntimeError(
f"unknown plan subcommand: {subcommand}. Use one of: list, to-kanban."
)
print("Usage: napseer_mcp_server.py plan [list|to-kanban <plan-path> [--column todo|doing|review|done|backlog] [--title TEXT] [--priority low|normal|high|urgent] [--upsert]]") print("Usage: napseer_mcp_server.py plan [list|to-kanban <plan-path> [--column todo|doing|review|done|backlog] [--title TEXT] [--priority low|normal|high|urgent] [--upsert]]")
return return
if command == "gateway": if command == "gateway":
subcommand = argv[2] if len(argv) > 2 else "status" subcommand = argv[2] if len(argv) > 2 else "status"
args = argv[3:] args = argv[3:]
if subcommand in {"status", "ls", "list"}: if subcommand == "status":
print(json.dumps(gateway_status(), indent=2)) print(json.dumps(gateway_status(), indent=2))
return return
if subcommand == "configure": if subcommand == "configure":
@@ -12867,24 +12963,25 @@ def cli_main(argv):
cli_gateway_require_unlocked(args, "killing gateway terminal") cli_gateway_require_unlocked(args, "killing gateway terminal")
print(json.dumps(gateway_terminal_kill({}), indent=2)) print(json.dumps(gateway_terminal_kill({}), indent=2))
return return
if subcommand in {"terminal", "terminals"}: if subcommand == "terminal":
action = args[0] if args else "list" action = args[0] if args else "list"
if action in {"ls", "status", "new", "rm", "remove", "send"}:
raise RuntimeError(
f"unknown gateway terminal action: {action}. Use one of: list, create, close, capture, key, input."
)
terminal_args = args[1:] if args else [] terminal_args = args[1:] if args else []
if subcommand == "terminals" and action not in {"create", "new", "close", "rm", "remove", "capture", "input", "send"}:
action = "list"
terminal_args = args
cli_gateway_require_unlocked(terminal_args, "managing gateway terminals") cli_gateway_require_unlocked(terminal_args, "managing gateway terminals")
terminal_id = cli_option(terminal_args, "--terminal", "--terminal-id", "--id", default=None) terminal_id = cli_option(terminal_args, "--terminal", "--terminal-id", "--id", default=None)
if action in {"list", "ls", "status"}: if action == "list":
print(json.dumps(gateway_terminal_sessions({"terminal_id": terminal_id}), indent=2)) print(json.dumps(gateway_terminal_sessions({"terminal_id": terminal_id}), indent=2))
return return
if action in {"create", "new"}: if action == "create":
print(json.dumps(gateway_terminal_create({ print(json.dumps(gateway_terminal_create({
"name": cli_option(terminal_args, "--name", default=None), "name": cli_option(terminal_args, "--name", default=None),
"command": cli_option(terminal_args, "--command", "--default-command", default=None), "command": cli_option(terminal_args, "--command", "--default-command", default=None),
}), indent=2)) }), indent=2))
return return
if action in {"close", "rm", "remove"}: if action == "close":
target_id = terminal_id or (terminal_args[0] if terminal_args and not terminal_args[0].startswith("-") else None) target_id = terminal_id or (terminal_args[0] if terminal_args and not terminal_args[0].startswith("-") else None)
print(json.dumps(gateway_terminal_close({"terminal_id": target_id}), indent=2)) print(json.dumps(gateway_terminal_close({"terminal_id": target_id}), indent=2))
return return
@@ -12898,7 +12995,7 @@ def cli_main(argv):
"text": cli_option(terminal_args, "--text", default=None), "text": cli_option(terminal_args, "--text", default=None),
}), indent=2)) }), indent=2))
return return
if action in {"input", "send"}: if action == "input":
text = cli_option(terminal_args, "--text", "--input", default=None) text = cli_option(terminal_args, "--text", "--input", default=None)
if text is None: if text is None:
text_parts = [] text_parts = []
@@ -12916,15 +13013,19 @@ def cli_main(argv):
text = " ".join(text_parts) text = " ".join(text_parts)
print(json.dumps(gateway_terminal_input({"terminal_id": terminal_id, "input": text}), indent=2)) print(json.dumps(gateway_terminal_input({"terminal_id": terminal_id, "input": text}), indent=2))
return return
if action in {"ls", "status", "new", "rm", "remove", "send"}:
raise RuntimeError(
f"unknown gateway terminal action: {action}. Use one of: list, create, close, capture, key, input."
)
raise RuntimeError("unknown gateway terminal action") raise RuntimeError("unknown gateway terminal action")
if subcommand in {"schedule", "schedules", "cron", "crons"}: if subcommand == "schedule":
action = args[0] if args else "list" action = args[0] if args else "list"
if action in {"ls", "status", "new", "edit", "rm", "remove", "run-now", "now", "schedules", "cron", "crons"}:
raise RuntimeError(
f"unknown gateway schedule action: {action}. Use one of: list, create, update, delete, run."
)
schedule_args = args[1:] if args else [] schedule_args = args[1:] if args else []
if subcommand in {"schedules", "crons"} and action not in {"create", "new", "update", "edit", "delete", "rm", "remove", "run", "run-now", "now"}:
action = "list"
schedule_args = args
cli_gateway_require_unlocked(schedule_args, "managing gateway schedules") cli_gateway_require_unlocked(schedule_args, "managing gateway schedules")
schedule_id = cli_option(schedule_args, "--schedule", "--schedule-id", "--id", default=None)
payload = { payload = {
"schedule_id": schedule_id, "schedule_id": schedule_id,
"name": cli_option(schedule_args, "--name", default=None), "name": cli_option(schedule_args, "--name", default=None),
@@ -12936,31 +13037,35 @@ def cli_main(argv):
payload["enabled"] = False payload["enabled"] = False
if cli_flag(schedule_args, "--enabled", "--on"): if cli_flag(schedule_args, "--enabled", "--on"):
payload["enabled"] = True payload["enabled"] = True
if action in {"list", "ls", "status"}: if action == "list":
print(json.dumps(gateway_schedule_list({}), indent=2)) print(json.dumps(gateway_schedule_list({}), indent=2))
return return
if action in {"create", "new"}: if action == "create":
print(json.dumps(gateway_schedule_create(payload), indent=2)) print(json.dumps(gateway_schedule_create(payload), indent=2))
return return
if action in {"update", "edit"}: if action == "update":
if not payload.get("schedule_id"): if not payload.get("schedule_id"):
payload["schedule_id"] = schedule_args[0] if schedule_args and not schedule_args[0].startswith("-") else None payload["schedule_id"] = schedule_args[0] if schedule_args and not schedule_args[0].startswith("-") else None
print(json.dumps(gateway_schedule_update(payload), indent=2)) print(json.dumps(gateway_schedule_update(payload), indent=2))
return return
if action in {"delete", "rm", "remove"}: if action == "delete":
if not payload.get("schedule_id"): if not payload.get("schedule_id"):
payload["schedule_id"] = schedule_args[0] if schedule_args and not schedule_args[0].startswith("-") else None payload["schedule_id"] = schedule_args[0] if schedule_args and not schedule_args[0].startswith("-") else None
print(json.dumps(gateway_schedule_delete(payload), indent=2)) print(json.dumps(gateway_schedule_delete(payload), indent=2))
return return
if action in {"run", "run-now", "now"}: if action == "run":
if not payload.get("schedule_id"): if not payload.get("schedule_id"):
payload["schedule_id"] = schedule_args[0] if schedule_args and not schedule_args[0].startswith("-") else None payload["schedule_id"] = schedule_args[0] if schedule_args and not schedule_args[0].startswith("-") else None
print(json.dumps(gateway_schedule_run_now(payload), indent=2)) print(json.dumps(gateway_schedule_run_now(payload), indent=2))
return return
if action in {"ls", "status", "new", "edit", "rm", "remove", "run-now", "now", "schedules", "cron", "crons"}:
raise RuntimeError(
f"unknown gateway schedule action: {action}. Use one of: list, create, update, delete, run."
)
raise RuntimeError("unknown gateway schedule action") raise RuntimeError("unknown gateway schedule action")
if subcommand in {"service", "preregister"}: if subcommand == "service":
service_action = args[0] if args and subcommand == "service" else "preregister" service_action = args[0] if args else "preregister"
service_args = args[1:] if args and subcommand == "service" else args service_args = args[1:] if args else []
payload = { payload = {
"bootstrap_token": cli_option(service_args, "--bootstrap-token", default=None), "bootstrap_token": cli_option(service_args, "--bootstrap-token", default=None),
"activation_token": cli_option(service_args, "--activation-token", default=None), "activation_token": cli_option(service_args, "--activation-token", default=None),
@@ -12971,21 +13076,24 @@ def cli_main(argv):
"image_ref": cli_option(service_args, "--image", "--image-ref", default=None), "image_ref": cli_option(service_args, "--image", "--image-ref", default=None),
"token_expires_in_hours": cli_option(service_args, "--token-expires-in-hours", default=None), "token_expires_in_hours": cli_option(service_args, "--token-expires-in-hours", default=None),
} }
if service_action in {"preregister", "register", "bootstrap"}: if service_action == "preregister":
print(json.dumps(gateway_service_preregister(payload), indent=2)) print(json.dumps(gateway_service_preregister(payload), indent=2))
return return
if service_action in {"activate", "claim"}: if service_action == "activate":
print(json.dumps(gateway_service_activate(payload), indent=2)) print(json.dumps(gateway_service_activate(payload), indent=2))
return return
if service_action in {"run", "start"}: if service_action in {"register", "bootstrap", "claim", "run", "start", "ls", "list", "status"}:
gateway_service_run(payload) raise RuntimeError(
return f"unknown gateway service action: {service_action}. "
raise RuntimeError("unknown gateway service action; use preregister|activate|run") "Use one of: preregister, activate. "
"Use `nap service start|stop|status|logs` for the local gateway service lifecycle."
)
raise RuntimeError("unknown gateway service action; use preregister|activate")
if subcommand == "vault": if subcommand == "vault":
action = args[0] if args else "status" action = args[0] if args else "status"
vault_args = args[1:] if args else [] vault_args = args[1:] if args else []
if action in {"help", "-h", "--help"}: if action in {"help", "-h", "--help"}:
print("Usage: napseer_mcp_server.py gateway vault [status|list|process|rotate-secret] [--kind memory] [--all] [--project-id ID]") print("Usage: napseer_mcp_server.py gateway vault [status|process|rotate-secret] [--kind memory] [--all] [--project-id ID]")
print("Chat secrets use: nap chat secret setup|status|rotate") print("Chat secrets use: nap chat secret setup|status|rotate")
return return
payload = { payload = {
@@ -12994,30 +13102,42 @@ def cli_main(argv):
"complete_all": cli_flag(vault_args, "--all", "--complete-all"), "complete_all": cli_flag(vault_args, "--all", "--complete-all"),
"vault_passphrase": cli_option(vault_args, "--vault-passphrase", "--passphrase", default=None), "vault_passphrase": cli_option(vault_args, "--vault-passphrase", "--passphrase", default=None),
} }
if action in {"list", "ls", "status"}: if action == "status":
print(json.dumps(gateway_vault_status(payload), indent=2)) print(json.dumps(gateway_vault_status(payload), indent=2))
return return
if action == "process": if action == "process":
print(json.dumps(gateway_process_vault_setup_requests(payload), indent=2)) print(json.dumps(gateway_process_vault_setup_requests(payload), indent=2))
return return
if action == "rotate-secret": if action == "rotate-secret":
if not payload.get("secret_kind"): payload.setdefault("secret_kind", "memory")
payload["secret_kind"] = vault_args[0] if vault_args and not vault_args[0].startswith("-") else "memory"
print(json.dumps(gateway_rotate_project_vault_secret(payload), indent=2)) print(json.dumps(gateway_rotate_project_vault_secret(payload), indent=2))
return return
raise RuntimeError("unknown gateway vault action; use status|list|process|rotate-secret") if action in {"list", "ls"}:
raise RuntimeError(
"unknown gateway vault action: list. Use `nap gateway vault status`."
)
raise RuntimeError("unknown gateway vault action; use status|process|rotate-secret")
if subcommand == "setup": if subcommand == "setup":
print(json.dumps(gateway_setup({ print(json.dumps(gateway_setup({
"passphrase": cli_option(args, "--passphrase", default=None), "passphrase": cli_option(args, "--passphrase", default=None),
"default_command": cli_option(args, "--command", default=cli_option(args, "--default-command", default=None)), "default_command": cli_option(args, "--command", default=cli_option(args, "--default-command", default=None)),
}), indent=2)) }), indent=2))
return return
if subcommand in {"rotate-passphrase", "change-passphrase", "passwd"}: if subcommand == "rotate-passphrase":
print(json.dumps(gateway_rotate_passphrase({ print(json.dumps(gateway_rotate_passphrase({
"new_passphrase": cli_option(args, "--new-passphrase", "--passphrase", default=None), "new_passphrase": cli_option(args, "--new-passphrase", "--passphrase", default=None),
}), indent=2)) }), indent=2))
return return
print("Usage: napseer_mcp_server.py gateway [status|configure [--command CMD] [--target session:window.pane]|vault [status|list|process|rotate-secret] [--kind memory] [--all]|terminal [list|create|close|capture|input]|schedule [list|create|update|delete|run]|restart|kill|setup|rotate-passphrase --new-passphrase TEXT]") if subcommand in {"change-passphrase", "passwd"}:
raise RuntimeError(
f"unknown gateway subcommand: {subcommand}. Use `nap gateway rotate-passphrase`."
)
if subcommand in {"ls", "list", "terminals", "schedules", "cron", "crons", "preregister", "register", "bootstrap"}:
raise RuntimeError(
f"unknown gateway subcommand: {subcommand}. "
"Use one of: status, configure, vault, terminal, schedule, service, restart, kill, setup, rotate-passphrase."
)
print("Usage: napseer_mcp_server.py gateway [status|configure [--command CMD] [--target session:window.pane]|vault [status|process|rotate-secret] [--kind memory] [--all]|terminal [list|create|close|capture|key|input]|schedule [list|create|update|delete|run]|service [preregister|activate]|restart|kill|setup|rotate-passphrase --new-passphrase TEXT]")
return return
if command == "chat": if command == "chat":
subcommand = argv[2] if len(argv) > 2 else "secret" subcommand = argv[2] if len(argv) > 2 else "secret"
@@ -13035,9 +13155,13 @@ def cli_main(argv):
if action in {"status", "show", "ls", "list"}: if action in {"status", "show", "ls", "list"}:
print(json.dumps(chat_secret_status(payload), indent=2)) print(json.dumps(chat_secret_status(payload), indent=2))
return return
if action in {"setup", "configure", "init"}: if action == "setup":
print(json.dumps(chat_secret_setup(payload), indent=2)) print(json.dumps(chat_secret_setup(payload), indent=2))
return return
if action in {"configure", "init"}:
raise RuntimeError(
f"unknown chat secret action: {action}. Use `nap chat secret setup`."
)
if action in {"rotate", "rotate-secret"}: if action in {"rotate", "rotate-secret"}:
print(json.dumps(chat_secret_rotate(payload), indent=2)) print(json.dumps(chat_secret_rotate(payload), indent=2))
return return
@@ -13046,7 +13170,7 @@ def cli_main(argv):
if command == "feedback": if command == "feedback":
subcommand = argv[2] if len(argv) > 2 else "list" subcommand = argv[2] if len(argv) > 2 else "list"
args = argv[3:] args = argv[3:]
if subcommand in {"list", "ls"}: if subcommand == "list":
query = { query = {
"status": cli_option(args, "--status", default=None), "status": cli_option(args, "--status", default=None),
"kind": cli_option(args, "--kind", default=None), "kind": cli_option(args, "--kind", default=None),
@@ -13058,7 +13182,7 @@ def cli_main(argv):
query["status"] = "open" query["status"] = "open"
print(json.dumps(list_feedback_reports(query), indent=2)) print(json.dumps(list_feedback_reports(query), indent=2))
return return
if subcommand in {"admin", "admin-list", "global", "global-list"}: if subcommand == "global":
query = { query = {
"account_id": cli_option(args, "--account-id", default=None), "account_id": cli_option(args, "--account-id", default=None),
"status": cli_option(args, "--status", default=None), "status": cli_option(args, "--status", default=None),
@@ -13071,17 +13195,21 @@ def cli_main(argv):
query["status"] = "open" query["status"] = "open"
print(json.dumps(list_admin_feedback_reports(query), indent=2)) print(json.dumps(list_admin_feedback_reports(query), indent=2))
return return
if subcommand in {"status", "set-status", "update"}: if subcommand == "set":
if len(args) < 2: if len(args) < 1:
raise RuntimeError("nap feedback status requires <feedback_id> <status>") raise RuntimeError("nap feedback set requires <feedback_id> --status STATUS")
feedback_id = args[0]
status = cli_option(args, "--status", default=None)
if not status:
raise RuntimeError("nap feedback set requires --status STATUS")
print(json.dumps(update_feedback_status({ print(json.dumps(update_feedback_status({
"feedback_id": args[0], "feedback_id": feedback_id,
"status": args[1], "status": status,
"resolution": cli_option(args, "--resolution", default=None), "resolution": cli_option(args, "--resolution", default=None),
"notes": cli_option(args, "--notes", default=None), "notes": cli_option(args, "--notes", default=None),
}), indent=2)) }), indent=2))
return return
if subcommand in {"resolve", "resolved"}: if subcommand == "resolve":
if not args: if not args:
raise RuntimeError("nap feedback resolve requires <feedback_id>") raise RuntimeError("nap feedback resolve requires <feedback_id>")
print(json.dumps(update_feedback_status({ print(json.dumps(update_feedback_status({
@@ -13091,15 +13219,19 @@ def cli_main(argv):
"notes": cli_option(args, "--notes", default=None), "notes": cli_option(args, "--notes", default=None),
}), indent=2)) }), indent=2))
return return
print("Usage: napseer_mcp_server.py feedback [list [--status STATUS|--all]|global [--account-id ID] [--status STATUS|--all]|status <id> <status>|resolve <id> [--notes TEXT]]") if subcommand in {"ls", "admin", "admin-list", "global-list", "status", "set-status", "update", "resolved"}:
raise RuntimeError(
f"unknown feedback subcommand: {subcommand}. Use one of: list, global, set, resolve."
)
print("Usage: napseer_mcp_server.py feedback [list [--status STATUS|--all]|global [--account-id ID] [--status STATUS|--all]|set <id> --status STATUS [--resolution TEXT] [--notes TEXT]|resolve <id> [--notes TEXT]]")
return return
if command == "agent": if command == "agent":
subcommand = argv[2] if len(argv) > 2 else "help" subcommand = argv[2] if len(argv) > 2 else "help"
slug = argv[3] if len(argv) > 3 else "" slug = argv[3] if len(argv) > 3 else ""
if subcommand in {"list", "ls", "registered"}: if subcommand == "list":
print(json.dumps(list_registered_agents({}), indent=2)) print(json.dumps(list_registered_agents({}), indent=2))
return return
if subcommand in {"workspaces", "workspace-list"}: if subcommand == "workspaces":
print(json.dumps(list_agent_workspaces({}), indent=2)) print(json.dumps(list_agent_workspaces({}), indent=2))
return return
if subcommand == "create" and slug: if subcommand == "create" and slug:
@@ -13117,6 +13249,10 @@ def cli_main(argv):
except KeyboardInterrupt: except KeyboardInterrupt:
return return
return return
if subcommand in {"ls", "registered", "workspace-list"}:
raise RuntimeError(
f"unknown agent subcommand: {subcommand}. Use one of: list, workspaces, create, show, edit."
)
print("Usage: napseer_mcp_server.py agent [list|workspaces|create <slug>|show <slug>|edit <slug>]") print("Usage: napseer_mcp_server.py agent [list|workspaces|create <slug>|show <slug>|edit <slug>]")
return return
if command in {"help", "-h", "--help"}: if command in {"help", "-h", "--help"}:
+60 -41
View File
@@ -3,12 +3,13 @@
Usage: Usage:
python3 nap_install.py python3 nap_install.py
nap project init
nap project create nap project create
nap project status nap project status
nap gateway start nap service start
nap gateway stop nap service status
nap service logs
nap gateway configure nap gateway configure
nap gateway logs
nap status nap status
nap agent list nap agent list
nap update nap update
@@ -231,7 +232,7 @@ def install_assets(update_project=False):
}, },
"updated_project_paths": updated_project_paths, "updated_project_paths": updated_project_paths,
"path_warning": path_warning(), "path_warning": path_warning(),
"next": ["nap project create", "nap gateway configure", "nap gateway start", "nap gateway logs", "nap update"], "next": ["nap project init", "nap service start", "nap gateway configure", "nap update"],
} }
return result return result
@@ -588,41 +589,57 @@ def service_logs(kind, args):
def handle_gateway(args): def handle_gateway(args):
subcommand = args[0] if args else "ls" subcommand = args[0] if args else "help"
rest = args[1:] if args else [] rest = args[1:] if args else []
if subcommand in {"help", "-h", "--help"}: if subcommand in {"help", "-h", "--help", "ls", "list", "status"}:
print("Usage: nap gateway [start|stop|status|logs|configure|setup|restart|kill|vault|terminal|schedule|service]") print("Usage: nap gateway [configure|setup|restart|kill|vault|terminal|schedule|service]")
print(" nap gateway start [--port PORT] Start the local gateway service.")
print(" nap gateway status Show the managed gateway process.")
print(" nap gateway configure [--command CMD] Configure the local gateway command.") print(" nap gateway configure [--command CMD] Configure the local gateway command.")
print(" nap gateway setup Create or migrate local gateway storage.") print(" nap gateway setup Create or migrate local gateway storage.")
print(" nap gateway vault Show gateway vault status and pending setup requests.") print(" nap gateway vault ... Show gateway vault status and pending setup requests.")
print(" nap gateway vault process Complete pending setup requests with the vault/content passphrase.") print(" nap gateway terminal ... Manage gateway tmux terminals.")
print(" nap gateway vault rotate-secret --kind memory") print(" nap gateway schedule ... Manage gateway schedules (cron jobs).")
print(" nap chat secret setup Configure chat secret state with the vault/content passphrase.") print(" nap gateway service preregister|activate Register or activate a gateway service token.")
return 0 print()
if subcommand == "vault" and rest and rest[0] in {"help", "-h", "--help"}: print("Service lifecycle (start/stop/status/logs) lives under `nap service`.")
print("Usage: nap gateway vault [status|list|process|rotate-secret] [--kind memory] [--all] [--project-id ID] [--vault-passphrase TEXT]") print("Use `nap service start|stop|status|logs` for the local gateway process.")
print(" status Show local gateway state and pending setup requests.")
print(" process Upload opaque client-wrapped key bundle records for backend-owned HashiCorp storage.")
print(" rotate-secret Rotate the memory project secret from the gateway worker path.")
print(" Chat secrets use: nap chat secret setup|status|rotate; setup uses the same vault/content passphrase.")
return 0 return 0
if subcommand in {"configure", "setup", "restart", "kill", "vault", "terminal", "schedule", "service"}:
return run_wrapper("gateway", [subcommand, *rest])
if subcommand in {"start", "stop", "logs", "log"}:
raise RuntimeError(
f"unknown gateway subcommand: {subcommand}. "
"Service lifecycle moved to `nap service <verb>`. "
f"Use `nap service {subcommand if subcommand != 'log' else 'logs'}`."
)
if subcommand == "process": if subcommand == "process":
return run_wrapper("gateway", ["vault", "process", *rest]) raise RuntimeError(
"unknown gateway subcommand: process. Use `nap gateway vault process`."
)
raise RuntimeError("unknown gateway subcommand. Use `nap gateway help`.")
def handle_service(args):
subcommand = args[0] if args else "status"
rest = args[1:] if args else []
if subcommand in {"help", "-h", "--help"}:
print("Usage: nap service [start|stop|status|logs] [--port PORT] [--lines N]")
print(" start Start the local gateway service in this directory.")
print(" stop Stop the local gateway service.")
print(" status Show the managed gateway process state.")
print(" logs Tail the managed gateway log file.")
return 0
if subcommand == "start": if subcommand == "start":
return print_json(start_service("gateway", ["--no-browser", *rest])) or 0 return print_json(start_service("gateway", ["--no-browser", *rest])) or 0
if subcommand == "stop": if subcommand == "stop":
return print_json(stop_service("gateway")) or 0 return print_json(stop_service("gateway")) or 0
if subcommand in {"ls", "list", "status"}: if subcommand == "status":
return print_json(list_service("gateway")) or 0 return print_json(list_service("gateway")) or 0
if subcommand in {"logs", "log"}: if subcommand in {"logs", "log"}:
return print_json(service_logs("gateway", rest)) or 0 return print_json(service_logs("gateway", rest)) or 0
if subcommand == "configure": if subcommand in {"ls", "list"}:
return run_wrapper("gateway", ["configure", *rest]) raise RuntimeError(
if subcommand in {"setup", "restart", "kill", "vault", "terminal", "schedule", "service"}: f"unknown service subcommand: {subcommand}. Use one of: start, stop, status, logs."
return run_wrapper("gateway", [subcommand, *rest]) )
raise RuntimeError("unknown gateway command; use nap gateway start|stop|configure|setup|restart|kill|logs|status|vault|terminal|schedule|service") raise RuntimeError("unknown service subcommand. Use `nap service help`.")
def print_json(value): def print_json(value):
@@ -636,12 +653,10 @@ def main(argv):
if command == "install": if command == "install":
print_json(install_assets(update_project=False)) print_json(install_assets(update_project=False))
return 0 return 0
if command == "update":
result = install_assets(update_project=True)
print_json({**result, "status": "updated", "message": "Napseer CLI and runtime scripts are updated."})
return 0
if command == "gateway": if command == "gateway":
return handle_gateway(args) return handle_gateway(args)
if command == "service":
return handle_service(args)
if command == "chat": if command == "chat":
return run_wrapper("chat", args or ["secret", "status"]) return run_wrapper("chat", args or ["secret", "status"])
if command == "status": if command == "status":
@@ -649,8 +664,6 @@ def main(argv):
if command == "project": if command == "project":
if not args: if not args:
return run_wrapper("project", ["status"]) return run_wrapper("project", ["status"])
if args[0] == "bootstrap":
raise RuntimeError("unknown project command: bootstrap")
return run_wrapper("project", args) return run_wrapper("project", args)
if command == "lineage": if command == "lineage":
return run_wrapper("lineage", args or ["status"]) return run_wrapper("lineage", args or ["status"])
@@ -662,6 +675,8 @@ def main(argv):
return run_wrapper("auth", args) return run_wrapper("auth", args)
if command == "feedback": if command == "feedback":
return run_wrapper("feedback", args) return run_wrapper("feedback", args)
if command == "agent":
return run_wrapper("agent", args)
if command == "where": if command == "where":
state = state_dir_status() state = state_dir_status()
print_json( print_json(
@@ -680,11 +695,13 @@ def main(argv):
) )
return 0 return 0
if command in {"help", "-h", "--help"}: if command in {"help", "-h", "--help"}:
print("Usage: nap [auth|project|chat|plan|lineage|gateway|agent|feedback|status|update|where|install]") print("Usage: nap [auth|project|chat|plan|lineage|gateway|service|agent|feedback|status|update|where|install]")
print(" auth Auth commands: login-local, status.") print(" auth Auth commands: login-local, status.")
print(" Example: nap auth login-local") print(" Example: nap auth login-local")
print(" project Project commands: init/attach existing, create, status, encryption.") print(" project Project commands: init, attach, create, claim, status, encryption.")
print(" Examples: nap project init, nap project create, nap project status") print(" First-time setup: nap project init (enrolls and creates a project, no browser).")
print(" Attach an existing project to a Napseer account: nap project attach (opens browser).")
print(" Upgrade an anonymous project to a claimed one: nap project claim.")
print(" Encrypted project setup is completed by the local gateway: nap gateway vault process") print(" Encrypted project setup is completed by the local gateway: nap gateway vault process")
print(" chat Chat secret commands: secret status, secret setup, secret rotate.") print(" chat Chat secret commands: secret status, secret setup, secret rotate.")
print(" Examples: nap chat secret setup --vault-passphrase TEXT, nap chat secret status") print(" Examples: nap chat secret setup --vault-passphrase TEXT, nap chat secret status")
@@ -692,13 +709,15 @@ def main(argv):
print(" Example: nap plan to-kanban /plans/example --column todo") print(" Example: nap plan to-kanban /plans/example --column todo")
print(" lineage Generic source-record -> plan -> kanban lineage checks.") print(" lineage Generic source-record -> plan -> kanban lineage checks.")
print(" Example: nap lineage status") print(" Example: nap lineage status")
print(" gateway Gateway commands: start, stop, configure, setup, restart, kill, logs, status, vault.") print(" gateway Gateway operations: configure, setup, restart, kill, vault, terminal, schedule, service.")
print(" Start auto-selects a local port; use --port to pin one.") print(" Examples: nap gateway configure, nap gateway vault process --all")
print(" Examples: nap gateway start, nap gateway vault, nap gateway vault process --all") print(" service Local gateway service lifecycle: start, stop, status, logs.")
print(" Start auto-selects a port; pass --port to pin one.")
print(" Examples: nap service start, nap service logs --lines 200")
print(" agent Agent commands: list, workspaces, create, show, edit.") print(" agent Agent commands: list, workspaces, create, show, edit.")
print(" list shows registered project agents; workspaces shows /agents folders.") print(" list shows registered project agents; workspaces shows /agents folders.")
print(" feedback Feedback commands: list, global, status, resolve.") print(" feedback Feedback commands: list, global, set, resolve.")
print(" Examples: nap feedback list, nap feedback global --all, nap feedback resolve <id> --notes TEXT") print(" Examples: nap feedback list, nap feedback set <id> --status open, nap feedback resolve <id> --notes TEXT")
print(" status Show cwd Napseer configuration and next setup commands.") print(" status Show cwd Napseer configuration and next setup commands.")
print(" update Refresh the globally installed nap command and wrapper.") print(" update Refresh the globally installed nap command and wrapper.")
print(" where Show installed nap paths and cwd auth location.") print(" where Show installed nap paths and cwd auth location.")
+204
View File
@@ -0,0 +1,204 @@
#!/usr/bin/env python3
"""Smoke-test the consolidated CLI dispatch surface.
Verifies that:
- canonical verbs (status, list, create, update, delete, run, preregister, activate)
are accepted and route to the right function.
- dropped aliases (ls, new, send, rm, set-status, admin, global-list, etc.)
now raise RuntimeError with a clear pointer to the canonical verb.
- `nap project encryption set` requires --state; positional-state shortcut is gone.
- `nap project encryption plaintext` (action-as-state) now errors.
"""
import importlib.util
import pathlib
import sys
def load_module():
script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py"
sys.path.insert(0, str(script_path.parent))
spec = importlib.util.spec_from_file_location("napseer_mcp_server_dispatch_test", script_path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module
def expect_unknown(mod, argv, expected_fragment):
"""Run cli_main(argv) and assert it raises RuntimeError with the fragment."""
try:
mod.cli_main(argv)
except RuntimeError as exc:
assert expected_fragment in str(exc), (argv, str(exc))
else:
raise AssertionError(f"expected RuntimeError for {argv}")
def test_chat_secret_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "chat", "secret", "init"],
["napseer_mcp_server.py", "chat", "secret", "configure"],
]:
expect_unknown(mod, args, "unknown chat secret action")
def test_lineage_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "lineage", "list"],
["napseer_mcp_server.py", "lineage", "ls"],
["napseer_mcp_server.py", "lineage", "check"],
]:
expect_unknown(mod, args, "unknown lineage subcommand")
def test_plan_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "plan", "ls"],
["napseer_mcp_server.py", "plan", "active"],
["napseer_mcp_server.py", "plan", "card-from"],
["napseer_mcp_server.py", "plan", "create-card"],
]:
expect_unknown(mod, args, "unknown plan subcommand")
def test_feedback_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "feedback", "ls"],
["napseer_mcp_server.py", "feedback", "admin"],
["napseer_mcp_server.py", "feedback", "admin-list"],
["napseer_mcp_server.py", "feedback", "global-list"],
["napseer_mcp_server.py", "feedback", "status", "abc", "open"],
["napseer_mcp_server.py", "feedback", "set-status", "abc", "open"],
["napseer_mcp_server.py", "feedback", "update", "abc", "open"],
["napseer_mcp_server.py", "feedback", "resolved", "abc"],
]:
expect_unknown(mod, args, "unknown feedback subcommand")
def test_agent_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "agent", "ls"],
["napseer_mcp_server.py", "agent", "registered"],
["napseer_mcp_server.py", "agent", "workspace-list"],
]:
expect_unknown(mod, args, "unknown agent subcommand")
def test_gateway_status_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "gateway", "ls"],
["napseer_mcp_server.py", "gateway", "list"],
]:
expect_unknown(mod, args, "unknown gateway subcommand")
def test_gateway_terminal_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "gateway", "terminals", "list"],
["napseer_mcp_server.py", "gateway", "terminal", "new"],
["napseer_mcp_server.py", "gateway", "terminal", "rm", "x"],
["napseer_mcp_server.py", "gateway", "terminal", "send", "x"],
]:
expect_unknown(mod, args, "unknown gateway")
def test_gateway_schedule_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "gateway", "schedules", "list"],
["napseer_mcp_server.py", "gateway", "cron", "list"],
["napseer_mcp_server.py", "gateway", "crons", "list"],
["napseer_mcp_server.py", "gateway", "schedule", "ls"],
["napseer_mcp_server.py", "gateway", "schedule", "new"],
["napseer_mcp_server.py", "gateway", "schedule", "edit", "x"],
["napseer_mcp_server.py", "gateway", "schedule", "rm", "x"],
["napseer_mcp_server.py", "gateway", "schedule", "run-now", "x"],
]:
expect_unknown(mod, args, "unknown gateway")
def test_gateway_service_run_rejected():
"""Lifecycle verbs (run, start, etc.) moved to `nap service`."""
mod = load_module()
for args in [
["napseer_mcp_server.py", "gateway", "service", "run"],
["napseer_mcp_server.py", "gateway", "service", "start"],
["napseer_mcp_server.py", "gateway", "service", "register"],
["napseer_mcp_server.py", "gateway", "service", "bootstrap"],
]:
expect_unknown(mod, args, "nap service start")
def test_gateway_vault_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "gateway", "vault", "list"],
["napseer_mcp_server.py", "gateway", "vault", "ls"],
]:
expect_unknown(mod, args, "gateway vault action: list")
def test_gateway_rotate_passphrase_aliases_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "gateway", "passwd"],
["napseer_mcp_server.py", "gateway", "change-passphrase"],
]:
expect_unknown(mod, args, "rotate-passphrase")
def test_project_encryption_state_flag_required():
"""nap project encryption set requires --state; positional state is rejected."""
mod = load_module()
# Without --state, the dispatch should raise
expect_unknown(
mod,
["napseer_mcp_server.py", "project", "encryption", "set", "plaintext"],
"--state plaintext|encrypted",
)
def test_project_encryption_action_as_state_rejected():
"""`nap project encryption plaintext` (the old shortcut) must error."""
mod = load_module()
expect_unknown(
mod,
["napseer_mcp_server.py", "project", "encryption", "plaintext"],
"project encryption",
)
def test_project_encryption_show_alias_rejected():
mod = load_module()
for args in [
["napseer_mcp_server.py", "project", "encryption", "state"],
["napseer_mcp_server.py", "project", "encryption", "show"],
["napseer_mcp_server.py", "project", "encryption", "encrypted"],
]:
expect_unknown(mod, args, "project encryption")
if __name__ == "__main__":
test_chat_secret_aliases_rejected()
test_lineage_aliases_rejected()
test_plan_aliases_rejected()
test_feedback_aliases_rejected()
test_agent_aliases_rejected()
test_gateway_status_aliases_rejected()
test_gateway_terminal_aliases_rejected()
test_gateway_schedule_aliases_rejected()
test_gateway_service_run_rejected()
test_gateway_vault_aliases_rejected()
test_gateway_rotate_passphrase_aliases_rejected()
test_project_encryption_state_flag_required()
test_project_encryption_action_as_state_rejected()
test_project_encryption_show_alias_rejected()
print("ok: cli dispatch consolidation tests passed")
+18 -11
View File
@@ -25,7 +25,7 @@ def run():
assert calls == [] assert calls == []
assert mod.handle_gateway(["vault", "--help"]) == 0 assert mod.handle_gateway(["vault", "--help"]) == 0
assert calls == [] assert calls[-1] == ("gateway", ["vault", "--help"])
assert mod.handle_gateway(["vault", "list"]) == 0 assert mod.handle_gateway(["vault", "list"]) == 0
assert calls[-1] == ("gateway", ["vault", "list"]) assert calls[-1] == ("gateway", ["vault", "list"])
@@ -33,23 +33,24 @@ def run():
assert mod.handle_gateway(["vault", "rotate-secret", "--kind", "memory"]) == 0 assert mod.handle_gateway(["vault", "rotate-secret", "--kind", "memory"]) == 0
assert calls[-1] == ("gateway", ["vault", "rotate-secret", "--kind", "memory"]) assert calls[-1] == ("gateway", ["vault", "rotate-secret", "--kind", "memory"])
assert mod.handle_gateway(["process", "--all"]) == 0 try:
assert calls[-1] == ("gateway", ["vault", "process", "--all"]) mod.handle_gateway(["process", "--all"])
except RuntimeError as exc:
assert "nap gateway vault process" in str(exc), str(exc)
else:
raise AssertionError("deprecated gateway process shortcut must not route")
assert mod.main(["nap", "chat", "secret", "setup"]) == 0 assert mod.main(["nap", "chat", "secret", "setup"]) == 0
assert calls[-1] == ("chat", ["secret", "setup"]) assert calls[-1] == ("chat", ["secret", "setup"])
try: try:
mod.handle_gateway(["config"]) mod.handle_gateway(["config"])
except RuntimeError as exc: except RuntimeError as exc:
assert "unknown gateway command" in str(exc) assert "unknown gateway subcommand" in str(exc)
else: else:
raise AssertionError("gateway config alias must not route") raise AssertionError("gateway config alias must not route")
try: try:
mod.handle_gateway(["vault-setup", "list"]) mod.handle_gateway(["vault-setup", "list"])
except RuntimeError as exc: except RuntimeError as exc:
assert "unknown gateway command" in str(exc) assert "unknown gateway subcommand" in str(exc)
else: else:
raise AssertionError("deprecated gateway vault-setup alias must not route") raise AssertionError("deprecated gateway vault-setup alias must not route")
@@ -60,20 +61,26 @@ def run():
assert mod.main(["nap", "status"]) == 0 assert mod.main(["nap", "status"]) == 0
assert calls[-1] == ("configure", []) assert calls[-1] == ("configure", [])
# Install-level dispatch should reject: empty / config / configure / bootstrap / create / ui
for argv in ( for argv in (
["nap", "config"], ["nap", "config"],
["nap", "configure"], ["nap", "configure"],
["nap", "bootstrap"], ["nap", "bootstrap"],
["nap", "create"], ["nap", "create"],
["nap", "ui"], ["nap", "ui"],
["nap", "project", "bootstrap"],
): ):
try: try:
mod.main(argv) mod.main(argv)
except RuntimeError as exc: except RuntimeError as exc:
assert "unknown" in str(exc) assert "unknown" in str(exc), (argv, str(exc))
else: else:
raise AssertionError(f"deprecated route must not be accepted: {argv}") raise AssertionError(f"deprecated route must not be accepted at install level: {argv}")
# nap project bootstrap is now a deprecated alias handled by the MCP server
# (the install side forwards it; the MCP side raises with a clear error message).
forwarded = []
mod.run_wrapper = lambda command, args: forwarded.append((command, args)) or 0
mod.main(["nap", "project", "bootstrap"])
assert forwarded == [("project", ["bootstrap"])], forwarded
print("ok: nap installer gateway route smoke passed") print("ok: nap installer gateway route smoke passed")
+192
View File
@@ -0,0 +1,192 @@
#!/usr/bin/env python3
"""Smoke-test the `nap project init` first-time setup wizard."""
import importlib.util
import pathlib
import sys
PROJECT_ID = "22222222-2222-2222-2222-222222222222"
def load_module():
script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py"
sys.path.insert(0, str(script_path.parent))
spec = importlib.util.spec_from_file_location("napseer_mcp_server_project_init_test", script_path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module
def test_init_fresh_directory():
mod = load_module()
enrollment_calls = []
bootstrap_calls = []
saved_payloads = []
mod.TOKEN = None
mod.DEFAULT_PROJECT_ID = None
mod.AUTH = {}
mod.AUTH_PATH = pathlib.Path("/tmp/napseer-test/auth.json")
mod.BASE_URL = "https://api.example.test"
mod.load_public_auth_file = lambda: {}
def fake_ensure_enrolled(args):
enrollment_calls.append(args)
mod.TOKEN = "ssh-token"
return {"status": "enrolled", "token_expires_at": "2099-01-01T00:00:00Z"}
def fake_bootstrap_project(args):
bootstrap_calls.append(args)
mod.DEFAULT_PROJECT_ID = PROJECT_ID
return {
"status": "created",
"project": {"id": PROJECT_ID, "slug": args["slug"], "name": args["name"]},
"project_id": PROJECT_ID,
"encryption_state": "plaintext",
"project_unlocked": True,
"message": "created",
}
mod.ensure_enrolled = fake_ensure_enrolled
mod.bootstrap_project = fake_bootstrap_project
mod.save_auth = lambda updates: saved_payloads.append(dict(updates))
result = mod.cli_project_init([])
assert result["status"] == "initialized", result
assert result["project_id"] == PROJECT_ID
assert result["project"]["slug"], result
assert any(step.get("step") == "enroll" for step in result["steps"]), result["steps"]
assert any(step.get("step") == "create_project" for step in result["steps"]), result["steps"]
assert enrollment_calls, "ensure_enrolled should have been called"
assert bootstrap_calls, "bootstrap_project should have been called"
print("ok: project init fresh-directory wizard passed")
def test_init_already_initialized():
mod = load_module()
mod.TOKEN = "existing-token"
mod.DEFAULT_PROJECT_ID = PROJECT_ID
mod.AUTH = {"project_slug": "my-existing"}
mod.AUTH_PATH = pathlib.Path("/tmp/napseer-test/auth.json")
mod.load_public_auth_file = lambda: {"project_slug": "my-existing"}
enrollment_called = []
bootstrap_called = []
def fake_ensure_enrolled(args):
enrollment_called.append(args)
return {"status": "enrolled"}
def fake_bootstrap_project(args):
bootstrap_called.append(args)
return {}
mod.ensure_enrolled = fake_ensure_enrolled
mod.bootstrap_project = fake_bootstrap_project
result = mod.cli_project_init([])
assert result["status"] == "already_initialized", result
assert result["project_slug"] == "my-existing", result
assert not enrollment_called, "ensure_enrolled should NOT have been called"
assert not bootstrap_called, "bootstrap_project should NOT have been called"
print("ok: project init no-op on already-initialized folder passed")
def test_init_token_only_creates_project():
mod = load_module()
mod.TOKEN = "already-have-token"
mod.DEFAULT_PROJECT_ID = None
mod.AUTH = {}
mod.AUTH_PATH = pathlib.Path("/tmp/napseer-test/auth.json")
mod.load_public_auth_file = lambda: {}
enrollment_called = []
def fake_ensure_enrolled(args):
enrollment_called.append(args)
return {"status": "enrolled"}
def fake_bootstrap_project(args):
mod.DEFAULT_PROJECT_ID = PROJECT_ID
return {
"status": "created",
"project": {"id": PROJECT_ID, "slug": args["slug"], "name": args["name"]},
"project_id": PROJECT_ID,
"encryption_state": "plaintext",
"project_unlocked": True,
"message": "created",
}
mod.ensure_enrolled = fake_ensure_enrolled
mod.bootstrap_project = fake_bootstrap_project
result = mod.cli_project_init(["--slug", "my-slug", "--name", "My Name"])
assert result["status"] == "initialized", result
assert result["project_id"] == PROJECT_ID, result
assert not enrollment_called, "ensure_enrolled should NOT be called when TOKEN is already set"
enroll_step = next(s for s in result["steps"] if s["step"] == "enroll")
assert enroll_step["status"] == "skipped", enroll_step
create_step = next(s for s in result["steps"] if s["step"] == "create_project")
assert create_step["status"] == "created", create_step
assert fake_bootstrap_project.__call__ is not None
print("ok: project init creates project when token exists passed")
def test_cli_main_dispatches_init():
"""Verify the cli_main dispatch routes `project init` to cli_project_init."""
import io
import contextlib
mod = load_module()
mod.TOKEN = "t"
mod.DEFAULT_PROJECT_ID = PROJECT_ID
mod.AUTH = {"project_slug": "x"}
mod.AUTH_PATH = pathlib.Path("/tmp/napseer-test/auth.json")
mod.load_public_auth_file = lambda: {"project_slug": "x"}
captured = []
mod.cli_project_init = lambda args: (captured.append(args) or {"status": "already_initialized"})
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
rc = mod.cli_main(["napseer_mcp_server.py", "project", "init", "--no-browser"])
assert rc is None
assert captured == [["--no-browser"]]
assert '"already_initialized"' in buf.getvalue(), buf.getvalue()
print("ok: cli_main routes `project init` to cli_project_init passed")
def test_cli_main_rejects_old_aliases():
"""Verify `auth login` / `project bootstrap-existing` are no longer accepted."""
mod = load_module()
cases = [
(["napseer_mcp_server.py", "auth", "login"], "unknown auth subcommand: login"),
(["napseer_mcp_server.py", "auth", "operator-login"], "unknown auth subcommand: operator-login"),
(["napseer_mcp_server.py", "project", "login"], "unknown project subcommand: login"),
(["napseer_mcp_server.py", "project", "bootstrap-existing"], "unknown project subcommand: bootstrap-existing"),
]
for args, expected_substring in cases:
try:
mod.cli_main(args)
except RuntimeError as exc:
assert expected_substring in str(exc), (args, str(exc))
else:
raise AssertionError(f"expected RuntimeError for {args}")
print("ok: cli_main rejects old auth/project aliases with clear errors passed")
if __name__ == "__main__":
test_init_fresh_directory()
test_init_already_initialized()
test_init_token_only_creates_project()
test_cli_main_dispatches_init()
test_cli_main_rejects_old_aliases()
print("ok: project init wizard smoke tests passed")