diff --git a/cbrain_cli/cli_utils.py b/cbrain_cli/cli_utils.py index 5e1c00f..4a67a0f 100644 --- a/cbrain_cli/cli_utils.py +++ b/cbrain_cli/cli_utils.py @@ -10,13 +10,8 @@ import urllib.request from pathlib import Path -from cbrain_cli.config import DEFAULT_HEADERS, DEFAULT_TIMEOUT, auth_headers, load_credentials - -credentials = load_credentials() or {} -cbrain_url = credentials.get("cbrain_url") -api_token = credentials.get("api_token") -user_id = credentials.get("user_id") -cbrain_timestamp = credentials.get("timestamp") +from cbrain_cli import config as cbrain_config +from cbrain_cli.config import DEFAULT_HEADERS, DEFAULT_TIMEOUT, auth_headers PAGINATABLE_ACTIONS = { ("file", "list"), @@ -62,11 +57,32 @@ class CliResponseError(Exception): """ +def get_auth(): + """ + Return (cbrain_url, api_token, user_id) from current credentials file. + """ + creds = cbrain_config.load_credentials() or {} + return creds.get("cbrain_url"), creds.get("api_token"), creds.get("user_id") + + +def _request_target(path, token=None): + """ + Resolve (url, token) from path + optional token using call-time credentials. + """ + if token is not None: + return path, token + base, token, _ = get_auth() + if path.startswith(("http://", "https://")): + return path, token + return f"{base}{path}", token + + def is_authenticated(): """ Check if the user is authenticated. """ # Check if user is logged in. + cbrain_url, api_token, user_id = get_auth() if not api_token or not cbrain_url or not user_id: print("Not logged in. Use 'cbrain login' to login first.") return False @@ -196,6 +212,7 @@ def handle_connection_error(error): "Check your connection or set CBRAIN_TIMEOUT env var." ) elif "Connection refused" in str(error): + cbrain_url, _, _ = get_auth() print(f"Error: Cannot connect to CBRAIN server at {cbrain_url}") print("Please check if the CBRAIN server is running and accessible.") else: @@ -274,10 +291,11 @@ def version_info(args): return 0 -def api_get(url, token, params=None): +def api_get(path, token=None, params=None): """ Execute an authenticated GET request and return parsed JSON. """ + url, token = _request_target(path, token) if params: url = f"{url}?{urllib.parse.urlencode(params)}" req = urllib.request.Request(url, headers=auth_headers(token), method="GET") @@ -296,10 +314,12 @@ def api_post_form(url, form_data, headers=None): return json.loads(r.read().decode()) -def api_send(url, token, method="POST", payload=None): +def api_send(path, token=None, method="POST", payload=None): """ - Execute an authenticated POST/PUT/DELETE request and return (data, status). + Authenticated POST/PUT/DELETE. ``path`` is root-relative or absolute URL. + When ``token`` is omitted, credentials load at call time via ``get_auth()``. """ + url, token = _request_target(path, token) headers = auth_headers(token) body = None if payload is not None: @@ -311,6 +331,19 @@ def api_send(url, token, method="POST", payload=None): return (json.loads(raw) if raw.strip() else {}), r.status +def api_post_multipart(path, body, content_type, token=None): + """ + Authenticated multipart/form-data POST. Returns (parsed JSON, status). + """ + url, token = _request_target(path, token) + headers = auth_headers(token) + headers["Content-Type"] = content_type + headers["Content-Length"] = str(len(body)) + req = urllib.request.Request(url, data=body, headers=headers, method="POST") + with urllib.request.urlopen(req, timeout=DEFAULT_TIMEOUT) as r: + return json.loads(r.read().decode()), r.status + + def output_json(args, data): """ Print data as JSON or JSONL if requested. Returns True if output was handled. diff --git a/cbrain_cli/data/background_activities.py b/cbrain_cli/data/background_activities.py index 44a31a1..59006e0 100644 --- a/cbrain_cli/data/background_activities.py +++ b/cbrain_cli/data/background_activities.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import CliValidationError, api_get, api_token, cbrain_url +from cbrain_cli.cli_utils import CliValidationError, api_get def list_background_activities(args): @@ -15,7 +15,7 @@ def list_background_activities(args): list or None List of background activity dictionaries if successful, None if error """ - return api_get(f"{cbrain_url}/background_activities", api_token) + return api_get("/background_activities") def show_background_activity(args): @@ -36,4 +36,4 @@ def show_background_activity(args): activity_id = getattr(args, "id", None) if not activity_id: raise CliValidationError("Background activity ID is required", field="id") - return api_get(f"{cbrain_url}/background_activities/{activity_id}", api_token) + return api_get(f"/background_activities/{activity_id}") diff --git a/cbrain_cli/data/data_providers.py b/cbrain_cli/data/data_providers.py index 4aa6436..2aafdfd 100644 --- a/cbrain_cli/data/data_providers.py +++ b/cbrain_cli/data/data_providers.py @@ -3,8 +3,6 @@ CliValidationError, api_get, api_send, - api_token, - cbrain_url, pagination, ) @@ -27,7 +25,7 @@ def show_data_provider(args): data_provider_id = getattr(args, "id", None) if not data_provider_id: return list_data_providers(args) - data = api_get(f"{cbrain_url}/data_providers/{data_provider_id}", api_token) + data = api_get(f"/data_providers/{data_provider_id}") if data.get("error"): raise CliApiError(data.get("error")) return data @@ -48,7 +46,7 @@ def list_data_providers(args): List of data provider dictionaries """ params = pagination(args, {}) - return api_get(f"{cbrain_url}/data_providers", api_token, params) + return api_get("/data_providers", params=params) def is_alive(args): @@ -63,7 +61,7 @@ def is_alive(args): data_provider_id = getattr(args, "id", None) if not data_provider_id: raise CliValidationError("Data provider ID is required", field="id") - return api_get(f"{cbrain_url}/data_providers/{data_provider_id}/is_alive", api_token) + return api_get(f"/data_providers/{data_provider_id}/is_alive") def delete_unregistered_files(args): @@ -78,5 +76,5 @@ def delete_unregistered_files(args): data_provider_id = getattr(args, "id", None) if not data_provider_id: raise CliValidationError("Data provider ID is required", field="id") - data, _ = api_send(f"{cbrain_url}/data_providers/{data_provider_id}/delete", api_token) + data, _ = api_send(f"/data_providers/{data_provider_id}/delete") return data diff --git a/cbrain_cli/data/files.py b/cbrain_cli/data/files.py index dcd48d4..6a2cde2 100644 --- a/cbrain_cli/data/files.py +++ b/cbrain_cli/data/files.py @@ -1,17 +1,13 @@ -import json import mimetypes import os -import urllib.request from cbrain_cli.cli_utils import ( CliValidationError, api_get, + api_post_multipart, api_send, - api_token, - cbrain_url, pagination, ) -from cbrain_cli.config import DEFAULT_TIMEOUT, auth_headers def show_file(args): @@ -32,7 +28,7 @@ def show_file(args): file_id = getattr(args, "file", None) if not file_id: raise CliValidationError("File ID is required", field="file") - return api_get(f"{cbrain_url}/userfiles/{file_id}", api_token) + return api_get(f"/userfiles/{file_id}") def upload_file(args): @@ -84,17 +80,10 @@ def upload_file(args): file_content = f.read() body = body_text.encode("utf-8") + file_content + f"\r\n--{boundary}--\r\n".encode() - - headers = auth_headers(api_token) - headers["Content-Type"] = f"multipart/form-data; boundary={boundary}" - headers["Content-Length"] = str(len(body)) - - request = urllib.request.Request( - f"{cbrain_url}/userfiles", data=body, headers=headers, method="POST" + response_data, status = api_post_multipart( + "/userfiles", body, f"multipart/form-data; boundary={boundary}" ) - with urllib.request.urlopen(request, timeout=DEFAULT_TIMEOUT) as response: - response_data = json.loads(response.read().decode("utf-8")) - return response_data, response.status, file_name, file_size, args.data_provider + return response_data, status, file_name, file_size, args.data_provider def _change_provider(args, operation): @@ -111,7 +100,7 @@ def _change_provider(args, operation): "data_provider_id_for_mv_cp": dest_provider_id, operation: "", } - return api_send(f"{cbrain_url}/userfiles/change_provider", api_token, payload=payload) + return api_send("/userfiles/change_provider", payload=payload) def copy_file(args): @@ -175,7 +164,7 @@ def list_files(args): params[key] = str(val) params = pagination(args, params) - return api_get(f"{cbrain_url}/userfiles", api_token, params) + return api_get("/userfiles", params=params) def delete_file(args): @@ -196,8 +185,7 @@ def delete_file(args): if not file_id: raise CliValidationError("File ID is required", field="file_id") data, _ = api_send( - f"{cbrain_url}/userfiles/delete_files", - api_token, + "/userfiles/delete_files", method="DELETE", payload={"file_ids": [str(file_id)]}, ) diff --git a/cbrain_cli/data/projects.py b/cbrain_cli/data/projects.py index 15e45b9..fd86891 100644 --- a/cbrain_cli/data/projects.py +++ b/cbrain_cli/data/projects.py @@ -5,8 +5,6 @@ CliValidationError, api_get, api_send, - api_token, - cbrain_url, ) from cbrain_cli.config import load_credentials, save_credentials @@ -42,10 +40,10 @@ def switch_project(args): f"Invalid group ID '{group_id}'. Must be a number or 'all'", field="group_id" ) from None - _, switch_status = api_send(f"{cbrain_url}/groups/switch?id={group_id}", api_token) + _, switch_status = api_send(f"/groups/switch?id={group_id}") if switch_status not in (200, 201, 204): raise CliApiError(f"Failed to switch project (HTTP {switch_status})") - group_data = api_get(f"{cbrain_url}/groups/{group_id}", api_token) + group_data = api_get(f"/groups/{group_id}") credentials = load_credentials() if credentials is not None: @@ -79,7 +77,7 @@ def unswitch_project(args): previous_group_name = credentials.get("current_group_name") if previous_group_id: - api_send(f"{cbrain_url}/groups/switch", api_token) + api_send("/groups/switch") if credentials is not None: credentials.pop("current_group_id", None) @@ -114,7 +112,7 @@ def show_project(args): if project_id: # Show specific project by ID try: - return api_get(f"{cbrain_url}/groups/{project_id}", api_token) + return api_get(f"/groups/{project_id}") except urllib.error.HTTPError as e: if e.code == 404: raise CliApiError(f"Project with ID {project_id} not found") from None @@ -129,7 +127,7 @@ def show_project(args): return None try: - return api_get(f"{cbrain_url}/groups/{current_group_id}", api_token) + return api_get(f"/groups/{current_group_id}") except urllib.error.HTTPError as e: if e.code == 404: credentials.pop("current_group_id", None) @@ -153,4 +151,4 @@ def list_projects(args): list List of project dictionaries """ - return api_get(f"{cbrain_url}/groups", api_token) + return api_get("/groups") diff --git a/cbrain_cli/data/remote_resources.py b/cbrain_cli/data/remote_resources.py index 01e2118..85e8716 100644 --- a/cbrain_cli/data/remote_resources.py +++ b/cbrain_cli/data/remote_resources.py @@ -1,4 +1,4 @@ -from cbrain_cli.cli_utils import CliValidationError, api_get, api_token, cbrain_url +from cbrain_cli.cli_utils import CliValidationError, api_get def list_remote_resources(args): @@ -15,7 +15,7 @@ def list_remote_resources(args): list List of remote resource dictionaries """ - return api_get(f"{cbrain_url}/bourreaux", api_token) + return api_get("/bourreaux") def show_remote_resource(args): @@ -35,4 +35,4 @@ def show_remote_resource(args): resource_id = getattr(args, "remote_resource", None) if not resource_id: raise CliValidationError("Remote resource ID is required", field="remote_resource") - return api_get(f"{cbrain_url}/bourreaux/{resource_id}", api_token) + return api_get(f"/bourreaux/{resource_id}") diff --git a/cbrain_cli/data/tags.py b/cbrain_cli/data/tags.py index 04b62c7..336bf65 100644 --- a/cbrain_cli/data/tags.py +++ b/cbrain_cli/data/tags.py @@ -2,8 +2,6 @@ CliValidationError, api_get, api_send, - api_token, - cbrain_url, pagination, ) @@ -36,7 +34,7 @@ def list_tags(args): List of tag dictionaries """ params = pagination(args, {}) - return api_get(f"{cbrain_url}/tags", api_token, params) + return api_get("/tags", params=params) def show_tag(args): @@ -57,7 +55,7 @@ def show_tag(args): tag_id = getattr(args, "id", None) if not tag_id: raise CliValidationError("Tag ID is required", field="id") - return api_get(f"{cbrain_url}/tags/{tag_id}", api_token) + return api_get(f"/tags/{tag_id}") def create_tag(args): @@ -76,7 +74,7 @@ def create_tag(args): """ # Get tag details from command line arguments payload = _tag_payload(args) - data, status = api_send(f"{cbrain_url}/tags", api_token, payload=payload) + data, status = api_send("/tags", payload=payload) success = status in (200, 201, 204) return data, success, None, status @@ -100,7 +98,7 @@ def update_tag(args): if not tag_id: raise CliValidationError("Tag ID is required", field="tag_id") payload = _tag_payload(args) - data, status = api_send(f"{cbrain_url}/tags/{tag_id}", api_token, method="PUT", payload=payload) + data, status = api_send(f"/tags/{tag_id}", method="PUT", payload=payload) success = status in (200, 201, 204) return data, success, None, status @@ -123,6 +121,6 @@ def delete_tag(args): tag_id = getattr(args, "tag_id", None) if not tag_id: raise CliValidationError("Tag ID is required", field="tag_id") - _, status = api_send(f"{cbrain_url}/tags/{tag_id}", api_token, method="DELETE") + _, status = api_send(f"/tags/{tag_id}", method="DELETE") success = status in (200, 201, 204) return success, None, status diff --git a/cbrain_cli/data/tasks.py b/cbrain_cli/data/tasks.py index a1c18f5..bfe2ee4 100644 --- a/cbrain_cli/data/tasks.py +++ b/cbrain_cli/data/tasks.py @@ -2,8 +2,6 @@ CliValidationError, api_get, api_send, - api_token, - cbrain_url, json_printer, pagination, ) @@ -28,7 +26,7 @@ def list_tasks(args): bourreau_id = getattr(args, "bourreau_id", None) if filter_name is not None: - if filter_name != "bourreau-id": + if filter_name != "bourreau_id": raise CliValidationError(f"Unsupported filter: {filter_name}", field="filter_name") if bourreau_id is None: raise CliValidationError( @@ -43,7 +41,7 @@ def list_tasks(args): ) params = pagination(args, params) - return api_get(f"{cbrain_url}/tasks", api_token, params) + return api_get("/tasks", params=params) def show_task(args): @@ -63,12 +61,12 @@ def show_task(args): task_id = getattr(args, "task", None) if not task_id: raise CliValidationError("Task ID is required", field="task") - return api_get(f"{cbrain_url}/tasks/{task_id}", api_token) + return api_get(f"/tasks/{task_id}") def operation_task(args): """ Operation on a task. """ - data, _ = api_send(f"{cbrain_url}/tasks/operation", api_token) + data, _ = api_send("/tasks/operation") json_printer(data) diff --git a/cbrain_cli/data/tool_configs.py b/cbrain_cli/data/tool_configs.py index 1e07758..4212635 100644 --- a/cbrain_cli/data/tool_configs.py +++ b/cbrain_cli/data/tool_configs.py @@ -1,8 +1,6 @@ from cbrain_cli.cli_utils import ( CliValidationError, api_get, - api_token, - cbrain_url, pagination, ) @@ -18,7 +16,7 @@ def list_tool_configs(args): configuration details. """ params = pagination(args, {}) - return api_get(f"{cbrain_url}/tool_configs", api_token, params) + return api_get("/tool_configs", params=params) def show_tool_config(args): @@ -33,7 +31,7 @@ def show_tool_config(args): config_id = getattr(args, "id", None) if not config_id: raise CliValidationError("Tool configuration ID is required", field="id") - return api_get(f"{cbrain_url}/tool_configs/{config_id}", api_token) + return api_get(f"/tool_configs/{config_id}") def tool_config_boutiques_descriptor(args): @@ -48,4 +46,4 @@ def tool_config_boutiques_descriptor(args): config_id = getattr(args, "id", None) if not config_id: raise CliValidationError("Tool configuration ID is required", field="id") - return api_get(f"{cbrain_url}/tool_configs/{config_id}/boutiques_descriptor", api_token) + return api_get(f"/tool_configs/{config_id}/boutiques_descriptor") diff --git a/cbrain_cli/data/tools.py b/cbrain_cli/data/tools.py index cd93751..77847ab 100644 --- a/cbrain_cli/data/tools.py +++ b/cbrain_cli/data/tools.py @@ -2,8 +2,6 @@ CliApiError, CliValidationError, api_get, - api_token, - cbrain_url, pagination, ) @@ -13,7 +11,7 @@ def list_tools(args): Get paginated list of tools from CBRAIN. """ params = pagination(args, {}) - return api_get(f"{cbrain_url}/tools", api_token, params) + return api_get("/tools", params=params) def show_tool(args): @@ -31,9 +29,8 @@ def show_tool(args): page = 1 while True: tools_page = api_get( - f"{cbrain_url}/tools", - api_token, - {"page": str(page), "per_page": str(per_page)}, + "/tools", + params={"page": str(page), "per_page": str(per_page)}, ) if not tools_page or not isinstance(tools_page, list): break diff --git a/cbrain_cli/main.py b/cbrain_cli/main.py index 0649055..c610304 100644 --- a/cbrain_cli/main.py +++ b/cbrain_cli/main.py @@ -96,14 +96,28 @@ def build_parser(): # file list file_list_parser = file_subparsers.add_parser("list", help="List files") - file_list_parser.add_argument("--group-id", type=int, help="Filter files by group ID") - file_list_parser.add_argument("--dp-id", type=int, help="Filter files by data provider ID") - file_list_parser.add_argument("--user-id", type=int, help="Filter files by user ID") - file_list_parser.add_argument("--parent-id", type=int, help="Filter files by parent ID") - file_list_parser.add_argument("--file-type", type=str, help="Filter files by type") + file_list_parser.add_argument( + "--group-id", dest="group_id", type=int, help="Filter files by group ID" + ) + file_list_parser.add_argument( + "--dp-id", dest="dp_id", type=int, help="Filter files by data provider ID" + ) + file_list_parser.add_argument( + "--user-id", dest="user_id", type=int, help="Filter files by user ID" + ) + file_list_parser.add_argument( + "--parent-id", dest="parent_id", type=int, help="Filter files by parent ID" + ) + file_list_parser.add_argument( + "--file-type", dest="file_type", type=str, help="Filter files by type" + ) file_list_parser.add_argument("--page", type=int, default=1, help="Page number (default: 1)") file_list_parser.add_argument( - "--per-page", type=int, default=25, help="Number of files per page (5-1000, default: 25)" + "--per-page", + dest="per_page", + type=int, + default=25, + help="Number of files per page (5-1000, default: 25)", ) file_list_parser.set_defaults(func=handle_errors(handle_file_list)) @@ -116,9 +130,13 @@ def build_parser(): file_upload_parser = file_subparsers.add_parser("upload", help="Upload a file to CBRAIN") file_upload_parser.add_argument("file_path", help="Path to the file to upload") file_upload_parser.add_argument( - "--data-provider", type=int, required=True, help="Data provider ID" + "--data-provider", + dest="data_provider", + type=int, + required=True, + help="Data provider ID", ) - file_upload_parser.add_argument("--group-id", type=int, help="Group ID") + file_upload_parser.add_argument("--group-id", dest="group_id", type=int, help="Group ID") file_upload_parser.set_defaults(func=handle_errors(handle_file_upload)) @@ -128,13 +146,18 @@ def build_parser(): ) file_copy_parser.add_argument( "--file-id", + dest="file_id", type=int, nargs="+", required=True, help="One or more file IDs to copy", ) file_copy_parser.add_argument( - "--dp-id", type=int, required=True, help="Destination data provider ID" + "--dp-id", + dest="dp_id", + type=int, + required=True, + help="Destination data provider ID", ) file_copy_parser.set_defaults(func=handle_errors(handle_file_copy)) @@ -144,13 +167,18 @@ def build_parser(): ) file_move_parser.add_argument( "--file-id", + dest="file_id", type=int, nargs="+", required=True, help="One or more file IDs to move", ) file_move_parser.add_argument( - "--dp-id", type=int, required=True, help="Destination data provider ID" + "--dp-id", + dest="dp_id", + type=int, + required=True, + help="Destination data provider ID", ) file_move_parser.set_defaults(func=handle_errors(handle_file_move)) @@ -316,8 +344,12 @@ def build_parser(): # tag create tag_create_parser = tag_subparsers.add_parser("create", help="Create a new tag") tag_create_parser.add_argument("--name", type=str, required=True, help="Tag name") - tag_create_parser.add_argument("--user-id", type=int, required=True, help="User ID") - tag_create_parser.add_argument("--group-id", type=int, required=True, help="Group ID") + tag_create_parser.add_argument( + "--user-id", dest="user_id", type=int, required=True, help="User ID" + ) + tag_create_parser.add_argument( + "--group-id", dest="group_id", type=int, required=True, help="Group ID" + ) tag_create_parser.set_defaults(func=handle_errors(handle_tag_create)) # tag update @@ -328,8 +360,12 @@ def build_parser(): help="Tag ID to update", ) tag_update_parser.add_argument("--name", type=str, required=True, help="Tag name") - tag_update_parser.add_argument("--user-id", type=int, required=True, help="User ID") - tag_update_parser.add_argument("--group-id", type=int, required=True, help="Group ID") + tag_update_parser.add_argument( + "--user-id", dest="user_id", type=int, required=True, help="User ID" + ) + tag_update_parser.add_argument( + "--group-id", dest="group_id", type=int, required=True, help="Group ID" + ) tag_update_parser.set_defaults(func=handle_errors(handle_tag_update)) # tag delete @@ -370,11 +406,20 @@ def build_parser(): # task list task_list_parser = task_subparsers.add_parser("list", help="List tasks") task_list_parser.add_argument( - "filter_name", nargs="?", choices=["bourreau-id"], help="Filter type (optional)" + "filter_name", + nargs="?", + type=lambda value: value.replace("-", "_"), + choices=["bourreau_id"], + metavar="bourreau-id", + help="Filter type (optional)", ) task_list_parser.add_argument("--page", type=int, default=1, help="Page number (default: 1)") task_list_parser.add_argument( - "--per-page", type=int, default=25, help="Number of tasks per page (5-1000, default: 25)" + "--per-page", + dest="per_page", + type=int, + default=25, + help="Number of tasks per page (5-1000, default: 25)", ) task_list_parser.add_argument( "bourreau_id", diff --git a/cbrain_cli/sessions.py b/cbrain_cli/sessions.py index 9f00cdb..763b596 100644 --- a/cbrain_cli/sessions.py +++ b/cbrain_cli/sessions.py @@ -2,14 +2,13 @@ import getpass import urllib.error +from cbrain_cli import config as cbrain_config from cbrain_cli.cli_utils import ( CliValidationError, api_post_form, api_send, - api_token, - cbrain_url, ) -from cbrain_cli.config import CREDENTIALS_FILE, DEFAULT_BASE_URL, load_credentials, save_credentials +from cbrain_cli.config import DEFAULT_BASE_URL # MARK: Create Session. @@ -28,8 +27,8 @@ def create_session(args): Exit code (0 on success, 1 on failure). """ - if CREDENTIALS_FILE.exists(): - creds = load_credentials() + if cbrain_config.CREDENTIALS_FILE.exists(): + creds = cbrain_config.load_credentials() if creds and creds.get("api_token") and creds.get("cbrain_url"): print("Already logged in. Use 'cbrain logout' to logout.") return 1 @@ -65,9 +64,9 @@ def create_session(args): "timestamp": datetime.datetime.now().isoformat(), } - save_credentials(credentials) + cbrain_config.save_credentials(credentials) - print(f"Connection successful, API token saved in {CREDENTIALS_FILE}") + print(f"Connection successful, API token saved in {cbrain_config.CREDENTIALS_FILE}") return 0 @@ -87,25 +86,27 @@ def logout_session(args): Exit code (0 on success). """ - if not CREDENTIALS_FILE.exists(): + if not cbrain_config.CREDENTIALS_FILE.exists(): print("Not logged in. Use 'cbrain login' to login first.") return 0 - credentials = load_credentials() + credentials = cbrain_config.load_credentials() if credentials is None: print("Invalid credentials file. Removing local session.") - CREDENTIALS_FILE.unlink(missing_ok=True) - print(f"Local session removed from {CREDENTIALS_FILE}") + 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.") - CREDENTIALS_FILE.unlink(missing_ok=True) - print(f"Local session removed from {CREDENTIALS_FILE}") + cbrain_config.CREDENTIALS_FILE.unlink(missing_ok=True) + print(f"Local session removed from {cbrain_config.CREDENTIALS_FILE}") return 0 try: - _, status = api_send(f"{cbrain_url}/session", api_token, method="DELETE") + _, status = api_send("/session", method="DELETE") if status == 200: print("Successfully logged out from CBRAIN server.") else: @@ -118,7 +119,7 @@ def logout_session(args): except urllib.error.URLError as e: print(f"Network error during logout: {e}") - if CREDENTIALS_FILE.exists(): - CREDENTIALS_FILE.unlink() - print(f"Local session removed from {CREDENTIALS_FILE}") + if cbrain_config.CREDENTIALS_FILE.exists(): + cbrain_config.CREDENTIALS_FILE.unlink() + print(f"Local session removed from {cbrain_config.CREDENTIALS_FILE}") return 0 diff --git a/cbrain_cli/users.py b/cbrain_cli/users.py index eae1bce..01e275a 100644 --- a/cbrain_cli/users.py +++ b/cbrain_cli/users.py @@ -1,15 +1,11 @@ -import json import urllib.error -import urllib.request from cbrain_cli.cli_utils import ( - api_token, - cbrain_url, + api_get, + get_auth, handle_connection_error, json_printer, - user_id, ) -from cbrain_cli.config import DEFAULT_TIMEOUT, auth_headers def user_details(user_id): @@ -26,17 +22,8 @@ def user_details(user_id): dict or None User data dictionary, or None if the request fails. """ - user_endpoint = f"{cbrain_url}/users/{user_id}" - - user_request = urllib.request.Request( - user_endpoint, headers=auth_headers(api_token), method="GET" - ) - try: - with urllib.request.urlopen(user_request, timeout=DEFAULT_TIMEOUT) as response: - user_data = json.loads(response.read().decode("utf-8")) - return user_data - + return api_get(f"/users/{user_id}") except (urllib.error.URLError, urllib.error.HTTPError) as e: handle_connection_error(e) return None @@ -61,6 +48,7 @@ def whoami_user(args): Exit code on credential or API failure; otherwise None after printing. """ version = getattr(args, "version", False) + cbrain_url, api_token, user_id = get_auth() # Check if we have credentials first if user_id is None or cbrain_url is None or api_token is None: @@ -88,25 +76,18 @@ def whoami_user(args): if version: # Verify token by making a session request. - session_endpoint = f"{cbrain_url}/session" - - session_request = urllib.request.Request( - session_endpoint, headers=auth_headers(api_token), method="GET" - ) - try: - with urllib.request.urlopen(session_request, timeout=DEFAULT_TIMEOUT) as response: - session_data = json.loads(response.read().decode("utf-8")) + session_data = api_get("/session") - # Verify local credentials match server response. - remote_user_id = session_data.get("user_id") - remote_token = session_data.get("cbrain_api_token") + # Verify local credentials match server response. + remote_user_id = session_data.get("user_id") + remote_token = session_data.get("cbrain_api_token") - if str(remote_user_id) != str(user_id): - print(f"WARNING: User ID mismatch - Local: {user_id}, Remote: {remote_user_id}") + if str(remote_user_id) != str(user_id): + print(f"WARNING: User ID mismatch - Local: {user_id}, Remote: {remote_user_id}") - if remote_token != api_token: - print("WARNING: Token mismatch - tokens don't match") + if remote_token != api_token: + print("WARNING: Token mismatch - tokens don't match") except (urllib.error.URLError, urllib.error.HTTPError) as e: handle_connection_error(e) diff --git a/tests/conftest.py b/tests/conftest.py index 118e0ff..04ffe6d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,20 +17,25 @@ def make_args(**kwargs): return argparse.Namespace(**defaults) -def patch_credentials_file(monkeypatch, path, *, sessions=False): +def patch_credentials_file(monkeypatch, path): """Redirect credential file path away from the real ~/.config/cbrain.""" monkeypatch.setattr("cbrain_cli.config.CREDENTIALS_FILE", path) - if sessions: - monkeypatch.setattr("cbrain_cli.sessions.CREDENTIALS_FILE", path) -def patch_module_locals(monkeypatch, *module_paths, user_id=None): - """Patch module-local api_token / cbrain_url (and optional user_id) copies.""" - for module_path in module_paths: - monkeypatch.setattr(f"{module_path}.api_token", TOKEN) - monkeypatch.setattr(f"{module_path}.cbrain_url", URL) - if user_id is not None: - monkeypatch.setattr(f"{module_path}.user_id", user_id) +def write_auth_credentials(path, *, user_id=42, **overrides): + """Write call-time auth credentials into an isolated credentials file.""" + credentials = {"api_token": TOKEN, "cbrain_url": URL, "user_id": user_id} + credentials.update(overrides) + path.write_text(json.dumps(credentials)) + return credentials + + +def install_auth(*, user_id=None): + """Write auth credentials so call-time get_auth() / api_* helpers pick them up.""" + from cbrain_cli import config + + uid = 42 if user_id is None else user_id + write_auth_credentials(config.CREDENTIALS_FILE, user_id=uid) def sample_credentials(**overrides): @@ -53,20 +58,24 @@ def parse_json_output(capsys): return json.loads(capsys.readouterr().out.strip()) -@pytest.fixture -def creds_file(tmp_path, monkeypatch): - """Temp credentials file patched on cbrain_cli.config.""" +@pytest.fixture(autouse=True) +def _isolate_credentials(tmp_path, monkeypatch): + """Redirect credentials file so real home config never leaks into tests.""" path = tmp_path / CREDS_FILE patch_credentials_file(monkeypatch, path) return path @pytest.fixture -def sessions_creds_file(tmp_path, monkeypatch): - """Temp credentials file patched on config and sessions modules.""" - path = tmp_path / CREDS_FILE - patch_credentials_file(monkeypatch, path, sessions=True) - return path +def creds_file(_isolate_credentials): + """Temp credentials file patched on cbrain_cli.config.""" + return _isolate_credentials + + +@pytest.fixture +def sessions_creds_file(_isolate_credentials): + """Alias of the isolated credentials file used by session tests.""" + return _isolate_credentials @pytest.fixture @@ -104,30 +113,10 @@ def fake_urlopen(request, **kwargs): return configure, captured -@pytest.fixture(autouse=True) -def _reset_globals(monkeypatch): - """Reset cli_utils module-level globals before every test. - - Prevents real credentials on disk from leaking into tests. - """ - monkeypatch.setattr("cbrain_cli.cli_utils.api_token", None) - monkeypatch.setattr("cbrain_cli.cli_utils.cbrain_url", None) - monkeypatch.setattr("cbrain_cli.cli_utils.user_id", None) - - @pytest.fixture -def fake_credentials(monkeypatch, _reset_globals): - """Set known credentials on cbrain_cli.cli_utils globals. - - Explicit dependency on _reset_globals guarantees ordering — _reset_globals - runs first (sets None), then this fixture overwrites with real values. - - Tests for data modules must ALSO patch the module-local copy, e.g.: - patch_module_locals(monkeypatch, "cbrain_cli.data.tasks") - """ - monkeypatch.setattr("cbrain_cli.cli_utils.api_token", TOKEN) - monkeypatch.setattr("cbrain_cli.cli_utils.cbrain_url", URL) - monkeypatch.setattr("cbrain_cli.cli_utils.user_id", 1) +def fake_credentials(_isolate_credentials): + """Write known credentials so get_auth() / is_authenticated() see them.""" + write_auth_credentials(_isolate_credentials, user_id=1) @pytest.fixture diff --git a/tests/test_background_activities.py b/tests/test_background_activities.py index 9f1b6be..f74cc76 100644 --- a/tests/test_background_activities.py +++ b/tests/test_background_activities.py @@ -5,13 +5,13 @@ list_background_activities, show_background_activity, ) +from tests.conftest import install_auth from tests.conftest import make_args as _args -from tests.conftest import patch_module_locals @pytest.fixture(autouse=True) -def _patch_locals(monkeypatch): - patch_module_locals(monkeypatch, "cbrain_cli.data.background_activities") +def _patch_locals(): + install_auth() def test_list_background_activities_returns_list(mock_urlopen): diff --git a/tests/test_cli_utils_output.py b/tests/test_cli_utils_output.py index 4c85559..a10c97c 100644 --- a/tests/test_cli_utils_output.py +++ b/tests/test_cli_utils_output.py @@ -64,8 +64,14 @@ def test_dynamic_table_print_header_mismatch_raises(): def test_version_info(capsys): - version_info(MagicMock(json=False, jsonl=False)) - assert f"cbrain cli client version {_setup_cfg_version()}" in capsys.readouterr().out + import importlib.metadata + + try: + expected = importlib.metadata.version("cbrain-cli") + except importlib.metadata.PackageNotFoundError: + expected = _setup_cfg_version() + assert version_info(MagicMock(json=False, jsonl=False)) == 0 + assert f"cbrain cli client version {expected}" in capsys.readouterr().out def test_version_info_prefers_package_metadata(monkeypatch, capsys): diff --git a/tests/test_data_api.py b/tests/test_data_api.py index c6c653d..11ed6b5 100644 --- a/tests/test_data_api.py +++ b/tests/test_data_api.py @@ -2,8 +2,8 @@ import pytest -from cbrain_cli.cli_utils import api_get, api_post_form, api_send -from tests.conftest import TOKEN, URL +from cbrain_cli.cli_utils import api_get, api_post_form, api_post_multipart, api_send +from tests.conftest import TOKEN, URL, install_auth def test_api_get_returns_parsed_json(mock_urlopen): @@ -82,3 +82,24 @@ def test_api_send_propagates_http_error(capture_urlopen): ) with pytest.raises(urllib.error.HTTPError): api_send(f"{URL}/tags", TOKEN) + + +def test_api_get_relative_path_uses_call_time_auth(capture_urlopen): + install_auth() + configure, captured = capture_urlopen + configure({}) + api_get("/tools") + assert captured["url"] == f"{URL}/tools" + assert captured["headers"].get("Authorization") == f"Bearer {TOKEN}" + + +def test_api_post_multipart_sets_content_type_and_auth(capture_urlopen): + install_auth() + configure, captured = capture_urlopen + configure({"id": 1}, status=201) + data, status = api_post_multipart("/userfiles", b"body", "multipart/form-data; boundary=x") + assert data == {"id": 1} + assert status == 201 + assert captured["url"] == f"{URL}/userfiles" + assert "multipart/form-data" in captured["headers"].get("Content-type", "") + assert captured["headers"].get("Authorization") == f"Bearer {TOKEN}" diff --git a/tests/test_data_providers.py b/tests/test_data_providers.py index 540a809..44ab259 100644 --- a/tests/test_data_providers.py +++ b/tests/test_data_providers.py @@ -7,13 +7,13 @@ list_data_providers, show_data_provider, ) +from tests.conftest import install_auth from tests.conftest import make_args as _args -from tests.conftest import patch_module_locals @pytest.fixture(autouse=True) -def _patch_locals(monkeypatch): - patch_module_locals(monkeypatch, "cbrain_cli.data.data_providers") +def _patch_locals(): + install_auth() def test_list_data_providers_returns_list(mock_urlopen): diff --git a/tests/test_exit_codes.py b/tests/test_exit_codes.py index 21188b6..314b3a1 100644 --- a/tests/test_exit_codes.py +++ b/tests/test_exit_codes.py @@ -51,21 +51,17 @@ def test_handle_connection_error_http(code, expected, capsys): assert expected in capsys.readouterr().out -def test_handle_connection_error_url_error_connection_refused(monkeypatch, capsys): - monkeypatch.setattr("cbrain_cli.cli_utils.cbrain_url", URL) +def test_handle_connection_error_url_error_connection_refused(fake_credentials, capsys): handle_connection_error(URLError("Connection refused")) assert "Cannot connect to CBRAIN server" in capsys.readouterr().out def test_is_authenticated_false_when_no_credentials(): - # _reset_globals autouse fixture leaves api_token=None assert is_authenticated() is False -def test_is_authenticated_false_when_cbrain_url_is_none(monkeypatch): - monkeypatch.setattr("cbrain_cli.cli_utils.api_token", "tok") - monkeypatch.setattr("cbrain_cli.cli_utils.user_id", 1) - # cbrain_url stays None from _reset_globals +def test_is_authenticated_false_when_cbrain_url_is_none(creds_file): + creds_file.write_text(json.dumps({"api_token": "tok", "user_id": 1})) assert is_authenticated() is False @@ -142,8 +138,7 @@ def test_handle_errors_bare_read_timeout(capsys): assert "timed out" in capsys.readouterr().out -def test_handle_connection_error_generic_url_error(capsys, monkeypatch): - monkeypatch.setattr("cbrain_cli.cli_utils.cbrain_url", URL) +def test_handle_connection_error_generic_url_error(capsys): handle_connection_error(URLError("timed out")) assert "Connection failed" in capsys.readouterr().out diff --git a/tests/test_files.py b/tests/test_files.py index 56384e1..4b34362 100644 --- a/tests/test_files.py +++ b/tests/test_files.py @@ -9,13 +9,13 @@ show_file, upload_file, ) +from tests.conftest import install_auth from tests.conftest import make_args as _args -from tests.conftest import patch_module_locals @pytest.fixture(autouse=True) -def _patch_locals(monkeypatch): - patch_module_locals(monkeypatch, "cbrain_cli.data.files") +def _patch_locals(): + install_auth() def test_show_file_missing_id_raises(): @@ -53,8 +53,12 @@ def fake_urlopen(req, **kwargs): return cm monkeypatch.setattr("urllib.request.urlopen", fake_urlopen) - list_files(_args(group_id=5, dp_id=None, user_id=None, parent_id=None, file_type=None)) + list_files(_args(group_id=5, dp_id=9, user_id=None, parent_id=None, file_type="TextFile")) assert "group_id=5" in captured["url"] + assert "data_provider_id=9" in captured["url"] + assert "type=TextFile" in captured["url"] + assert "dp_id=" not in captured["url"] + assert "file_type=" not in captured["url"] def test_delete_file_missing_id_raises(): diff --git a/tests/test_handlers.py b/tests/test_handlers.py index bfc2fda..fbf97fd 100644 --- a/tests/test_handlers.py +++ b/tests/test_handlers.py @@ -11,7 +11,7 @@ handle_task_show, ) from cbrain_cli.users import user_details, whoami_user -from tests.conftest import make_args, parse_json_output, patch_module_locals +from tests.conftest import install_auth, make_args, parse_json_output LIST_HANDLER_CASES = [ ( @@ -153,10 +153,11 @@ def test_list_handler_validation_error_returns_1(monkeypatch): assert handle_errors(handle_task_list)(make_args()) == 1 -def test_user_details_sends_current_token_in_header(monkeypatch, capture_urlopen): - """auth_headers(api_token) is called inside user_details, not at import time.""" - patch_module_locals(monkeypatch, "cbrain_cli.users") - monkeypatch.setattr("cbrain_cli.users.api_token", "new-token") +def test_user_details_sends_current_token_in_header(monkeypatch, capture_urlopen, creds_file): + """auth_headers(api_token) uses call-time credentials, not import-time globals.""" + from tests.conftest import write_auth_credentials + + write_auth_credentials(creds_file, api_token="new-token") configure, captured = capture_urlopen configure({"id": 1, "login": "admin"}) @@ -167,7 +168,7 @@ def test_user_details_sends_current_token_in_header(monkeypatch, capture_urlopen def test_whoami_user_version_does_not_print_debug_lines(monkeypatch, capsys): """whoami_user with version=True must not print DEBUG: lines.""" - patch_module_locals(monkeypatch, "cbrain_cli.users", user_id=1) + install_auth(user_id=1) monkeypatch.setattr( "cbrain_cli.users.user_details", diff --git a/tests/test_main_dispatch.py b/tests/test_main_dispatch.py index 15652ce..79c81b2 100644 --- a/tests/test_main_dispatch.py +++ b/tests/test_main_dispatch.py @@ -1,7 +1,7 @@ import importlib from pathlib import Path -from tests.conftest import patch_module_locals, run_main +from tests.conftest import install_auth, run_main def test_main_invalid_pagination_skips_network(monkeypatch): @@ -51,7 +51,7 @@ def test_main_version_does_not_create_config_dir(monkeypatch, tmp_path): def test_main_task_list_bourreau_id_parses_and_dispatches( monkeypatch, fake_credentials, capture_urlopen ): - patch_module_locals(monkeypatch, "cbrain_cli.data.tasks") + install_auth() configure, captured = capture_urlopen configure([]) result = run_main(monkeypatch, ["cbrain", "task", "list", "bourreau-id", "7"]) diff --git a/tests/test_parser.py b/tests/test_parser.py index bcd2612..9a0ed01 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -14,10 +14,43 @@ def test_task_list_bourreau_id_args(): args = parser.parse_args(["task", "list", "bourreau-id", "7"]) assert args.command == "task" assert args.action == "list" - assert args.filter_name == "bourreau-id" + assert args.filter_name == "bourreau_id" assert args.bourreau_id == 7 +def test_file_list_kebab_options_normalize_to_snake_case(): + parser, _command_parsers = build_parser() + args = parser.parse_args( + [ + "file", + "list", + "--group-id", + "5", + "--dp-id", + "9", + "--file-type", + "TextFile", + "--per-page", + "50", + ] + ) + assert args.group_id == 5 + assert args.dp_id == 9 + assert args.file_type == "TextFile" + assert args.per_page == 50 + + +def test_tag_create_kebab_options_normalize_to_snake_case(): + parser, _command_parsers = build_parser() + args = parser.parse_args( + ["tag", "create", "--name", "demo", "--user-id", "1", "--group-id", "2"] + ) + assert args.user_id == 1 + assert args.group_id == 2 + assert not hasattr(args, "user-id") + assert not hasattr(args, "group-id") + + def test_project_unswitch_subcommand(): parser, _command_parsers = build_parser() args = parser.parse_args(["project", "unswitch"]) diff --git a/tests/test_projects.py b/tests/test_projects.py index 1cfd0c4..47fa14f 100644 --- a/tests/test_projects.py +++ b/tests/test_projects.py @@ -7,17 +7,13 @@ from cbrain_cli.cli_utils import CliApiError, CliValidationError from cbrain_cli.data.data_providers import show_data_provider from cbrain_cli.data.projects import list_projects, show_project, switch_project, unswitch_project -from tests.conftest import TOKEN, URL, make_args, patch_module_locals +from tests.conftest import TOKEN, URL, install_auth, make_args @pytest.fixture(autouse=True) -def _patch_projects_locals(monkeypatch): - """Patch data.projects module-local copies of api_token / cbrain_url.""" - patch_module_locals( - monkeypatch, - "cbrain_cli.data.projects", - "cbrain_cli.data.data_providers", - ) +def _patch_projects_locals(): + """Write auth credentials so call-time get_auth() picks them up.""" + install_auth() def test_list_projects_returns_list(mock_urlopen): @@ -64,7 +60,9 @@ def test_show_project_no_current_group_returns_none(creds_file): def test_show_project_stale_group_raises_and_cleans_credentials(monkeypatch, creds_file): """When saved current_group_id returns 404, credentials cleaned and CliApiError raised.""" - creds_file.write_text(json.dumps({"api_token": TOKEN, "current_group_id": 99})) + from tests.conftest import write_auth_credentials + + write_auth_credentials(creds_file, current_group_id=99) monkeypatch.setattr( "urllib.request.urlopen", MagicMock(side_effect=urllib.error.HTTPError(URL, 404, "Not Found", {}, None)), @@ -91,7 +89,9 @@ def test_switch_project_invalid_string_raises(): def test_switch_project_saves_credentials(monkeypatch, creds_file): - creds_file.write_text(json.dumps({"api_token": TOKEN})) + from tests.conftest import write_auth_credentials + + write_auth_credentials(creds_file) session_delete_response = MagicMock() session_delete_response.__enter__.return_value.read.return_value = b"" @@ -113,9 +113,9 @@ def test_switch_project_saves_credentials(monkeypatch, creds_file): def test_unswitch_project_removes_group_from_credentials(creds_file, mock_urlopen): - creds_file.write_text( - json.dumps({"api_token": TOKEN, "current_group_id": 5, "current_group_name": "G"}) - ) + from tests.conftest import write_auth_credentials + + write_auth_credentials(creds_file, current_group_id=5, current_group_name="G") mock_urlopen({}) result = unswitch_project(make_args()) assert result["current_group_id"] is None diff --git a/tests/test_remote_resources.py b/tests/test_remote_resources.py index 5f2697f..903f066 100644 --- a/tests/test_remote_resources.py +++ b/tests/test_remote_resources.py @@ -2,13 +2,13 @@ from cbrain_cli.cli_utils import CliValidationError from cbrain_cli.data.remote_resources import list_remote_resources, show_remote_resource +from tests.conftest import install_auth from tests.conftest import make_args as _args -from tests.conftest import patch_module_locals @pytest.fixture(autouse=True) -def _patch_locals(monkeypatch): - patch_module_locals(monkeypatch, "cbrain_cli.data.remote_resources") +def _patch_locals(): + install_auth() def test_list_remote_resources_returns_list(mock_urlopen): diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 6628392..52e45cf 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -84,14 +84,9 @@ def test_logout_session_corrupt_file_removes_it(sessions_creds_file, capsys): assert not sessions_creds_file.exists() -def test_logout_session_no_api_token_removes_file_without_http( - monkeypatch, sessions_creds_file, capsys -): - """When sessions module-local api_token is None, logout removes file with no HTTP call.""" - sessions_creds_file.write_text('{"api_token": "tok", "cbrain_url": "http://localhost:3000"}') - # explicitly null out the module-local copies (not reset by _reset_globals) - monkeypatch.setattr("cbrain_cli.sessions.api_token", None) - monkeypatch.setattr("cbrain_cli.sessions.cbrain_url", None) +def test_logout_session_no_api_token_removes_file_without_http(sessions_creds_file, capsys): + """Incomplete credentials file is removed with no HTTP call.""" + sessions_creds_file.write_text('{"cbrain_url": "http://localhost:3000"}') result = logout_session(argparse.Namespace()) assert result == 0 assert not sessions_creds_file.exists() @@ -101,8 +96,6 @@ def test_logout_session_success_sends_delete_and_removes_file( monkeypatch, sessions_creds_file, capsys ): sessions_creds_file.write_text('{"api_token": "tok", "cbrain_url": "http://localhost:3000"}') - monkeypatch.setattr("cbrain_cli.sessions.api_token", "tok") - monkeypatch.setattr("cbrain_cli.sessions.cbrain_url", "http://localhost:3000") monkeypatch.setattr("cbrain_cli.sessions.api_send", lambda *_, **__: ({}, 200)) result = logout_session(argparse.Namespace()) assert result == 0 @@ -112,8 +105,6 @@ def test_logout_session_success_sends_delete_and_removes_file( def test_logout_session_server_401_still_removes_file(monkeypatch, sessions_creds_file, capsys): sessions_creds_file.write_text('{"api_token": "tok", "cbrain_url": "http://localhost:3000"}') - monkeypatch.setattr("cbrain_cli.sessions.api_token", "tok") - monkeypatch.setattr("cbrain_cli.sessions.cbrain_url", "http://localhost:3000") monkeypatch.setattr( "cbrain_cli.sessions.api_send", MagicMock( diff --git a/tests/test_tags.py b/tests/test_tags.py index 39baec5..3179970 100644 --- a/tests/test_tags.py +++ b/tests/test_tags.py @@ -2,13 +2,13 @@ from cbrain_cli.cli_utils import CliValidationError from cbrain_cli.data.tags import create_tag, delete_tag, list_tags, show_tag, update_tag +from tests.conftest import install_auth from tests.conftest import make_args as _args -from tests.conftest import patch_module_locals @pytest.fixture(autouse=True) -def _patch_locals(monkeypatch): - patch_module_locals(monkeypatch, "cbrain_cli.data.tags") +def _patch_locals(): + install_auth() def _tag_args(**kwargs): diff --git a/tests/test_tasks.py b/tests/test_tasks.py index b796e96..9b4be7e 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -2,12 +2,12 @@ from cbrain_cli.cli_utils import CliValidationError from cbrain_cli.data.tasks import list_tasks, show_task -from tests.conftest import make_args, patch_module_locals +from tests.conftest import install_auth, make_args @pytest.fixture(autouse=True) -def _patch_tasks_locals(monkeypatch): - patch_module_locals(monkeypatch, "cbrain_cli.data.tasks") +def _patch_tasks_locals(): + install_auth() def make_task_args(**kwargs): @@ -20,8 +20,8 @@ def make_task_args(**kwargs): "filter_name,bourreau_id,raises", [ (None, None, False), - ("bourreau-id", 7, False), - ("bourreau-id", None, True), + ("bourreau_id", 7, False), + ("bourreau_id", None, True), (None, 7, True), ], ) @@ -57,10 +57,15 @@ def test_show_task_returns_dict(mock_urlopen): def test_list_tasks_sends_bourreau_id_query(capture_urlopen): configure, captured = capture_urlopen configure([]) - list_tasks(make_task_args(filter_name="bourreau-id", bourreau_id=7)) + list_tasks(make_task_args(filter_name="bourreau_id", bourreau_id=7)) assert "bourreau_id=7" in captured["url"] +def test_list_tasks_rejects_kebab_filter_name(): + with pytest.raises(CliValidationError): + list_tasks(make_task_args(filter_name="bourreau-id", bourreau_id=7)) + + def test_list_tasks_unsupported_filter_raises(): with pytest.raises(CliValidationError): list_tasks(make_task_args(filter_name="other", bourreau_id=1)) diff --git a/tests/test_tool_configs.py b/tests/test_tool_configs.py index 128dfb1..0255b97 100644 --- a/tests/test_tool_configs.py +++ b/tests/test_tool_configs.py @@ -6,13 +6,13 @@ show_tool_config, tool_config_boutiques_descriptor, ) +from tests.conftest import install_auth from tests.conftest import make_args as _args -from tests.conftest import patch_module_locals @pytest.fixture(autouse=True) -def _patch_locals(monkeypatch): - patch_module_locals(monkeypatch, "cbrain_cli.data.tool_configs") +def _patch_locals(): + install_auth() def test_list_tool_configs_returns_list(mock_urlopen): diff --git a/tests/test_tools.py b/tests/test_tools.py index 62370c1..aee9592 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -5,12 +5,12 @@ from cbrain_cli.cli_utils import CliApiError, CliValidationError from cbrain_cli.data.tools import list_tools, show_tool -from tests.conftest import make_args, patch_module_locals +from tests.conftest import install_auth, make_args @pytest.fixture(autouse=True) -def _patch_tools_locals(monkeypatch): - patch_module_locals(monkeypatch, "cbrain_cli.data.tools") +def _patch_tools_locals(): + install_auth() def test_list_tools_passes_page_and_per_page(monkeypatch): diff --git a/tests/test_users.py b/tests/test_users.py index 4278b91..772a498 100644 --- a/tests/test_users.py +++ b/tests/test_users.py @@ -3,11 +3,11 @@ from unittest.mock import MagicMock from cbrain_cli.users import user_details, whoami_user -from tests.conftest import URL, make_args, parse_json_output, patch_module_locals +from tests.conftest import URL, install_auth, make_args, parse_json_output def test_user_details_http_error_returns_none(monkeypatch, capsys): - patch_module_locals(monkeypatch, "cbrain_cli.users") + install_auth() monkeypatch.setattr( "urllib.request.urlopen", MagicMock(side_effect=urllib.error.HTTPError(URL, 500, "Err", {}, None)), @@ -17,7 +17,7 @@ def test_user_details_http_error_returns_none(monkeypatch, capsys): def test_user_details_unexpected_error_returns_none(monkeypatch, capsys): - patch_module_locals(monkeypatch, "cbrain_cli.users") + install_auth() def boom(_req): raise ValueError("parse fail") @@ -27,24 +27,18 @@ def boom(_req): assert "Error getting user details" in capsys.readouterr().out -def test_whoami_missing_credentials_json(capsys, monkeypatch): - monkeypatch.setattr("cbrain_cli.users.user_id", None) - monkeypatch.setattr("cbrain_cli.users.cbrain_url", None) - monkeypatch.setattr("cbrain_cli.users.api_token", None) +def test_whoami_missing_credentials_json(capsys): assert whoami_user(make_args(json=True)) == 1 assert parse_json_output(capsys)["logged_in"] is False -def test_whoami_missing_credentials_plain(capsys, monkeypatch): - monkeypatch.setattr("cbrain_cli.users.user_id", None) - monkeypatch.setattr("cbrain_cli.users.cbrain_url", None) - monkeypatch.setattr("cbrain_cli.users.api_token", None) +def test_whoami_missing_credentials_plain(capsys): assert whoami_user(make_args()) == 1 assert "Credential file is missing" in capsys.readouterr().out def test_whoami_json_output(monkeypatch, capsys): - patch_module_locals(monkeypatch, "cbrain_cli.users", user_id=1) + install_auth(user_id=1) monkeypatch.setattr( "cbrain_cli.users.user_details", lambda _: {"login": "admin", "full_name": "Admin User"}, @@ -56,7 +50,7 @@ def test_whoami_json_output(monkeypatch, capsys): def test_whoami_plain_output(monkeypatch, capsys): - patch_module_locals(monkeypatch, "cbrain_cli.users", user_id=1) + install_auth(user_id=1) monkeypatch.setattr( "cbrain_cli.users.user_details", lambda _: {"login": "admin", "full_name": "Admin User"}, @@ -66,7 +60,7 @@ def test_whoami_plain_output(monkeypatch, capsys): def test_whoami_version_token_mismatch_warning(monkeypatch, capsys): - patch_module_locals(monkeypatch, "cbrain_cli.users", user_id=1) + install_auth(user_id=1) monkeypatch.setattr( "cbrain_cli.users.user_details", lambda _: {"login": "admin", "full_name": "Admin User"}, @@ -81,3 +75,39 @@ def test_whoami_version_token_mismatch_warning(monkeypatch, capsys): out = capsys.readouterr().out assert "WARNING: User ID mismatch" in out assert "Token mismatch" in out + + +def test_login_then_whoami_uses_fresh_credentials(monkeypatch, sessions_creds_file, capsys): + """Verify login/logout/whoami see current credential state in one process.""" + import argparse + + from cbrain_cli.cli_utils import get_auth, is_authenticated + from cbrain_cli.sessions import create_session, logout_session + + assert get_auth() == (None, None, None) + assert is_authenticated() is False + capsys.readouterr() + + inputs = iter(["", "admin"]) + monkeypatch.setattr("builtins.input", lambda _: next(inputs)) + monkeypatch.setattr("getpass.getpass", lambda _: "secret") + monkeypatch.setattr( + "cbrain_cli.sessions.api_post_form", + lambda *_: {"cbrain_api_token": "fresh-tok", "user_id": 7}, + ) + assert create_session(argparse.Namespace()) == 0 + assert get_auth() == (URL, "fresh-tok", 7) + assert is_authenticated() is True + capsys.readouterr() + + monkeypatch.setattr( + "cbrain_cli.users.user_details", + lambda uid: {"login": "admin", "full_name": "Admin"}, + ) + assert whoami_user(make_args(json=True)) == 0 + assert parse_json_output(capsys)["login"] == "admin" + + monkeypatch.setattr("cbrain_cli.sessions.api_send", lambda *_, **__: ({}, 200)) + assert logout_session(argparse.Namespace()) == 0 + assert get_auth() == (None, None, None) + assert is_authenticated() is False