diff --git a/capture_tests/cbrain_cli_commands b/capture_tests/cbrain_cli_commands index 028629f..ad8b116 100755 --- a/capture_tests/cbrain_cli_commands +++ b/capture_tests/cbrain_cli_commands @@ -43,6 +43,9 @@ cbrain --json version cbrain whoami cbrain --json whoami +# Named session management +cbrain session list + # Bourreaux cbrain remote-resource list cbrain --json remote-resource list @@ -120,6 +123,7 @@ cbrain task operation hold # missing --task-id / --batch-id # ToolConfigs, as admin user ./switch_session admin +cbrain session list cbrain tool-config list cbrain --json tool-config list cbrain --jsonl tool-config list @@ -128,8 +132,9 @@ cbrain tool-config show 19 # not visible to normal user norm cbrain --json tool-config show 19 cbrain --jsonl tool-config show 19 -# ToolConfigs, as normal user -./switch_session norm +# ToolConfigs, as normal user (CLI switch_session; both sessions already planted) +cbrain switch_session norm +cbrain session list cbrain tool-config list cbrain --json tool-config list cbrain --jsonl tool-config list diff --git a/capture_tests/expected_captures.txt b/capture_tests/expected_captures.txt index da977c9..56f7fa6 100644 --- a/capture_tests/expected_captures.txt +++ b/capture_tests/expected_captures.txt @@ -123,6 +123,21 @@ Stdout: Stderr: (No output) +############################ +Command: cbrain session list +Status: 0 +Stdout: 337 bytes +Stderr: 0 bytes + +Stdout: +# SESSION USERNAME USER ID SERVER TIMESTAMP +------------------------------------------------------------------------------------------ +*1 norm norm 2 http://localhost:3000 2222-22-22T44:44 + +Active session: norm (* = active) +Stderr: +(No output) + ############################ Command: cbrain remote-resource list Status: 0 @@ -1073,6 +1088,22 @@ Stdout: Stderr: (No output) +############################ +Command: cbrain session list +Status: 0 +Stdout: 448 bytes +Stderr: 0 bytes + +Stdout: +# SESSION USERNAME USER ID SERVER TIMESTAMP +------------------------------------------------------------------------------------------ + 1 norm norm 2 http://localhost:3000 2222-22-22T44:44 +*2 admin admin 1 http://localhost:3000 2222-22-22T44:44 + +Active session: admin (* = active) +Stderr: +(No output) + ############################ Command: cbrain tool-config list Status: 0 @@ -1179,13 +1210,29 @@ Stderr: (No output) ############################ -Command: ./switch_session norm +Command: cbrain switch_session norm Status: 0 -Stdout: 0 bytes +Stdout: 71 bytes Stderr: 0 bytes Stdout: +Switched to session 'norm'. All future commands will use this session. +Stderr: (No output) + +############################ +Command: cbrain session list +Status: 0 +Stdout: 447 bytes +Stderr: 0 bytes + +Stdout: +# SESSION USERNAME USER ID SERVER TIMESTAMP +------------------------------------------------------------------------------------------ +*1 norm norm 2 http://localhost:3000 2222-22-22T44:44 + 2 admin admin 1 http://localhost:3000 2222-22-22T44:44 + +Active session: norm (* = active) Stderr: (No output) @@ -1720,12 +1767,14 @@ Stderr: ############################ Command: cbrain logout Status: 0 -Stdout: 116 bytes +Stdout: 266 bytes Stderr: 0 bytes Stdout: -Successfully logged out from CBRAIN server. -Local session removed from /home/runner/.config/cbrain/credentials.json +Successfully logged out from CBRAIN server as norm. +Local session 'norm' removed from /home/runner/.config/cbrain/credentials.json. +Successfully logged out from CBRAIN server as admin. +Local session 'admin' removed from /home/runner/.config/cbrain/credentials.json. Stderr: (No output) diff --git a/capture_tests/switch_session b/capture_tests/switch_session index fb0aaf2..c5afaf7 100755 --- a/capture_tests/switch_session +++ b/capture_tests/switch_session @@ -16,6 +16,7 @@ DEL_TOKEN="0123456789abcdefffffffffffffffff"; # cbrain client JSON file with credentials to update cbrain_cred_file=$HOME/.config/cbrain/credentials.json +mkdir -p "$(dirname "$cbrain_cred_file")" # Timestamp as expected by the cbrain client program timestamp=$(date +"%Y-%m-%dT%H:%M:%S") @@ -27,8 +28,15 @@ timestamp=$(date +"%Y-%m-%dT%H:%M:%S") # thus logging out the cbrain command. testses="$1" -# Stdout from now on goes into the JSON -exec 1> $cbrain_cred_file +# Keep prior named sessions so planting admin after norm does not wipe norm. +existing="{}" +if test -s "$cbrain_cred_file" ; then + existing=$(cat "$cbrain_cred_file") +fi + +# Stdout from now on goes into a temp JSON blob for this session only +tmp_cred=$(mktemp) +exec 1> "$tmp_cred" # Session for normal user if test "X$1" = "Xnorm" ; then @@ -37,6 +45,7 @@ if test "X$1" = "Xnorm" ; then "cbrain_url": "http://localhost:3000", "api_token": "$NORMAL_TOKEN", "user_id": 2, + "username": "norm", "timestamp": "$timestamp" } CREDJSON @@ -49,6 +58,7 @@ if test "X$1" = "Xadmin" ; then "cbrain_url": "http://localhost:3000", "api_token": "$ADMIN_TOKEN", "user_id": 1, + "username": "admin", "timestamp": "$timestamp" } CREDJSON @@ -61,6 +71,7 @@ if test "X$1" = "Xnormdel" ; then "cbrain_url": "http://localhost:3000", "api_token": "$DEL_TOKEN", "user_id": 2, + "username": "norm", "timestamp": "$timestamp" } CREDJSON @@ -70,6 +81,30 @@ fi exec 1>& - # Remove outright if it's empty -if ! test -s "$cbrain_cred_file" ; then - rm -f "$cbrain_cred_file" +if ! test -s "$tmp_cred" ; then + rm -f "$tmp_cred" "$cbrain_cred_file" + exit 0 fi + +# Merge this session into the named credentials map (multi-session). +python3 - "$cbrain_cred_file" "$testses" "$existing" "$tmp_cred" <<'PY' +import json, sys + +path, name, existing_raw, new_path = sys.argv[1:5] +with open(new_path) as f: + new = json.load(f) +try: + data = json.loads(existing_raw) if existing_raw.strip() else {} +except json.JSONDecodeError: + data = {} +# Promote legacy flat file to named map. +if "api_token" in data or "cbrain_url" in data: + if not any(isinstance(v, dict) and "api_token" in v for k, v in data.items() if k != "_active_session"): + data = {"default": {k: v for k, v in data.items() if k != "_active_session"}} +data[name] = new +data["_active_session"] = name +with open(path, "w") as f: + json.dump(data, f, indent=2) + f.write("\n") +PY +rm -f "$tmp_cred" diff --git a/cbrain_cli/cli_utils.py b/cbrain_cli/cli_utils.py index c8e616d..3618f37 100644 --- a/cbrain_cli/cli_utils.py +++ b/cbrain_cli/cli_utils.py @@ -11,10 +11,19 @@ from pathlib import Path from cbrain_cli import config as cbrain_config -from cbrain_cli.config import DEFAULT_HEADERS, DEFAULT_TIMEOUT, auth_headers +from cbrain_cli.config import ( + DEFAULT_HEADERS, + DEFAULT_TIMEOUT, + auth_headers, + cli_session_not_found, + resolve_session_credentials, +) _debug = False +session_name = "default" +session_specified = False + def set_debug(flag: bool) -> None: """Enable or disable debug output.""" @@ -44,7 +53,9 @@ def from_credentials(cls, timeout=None): """ Build a client from the saved credentials file. """ - creds = cbrain_config.load_credentials() or {} + all_creds = cbrain_config.load_credentials() or {} + name = session_name if session_specified else None + creds = resolve_session_credentials(all_creds, name) return cls( creds.get("cbrain_url", ""), creds.get("api_token", ""), @@ -84,7 +95,8 @@ def get(self, path, params=None): Authenticated GET; returns parsed JSON. """ raw, _ = self._request("GET", path, params=params) - return json.loads(raw.decode()) + decoded = raw.decode() + return json.loads(decoded) if decoded.strip() else {} def send(self, method, path, payload=None): """ @@ -173,6 +185,10 @@ def is_authenticated(): """ Check if the user is authenticated. """ + missing = cli_session_not_found() + if missing: + print(f"Session '{missing}' not found.") + return False client = CbrainClient.from_credentials() if not client.token or not client.base_url or not client.user_id: print("Not logged in. Use 'cbrain login' to login first.") @@ -211,7 +227,7 @@ def get_status_code_description(status_code): return f"HTTP error ({status_code})" -def handle_connection_error(error): +def handle_connection_error(error, server_url=None): """ Handle connection errors with informative messages including server URL. @@ -219,6 +235,8 @@ def handle_connection_error(error): ---------- error : Exception The connection error that occurred + server_url : str, optional + Server URL for errors before credentials exist (e.g. during login) Returns ------- @@ -303,10 +321,11 @@ def handle_connection_error(error): "Check your connection or set CBRAIN_TIMEOUT env var." ) elif "Connection refused" in str(error): - print( - "Error: Cannot connect to CBRAIN server at " - f"{CbrainClient.from_credentials().base_url}" - ) + url = server_url or CbrainClient.from_credentials().base_url + if url: + print(f"Error: Cannot connect to CBRAIN server at {url}") + else: + print("Error: Cannot connect to CBRAIN server.") print("Please check if the CBRAIN server is running and accessible.") else: print(f"Connection failed: {error.reason}") diff --git a/cbrain_cli/config.py b/cbrain_cli/config.py index 17af962..3196977 100644 --- a/cbrain_cli/config.py +++ b/cbrain_cli/config.py @@ -21,6 +21,10 @@ except ValueError: DEFAULT_TIMEOUT = 30 +# Key used inside credentials.json to track the currently active session. +# Prefixed with "_" so it is clearly not a session name. +ACTIVE_SESSION_KEY = "_active_session" + # HTTP headers. DEFAULT_HEADERS = { "Content-Type": "application/x-www-form-urlencoded", @@ -56,6 +60,84 @@ def load_credentials(): return None +def is_flat_credentials(data): + """True when file is single-session (api_token at top level), not named map.""" + if not isinstance(data, dict) or not data: + return False + if "api_token" in data or "cbrain_url" in data: + return not any( + k != ACTIVE_SESSION_KEY + and isinstance(v, dict) + and ("api_token" in v or "cbrain_url" in v) + for k, v in data.items() + ) + return False + + +def get_named_sessions(data): + """Return {name: creds} for flat or multi-session files.""" + if not data: + return {} + if is_flat_credentials(data): + return {"default": {k: v for k, v in data.items() if k != ACTIVE_SESSION_KEY}} + return {k: v for k, v in data.items() if k != ACTIVE_SESSION_KEY and isinstance(v, dict)} + + +def _cli_session_override(): + """Return --session name from argv when present.""" + from cbrain_cli.cli_utils import session_name, session_specified + + return session_name if session_specified else None + + +def cli_session_not_found(): + """Return session name when --session was given but is not saved locally.""" + from cbrain_cli.cli_utils import session_name, session_specified + + if not session_specified: + return None + if session_name in get_named_sessions(load_credentials() or {}): + return None + return session_name + + +def resolve_session_credentials(data, session_name=None): + """Pick active session dict from flat or multi-session credentials file.""" + if not data: + return {} + if session_name is None: + session_name = _cli_session_override() + if is_flat_credentials(data): + return {k: v for k, v in data.items() if k != ACTIVE_SESSION_KEY} + name = session_name or data.get(ACTIVE_SESSION_KEY) or "default" + entry = data.get(name) + return dict(entry) if isinstance(entry, dict) else {} + + +def update_active_credentials(updates=None, remove_keys=None, session_name=None): + """Patch fields on the active (or named) session; supports flat + nested files.""" + data = load_credentials() + if data is None: + return + updates = updates or {} + remove_keys = remove_keys or [] + if session_name is None: + session_name = _cli_session_override() + if is_flat_credentials(data): + data.update(updates) + for k in remove_keys: + data.pop(k, None) + save_credentials(data) + return + name = session_name or data.get(ACTIVE_SESSION_KEY) or "default" + entry = dict(data.get(name) or {}) + entry.update(updates) + for k in remove_keys: + entry.pop(k, None) + data[name] = entry + save_credentials(data) + + def save_credentials(credentials): """ Save credentials to the session file. diff --git a/cbrain_cli/data/data_providers.py b/cbrain_cli/data/data_providers.py index fdb7de2..9cf940a 100644 --- a/cbrain_cli/data/data_providers.py +++ b/cbrain_cli/data/data_providers.py @@ -22,8 +22,8 @@ def show_data_provider(args): """ # Get the data provider ID from the --id argument. data_provider_id = getattr(args, "id", None) - if not data_provider_id: - return list_data_providers(args) + if data_provider_id is None: + raise CliValidationError("Data provider ID is required", field="id") data = CbrainClient.from_credentials().get(f"/data_providers/{data_provider_id}") if data.get("error"): raise CliApiError(data.get("error")) diff --git a/cbrain_cli/data/projects.py b/cbrain_cli/data/projects.py index 66b7fb4..31d6a98 100644 --- a/cbrain_cli/data/projects.py +++ b/cbrain_cli/data/projects.py @@ -3,7 +3,11 @@ CliApiError, CliValidationError, ) -from cbrain_cli.config import load_credentials, save_credentials +from cbrain_cli.config import ( + load_credentials, + resolve_session_credentials, + update_active_credentials, +) def switch_project(args): @@ -45,11 +49,13 @@ def switch_project(args): else: group_data = client.get(f"/groups/{group_id}") - credentials = load_credentials() - if credentials is not None: - credentials["current_group_id"] = group_id - credentials["current_group_name"] = group_data.get("name", "Unknown") - save_credentials(credentials) + if load_credentials() is not None: + update_active_credentials( + { + "current_group_id": group_id, + "current_group_name": group_data.get("name", "Unknown"), + } + ) return group_data @@ -73,16 +79,15 @@ def unswitch_project(args): previous_group_name = None if credentials is not None: - previous_group_id = credentials.get("current_group_id") - previous_group_name = credentials.get("current_group_name") + active = resolve_session_credentials(credentials) + previous_group_id = active.get("current_group_id") + previous_group_name = active.get("current_group_name") if previous_group_id: CbrainClient.from_credentials().send("POST", "/groups/switch") if credentials is not None: - credentials.pop("current_group_id", None) - credentials.pop("current_group_name", None) - save_credentials(credentials) + update_active_credentials(remove_keys=["current_group_id", "current_group_name"]) return { "previous_group_id": previous_group_id, @@ -122,7 +127,8 @@ def show_project(args): if credentials is None: return None - current_group_id = credentials.get("current_group_id") + active = resolve_session_credentials(credentials) + current_group_id = active.get("current_group_id") if not current_group_id: return None @@ -130,16 +136,14 @@ def show_project(args): if current_group_id == "all": return { "id": "all", - "name": credentials.get("current_group_name") or "all", + "name": active.get("current_group_name") or "all", } try: return CbrainClient.from_credentials().get(f"/groups/{current_group_id}") except CliApiError as e: if e.status == 404: - credentials.pop("current_group_id", None) - credentials.pop("current_group_name", None) - save_credentials(credentials) + update_active_credentials(remove_keys=["current_group_id", "current_group_name"]) raise CliApiError(f"Current project (ID {current_group_id}) no longer exists") from None raise diff --git a/cbrain_cli/main.py b/cbrain_cli/main.py index 18ffbfc..9eddc7b 100644 --- a/cbrain_cli/main.py +++ b/cbrain_cli/main.py @@ -5,6 +5,8 @@ import argparse import sys +from cbrain_cli import cli_utils +from cbrain_cli import config as cbrain_config from cbrain_cli.cli_utils import ( PAGINATABLE_ACTIONS, CliValidationError, @@ -48,7 +50,7 @@ handle_tool_list, handle_tool_show, ) -from cbrain_cli.sessions import create_session, logout_session +from cbrain_cli.sessions import create_session, list_sessions, logout_session, switch_session from cbrain_cli.users import whoami_user @@ -77,6 +79,12 @@ def build_parser(): action="store_true", help="Print sanitized request/response diagnostics to stderr", ) + parser.add_argument( + "--session", + type=str, + default=None, + help="Session name to use (default: active session, or 'default')", + ) subparsers = parser.add_subparsers(dest="command", help="Available commands") @@ -87,17 +95,49 @@ def build_parser(): # MARK: Session commands (top-level) # Create new session. login_parser = subparsers.add_parser("login", help="Login to CBRAIN") + login_parser.add_argument( + "--session", type=str, default=argparse.SUPPRESS, help="Session name to use" + ) + login_parser.add_argument("-u", "--username", type=str, help="CBRAIN username") + login_parser.add_argument("-s", "--server", type=str, help="CBRAIN server URL") login_parser.set_defaults(func=handle_errors(create_session)) # Logout session. logout_parser = subparsers.add_parser("logout", help="Logout from CBRAIN") + logout_parser.add_argument( + "--session", + type=str, + default=argparse.SUPPRESS, + help="Session name to logout (default: all sessions)", + ) logout_parser.set_defaults(func=handle_errors(logout_session)) # Show current session. whoami_parser = subparsers.add_parser("whoami", help="Show current session") + whoami_parser.add_argument( + "--session", type=str, default=argparse.SUPPRESS, help="Session name to show" + ) whoami_parser.add_argument("-v", "--version", action="store_true", help="Show version") whoami_parser.set_defaults(func=handle_errors(whoami_user)) + # Switch active session. + switch_session_parser = subparsers.add_parser( + "switch_session", + help="Switch the default session (e.g. cbrain switch_session prod)", + ) + switch_session_parser.add_argument( + "session_target", + type=str, + help="Name of the session to make the default", + ) + switch_session_parser.set_defaults(func=handle_errors(switch_session)) + + # Session management sub-commands. + session_parser = subparsers.add_parser("session", help="Session management") + session_subparsers = session_parser.add_subparsers(dest="action", help="Session actions") + session_list_parser = session_subparsers.add_parser("list", help="List all saved sessions") + session_list_parser.set_defaults(func=handle_errors(list_sessions)) + # MARK: Model-based commands # File commands file_parser = subparsers.add_parser("file", help="File operations") @@ -536,6 +576,7 @@ def build_parser(): "background": background_parser, "task": task_parser, "remote-resource": remote_resource_parser, + "session": session_parser, } return parser, command_parsers @@ -558,6 +599,13 @@ def main(argv=None): args = parser.parse_args(argv) set_debug(getattr(args, "debug", False)) + explicit_session = getattr(args, "session", None) + session_val = explicit_session + if not session_val: + _all = cbrain_config.load_credentials() or {} + session_val = _all.get(cbrain_config.ACTIVE_SESSION_KEY) or None + cli_utils.session_specified = bool(explicit_session) + cli_utils.session_name = session_val or "default" if not args.command: parser.print_help() @@ -582,6 +630,13 @@ def main(argv=None): return handle_errors(version_info)(args) elif args.command == "whoami": return handle_errors(whoami_user)(args) + elif args.command == "switch_session": + return handle_errors(switch_session)(args) + elif args.command == "session": + if not getattr(args, "action", None): + command_parsers["session"].print_help() + return 1 + return args.func(args) # All other commands require authentication. if not is_authenticated(): diff --git a/cbrain_cli/sessions.py b/cbrain_cli/sessions.py index 6d27cec..1893ad9 100644 --- a/cbrain_cli/sessions.py +++ b/cbrain_cli/sessions.py @@ -1,14 +1,97 @@ import datetime import getpass +import os import urllib.error +from cbrain_cli import cli_utils from cbrain_cli import config as cbrain_config from cbrain_cli.cli_utils import ( CbrainClient, CliApiError, CliValidationError, + handle_connection_error, ) -from cbrain_cli.config import DEFAULT_BASE_URL +from cbrain_cli.config import ( + ACTIVE_SESSION_KEY, + DEFAULT_BASE_URL, + get_named_sessions, + is_flat_credentials, + resolve_session_credentials, +) + +# MARK: Switch Session + + +def switch_session(args): + """Switch the default session used by bare commands.""" + target = getattr(args, "session_target", None) + if not target: + print("Usage: cbrain switch_session ") + return 1 + + if not cbrain_config.CREDENTIALS_FILE.exists(): + print("No saved sessions. Use 'cbrain login' to create one.") + return 1 + + all_creds = cbrain_config.load_credentials() + if all_creds is None: + print(f"Error: credentials file is corrupted ({cbrain_config.CREDENTIALS_FILE}).") + return 1 + + sessions = get_named_sessions(all_creds) + if target not in sessions: + available = ", ".join(sessions) or "(none)" + print(f"Session '{target}' not found. Available sessions: {available}") + return 1 + + # Promote flat file to named map so _active_session can live alongside entries. + if is_flat_credentials(all_creds): + all_creds = {ACTIVE_SESSION_KEY: target, "default": sessions["default"]} + else: + all_creds[ACTIVE_SESSION_KEY] = target + + cbrain_config.save_credentials(all_creds) + print(f"Switched to session '{target}'. All future commands will use this session.") + return 0 + + +# MARK: List Sessions + + +def list_sessions(args): + """List all saved sessions, marking the currently active one with '*'.""" + if not cbrain_config.CREDENTIALS_FILE.exists(): + print("No saved sessions. Use 'cbrain login' to create one.") + return 0 + + all_creds = cbrain_config.load_credentials() + if all_creds is None: + print(f"Error: credentials file is corrupted ({cbrain_config.CREDENTIALS_FILE}).") + return 1 + + active = ( + all_creds.get(ACTIVE_SESSION_KEY, "default") + if not is_flat_credentials(all_creds) + else "default" + ) + sessions = get_named_sessions(all_creds) + + if not sessions: + print("No saved sessions. Use 'cbrain login' to create one.") + return 0 + + print(f"{'#':<4} {'SESSION':<20} {'USERNAME':<16} {'USER ID':<10} {'SERVER':<35} {'TIMESTAMP'}") + print("-" * 90) + for idx, (name, c) in enumerate(sessions.items(), start=1): + marker = "*" if name == active else " " + print( + f"{marker}{idx:<3} {name:<20} {c.get('username', '(unknown)'):<16} " + f"{c.get('user_id', 'N/A')!s:<10} {c.get('cbrain_url', 'N/A'):<35} " + f"{c.get('timestamp', 'N/A')}" + ) + + print(f"\nActive session: {active} (* = active)") + return 0 # MARK: Create Session. @@ -26,48 +109,88 @@ def create_session(args): int Exit code (0 on success, 1 on failure). """ + target_session = getattr(args, "session", None) or ( + cli_utils.session_name if cli_utils.session_specified else "default" + ) if cbrain_config.CREDENTIALS_FILE.exists(): - creds = cbrain_config.load_credentials() - if creds and creds.get("api_token") and creds.get("cbrain_url"): - # File alone is not enough, probe server to detect expired tokens. - try: - CbrainClient.from_credentials().get("/session") - except CliApiError as e: - if e.status == 401: - print("Saved session expired. Please log in again.") - elif e.status >= 500: - print(f"Server returned HTTP {e.status} during session check.") - print("The server may be temporarily unavailable. Try again later.") + all_creds = cbrain_config.load_credentials() + if all_creds: + named = get_named_sessions(all_creds) + if target_session in named: + existing = named[target_session] + elif is_flat_credentials(all_creds) and target_session == "default": + existing = resolve_session_credentials(all_creds) + else: + existing = {} + if existing.get("api_token") and existing.get("cbrain_url"): + # File alone is not enough, probe server to detect expired tokens. + try: + CbrainClient( + existing["cbrain_url"], + existing.get("api_token"), + existing.get("user_id"), + ).get("/session") + except CliApiError as e: + if e.status == 401: + print("Saved session expired. Please log in again.") + elif e.status >= 500: + print(f"Server returned HTTP {e.status} during session check.") + print("The server may be temporarily unavailable. Try again later.") + return 1 + else: + print(f"Server returned HTTP {e.status} during session check.") + print("Use 'cbrain logout' to reset local credentials.") + return 1 + except urllib.error.URLError: + print(f"Cannot reach CBRAIN server at {existing['cbrain_url']}.") + print("Check your connection. Use 'cbrain logout' to reset local credentials.") return 1 else: - print(f"Server returned HTTP {e.status} during session check.") - print("Use 'cbrain logout' to reset local credentials.") + label = f" to session '{target_session}'" if target_session != "default" else "" + print(f"Already logged in{label}. Use 'cbrain logout' to logout.") return 1 - except urllib.error.URLError: - print(f"Cannot reach CBRAIN server at {creds['cbrain_url']}.") - print("Check your connection. Use 'cbrain logout' to reset local credentials.") - return 1 - else: - print("Already logged in. Use 'cbrain logout' to logout.") - return 1 - # Get user input. - cbrain_url = input("Enter CBRAIN server base URL [default: localhost:3000]: ").strip() - if not cbrain_url: - cbrain_url = DEFAULT_BASE_URL + cbrain_url = getattr(args, "server", None) + if cbrain_url is None: + cbrain_url = ( + input("Enter CBRAIN server base URL [default: localhost:3000]: ").strip() + or DEFAULT_BASE_URL + ) + else: + cbrain_url = str(cbrain_url).strip() + if not cbrain_url: + raise CliValidationError("Server URL is required", field="server") - username = input("Enter CBRAIN username: ").strip() + username = getattr(args, "username", None) + if username is None: + username = input("Enter CBRAIN username: ").strip() + else: + username = str(username).strip() if not username: raise CliValidationError("Username is required", field="username") - password = getpass.getpass("Enter CBRAIN password: ") + password = getattr(args, "password", None) + if password is None and "CBRAIN_PASSWORD" in os.environ: + password = os.environ["CBRAIN_PASSWORD"] + if password is None: + password = getpass.getpass("Enter CBRAIN password: ") if not password: raise CliValidationError("Password is required", field="password") - response_data = CbrainClient(cbrain_url).post_form( - "/session", {"login": username, "password": password} - ) + try: + response_data = CbrainClient(cbrain_url).post_form( + "/session", {"login": username, "password": password} + ) + except CliApiError as e: + if e.status == 401: + print("Authentication error (401): Unauthorized") + print("Error: Login failed. Check username and password.") + return 1 + raise + except urllib.error.URLError as e: + handle_connection_error(e, server_url=cbrain_url) + return 1 cbrain_api_token = response_data.get("cbrain_api_token") cbrain_user_id = response_data.get("user_id") @@ -80,65 +203,133 @@ def create_session(args): "cbrain_url": cbrain_url, "api_token": cbrain_api_token, "user_id": cbrain_user_id, + "username": username, "timestamp": datetime.datetime.now().isoformat(), } - cbrain_config.save_credentials(credentials) - - print(f"Connection successful, API token saved in {cbrain_config.CREDENTIALS_FILE}") + # Named sessions → nested map; bare login keeps flat file (main-compatible). + if target_session != "default" or cli_utils.session_specified: + on_disk = cbrain_config.load_credentials() or {} + if is_flat_credentials(on_disk): + on_disk = {"default": {k: v for k, v in on_disk.items() if k != ACTIVE_SESSION_KEY}} + on_disk[target_session] = credentials + on_disk[ACTIVE_SESSION_KEY] = target_session + cbrain_config.save_credentials(on_disk) + print( + f"Connection successful, API token saved in {cbrain_config.CREDENTIALS_FILE} " + f"for session '{target_session}'" + ) + else: + cbrain_config.save_credentials(credentials) + print(f"Connection successful, API token saved in {cbrain_config.CREDENTIALS_FILE}") return 0 # MARK: Logout -def logout_session(args): - """ - Logout from CBRAIN by deleting the session file. - Parameters - ---------- - args : argparse.Namespace - Parsed command-line arguments (unused). - Returns - ------- - int - Exit code (0 on success). +def logout_session(args): """ + Logout from CBRAIN. + Without ``--session``: logout all sessions (or the single flat session). + With ``--session ``: logout only that session. + """ if not cbrain_config.CREDENTIALS_FILE.exists(): print("Not logged in. Use 'cbrain login' to login first.") return 0 - credentials = cbrain_config.load_credentials() - if credentials is None: + all_creds = cbrain_config.load_credentials() + if all_creds is None: print("Invalid credentials file. Removing local session.") cbrain_config.CREDENTIALS_FILE.unlink(missing_ok=True) print(f"Local session removed from {cbrain_config.CREDENTIALS_FILE}") return 0 - cbrain_url = credentials.get("cbrain_url") - api_token = credentials.get("api_token") - if not cbrain_url or not api_token: - print("Invalid credentials file. Removing local session.") - cbrain_config.CREDENTIALS_FILE.unlink(missing_ok=True) - print(f"Local session removed from {cbrain_config.CREDENTIALS_FILE}") + sessions = get_named_sessions(all_creds) + flat = is_flat_credentials(all_creds) + + target = getattr(args, "session", None) or ( + cli_utils.session_name if cli_utils.session_specified else None + ) + if target: + sessions_to_logout = [target] + else: + sessions_to_logout = list(sessions) + + if not sessions_to_logout: + print("Not logged in. Use 'cbrain login' to login first.") return 0 - try: - _, status = CbrainClient.from_credentials().send("DELETE", "/session") - if status == 200: - print("Successfully logged out from CBRAIN server.") - else: - print("Logout failed") - except CliApiError as e: - if e.status == 401: - print("Session already expired on server.") - else: - print(f"Logout request failed: HTTP {e.status}") - except urllib.error.URLError as e: - print(f"Network error during logout: {e}") + for s_name in sessions_to_logout: + creds = sessions.get(s_name, {}) + s_url, s_token = creds.get("cbrain_url"), creds.get("api_token") - if cbrain_config.CREDENTIALS_FILE.exists(): - cbrain_config.CREDENTIALS_FILE.unlink() - print(f"Local session removed from {cbrain_config.CREDENTIALS_FILE}") + if not s_url or not s_token: + if s_name in sessions: + print(f"Invalid credentials for session '{s_name}'. Removing local session.") + if flat: + cbrain_config.CREDENTIALS_FILE.unlink(missing_ok=True) + print(f"Local session removed from {cbrain_config.CREDENTIALS_FILE}") + return 0 + all_creds.pop(s_name, None) + elif target: + print(f"Not logged in to session '{s_name}'.") + elif len(sessions_to_logout) == 1: + print("Not logged in. Use 'cbrain login' to login first.") + if flat: + cbrain_config.CREDENTIALS_FILE.unlink(missing_ok=True) + print(f"Local session removed from {cbrain_config.CREDENTIALS_FILE}") + return 0 + continue + + display_name = creds.get("username", s_name) + try: + _, status = CbrainClient(s_url, s_token, creds.get("user_id")).send( + "DELETE", "/session" + ) + if status == 200: + if flat or not target and len(sessions_to_logout) == 1: + print("Successfully logged out from CBRAIN server.") + else: + print(f"Successfully logged out from CBRAIN server as {display_name}.") + else: + print(f"Logout failed for session '{s_name}'." if not flat else "Logout failed") + except CliApiError as e: + if e.status == 401: + print( + "Session already expired on server." + if flat + else f"Session '{s_name}' already expired on server." + ) + else: + print( + f"Logout request failed: HTTP {e.status}" + if flat + else f"Logout request failed for '{s_name}': HTTP {e.status}" + ) + except urllib.error.URLError as e: + print( + f"Network error during logout: {e}" + if flat + else f"Network error during logout for '{s_name}': {e}" + ) + + if flat: + if cbrain_config.CREDENTIALS_FILE.exists(): + cbrain_config.CREDENTIALS_FILE.unlink() + print(f"Local session removed from {cbrain_config.CREDENTIALS_FILE}") + return 0 + + all_creds.pop(s_name, None) + print(f"Local session '{s_name}' removed from {cbrain_config.CREDENTIALS_FILE}.") + + if not flat: + remaining = get_named_sessions(all_creds) + if not remaining: + cbrain_config.CREDENTIALS_FILE.unlink(missing_ok=True) + else: + if all_creds.get(ACTIVE_SESSION_KEY) not in remaining: + all_creds[ACTIVE_SESSION_KEY] = next(iter(remaining)) + cbrain_config.save_credentials(all_creds) return 0 diff --git a/cbrain_cli/users.py b/cbrain_cli/users.py index 8270d87..5c41376 100644 --- a/cbrain_cli/users.py +++ b/cbrain_cli/users.py @@ -1,7 +1,9 @@ from cbrain_cli.cli_utils import ( CbrainClient, json_printer, + output_json, ) +from cbrain_cli.config import cli_session_not_found def user_details(user_id): @@ -37,6 +39,15 @@ def whoami_user(args): Exit code on credential or API failure; otherwise None after printing. """ version = getattr(args, "version", False) + missing = cli_session_not_found() + if missing: + msg = f"Session '{missing}' not found." + if getattr(args, "json", False): + json_printer({"error": msg, "logged_in": False}) + else: + print(msg) + return 1 + client = CbrainClient.from_credentials() # Check if we have credentials first @@ -50,13 +61,12 @@ def whoami_user(args): user_data = user_details(client.user_id) # Handle JSON output first - if getattr(args, "json", False): - output = { - "login": user_data.get("login", ""), - "full_name": user_data.get("full_name", ""), - "server": client.base_url, - } - json_printer(output) + output = { + "login": user_data.get("login", ""), + "full_name": user_data.get("full_name", ""), + "server": client.base_url, + } + if output_json(args, output): return 0 if version: diff --git a/switch_session b/switch_session new file mode 100755 index 0000000..075934f --- /dev/null +++ b/switch_session @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +# Convenience wrapper: ./switch_session +# Equivalent to: cbrain switch_session +set -e +exec "$(dirname "$0")/cbrain" switch_session "$@" diff --git a/tests/test_config.py b/tests/test_config.py index a9922cf..ca02132 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -3,7 +3,15 @@ import pytest -from cbrain_cli.config import auth_headers, load_credentials, save_credentials +from cbrain_cli.config import ( + ACTIVE_SESSION_KEY, + auth_headers, + cli_session_not_found, + load_credentials, + resolve_session_credentials, + save_credentials, + update_active_credentials, +) from tests.conftest import CREDS_FILE, patch_credentials_file, sample_credentials @@ -63,3 +71,43 @@ def test_save_credentials_non_posix_branch(monkeypatch, creds_file): monkeypatch.setattr("os.name", "nt") save_credentials({"key": "value"}) assert load_credentials() == {"key": "value"} + + +def test_update_active_credentials_honors_cli_session_override(monkeypatch, creds_file): + creds_file.write_text( + json.dumps( + { + ACTIVE_SESSION_KEY: "default", + "default": {"api_token": "a", "cbrain_url": "http://localhost:3000"}, + "prod": {"api_token": "b", "cbrain_url": "http://localhost:3000"}, + } + ) + ) + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "prod") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + + update_active_credentials({"current_group_id": 7}) + + saved = load_credentials() + assert saved["prod"]["current_group_id"] == 7 + assert "current_group_id" not in saved["default"] + + +def test_resolve_session_credentials_honors_cli_session_override(monkeypatch, creds_file): + data = { + ACTIVE_SESSION_KEY: "default", + "default": {"api_token": "a", "user_id": 1}, + "prod": {"api_token": "b", "user_id": 2}, + } + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "prod") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + + resolved = resolve_session_credentials(data) + assert resolved["api_token"] == "b" + assert resolved["user_id"] == 2 + + +def test_cli_session_not_found(monkeypatch): + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "ghost") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + assert cli_session_not_found() == "ghost" diff --git a/tests/test_data_providers.py b/tests/test_data_providers.py index ed27a43..e576dc7 100644 --- a/tests/test_data_providers.py +++ b/tests/test_data_providers.py @@ -34,14 +34,10 @@ def test_show_data_provider_with_id_returns_dict(mock_urlopen): assert result["id"] == 2 -def test_show_data_provider_id_none_falls_back_to_list(capture_urlopen): - """Missing id falls back to list_data_providers hits list endpoint, does not raise.""" - configure, captured = capture_urlopen - configure(raw_body=b'[{"id": 1}, {"id": 2}]') - result = show_data_provider(_args(id=None)) - assert isinstance(result, list) - assert "/data_providers" in captured["url"] - assert "/data_providers/" not in captured["url"].split("?")[0] +def test_show_data_provider_id_none_raises(): + """Missing id raises CliValidationError instead of silently listing.""" + with pytest.raises(CliValidationError, match="Data provider ID is required"): + show_data_provider(_args(id=None)) def test_show_data_provider_api_error_in_body_raises(mock_urlopen): diff --git a/tests/test_exit_codes.py b/tests/test_exit_codes.py index 2d65df4..ae2f4fd 100644 --- a/tests/test_exit_codes.py +++ b/tests/test_exit_codes.py @@ -56,6 +56,11 @@ def test_handle_connection_error_url_error_connection_refused(fake_credentials, assert "Cannot connect to CBRAIN server" in capsys.readouterr().out +def test_handle_connection_error_url_error_uses_explicit_server_url(capsys): + handle_connection_error(URLError("Connection refused"), server_url="http://127.0.0.1:59999") + assert "127.0.0.1:59999" in capsys.readouterr().out + + def test_is_authenticated_false_when_no_credentials(): assert is_authenticated() is False @@ -69,6 +74,27 @@ def test_is_authenticated_true_with_fake_credentials(fake_credentials): assert is_authenticated() is True +def test_is_authenticated_unknown_session(monkeypatch, creds_file, capsys): + creds_file.write_text( + json.dumps({"default": {"api_token": "tok", "cbrain_url": URL, "user_id": 1}}) + ) + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "ghost") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + assert is_authenticated() is False + assert "Session 'ghost' not found" in capsys.readouterr().out + + +def test_main_file_list_unknown_session_returns_1(monkeypatch, creds_file, capsys): + creds_file.write_text( + json.dumps({"default": {"api_token": "tok", "cbrain_url": URL, "user_id": 1}}) + ) + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "ghost") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + result = run_main(monkeypatch, ["cbrain", "--session", "ghost", "file", "list"]) + assert result == 1 + assert "Session 'ghost' not found" in capsys.readouterr().out + + def test_main_file_list_unauthenticated_returns_1(monkeypatch, capsys): result = run_main(monkeypatch, ["cbrain", "file", "list"]) assert result == 1 diff --git a/tests/test_main_dispatch.py b/tests/test_main_dispatch.py index 6df1377..7e73df7 100644 --- a/tests/test_main_dispatch.py +++ b/tests/test_main_dispatch.py @@ -1,4 +1,6 @@ import importlib +import json +import sys from pathlib import Path from tests.conftest import install_auth, run_main @@ -87,3 +89,58 @@ def test_main_data_provider_canonical_dispatches(monkeypatch, fake_credentials, result = run_main(monkeypatch, ["cbrain", "data-provider", "list"]) assert result is None assert "/data_providers" in captured["url"] + + +def test_main_honors_argv_session_not_sys_argv(monkeypatch): + from cbrain_cli import cli_utils + from cbrain_cli.main import main + + monkeypatch.setattr(sys, "argv", ["pytest", "--session", "ghost", "whoami"]) + + main(["version"]) + assert cli_utils.session_specified is False + assert cli_utils.session_name == "default" + + main(["--session", "prod", "version"]) + assert cli_utils.session_specified is True + assert cli_utils.session_name == "prod" + + main(["--session=staging", "version"]) + assert cli_utils.session_specified is True + assert cli_utils.session_name == "staging" + + main(["whoami"]) + assert cli_utils.session_specified is False + + +def test_main_bare_logout_removes_all_named_sessions(monkeypatch, sessions_creds_file): + from cbrain_cli.main import main + + sessions_creds_file.write_text( + json.dumps( + { + "_active_session": "norm", + "norm": { + "api_token": "norm-token", + "cbrain_url": "http://localhost:3000", + "user_id": 2, + }, + "admin": { + "api_token": "admin-token", + "cbrain_url": "http://localhost:3000", + "user_id": 1, + }, + } + ) + ) + logged_out_tokens = [] + + def fake_send(client, method, path, payload=None): + logged_out_tokens.append(client.token) + return {}, 200 + + monkeypatch.setattr("cbrain_cli.cli_utils.CbrainClient.send", fake_send) + + assert main(["logout"]) == 0 + assert logged_out_tokens == ["norm-token", "admin-token"] + assert not sessions_creds_file.exists() diff --git a/tests/test_parser.py b/tests/test_parser.py index b6c9dd0..3a2f642 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -1,3 +1,5 @@ +import pytest + from cbrain_cli.main import build_parser @@ -71,7 +73,7 @@ def test_file_dp_id_aliases(): def test_file_upload_data_provider_aliases(): parser, _command_parsers = build_parser() for flag in ("--data-provider-id", "--data-provider", "--dp-id"): - args = parser.parse_args(["file", "upload", "/tmp/x", flag, "15"]) + args = parser.parse_args(["file", "upload", "/tmp/x", flag, "15", "--group-id", "1"]) assert args.data_provider == 15 @@ -151,3 +153,17 @@ def test_command_parsers_include_model_commands(): "remote-resource", ): assert command in command_parsers + + +def test_login_rejects_password_flag(): + parser, _command_parsers = build_parser() + with pytest.raises(SystemExit): + parser.parse_args(["login", "--username", "admin", "--password", "secret"]) + + +def test_global_session_omitted_is_none(): + parser, _command_parsers = build_parser() + assert parser.parse_args(["version"]).session is None + assert parser.parse_args(["--session", "prod", "version"]).session == "prod" + assert parser.parse_args(["--session=staging", "whoami"]).session == "staging" + assert parser.parse_args(["whoami", "--session", "prod"]).session == "prod" diff --git a/tests/test_projects.py b/tests/test_projects.py index f97605f..798ced5 100644 --- a/tests/test_projects.py +++ b/tests/test_projects.py @@ -132,3 +132,70 @@ def test_unswitch_project_removes_group_from_credentials(creds_file, mock_urlope assert result["current_group_id"] is None saved = json.loads(creds_file.read_text()) assert "current_group_id" not in saved + + +def test_switch_project_updates_named_session_not_active(monkeypatch, creds_file): + creds_file.write_text( + json.dumps( + { + "_active_session": "default", + "default": {"api_token": TOKEN, "cbrain_url": URL, "user_id": 42}, + "prod": {"api_token": TOKEN, "cbrain_url": URL, "user_id": 42}, + } + ) + ) + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "prod") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + + switch_response = MagicMock() + switch_response.__enter__.return_value.read.return_value = b"" + switch_response.__enter__.return_value.status = 200 + project_details_response = MagicMock() + project_details_response.__enter__.return_value.read.return_value = json.dumps( + {"id": 5, "name": "MyGroup"} + ).encode() + project_details_response.__enter__.return_value.status = 200 + monkeypatch.setattr( + "urllib.request.urlopen", + MagicMock(side_effect=[switch_response, project_details_response]), + ) + + switch_project(make_args(group_id="5")) + saved = json.loads(creds_file.read_text()) + assert saved["prod"]["current_group_id"] == 5 + assert "current_group_id" not in saved["default"] + + +def test_unswitch_project_clears_named_session_not_active(monkeypatch, creds_file): + creds_file.write_text( + json.dumps( + { + "_active_session": "default", + "default": { + "api_token": TOKEN, + "cbrain_url": URL, + "user_id": 42, + "current_group_id": 9, + "current_group_name": "Other", + }, + "prod": { + "api_token": TOKEN, + "cbrain_url": URL, + "user_id": 42, + "current_group_id": 5, + "current_group_name": "ProdGroup", + }, + } + ) + ) + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "prod") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + mock_urlopen = MagicMock() + mock_urlopen.__enter__.return_value.read.return_value = b"{}" + mock_urlopen.__enter__.return_value.status = 200 + monkeypatch.setattr("urllib.request.urlopen", MagicMock(return_value=mock_urlopen)) + + unswitch_project(make_args()) + saved = json.loads(creds_file.read_text()) + assert "current_group_id" not in saved["prod"] + assert saved["default"]["current_group_id"] == 9 diff --git a/tests/test_sessions.py b/tests/test_sessions.py index e031c31..d3426aa 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -4,7 +4,7 @@ import pytest from cbrain_cli.cli_utils import CliApiError, CliValidationError -from cbrain_cli.sessions import create_session, logout_session +from cbrain_cli.sessions import create_session, list_sessions, logout_session, switch_session def test_create_session_already_logged_in(sessions_creds_file, monkeypatch, capsys): @@ -190,3 +190,96 @@ def test_create_session_uses_default_url(monkeypatch, sessions_creds_file): ) result = create_session(argparse.Namespace()) assert result == 0 + + +def test_create_session_flat_file_allows_first_named_session( + sessions_creds_file, monkeypatch, capsys +): + import json + + sessions_creds_file.write_text( + json.dumps({"api_token": "tok", "cbrain_url": "http://localhost:3000", "user_id": 1}) + ) + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "prod") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + monkeypatch.setattr("builtins.input", lambda _: "admin") + monkeypatch.setattr("getpass.getpass", lambda _: "secret") + monkeypatch.setattr( + "cbrain_cli.cli_utils.CbrainClient.post_form", + lambda self, *_: {"cbrain_api_token": "newtok", "user_id": 1}, + ) + + result = create_session(argparse.Namespace()) + assert result == 0 + saved = json.loads(sessions_creds_file.read_text()) + assert saved["prod"]["api_token"] == "newtok" + assert saved["default"]["api_token"] == "tok" + assert "Already logged in" not in capsys.readouterr().out + + +def test_list_sessions_no_credentials_file(sessions_creds_file, capsys): + result = list_sessions(argparse.Namespace()) + assert result == 0 + assert "No saved sessions" in capsys.readouterr().out + + +def test_switch_session_no_credentials_file(sessions_creds_file, capsys): + result = switch_session(argparse.Namespace(session_target="default")) + assert result == 1 + assert "No saved sessions" in capsys.readouterr().out + + +def test_create_session_empty_username_arg_raises(sessions_creds_file): + with pytest.raises(CliValidationError, match="[Uu]sername"): + create_session( + argparse.Namespace(server="http://localhost:3000", username="", password="x") + ) + + +def test_create_session_empty_server_arg_raises(sessions_creds_file): + with pytest.raises(CliValidationError, match="[Ss]erver"): + create_session(argparse.Namespace(server="", username="admin", password="x")) + + +def test_create_session_bad_password_reports_login_failure( + monkeypatch, sessions_creds_file, capsys +): + def _raise_401(self, *_a, **_k): + raise CliApiError("Unauthorized", status=401) + + monkeypatch.setattr("cbrain_cli.cli_utils.CbrainClient.post_form", _raise_401) + result = create_session( + argparse.Namespace(server="http://127.0.0.1:3000", username="admin", password="wrong") + ) + assert result == 1 + out = capsys.readouterr().out + assert "Login failed" in out + assert "Session expired" not in out + + +def test_create_session_unreachable_login_server_shows_url( + monkeypatch, sessions_creds_file, capsys +): + def _raise(self, *_a, **_k): + raise urllib.error.URLError("Connection refused") + + monkeypatch.setattr("cbrain_cli.cli_utils.CbrainClient.post_form", _raise) + result = create_session( + argparse.Namespace(server="http://127.0.0.1:59999", username="admin", password="x") + ) + assert result == 1 + assert "127.0.0.1:59999" in capsys.readouterr().out + + +def test_create_session_password_from_env(monkeypatch, sessions_creds_file): + monkeypatch.setenv("CBRAIN_PASSWORD", "secret") + monkeypatch.setattr( + "cbrain_cli.sessions.getpass.getpass", + lambda _: (_ for _ in ()).throw(AssertionError("getpass must not run")), + ) + monkeypatch.setattr( + "cbrain_cli.cli_utils.CbrainClient.post_form", + lambda self, *_: {"cbrain_api_token": "tok", "user_id": 1}, + ) + result = create_session(argparse.Namespace(server="http://localhost:3000", username="admin")) + assert result == 0 diff --git a/tests/test_users.py b/tests/test_users.py index ec3138a..56269b3 100644 --- a/tests/test_users.py +++ b/tests/test_users.py @@ -41,6 +41,16 @@ def test_whoami_missing_credentials_plain(capsys): assert "Credential file is missing" in capsys.readouterr().out +def test_whoami_unknown_session_name(monkeypatch, creds_file, capsys): + creds_file.write_text( + json.dumps({"default": {"api_token": "tok", "cbrain_url": URL, "user_id": 1}}) + ) + monkeypatch.setattr("cbrain_cli.cli_utils.session_name", "ghost") + monkeypatch.setattr("cbrain_cli.cli_utils.session_specified", True) + assert whoami_user(make_args()) == 1 + assert "Session 'ghost' not found" in capsys.readouterr().out + + def test_whoami_json_output(monkeypatch, capsys): install_auth(user_id=1) monkeypatch.setattr(