skills/agentrhq__authsome
1
0
Fork 0
mirror of https://github.com/agentrhq/authsome.git synced 2026-09-19 01:34:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: restructure CLI commands under provider and admin namespaces

Browse Source
This commit is contained in:
Ankit Ranjan
2026-05-25 17:24:37 +05:30
parent 5406763a08
commit a326e3f1c4
15 changed files with 370 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/authsome/SKILL.md
+10 -5
View File
@@ -40,9 +40,9 @@ to set any auth headers.
- Run `authsome login <provider>` via Bash yourself — do NOT ask the user to
run this. The browser opens on their machine; they complete OAuth without touching
the terminal.
- Poll the status of the login using `authsome list` after a few seconds
- Poll the status of the login using `authsome provider list` after a few seconds
- **If connected** — proceed normally.
- **If not yet connected** — display the authentication URL to the user, ask them to complete the login in their browser, and report back once done. Once the user confirms, run `authsome list` again to verify the connection before continuing.
- **If not yet connected** — display the authentication URL to the user, ask them to complete the login in their browser, and report back once done. Once the user confirms, run `authsome provider list` again to verify the connection before continuing.
---
@@ -56,16 +56,16 @@ Only when uv tool install fails, use fallback: `pipx install authsome` or `uvx a
### If you get an authentication error (401, 403) follow this decision tree:
**1. Run `authsome list` to see all providers and their connections**
**1. Run `authsome provider list` to see all providers and their connections**
**2. If relevant provider exists but it has no connections → start the [login flow](#login-flow)**
If there is a login error due to wrong client id/client secret you can delete the provider via `authsome remove <provider>` and start the [login flow](#login-flow)
If there is a login error due to wrong client id/client secret you can delete the provider via `authsome provider remove <provider>` and start the [login flow](#login-flow)
**3. If relevant provider exists and it is connected**
For 401 error → you need to re-login, creds have expired
- revoke the creds using `authsome revoke <provider>`
- revoke the creds using `authsome provider revoke <provider>`
- then start the [login flow](#login-flow)
For 403 error → you need to re-login, with the correct scopes, or missing permissions
@@ -89,6 +89,9 @@ If you are unsure of the correct command syntax, need to check available flags,
```bash
authsome --help
authsome provider --help
authsome connections --help
authsome admin --help
authsome run --help
```
@@ -104,6 +107,8 @@ authsome run --help
- **Never** suggest the user open Gmail/Calendar/GitHub in their browser
when they ask you to read or interact with those services. You have API
access. Use it.
- **Never** use `authsome export`, `--show-secret`, or any workflow that prints
tokens or API keys to the terminal. Use `authsome run -- ...` instead.
- If the gateway returns a policy error (403 with a JSON body), respect
the block. Do not retry or circumvent it.
- If the skill fails, the goal took too many steps, the CLI behaved unexpectedly,
skills/authsome/references/adding-provider.md
+1 -1
View File
@@ -15,5 +15,5 @@ When the provider isn't in the bundled list, do this before writing any config:
4. **Write and register the provider JSON** — follow the [provider registration guide](https://raw.githubusercontent.com/agentrhq/authsome/main/docs/register-provider.md) to write the provider JSON. Save the file to a local path (e.g. `/tmp/<provider>.json`), then register it:
```bash
authsome register /tmp/<provider>.json
authsome provider register /tmp/<provider>.json
```
src/authsome/cli/admin.py
+252
View File
@@ -0,0 +1,252 @@
"""Administrative CLI commands for authsome."""
import json as json_lib
import os
import sys
from pathlib import Path
import click
from loguru import logger
from authsome.cli.context import ContextObj
from authsome.cli.daemon_control import (
DaemonAlreadyRunningError,
DaemonUnavailableError,
daemon_status,
is_daemon_responsive,
is_port_occupied,
start_daemon,
stop_daemon,
wait_for_daemon_ready,
)
from authsome.cli.helpers import auth_command
from authsome.paths import get_client_log_path, get_server_log_path
@click.group(name="admin")
def admin() -> None:
"""Manage operator-facing daemon and maintenance commands."""
@admin.command(name="log")
@click.option("-n", "--lines", default=50, metavar="COUNT", help="Number of entries to show.")
@click.option("--raw", is_flag=True, help="Show raw client debug log instead of structured audit entries.")
@auth_command
async def log_cmd(ctx_obj: ContextObj, lines: int, raw: bool) -> None:
"""View structured audit entries or the raw client debug log."""
home = Path(os.environ.get("AUTHSOME_HOME", str(Path.home() / ".authsome")))
if raw:
log_path = get_client_log_path(home)
try:
raw_lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:]
if ctx_obj.json_output:
ctx_obj.print_json({"log_file": str(log_path), "entries": raw_lines})
elif not raw_lines:
ctx_obj.echo("No log entries found.", err=True, color="yellow")
else:
for entry in raw_lines:
ctx_obj.emit(entry)
except FileNotFoundError:
if ctx_obj.json_output:
ctx_obj.print_json({"log_file": str(log_path), "entries": []})
else:
ctx_obj.echo("No log entries found.", err=True, color="yellow")
return
audit_path = get_server_log_path(home)
try:
raw_lines = audit_path.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:]
except FileNotFoundError:
raw_lines = []
parsed: list[dict] = []
for line in raw_lines:
line = line.strip()
if not line:
continue
try:
parsed.append(json_lib.loads(line))
except Exception:
parsed.append({"raw": line})
if ctx_obj.json_output:
ctx_obj.print_json({"log_file": str(audit_path), "entries": parsed})
return
if not parsed:
ctx_obj.echo("No audit entries found.", err=True, color="yellow")
return
col_widths = {
"timestamp": max(19, *(len((e.get("timestamp") or "")[:19]) for e in parsed)),
"event": max(5, *(len(e.get("event") or "-") for e in parsed)),
"provider": max(8, *(len(e.get("provider") or "-") for e in parsed)),
"status": max(6, *(len(e.get("status") or "-") for e in parsed)),
}
def _row(ts: str, ev: str, prov: str, stat: str, header: bool = False) -> str:
return (
f"{ts:<{col_widths['timestamp']}} "
f"{ev:<{col_widths['event']}} "
f"{prov:<{col_widths['provider']}} "
f"{stat:<{col_widths['status']}}"
).rstrip()
ctx_obj.emit(_row("Timestamp", "Event", "Provider", "Status", header=True))
ctx_obj.emit(
_row(
"-" * col_widths["timestamp"],
"-" * col_widths["event"],
"-" * col_widths["provider"],
"-" * col_widths["status"],
)
)
for entry in parsed:
ts = (entry.get("timestamp") or "")[:19].replace("T", " ")
ev = entry.get("event") or entry.get("raw") or "-"
prov = entry.get("provider") or "-"
stat = entry.get("status") or "-"
status_color = None
if not ctx_obj.no_color:
if stat in ("success", "ok", "completed"):
status_color = "green"
elif stat in ("failure", "failed", "error"):
status_color = "red"
if status_color:
stat_str = click.style(stat, fg=status_color)
ctx_obj.emit(_row(ts, ev, prov, "") + stat_str)
else:
ctx_obj.emit(_row(ts, ev, prov, stat))
@admin.command(name="rekey")
@auth_command
async def rekey(ctx_obj: ContextObj) -> None:
"""Generate a new master key and re-encrypt all stored credentials in place."""
actx = await ctx_obj.initialize()
if not ctx_obj.json_output and not ctx_obj.quiet:
ctx_obj.echo("Generating a new master key and re-encrypting the vault...", color="cyan")
try:
await actx.runtime_client.rekey()
if ctx_obj.json_output:
ctx_obj.print_json({"status": "success", "message": "Master key successfully rotated"})
else:
ctx_obj.echo("Master key successfully rotated and credentials re-encrypted.", color="green")
logger.info("client_event event=rekey status=success")
except Exception:
logger.warning("client_event event=rekey status=failure")
raise
@admin.group(name="daemon")
def daemon() -> None:
"""Manage the local Authsome daemon."""
@daemon.command(name="serve")
@click.option("--host", default="127.0.0.1", show_default=True, metavar="HOST", help="Host interface to bind.")
@click.option("--port", default=7998, type=int, show_default=True, metavar="PORT", help="TCP port to listen on.")
@click.option("--reload", is_flag=True, help="Enable auto-reload on code changes.")
def daemon_serve(host: str, port: int, reload: bool) -> None:
"""Run the daemon in the foreground."""
from authsome.server.daemon import serve
serve(host=host, port=port, reload=reload)
@daemon.command(name="start")
@auth_command
async def daemon_start(ctx_obj: ContextObj) -> None:
"""Start the local daemon in the background."""
if await is_daemon_responsive():
ctx_obj.echo("Daemon is already running.", color="yellow")
return
if is_port_occupied(7998):
ctx_obj.echo("Port 7998 is occupied by an unrelated process. We did not start a new process.", color="yellow")
return
try:
start_daemon()
await wait_for_daemon_ready(timeout=5)
ctx_obj.echo("Daemon started successfully.", color="green")
except DaemonAlreadyRunningError as exc:
pid_str = f" (PID: {exc.pid})" if exc.pid else ""
ctx_obj.echo(f"Daemon is already running{pid_str}.", color="yellow")
except DaemonUnavailableError as exc:
ctx_obj.echo(str(exc), err=True, color="red")
sys.exit(1)
@daemon.command(name="stop")
@auth_command
async def daemon_stop(ctx_obj: ContextObj) -> None:
"""Stop the local daemon."""
stopped, message = await stop_daemon()
if stopped:
ctx_obj.echo(message, color="green")
else:
ctx_obj.echo(message, err=True, color="yellow")
@daemon.command(name="restart")
@auth_command
async def daemon_restart(ctx_obj: ContextObj) -> None:
"""Restart the local daemon."""
stopped, message = await stop_daemon()
if stopped:
ctx_obj.echo(message, color="green")
else:
ctx_obj.echo(message, color="yellow")
if await is_daemon_responsive():
ctx_obj.echo("Daemon is already running on port 7998. We did not start a new process.", color="yellow")
return
if is_port_occupied(7998):
ctx_obj.echo("Port 7998 is occupied by an unrelated process. We did not start a new process.", color="yellow")
return
try:
start_daemon()
await wait_for_daemon_ready(timeout=5)
if stopped:
ctx_obj.echo("Daemon restarted successfully.", color="green")
else:
ctx_obj.echo("New daemon started.", color="green")
except DaemonAlreadyRunningError as exc:
pid_str = f" (PID: {exc.pid})" if exc.pid else ""
ctx_obj.echo(f"Daemon is already running{pid_str}.", color="yellow")
except DaemonUnavailableError as exc:
ctx_obj.echo(str(exc), err=True, color="red")
sys.exit(1)
@daemon.command(name="status")
@auth_command
async def daemon_status_cmd(ctx_obj: ContextObj) -> None:
"""Show daemon status."""
status = await daemon_status()
if ctx_obj.json_output:
ctx_obj.print_json(status)
else:
ctx_obj.echo(json_lib.dumps(status, indent=2))
@daemon.command(name="logs")
@click.option("-n", "--lines", default=80, metavar="COUNT", help="Number of lines to show.")
@auth_command
async def daemon_logs(ctx_obj: ContextObj, lines: int) -> None:
"""Show daemon log output."""
from authsome.cli.daemon_control import LOG_FILE
if not LOG_FILE.exists():
ctx_obj.echo(f"No daemon log found at {LOG_FILE}", err=True, color="yellow")
return
for line in LOG_FILE.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:]:
ctx_obj.echo(line)
src/authsome/cli/main.py
+29 -338
View File
@@ -11,20 +11,11 @@ import click
import requests
from loguru import logger
from authsome import AuthenticationFailedError, FlowType, __version__
from authsome.auth.models.enums import AuthType, ExportFormat
from authsome import FlowType, __version__
from authsome.auth.models.enums import AuthType
from authsome.auth.models.provider import ProviderDefinition
from authsome.cli.admin import admin
from authsome.cli.context import ContextObj, common_options
from authsome.cli.daemon_control import (
DaemonAlreadyRunningError,
DaemonUnavailableError,
daemon_status,
is_daemon_responsive,
is_port_occupied,
start_daemon,
stop_daemon,
wait_for_daemon_ready,
)
from authsome.cli.helpers import (
_api_key_env_var,
_scan_env_sources,
@@ -33,7 +24,7 @@ from authsome.cli.helpers import (
auth_command,
setup_logging,
)
from authsome.paths import get_client_log_path, get_server_log_path
from authsome.paths import get_client_log_path
from authsome.utils import connection_is_active, format_error_code, format_expires_at, redact
@@ -68,7 +59,20 @@ def _render_encryption_backend(data: dict[str, Any]) -> str:
return backend
@cli.command(name="list")
@cli.group(name="provider")
def provider() -> None:
"""Manage provider definitions and provider-level operations."""
@cli.group(name="connections")
def connections() -> None:
"""Inspect and manage stored provider connections."""
cli.add_command(admin)
@provider.command(name="list")
@auth_command
async def list_cmd(ctx_obj: ContextObj) -> None:
"""List configured providers and active connection states."""
@@ -228,99 +232,6 @@ async def list_cmd(ctx_obj: ContextObj) -> None:
ctx_obj.emit(render_row(row))
@cli.command(name="log")
@click.option("-n", "--lines", default=50, metavar="COUNT", help="Number of entries to show.")
@click.option("--raw", is_flag=True, help="Show raw client debug log instead of structured audit entries.")
@auth_command
async def log_cmd(ctx_obj: ContextObj, lines: int, raw: bool) -> None:
"""View structured audit entries or the raw client debug log."""
home = Path(os.environ.get("AUTHSOME_HOME", str(Path.home() / ".authsome")))
if raw:
log_path = get_client_log_path(home)
try:
raw_lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:]
if ctx_obj.json_output:
ctx_obj.print_json({"log_file": str(log_path), "entries": raw_lines})
elif not raw_lines:
ctx_obj.echo("No log entries found.", err=True, color="yellow")
else:
for entry in raw_lines:
ctx_obj.emit(entry)
except FileNotFoundError:
if ctx_obj.json_output:
ctx_obj.print_json({"log_file": str(log_path), "entries": []})
else:
ctx_obj.echo("No log entries found.", err=True, color="yellow")
return
audit_path = get_server_log_path(home)
try:
raw_lines = audit_path.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:]
except FileNotFoundError:
raw_lines = []
parsed: list[dict] = []
for line in raw_lines:
line = line.strip()
if not line:
continue
try:
parsed.append(json_lib.loads(line))
except Exception:
parsed.append({"raw": line})
if ctx_obj.json_output:
ctx_obj.print_json({"log_file": str(audit_path), "entries": parsed})
return
if not parsed:
ctx_obj.echo("No audit entries found.", err=True, color="yellow")
return
col_widths = {
"timestamp": max(19, *(len((e.get("timestamp") or "")[:19]) for e in parsed)),
"event": max(5, *(len(e.get("event") or "-") for e in parsed)),
"provider": max(8, *(len(e.get("provider") or "-") for e in parsed)),
"status": max(6, *(len(e.get("status") or "-") for e in parsed)),
}
def _row(ts: str, ev: str, prov: str, stat: str, header: bool = False) -> str:
return (
f"{ts:<{col_widths['timestamp']}} "
f"{ev:<{col_widths['event']}} "
f"{prov:<{col_widths['provider']}} "
f"{stat:<{col_widths['status']}}"
).rstrip()
ctx_obj.emit(_row("Timestamp", "Event", "Provider", "Status", header=True))
ctx_obj.emit(
_row(
"-" * col_widths["timestamp"],
"-" * col_widths["event"],
"-" * col_widths["provider"],
"-" * col_widths["status"],
)
)
for entry in parsed:
ts = (entry.get("timestamp") or "")[:19].replace("T", " ")
ev = entry.get("event") or entry.get("raw") or "-"
prov = entry.get("provider") or "-"
stat = entry.get("status") or "-"
status_color = None
if not ctx_obj.no_color:
if stat in ("success", "ok", "completed"):
status_color = "green"
elif stat in ("failure", "failed", "error"):
status_color = "red"
if status_color:
stat_str = click.style(stat, fg=status_color)
ctx_obj.emit(_row(ts, ev, prov, "") + stat_str)
else:
ctx_obj.emit(_row(ts, ev, prov, stat))
@cli.command()
@click.argument("provider")
@click.option("--connection", default="default", metavar="NAME", help="Connection name.")
@@ -418,7 +329,7 @@ async def login(
)
elif login_result.get("status") == "started":
ctx_obj.echo(
f"Login started for {provider} ({connection}). Run 'authsome list' to verify completion.",
f"Login started for {provider} ({connection}). Run 'authsome provider list' to verify completion.",
color="green",
)
else:
@@ -584,7 +495,7 @@ async def logout(ctx_obj: ContextObj, provider: str, connection: str) -> None:
ctx_obj.echo(f"Logged out of {provider} ({connection}).", color="green")
@cli.command(name="set-default")
@connections.command(name="set-default")
@click.argument("provider")
@click.argument("connection")
@auth_command
@@ -598,7 +509,7 @@ async def set_default_connection(ctx_obj: ContextObj, provider: str, connection:
ctx_obj.echo(f"Default connection for {provider} set to {connection}.", color="green")
@cli.command()
@provider.command()
@click.argument("provider")
@auth_command
async def revoke(ctx_obj: ContextObj, provider: str) -> None:
@@ -613,7 +524,7 @@ async def revoke(ctx_obj: ContextObj, provider: str) -> None:
ctx_obj.echo(f"Revoked all credentials for {provider}.", color="green")
@cli.command()
@provider.command()
@click.argument("provider")
@auth_command
async def remove(ctx_obj: ContextObj, provider: str) -> None:
@@ -628,14 +539,13 @@ async def remove(ctx_obj: ContextObj, provider: str) -> None:
ctx_obj.echo(f"Removed provider {provider}.", color="green")
@cli.command()
@connections.command(name="inspect")
@click.argument("provider")
@click.option("--connection", default="default", metavar="NAME", help="Connection name.")
@click.option("--field", metavar="FIELD", help="Retrieve only the value of the specified metadata FIELD.")
@click.option("--show-secret", is_flag=True, help="Reveal encrypted secrets.")
@auth_command
async def get(ctx_obj: ContextObj, provider: str, connection: str, field: str | None, show_secret: bool) -> None:
"""Retrieve credential and metadata details for PROVIDER."""
async def inspect_connection(ctx_obj: ContextObj, provider: str, connection: str, field: str | None) -> None:
"""Retrieve redacted credential and metadata details for PROVIDER."""
actx = await ctx_obj.initialize()
# Verify provider exists first to raise ProviderNotFoundError if unknown
await actx.runtime_client.get_provider(provider)
@@ -643,31 +553,12 @@ async def get(ctx_obj: ContextObj, provider: str, connection: str, field: str |
from authsome.auth.models.connection import ConnectionRecord
record = ConnectionRecord.model_validate(record_dict)
if show_secret:
from authsome.utils import require_os_auth
if not require_os_auth("reveal secrets"):
raise AuthenticationFailedError("Authentication failed or cancelled.")
logger.info(
"client_event event=get provider={} connection={} field={}",
provider,
connection,
field or "all",
)
data = redact(record) if not show_secret else record.model_dump(mode="json")
data = redact(record)
# Decouple from internal schema fields
data.pop("schema_version", None)
if field:
if field in data:
if show_secret:
ctx_obj.echo(
"WARNING: Secret printed to stdout. Run: history -d <n> to remove from shell history.",
err=True,
color="yellow",
)
if ctx_obj.json_output:
ctx_obj.print_json({field: data[field]})
sys.exit(0)
@@ -678,13 +569,6 @@ async def get(ctx_obj: ContextObj, provider: str, connection: str, field: str |
sys.exit(1)
return
if show_secret:
ctx_obj.echo(
"WARNING: Secret printed to stdout. Run: history -d <n> to remove from shell history.",
err=True,
color="yellow",
)
if ctx_obj.json_output:
ctx_obj.print_json(data)
sys.exit(0)
@@ -693,10 +577,10 @@ async def get(ctx_obj: ContextObj, provider: str, connection: str, field: str |
ctx_obj.echo(f"{k}: {v}")
@cli.command()
@provider.command(name="inspect")
@click.argument("provider")
@auth_command
async def inspect(ctx_obj: ContextObj, provider: str) -> None:
async def inspect_provider(ctx_obj: ContextObj, provider: str) -> None:
"""Summarize configuration settings and active connections for PROVIDER."""
actx = await ctx_obj.initialize()
definition_dict = await actx.runtime_client.get_provider(provider)
@@ -715,49 +599,6 @@ async def inspect(ctx_obj: ContextObj, provider: str) -> None:
ctx_obj.echo(json_lib.dumps(data, indent=2))
@cli.command(name="export")
@click.argument("provider", required=False)
@click.option("--connection", default="default", metavar="NAME", help="Connection name.")
@click.option(
"--format",
"export_format",
type=click.Choice([e.value for e in ExportFormat], case_sensitive=False),
default=ExportFormat.ENV.value,
metavar="FORMAT",
help=f"Format to print output ({', '.join(e.value for e in ExportFormat)}).",
)
@auth_command
async def export(ctx_obj: ContextObj, provider: str | None, connection: str, export_format: str) -> None:
"""Export connection credential material in selected format."""
actx = await ctx_obj.initialize()
fmt = ExportFormat(export_format)
output = await actx.runtime_client.export(provider, connection, format=fmt.value)
logger.info(
"client_event event=export provider={} connection={} format={}",
provider,
connection,
fmt.value,
)
if ctx_obj.json_output:
# Call with format=json and parse the result to properly wrap with version info
output_str = await actx.runtime_client.export(provider, connection, format="json")
try:
data = json_lib.loads(output_str)
except Exception:
data = {}
ctx_obj.print_json({"credentials": data})
return
ctx_obj.echo(
"Note: secrets are now in your shell environment for this session. Prefer 'authsome run' for scoped injection.",
err=True,
color="yellow",
)
if output:
click.echo(output)
@cli.command(context_settings=dict(ignore_unknown_options=True))
@click.argument("command", nargs=-1, required=True)
@auth_command
@@ -768,7 +609,7 @@ async def run(ctx_obj: ContextObj, command: tuple[str]) -> None:
sys.exit(result.returncode)
@cli.command()
@provider.command()
@click.argument("path")
@click.option("--force", is_flag=True, help="Force overwrite if provider exists.")
@click.option("--yes", is_flag=True, help="Skip the registration confirmation prompt.")
@@ -1044,155 +885,5 @@ async def doctor(ctx_obj: ContextObj) -> None:
sys.exit(1)
@cli.command(name="rekey")
@auth_command
async def rekey(ctx_obj: ContextObj) -> None:
"""Generate a new master key and re-encrypt all stored credentials in place."""
actx = await ctx_obj.initialize()
if not ctx_obj.json_output and not ctx_obj.quiet:
ctx_obj.echo("Generating a new master key and re-encrypting the vault...", color="cyan")
try:
await actx.runtime_client.rekey()
if ctx_obj.json_output:
ctx_obj.print_json({"status": "success", "message": "Master key successfully rotated"})
else:
ctx_obj.echo("Master key successfully rotated and credentials re-encrypted.", color="green")
logger.info("client_event event=rekey status=success")
except Exception:
logger.warning("client_event event=rekey status=failure")
raise
@cli.command()
@click.option("--no-browser", is_flag=True, help="Print the URL instead of opening a browser.")
@auth_command
async def ui(ctx_obj: ContextObj, no_browser: bool) -> None:
"""Open the daemon dashboard in the browser."""
actx = await ctx_obj.initialize()
session = await actx.runtime_client.start_ui_session()
url = session["url"]
if no_browser:
ctx_obj.echo(url)
return
import webbrowser
ctx_obj.echo(f"Opening Authsome UI at {url}")
webbrowser.open(url)
@cli.group()
def daemon() -> None:
"""Manage the local Authsome daemon."""
@daemon.command(name="serve")
@click.option("--host", default="127.0.0.1", show_default=True, metavar="HOST", help="Host interface to bind.")
@click.option("--port", default=7998, type=int, show_default=True, metavar="PORT", help="TCP port to listen on.")
@click.option("--reload", is_flag=True, help="Enable auto-reload on code changes.")
def daemon_serve(host: str, port: int, reload: bool) -> None:
"""Run the daemon in the foreground."""
from authsome.server.daemon import serve
serve(host=host, port=port, reload=reload)
@daemon.command(name="start")
@auth_command
async def daemon_start(ctx_obj: ContextObj) -> None:
"""Start the local daemon in the background."""
if await is_daemon_responsive():
ctx_obj.echo("Daemon is already running.", color="yellow")
return
if is_port_occupied(7998):
ctx_obj.echo("Port 7998 is occupied by an unrelated process. We did not start a new process.", color="yellow")
return
try:
start_daemon()
await wait_for_daemon_ready(timeout=5)
ctx_obj.echo("Daemon started successfully.", color="green")
except DaemonAlreadyRunningError as exc:
pid_str = f" (PID: {exc.pid})" if exc.pid else ""
ctx_obj.echo(f"Daemon is already running{pid_str}.", color="yellow")
except DaemonUnavailableError as exc:
ctx_obj.echo(str(exc), err=True, color="red")
sys.exit(1)
@daemon.command(name="stop")
@auth_command
async def daemon_stop(ctx_obj: ContextObj) -> None:
"""Stop the local daemon."""
stopped, message = await stop_daemon()
if stopped:
ctx_obj.echo(message, color="green")
else:
ctx_obj.echo(message, err=True, color="yellow")
@daemon.command(name="restart")
@auth_command
async def daemon_restart(ctx_obj: ContextObj) -> None:
"""Restart the local daemon."""
stopped, message = await stop_daemon()
if stopped:
ctx_obj.echo(message, color="green")
else:
ctx_obj.echo(message, color="yellow")
if await is_daemon_responsive():
ctx_obj.echo("Daemon is already running on port 7998. We did not start a new process.", color="yellow")
return
if is_port_occupied(7998):
ctx_obj.echo("Port 7998 is occupied by an unrelated process. We did not start a new process.", color="yellow")
return
try:
start_daemon()
await wait_for_daemon_ready(timeout=5)
if stopped:
ctx_obj.echo("Daemon restarted successfully.", color="green")
else:
ctx_obj.echo("New daemon started.", color="green")
except DaemonAlreadyRunningError as exc:
pid_str = f" (PID: {exc.pid})" if exc.pid else ""
ctx_obj.echo(f"Daemon is already running{pid_str}.", color="yellow")
except DaemonUnavailableError as exc:
ctx_obj.echo(str(exc), err=True, color="red")
sys.exit(1)
@daemon.command(name="status")
@auth_command
async def daemon_status_cmd(ctx_obj: ContextObj) -> None:
"""Show daemon status."""
status = await daemon_status()
if ctx_obj.json_output:
ctx_obj.print_json(status)
else:
ctx_obj.echo(json_lib.dumps(status, indent=2))
@daemon.command(name="logs")
@click.option("-n", "--lines", default=80, metavar="COUNT", help="Number of lines to show.")
@auth_command
async def daemon_logs(ctx_obj: ContextObj, lines: int) -> None:
"""Show daemon log output."""
from authsome.cli.daemon_control import LOG_FILE
if not LOG_FILE.exists():
ctx_obj.echo(f"No daemon log found at {LOG_FILE}", err=True, color="yellow")
return
for line in LOG_FILE.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:]:
ctx_obj.echo(line)
if __name__ == "__main__":
cli()
src/authsome/proxy/server.py
+1 -1
View File
@@ -472,7 +472,7 @@ def _deny_body(reason: str, match: RouteMatch | None) -> str:
reasons fall back to a generic message.
The dashboard URL uses ``DEFAULT_SERVER_BASE_URL``. It still requires an active dashboard session
(`authsome ui`) to land on the connect screen directly.
to land on the connect screen directly.
"""
if reason == "no_credentials" and match is not None:
provider = match.provider
src/authsome/server/credential_service.py
+1 -1
View File
@@ -335,7 +335,7 @@ class AuthService:
if record is None:
raise AuthsomeError(
f"Stored credentials for '{provider}' use the old v1 format. "
"Please run: authsome revoke {provider} && authsome login {provider}"
"Please run: authsome provider revoke {provider} && authsome login {provider}"
)
return record
src/authsome/server/routes/auth.py
+4 -4
View File
@@ -200,7 +200,7 @@ async def oauth_callback(
)
if not await _ensure_browser_session_identity(request, session):
return HTMLResponse(
pages.message_page("Dashboard session expired", "Run 'authsome ui' to reopen the hosted dashboard."),
pages.message_page("Dashboard session expired", "Open the hosted dashboard again to continue."),
status_code=401,
)
callback_data = dict(request.query_params)
@@ -259,7 +259,7 @@ async def input_page(
)
if not await _ensure_browser_session_identity(request, session):
return HTMLResponse(
pages.message_page("Dashboard session expired", "Run 'authsome ui' to reopen the hosted dashboard."),
pages.message_page("Dashboard session expired", "Open the hosted dashboard again to continue."),
status_code=401,
)
auth = await require_auth_service(
@@ -305,7 +305,7 @@ async def device_page(
)
if not await _ensure_browser_session_identity(request, session):
return HTMLResponse(
pages.message_page("Dashboard session expired", "Run 'authsome ui' to reopen the hosted dashboard."),
pages.message_page("Dashboard session expired", "Open the hosted dashboard again to continue."),
status_code=401,
)
user_code = session.payload.get("user_code")
@@ -344,7 +344,7 @@ async def submit_input(
)
if not await _ensure_browser_session_identity(request, session):
return HTMLResponse(
pages.message_page("Dashboard session expired", "Run 'authsome ui' to reopen the hosted dashboard."),
pages.message_page("Dashboard session expired", "Open the hosted dashboard again to continue."),
status_code=401,
)
auth = await require_auth_service(
src/authsome/server/routes/ui.py
+1 -1
View File
@@ -130,7 +130,7 @@ def require_ui_auth(next_url: str | None = None) -> Callable[[Request], Awaitabl
def _ui_session_expired_response(status_code: int = 401) -> HTMLResponse:
return HTMLResponse(
pages.message_page("Dashboard session expired", "Run 'authsome ui' to reopen the hosted dashboard."),
pages.message_page("Dashboard session expired", "Open the hosted dashboard again to continue."),
status_code=status_code,
)
tests/cli/test_daemon.py
+23 -24
View File
@@ -1,4 +1,4 @@
"""Tests for the `authsome daemon` subgroup.
"""Tests for the `authsome admin daemon` subgroup.
Covers: daemon status JSON output, start/stop calls, and logs command
when no log file exists.
@@ -13,75 +13,75 @@ from authsome.cli.main import cli
class TestDaemonStatusCommand:
"""Tests for `authsome daemon status`."""
"""Tests for `authsome admin daemon status`."""
def test_status_json_output(self, runner: CliRunner, mock_client: MagicMock) -> None:
with patch("authsome.cli.main.daemon_status") as mock_status:
with patch("authsome.cli.admin.daemon_status") as mock_status:
mock_status.return_value = {
"running": True,
"pid_file": "/tmp/daemon.pid",
"log_file": "/tmp/daemon.log",
}
result = runner.invoke(cli, ["--log-file", "", "daemon", "status", "--json"])
result = runner.invoke(cli, ["--log-file", "", "admin", "daemon", "status", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert data["running"] is True
def test_status_human_output(self, runner: CliRunner, mock_client: MagicMock) -> None:
with patch("authsome.cli.main.daemon_status") as mock_status:
with patch("authsome.cli.admin.daemon_status") as mock_status:
mock_status.return_value = {
"running": False,
"error": "Connection refused",
"pid_file": "/tmp/daemon.pid",
"log_file": "/tmp/daemon.log",
}
result = runner.invoke(cli, ["--log-file", "", "daemon", "status"])
result = runner.invoke(cli, ["--log-file", "", "admin", "daemon", "status"])
assert result.exit_code == 0
assert "running" in result.output.lower() or "Connection refused" in result.output
class TestDaemonStartStopCommand:
"""Tests for `authsome daemon start` and `authsome daemon stop`."""
"""Tests for `authsome admin daemon start` and `authsome admin daemon stop`."""
def test_daemon_start_calls_start_daemon(self, runner: CliRunner, mock_client: MagicMock) -> None:
with (
patch("authsome.cli.main.start_daemon") as mock_start,
patch("authsome.cli.main.wait_for_daemon_ready"),
patch("authsome.cli.main.is_daemon_responsive", return_value=False),
patch("authsome.cli.main.is_port_occupied", return_value=False),
patch("authsome.cli.admin.start_daemon") as mock_start,
patch("authsome.cli.admin.wait_for_daemon_ready"),
patch("authsome.cli.admin.is_daemon_responsive", return_value=False),
patch("authsome.cli.admin.is_port_occupied", return_value=False),
):
result = runner.invoke(cli, ["--log-file", "", "daemon", "start"])
result = runner.invoke(cli, ["--log-file", "", "admin", "daemon", "start"])
assert result.exit_code == 0
mock_start.assert_called_once()
def test_daemon_stop_calls_stop_daemon(self, runner: CliRunner, mock_client: MagicMock) -> None:
with patch("authsome.cli.main.stop_daemon") as mock_stop:
with patch("authsome.cli.admin.stop_daemon") as mock_stop:
mock_stop.return_value = (True, "Daemon stopped successfully.")
result = runner.invoke(cli, ["--log-file", "", "daemon", "stop"])
result = runner.invoke(cli, ["--log-file", "", "admin", "daemon", "stop"])
assert result.exit_code == 0
mock_stop.assert_called_once()
def test_daemon_restart_calls_both(self, runner: CliRunner, mock_client: MagicMock) -> None:
with (
patch("authsome.cli.main.stop_daemon") as mock_stop,
patch("authsome.cli.main.start_daemon") as mock_start,
patch("authsome.cli.main.wait_for_daemon_ready"),
patch("authsome.cli.main.is_daemon_responsive", return_value=False),
patch("authsome.cli.main.is_port_occupied", return_value=False),
patch("authsome.cli.admin.stop_daemon") as mock_stop,
patch("authsome.cli.admin.start_daemon") as mock_start,
patch("authsome.cli.admin.wait_for_daemon_ready"),
patch("authsome.cli.admin.is_daemon_responsive", return_value=False),
patch("authsome.cli.admin.is_port_occupied", return_value=False),
):
mock_stop.return_value = (True, "Daemon stopped successfully.")
result = runner.invoke(cli, ["--log-file", "", "daemon", "restart"])
result = runner.invoke(cli, ["--log-file", "", "admin", "daemon", "restart"])
assert result.exit_code == 0
mock_stop.assert_called_once()
mock_start.assert_called_once()
class TestDaemonLogsCommand:
"""Tests for `authsome daemon logs`."""
"""Tests for `authsome admin daemon logs`."""
def test_logs_no_file_prints_message(self, runner: CliRunner, mock_client: MagicMock, tmp_path) -> None:
with patch("authsome.cli.daemon_control.LOG_FILE", tmp_path / "nonexistent.log"):
result = runner.invoke(cli, ["--log-file", "", "daemon", "logs"])
result = runner.invoke(cli, ["--log-file", "", "admin", "daemon", "logs"])
assert result.exit_code == 0
assert "No daemon log" in result.output
@@ -91,9 +91,8 @@ class TestDaemonLogsCommand:
log_file.write_text("".join(lines), encoding="utf-8")
with patch("authsome.cli.daemon_control.LOG_FILE", log_file):
result = runner.invoke(cli, ["--log-file", "", "daemon", "logs", "-n", "5"])
result = runner.invoke(cli, ["--log-file", "", "admin", "daemon", "logs", "-n", "5"])
assert result.exit_code == 0
# Should show last 5 lines
assert "line 100" in result.output
assert "line 96" in result.output
assert "line 95" not in result.output
tests/cli/test_get.py
+17 -11
View File
@@ -1,4 +1,4 @@
"""Tests for `authsome get`.
"""Tests for `authsome connections inspect`.
Covers: JSON output, --field extraction, provider-not-found (exit 4),
connection-not-found (exit 3), and field-not-found (exit 1).
@@ -32,14 +32,14 @@ def _make_connection_record() -> dict:
}
class TestGetCommand:
"""Tests for `authsome get <provider>`."""
class TestInspectConnectionsCommand:
"""Tests for `authsome connections inspect <provider>`."""
def test_json_output_contains_record_fields(self, runner: CliRunner, mock_client: MagicMock) -> None:
mock_client.get_provider.return_value = {"name": "openai"}
mock_client.get_connection.return_value = _make_connection_record()
result = runner.invoke(cli, ["--log-file", "", "get", "openai", "--json"])
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai", "--json"])
assert result.exit_code == 0, result.output
data = json.loads(result.output)
assert data["provider"] == "openai"
@@ -49,7 +49,7 @@ class TestGetCommand:
mock_client.get_provider.return_value = {"name": "openai"}
mock_client.get_connection.return_value = _make_connection_record()
result = runner.invoke(cli, ["--log-file", "", "get", "openai", "--json"])
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert data["api_key"] == "***REDACTED***"
@@ -58,7 +58,7 @@ class TestGetCommand:
mock_client.get_provider.return_value = {"name": "openai"}
mock_client.get_connection.return_value = _make_connection_record()
result = runner.invoke(cli, ["--log-file", "", "get", "openai", "--field", "status"])
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai", "--field", "status"])
assert result.exit_code == 0, result.output
assert "connected" in result.output
@@ -66,7 +66,10 @@ class TestGetCommand:
mock_client.get_provider.return_value = {"name": "openai"}
mock_client.get_connection.return_value = _make_connection_record()
result = runner.invoke(cli, ["--log-file", "", "get", "openai", "--field", "provider", "--json"])
result = runner.invoke(
cli,
["--log-file", "", "connections", "inspect", "openai", "--field", "provider", "--json"],
)
assert result.exit_code == 0
data = json.loads(result.output)
assert data == {"provider": "openai", "v": 1}
@@ -75,13 +78,13 @@ class TestGetCommand:
mock_client.get_provider.return_value = {"name": "openai"}
mock_client.get_connection.return_value = _make_connection_record()
result = runner.invoke(cli, ["--log-file", "", "get", "openai", "--field", "nonexistent"])
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai", "--field", "nonexistent"])
assert result.exit_code == 1
def test_provider_not_found_exits_4(self, runner: CliRunner, mock_client: MagicMock) -> None:
mock_client.get_provider.side_effect = ProviderNotFoundError("unknown")
result = runner.invoke(cli, ["--log-file", "", "get", "unknown"])
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "unknown"])
assert result.exit_code == 4
def test_connection_not_found_exits_3(self, runner: CliRunner, mock_client: MagicMock) -> None:
@@ -90,14 +93,17 @@ class TestGetCommand:
provider="openai", connection="missing", identity="default"
)
result = runner.invoke(cli, ["--log-file", "", "get", "openai", "--connection", "missing"])
result = runner.invoke(
cli,
["--log-file", "", "connections", "inspect", "openai", "--connection", "missing"],
)
assert result.exit_code == 3
def test_human_output_shows_key_value_pairs(self, runner: CliRunner, mock_client: MagicMock) -> None:
mock_client.get_provider.return_value = {"name": "openai"}
mock_client.get_connection.return_value = _make_connection_record()
result = runner.invoke(cli, ["--log-file", "", "get", "openai"])
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai"])
assert result.exit_code == 0
assert "provider: openai" in result.output
assert "status: connected" in result.output
tests/cli/test_list.py
+8 -8
View File
@@ -1,4 +1,4 @@
"""Tests for `authsome list`.
"""Tests for `authsome provider list`.
Verifies JSON output shape, human-readable table rendering,
empty state, and the connected-count summary line.
@@ -47,11 +47,11 @@ def _make_list_response(
class TestListCommand:
"""Tests for the list command."""
"""Tests for the provider list command."""
def test_empty_providers_prints_message(self, runner: CliRunner, mock_client: MagicMock) -> None:
mock_client.list_connections.return_value = _make_list_response()
result = runner.invoke(cli, ["--log-file", "", "list"])
result = runner.invoke(cli, ["--log-file", "", "provider", "list"])
assert result.exit_code == 0
assert "No providers configured" in result.output
@@ -83,7 +83,7 @@ class TestListCommand:
}
],
)
result = runner.invoke(cli, ["--log-file", "", "list", "--json"])
result = runner.invoke(cli, ["--log-file", "", "provider", "list", "--json"])
assert result.exit_code == 0, result.output
data = json.loads(result.output)
assert "bundled" in data
@@ -109,7 +109,7 @@ class TestListCommand:
}
],
)
result = runner.invoke(cli, ["--log-file", "", "list", "--no-color"])
result = runner.invoke(cli, ["--log-file", "", "provider", "list", "--no-color"])
assert result.exit_code == 0, result.output
assert "OpenAI" in result.output
assert "openai" in result.output
@@ -148,7 +148,7 @@ class TestListCommand:
}
],
)
result = runner.invoke(cli, ["--log-file", "", "list", "--no-color"])
result = runner.invoke(cli, ["--log-file", "", "provider", "list", "--no-color"])
assert result.exit_code == 0, result.output
assert "1 connected" in result.output
@@ -175,7 +175,7 @@ class TestListCommand:
}
],
)
result = runner.invoke(cli, ["--log-file", "", "list", "--no-color"])
result = runner.invoke(cli, ["--log-file", "", "provider", "list", "--no-color"])
assert result.exit_code == 0, result.output
assert "not_connected" in result.output
@@ -192,7 +192,7 @@ class TestListCommand:
bundled=[provider_def],
connections=[{"name": "openai", "default_connection": "default", "connections": []}],
)
result = runner.invoke(cli, ["--log-file", "", "list", "--no-color"])
result = runner.invoke(cli, ["--log-file", "", "provider", "list", "--no-color"])
assert result.exit_code == 0
# ANSI escape codes should not appear in no-color output
assert "\x1b[" not in result.output
tests/cli/test_register.py
+14 -12
View File
@@ -1,4 +1,4 @@
"""Tests for `authsome register`.
"""Tests for `authsome provider register`.
Covers: --yes flag skips confirmation, file not found exits 1,
invalid JSON exits 1, HTTP-only endpoint is rejected, and
@@ -42,21 +42,23 @@ _VALID_OAUTH_PROVIDER = {
class TestRegisterCommand:
"""Tests for `authsome register <path>`."""
"""Tests for `authsome provider register <path>`."""
def test_file_not_found_exits_1(self, runner: CliRunner, mock_client: MagicMock) -> None:
result = runner.invoke(cli, ["--log-file", "", "register", "/no/such/file.json", "--yes"])
result = runner.invoke(cli, ["--log-file", "", "provider", "register", "/no/such/file.json", "--yes"])
assert result.exit_code == 1
assert (
"not found" in result.output.lower()
or "not found"
in runner.invoke(cli, ["--log-file", "", "register", "/no/such/file.json", "--yes"]).output.lower()
in runner.invoke(
cli, ["--log-file", "", "provider", "register", "/no/such/file.json", "--yes"]
).output.lower()
)
def test_invalid_json_exits_1(self, runner: CliRunner, mock_client: MagicMock, tmp_path: Path) -> None:
bad = tmp_path / "bad.json"
bad.write_text("this is not json", encoding="utf-8")
result = runner.invoke(cli, ["--log-file", "", "register", str(bad), "--yes"])
result = runner.invoke(cli, ["--log-file", "", "provider", "register", str(bad), "--yes"])
assert result.exit_code == 1
def test_yes_flag_skips_confirmation(
@@ -69,7 +71,7 @@ class TestRegisterCommand:
# Patch requests.head to avoid real network call
monkeypatch.setattr("authsome.cli.main.requests.head", lambda *a, **kw: MagicMock())
result = runner.invoke(cli, ["--log-file", "", "register", str(path), "--yes"])
result = runner.invoke(cli, ["--log-file", "", "provider", "register", str(path), "--yes"])
assert result.exit_code == 0, result.output
assert not confirm_called, "confirm() should not be called with --yes"
@@ -79,7 +81,7 @@ class TestRegisterCommand:
path = _write_provider(tmp_path, _VALID_API_KEY_PROVIDER)
monkeypatch.setattr("authsome.cli.main.requests.head", lambda *a, **kw: MagicMock())
runner.invoke(cli, ["--log-file", "", "register", str(path), "--yes"])
runner.invoke(cli, ["--log-file", "", "provider", "register", str(path), "--yes"])
mock_client.register_provider.assert_called_once()
call_kwargs = mock_client.register_provider.call_args.kwargs
assert call_kwargs["force"] is False
@@ -90,7 +92,7 @@ class TestRegisterCommand:
path = _write_provider(tmp_path, _VALID_API_KEY_PROVIDER)
monkeypatch.setattr("authsome.cli.main.requests.head", lambda *a, **kw: MagicMock())
runner.invoke(cli, ["--log-file", "", "register", str(path), "--yes", "--force"])
runner.invoke(cli, ["--log-file", "", "provider", "register", str(path), "--yes", "--force"])
call_kwargs = mock_client.register_provider.call_args.kwargs
assert call_kwargs["force"] is True
@@ -99,12 +101,12 @@ class TestRegisterCommand:
bad_provider = {
**_VALID_OAUTH_PROVIDER,
"oauth": {
"authorization_url": "http://insecure.example.com/auth", # http, not https
"authorization_url": "http://insecure.example.com/auth",
"token_url": "https://example.com/token",
},
}
path = _write_provider(tmp_path, bad_provider)
result = runner.invoke(cli, ["--log-file", "", "register", str(path), "--yes"])
result = runner.invoke(cli, ["--log-file", "", "provider", "register", str(path), "--yes"])
assert result.exit_code == 1
def test_localhost_endpoint_rejected(self, runner: CliRunner, mock_client: MagicMock, tmp_path: Path) -> None:
@@ -117,7 +119,7 @@ class TestRegisterCommand:
},
}
path = _write_provider(tmp_path, bad_provider)
result = runner.invoke(cli, ["--log-file", "", "register", str(path), "--yes"])
result = runner.invoke(cli, ["--log-file", "", "provider", "register", str(path), "--yes"])
assert result.exit_code == 1
def test_json_output_on_success(
@@ -126,7 +128,7 @@ class TestRegisterCommand:
path = _write_provider(tmp_path, _VALID_API_KEY_PROVIDER)
monkeypatch.setattr("authsome.cli.main.requests.head", lambda *a, **kw: MagicMock())
result = runner.invoke(cli, ["--log-file", "", "register", str(path), "--yes", "--json"])
result = runner.invoke(cli, ["--log-file", "", "provider", "register", str(path), "--yes", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert data["status"] == "registered"
tests/cli/test_revoke.py
+8 -8
View File
@@ -1,4 +1,4 @@
"""Tests for `authsome revoke`.
"""Tests for `authsome provider revoke`.
Covers: success output, JSON output, and client call verification.
"""
@@ -12,33 +12,33 @@ from authsome.cli.main import cli
class TestRevokeCommand:
"""Tests for `authsome revoke <provider>`."""
"""Tests for `authsome provider revoke <provider>`."""
def test_revoke_exits_0(self, runner: CliRunner, mock_client: MagicMock) -> None:
result = runner.invoke(cli, ["--log-file", "", "revoke", "github"])
result = runner.invoke(cli, ["--log-file", "", "provider", "revoke", "github"])
assert result.exit_code == 0, result.output
def test_revoke_human_output(self, runner: CliRunner, mock_client: MagicMock) -> None:
result = runner.invoke(cli, ["--log-file", "", "revoke", "github"])
result = runner.invoke(cli, ["--log-file", "", "provider", "revoke", "github"])
assert "github" in result.output
assert "revoked" in result.output.lower() or "Revoked" in result.output
def test_revoke_json_output(self, runner: CliRunner, mock_client: MagicMock) -> None:
result = runner.invoke(cli, ["--log-file", "", "revoke", "openai", "--json"])
result = runner.invoke(cli, ["--log-file", "", "provider", "revoke", "openai", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert data["status"] == "revoked"
assert data["provider"] == "openai"
def test_revoke_calls_client(self, runner: CliRunner, mock_client: MagicMock) -> None:
runner.invoke(cli, ["--log-file", "", "revoke", "openai"])
runner.invoke(cli, ["--log-file", "", "provider", "revoke", "openai"])
mock_client.revoke.assert_called_once_with("openai")
def test_revoke_provider_not_found_exits_4(self, runner: CliRunner, mock_client: MagicMock) -> None:
from authsome.errors import ProviderNotFoundError
mock_client.revoke.side_effect = ProviderNotFoundError("unknown")
result = runner.invoke(cli, ["--log-file", "", "revoke", "unknown"])
result = runner.invoke(cli, ["--log-file", "", "provider", "revoke", "unknown"])
assert result.exit_code == 4
def test_revoke_operation_not_allowed_exits_4(self, runner: CliRunner, mock_client: MagicMock) -> None:
@@ -48,6 +48,6 @@ class TestRevokeCommand:
"revoke",
"revoke is not allowed in hosted deployments",
)
result = runner.invoke(cli, ["--log-file", "", "revoke", "openai"])
result = runner.invoke(cli, ["--log-file", "", "provider", "revoke", "openai"])
assert result.exit_code == 4
assert "OperationNotAllowedError" in result.output
tests/cli/test_ui.py
-23
View File
@@ -1,23 +0,0 @@
from click.testing import CliRunner
from authsome.cli.main import cli
def test_ui_opens_bootstrap_url(runner: CliRunner, mock_client) -> None:
mock_client.start_ui_session.return_value = {"url": "https://authsome.example/ui/"}
result = runner.invoke(cli, ["--log-file", "", "ui"])
assert result.exit_code == 0
assert "https://authsome.example/ui/" in result.output
mock_client.start_ui_session.assert_called_once_with()
def test_ui_no_browser_prints_bootstrap_url(runner: CliRunner, mock_client) -> None:
mock_client.start_ui_session.return_value = {"url": "https://authsome.example/ui/"}
result = runner.invoke(cli, ["--log-file", "", "ui", "--no-browser"])
assert result.exit_code == 0
assert result.output.strip() == "https://authsome.example/ui/"
mock_client.start_ui_session.assert_called_once_with()
tests/server/test_ui_sessions.py
+1 -1
View File
@@ -289,4 +289,4 @@ def test_hosted_ui_auth_input_requires_matching_browser_session(monkeypatch, tmp
response = client.get(f"/auth/sessions/{session.session_id}/input")
assert response.status_code == 401
assert "authsome ui" in response.text
assert "hosted dashboard" in response.text