diff --git a/bfabric/docs/design/oauth_integration.md b/bfabric/docs/design/oauth_integration.md index efa8d125..3c798b51 100644 --- a/bfabric/docs/design/oauth_integration.md +++ b/bfabric/docs/design/oauth_integration.md @@ -45,8 +45,8 @@ All commands registered under `bfabric-cli auth` via cyclopts. | Command | File | Description | |---------|------|-------------| -| `auth login ` | `cli/login/pkce.py` | Browser-based OAuth login. Caches tokens + writes config. | -| `auth device-code ` | `cli/login/device_code.py` | Headless OAuth login. | +| `auth login ` | `cli/login/oauth_login.py` | Browser-based OAuth login. Caches tokens + writes config. | +| `auth device-code ` | `cli/login/oauth_login.py` | Headless OAuth login. | | `auth pat ` | `cli/login/pat.py` | Personal Access Token login. | | `auth register ` | `cli/login/register.py` | RFC 7591 dynamic client registration. Outputs JSON. | | `auth status` | `cli/login/status.py` | Show current auth status for an environment. | diff --git a/bfabric/src/bfabric/config/config_writer.py b/bfabric/src/bfabric/config/config_writer.py index a12a194b..ba6b399e 100644 --- a/bfabric/src/bfabric/config/config_writer.py +++ b/bfabric/src/bfabric/config/config_writer.py @@ -1,7 +1,6 @@ """Write environment entries to the bfabricpy YAML config file. -Note: ``yaml.dump`` does not preserve YAML comments — any comments in the -existing config file will be lost when the file is rewritten. +Note: rewriting the file drops any YAML comments in it (``yaml.dump`` doesn't preserve them). """ from __future__ import annotations @@ -9,7 +8,7 @@ import copy import os from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast import yaml @@ -20,12 +19,11 @@ def _write_config_file(config_path: Path, data: Mapping[str, object]) -> None: - """Serialize *data* to *config_path* as YAML with ``0o600`` permissions. - - Centralizes the secret-safe write shared by every config mutation: creates parent - directories, dumps the mapping, and forces ``0o600`` even on a pre-existing file (whose - permissions ``os.open`` would otherwise leave untouched) so a secret never lands in a + """Serialize *data* to *config_path* as YAML, always ``0o600`` so a secret never lands in a group/world-readable file. + + The explicit ``fchmod`` forces the mode even on a pre-existing file, whose permissions + ``os.open`` would otherwise leave untouched. """ config_path = Path(config_path).expanduser() config_path.parent.mkdir(parents=True, exist_ok=True) @@ -39,16 +37,12 @@ def _write_config_file(config_path: Path, data: Mapping[str, object]) -> None: def _validate_round_trip(env_name: str, env_data: Mapping[str, object]) -> None: - """Reject an environment that the reader could not load back. + """Reject an environment the reader could not load back, so a write either persists a parseable + config or fails without touching the file. - Mirrors the reader's invariants so a write either persists a parseable config or fails - cleanly without touching the file. Checks two things: - - * The name is not reserved -- the reader treats ``GENERAL`` as the general section and - forbids ``default``, so neither can be an environment. - * The fields form a valid :class:`EnvironmentConfig` (e.g. ``base_url`` present, auth - combination consistent). Only the single environment is validated, not the merged file, - so a pre-existing legacy environment can't block writing a new, valid one. + Rejects the reserved names (``GENERAL`` is the general section, ``default`` is forbidden) and + anything that isn't a valid :class:`EnvironmentConfig`. Only this one environment is validated, + not the merged file, so a pre-existing legacy environment can't block writing a new valid one. """ if env_name in ("GENERAL", "default"): raise ValueError(f"Environment name {env_name!r} is reserved and cannot be used.") @@ -62,26 +56,19 @@ def write_environment_to_config( *, set_default: bool, ) -> None: - """Write (or update) an environment section in the bfabricpy YAML config. - - * Creates the file if it does not exist. - * Merges *env_data* into the target environment, preserving other - environments. - * Sets ``GENERAL.default_config`` to *env_name* when *set_default* is - ``True``. - * File permissions are set to ``0o600``. + """Write (or update) an environment section in the bfabricpy YAML config, creating the file + (mode ``0o600``) if needed and preserving other environments. Sets ``GENERAL.default_config`` + to *env_name* when *set_default*. :param config_path: Path to the YAML config file (will be expanded). :param env_name: Name of the environment to create / update. - :param env_data: Dictionary of fields for the environment. - :param set_default: If ``True``, set this environment as the default. Required: callers must - decide explicitly whether the new environment becomes the default. - :raises pydantic.ValidationError: If *env_data* would not parse back through the reader - (e.g. a missing ``base_url`` or an invalid auth combination). Validated before any - filesystem change, so a rejected write leaves an existing config untouched. + :param env_data: Fields for the environment. + :param set_default: Whether this environment becomes the default. Required: callers must decide + explicitly. + :raises pydantic.ValidationError: If *env_data* would not parse back through the reader (e.g. a + missing ``base_url`` or an invalid auth combination). Checked before any filesystem change, + so a rejected write leaves an existing config untouched. """ - # Guarantee the round-trip before any filesystem change, so a rejected write leaves an - # existing config untouched. _validate_round_trip(env_name, env_data) config_path = Path(config_path).expanduser() @@ -108,15 +95,13 @@ def write_environment_to_config( def set_default_config(config_path: Path, env_name: str) -> None: """Set ``GENERAL.default_config`` to an already-defined environment. - Unlike :func:`write_environment_to_config`, this only flips the default -- it never - creates or modifies an environment. Other environments and the general section are - preserved. + Only flips the default; never creates or modifies an environment. :param config_path: Path to the YAML config file (will be expanded). :param env_name: Name of an existing environment to mark as default. :raises FileNotFoundError: If the config file does not exist. - :raises ValueError: If *env_name* is not among the configured environments; the file is - left untouched in that case. + :raises ValueError: If *env_name* is not among the configured environments; the file is left + untouched. """ config_path = Path(config_path).expanduser() if not config_path.is_file(): @@ -126,13 +111,10 @@ def set_default_config(config_path: Path, env_name: str) -> None: existing: dict[str, object] existing = loaded if isinstance(loaded, dict) else {} # pyright: ignore[reportUnknownVariableType] - # Enumerate the configured environments through the reader so the check matches how the - # file will actually load back. ConfigFile's "before" validators mutate their input in - # place, so always validate a deep copy and keep ``existing`` pristine for the write. - # This also doubles as the round-trip guard: since env_name is confirmed to be one of - # config_file_obj.environments, and environments are otherwise untouched below, setting - # GENERAL.default_config to env_name cannot fail ConfigFile's own default-config-must-exist - # validator -- so no second validation pass is needed after the mutation. + # Validate through the reader (on a deep copy — ConfigFile's "before" validators mutate their + # input in place) so the membership check matches how the file loads back, keeping ``existing`` + # pristine for the write. This doubles as the round-trip guard: env_name is a known environment + # and the environments are untouched below, so setting the default can't fail the reader after. config_file_obj = ConfigFile.model_validate(copy.deepcopy(existing)) if env_name not in config_file_obj.environments: available = ", ".join(sorted(config_file_obj.environments)) or "(none)" @@ -144,3 +126,41 @@ def set_default_config(config_path: Path, env_name: str) -> None: general["default_config"] = env_name _write_config_file(config_path, existing) + + +def remove_environment_from_config(config_path: Path, env_name: str) -> None: + """Delete an environment section from the bfabricpy YAML config. + + Also clears ``GENERAL.default_config`` if it pointed at *env_name* — otherwise the reader would + refuse to load a file whose default names a missing environment. + + :param config_path: Path to the YAML config file (will be expanded). + :param env_name: Name of an existing environment to remove. + :raises FileNotFoundError: If the config file does not exist. + :raises ValueError: If *env_name* is not among the configured environments; the file is left + untouched. + """ + config_path = Path(config_path).expanduser() + if not config_path.is_file(): + raise FileNotFoundError(f"Config file not found: {config_path}") + + loaded: object = yaml.safe_load(config_path.read_text()) # pyright: ignore[reportAny] + existing: dict[str, object] + existing = loaded if isinstance(loaded, dict) else {} # pyright: ignore[reportUnknownVariableType] + + # Enumerate through the reader so the membership check matches how the file loads back. + # ConfigFile's "before" validators mutate their input in place, so validate a deep copy and + # keep ``existing`` pristine for the write. + config_file_obj = ConfigFile.model_validate(copy.deepcopy(existing)) + if env_name not in config_file_obj.environments: + available = ", ".join(sorted(config_file_obj.environments)) or "(none)" + raise ValueError(f"Environment {env_name!r} is not defined. Available environments: {available}") + + _ = existing.pop(env_name, None) + general = existing.get("GENERAL") + if isinstance(general, dict): + general_map = cast("dict[str, object]", general) + if general_map.get("default_config") == env_name: + _ = general_map.pop("default_config", None) + + _write_config_file(config_path, existing) diff --git a/bfabric_scripts/docs/changelog.md b/bfabric_scripts/docs/changelog.md index 089784a8..aca41ed2 100644 --- a/bfabric_scripts/docs/changelog.md +++ b/bfabric_scripts/docs/changelog.md @@ -10,21 +10,13 @@ Versioning currently follows `X.Y.Z` semantic versioning, independent of the `bf ## \[Unreleased\] -- `auth login` / `auth device-code` now **require** an explicit `--scope` (no default). Named presets document the useful sets — `read-only` (`api:read`), `read-write` (`api:write`), `upload` (`api:write tus`) — or pass any scope string. Client/webapp registration (`auth register` / `auth register-webapp`) keeps the OIDC-inclusive default, which webapps need for user-identity claims. - ## \[1.16.0rc2\] - 2026-07-15 -- `bfabric-cli auth` command group for OAuth authentication and client management: - - Login: `auth login` (browser), `auth device-code` (headless), `auth pat` (Personal Access Token). - - `auth register` / `auth register-webapp` — dynamic client registration, optionally with a linked B-Fabric app. - - `auth default` / `auth default set [CONFIG_ENV]` — show and set the default environment (interactive picker when no value is given). - - `auth status` and `auth logout`. - - `auth pat` stores the token under a `pat` key with `auth_method: pat` (not `login: __oauth__` / `password:`), keeping the shared config parseable by older (≤1.19.0) clients; `auth status` reports these as `pat`. -- `bfabric-cli workunit upload FILES...` — upload files and directories to a workunit over tus (resumable, large-file capable), creating a new workunit or targeting `--workunit-id`. One resource per file, skips duplicates (unless `--force`), expands directories, live progress (`--no-progress`); `--track-job` records an `UPLOAD` job. Requires an OAuth client with the `tus` scope. -- `bfabric-cli api create` / `api update` accept `--format json|yaml|tsv|table_rich` (default `json`), matching `api read`. -- `bfabric-cli dataset update` — update an existing dataset with a change preview before confirming (`csv`/`tsv`/`xlsx`/`parquet`, same validation flags as `dataset upload`). -- `api create` / `api update` now emit valid JSON (was Python `repr` via `rich.pretty.pprint`, breaking `jq`) and serialise `datetime` / `Decimal` to strings instead of raising `TypeError` ([#503](https://github.com/fgcz/bfabricPy/issues/503)). -- Internal: `dataset upload` and `bfabric_save_csv2dataset.py` now use `bfabric.operations.dataset.create_dataset`; API error handling is centralized in `@use_client` (no more `@logger.catch`); `--config-env` naming unified. Declares `lxml` as an explicit dependency (was transitive via the now-optional `zeep`) and requires `bfabric[transfer]` >= 1.20 (for `upload_files` / tus). +- `bfabric-cli auth` — OAuth authentication & client management. Login: `login` (browser), `device-code` (headless), `pat`; client registration: `register` / `register-webapp`; environment management: `default`, `list`, `status`, `logout`. Scope presets (`read-only` / `read-write` / `upload`) or a raw scope, via an interactive picker when `--scope` is omitted in a terminal; no baked-in default scope, so a headless run must pass `--scope` (registration keeps the OIDC-inclusive default webapps need). When `--config-env` is omitted it prompts for the environment (else targets the current default / `PRODUCTION`); unless `--set-default` / `--no-set-default` is given it asks (default yes) whether to make the env the default, and cancelling that prompt aborts the login. `status` reports an OAuth env's cached-token freshness and granted scope (annotated with the matching preset); `logout` removes an env's config entry and cached tokens (confirmation required). PATs are stored under a `pat` key (`auth_method: pat`), keeping the config parseable by ≤1.19.0 clients. +- `bfabric-cli workunit upload FILES...` — upload files/directories to a workunit over tus (resumable, large-file capable): new or `--workunit-id`, one resource per file, skips duplicates (`--force`), live progress (`--no-progress`), optional `--track-job`. Requires an OAuth client with the `tus` scope. +- `bfabric-cli api create` / `api update` — accept `--format json|yaml|tsv|table_rich` (default `json`); now emit valid JSON and serialise `datetime` / `Decimal` (was Python `repr`, breaking `jq`) ([#503](https://github.com/fgcz/bfabricPy/issues/503)). +- `bfabric-cli dataset update` — update an existing dataset with a change preview before confirming (`csv`/`tsv`/`xlsx`/`parquet`). +- Internal: `dataset upload` / `bfabric_save_csv2dataset.py` use `bfabric.operations.dataset.create_dataset`; API error handling centralized in `@use_client` (dropped `@logger.catch`); `--config-env` naming unified; `lxml` now an explicit dep (was transitive via optional `zeep`); requires `bfabric[transfer]>=1.20`; adds `questionary`-based `cli.interactive` helpers and `config_writer.remove_environment_from_config`. ## \[1.15.0\] - 2026-04-20 diff --git a/bfabric_scripts/pyproject.toml b/bfabric_scripts/pyproject.toml index 0c1d0bb1..de339d6e 100644 --- a/bfabric_scripts/pyproject.toml +++ b/bfabric_scripts/pyproject.toml @@ -12,6 +12,7 @@ dependencies = [ "bfabric[transfer]>=1.20.0rc2,<2", "cyclopts>=4.0,<5.0", "Flask>=3.0.3,<4.0", + "questionary>=2.0,<3", "rich>=13.0,<15", "lxml>=5.0,<7.0", "xmltodict>=1.0.0,<2.0.0", @@ -35,7 +36,7 @@ excel = [ ] [project.scripts] -"bfabric-cli" = "bfabric_scripts.cli.__main__:app" +"bfabric-cli" = "bfabric_scripts.cli.__main__:main" # TODO scripts for feeder "bfabric_list_not_existing_storage_directories.py" = "bfabric_scripts.bfabric_list_not_existing_storage_directories:app" diff --git a/bfabric_scripts/src/bfabric_scripts/cli/__main__.py b/bfabric_scripts/src/bfabric_scripts/cli/__main__.py index 511107af..8ef44dbd 100644 --- a/bfabric_scripts/src/bfabric_scripts/cli/__main__.py +++ b/bfabric_scripts/src/bfabric_scripts/cli/__main__.py @@ -2,6 +2,7 @@ import cyclopts +from bfabric.utils.cli_integration import setup_script_logging from bfabric_scripts.cli.cli_api import cmd_api from bfabric_scripts.cli.cli_dataset import cmd_dataset from bfabric_scripts.cli.cli_executable import cmd_executable @@ -23,5 +24,17 @@ # TODO delete after transitory release _ = app.command(_app_external_job, name="external-job") -if __name__ == "__main__": + +def main() -> None: + """CLI entry point: configure logging once, then dispatch. + + Commands using ``@use_client`` set logging up themselves, but those that don't (e.g. ``auth``) + would otherwise run under loguru's default DEBUG handler. Setting up here covers every command; + the per-command call is idempotent. + """ + setup_script_logging() app() + + +if __name__ == "__main__": + main() diff --git a/bfabric_scripts/src/bfabric_scripts/cli/cli_auth.py b/bfabric_scripts/src/bfabric_scripts/cli/cli_auth.py index 0dd0d499..1d35a887 100644 --- a/bfabric_scripts/src/bfabric_scripts/cli/cli_auth.py +++ b/bfabric_scripts/src/bfabric_scripts/cli/cli_auth.py @@ -1,13 +1,10 @@ import cyclopts -from bfabric_scripts.cli.login.default_config import cmd_auth_default -from bfabric_scripts.cli.login.device_code import cmd_login_device_code -from bfabric_scripts.cli.login.logout import cmd_login_logout +from bfabric_scripts.cli.login.manage import cmd_auth_default, cmd_auth_list, cmd_login_logout, cmd_login_status +from bfabric_scripts.cli.login.oauth_login import cmd_auth_login, cmd_login_device_code from bfabric_scripts.cli.login.pat import cmd_login_pat -from bfabric_scripts.cli.login.pkce import cmd_auth_login from bfabric_scripts.cli.login.register import cmd_login_register from bfabric_scripts.cli.login.register_webapp import cmd_login_register_webapp -from bfabric_scripts.cli.login.status import cmd_login_status cmd_auth = cyclopts.App(help="Authentication commands for B-Fabric.") _ = cmd_auth.command(cmd_login_pat, name="pat") @@ -18,3 +15,4 @@ _ = cmd_auth.command(cmd_login_status, name="status") _ = cmd_auth.command(cmd_login_logout, name="logout") _ = cmd_auth.command(cmd_auth_default, name="default") +_ = cmd_auth.command(cmd_auth_list, name="list") diff --git a/bfabric_scripts/src/bfabric_scripts/cli/interactive.py b/bfabric_scripts/src/bfabric_scripts/cli/interactive.py new file mode 100644 index 00000000..fc4d63d5 --- /dev/null +++ b/bfabric_scripts/src/bfabric_scripts/cli/interactive.py @@ -0,0 +1,61 @@ +"""Interactive prompt helpers built on ``questionary``. + +Prompts require a TTY; callers guard with :func:`is_interactive` and handle the ``None`` +returned on cancel (or empty input, where noted). +""" + +from __future__ import annotations + +import sys +from collections.abc import Callable, Sequence +from typing import cast + +import questionary + + +def is_interactive() -> bool: + """Whether both stdin and stdout are attached to a terminal.""" + return sys.stdin.isatty() and sys.stdout.isatty() + + +def select_choice( + message: str, + choices: Sequence[str], + *, + default: str | None = None, + describe: Callable[[str], str] | None = None, + search: bool = False, +) -> str | None: + """Arrow-key menu over *choices*; returns the picked value or ``None`` on cancel. + + *default* must be one of *choices*. *describe* maps a value to its display label. *search* + lets the user type to filter. + """ + items: list[str | questionary.Choice] + if describe is not None: + items = [questionary.Choice(title=describe(choice), value=choice) for choice in choices] + else: + items = list(choices) + return cast( + "str | None", + questionary.select(message, choices=items, default=default, use_search_filter=search, use_jk_keys=False).ask(), + ) + + +def select_or_input(message: str, choices: Sequence[str], *, default: str | None = None) -> str | None: + """Autocomplete over *choices* that also accepts a value not in the list; ``None`` on cancel/empty.""" + items = list(choices) + prompt = f"{message} (Tab to autocomplete)" if items else message + answer = cast("str | None", questionary.autocomplete(prompt, choices=items, default=default or "").ask()) + return answer or None + + +def text_input(message: str, *, default: str = "") -> str | None: + """Free-text prompt prefilled with *default*; ``None`` on cancel/empty.""" + answer = cast("str | None", questionary.text(message, default=default).ask()) + return answer or None + + +def confirm(message: str, *, default: bool = False) -> bool | None: + """Yes/no prompt; ``True``/``False``, or ``None`` on cancel (distinct from a declined "no").""" + return cast("bool | None", questionary.confirm(message, default=default).ask()) diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/_common.py b/bfabric_scripts/src/bfabric_scripts/cli/login/_common.py new file mode 100644 index 00000000..9b324966 --- /dev/null +++ b/bfabric_scripts/src/bfabric_scripts/cli/login/_common.py @@ -0,0 +1,91 @@ +"""Shared login-parameter resolution for the ``auth`` login commands. + +An environment name and OAuth scope may be given on the command line, picked interactively, or -- +for the environment -- typed as a new value; whether the environment becomes the config default +is likewise resolved here. Used by ``auth pat`` / ``auth login`` / ``auth device-code``. +""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +from bfabric.config.config_file import ConfigFile +from bfabric_scripts.cli.interactive import confirm, is_interactive, select_choice, select_or_input, text_input +from bfabric_scripts.cli.login._constants import SCOPE_PRESETS, SCOPE_PRESETS_BY_NAME + +# Interactive-only sentinel: choosing it opens a free-text prompt for raw scopes. +_CUSTOM = "custom" + +# Fallback environment name when none is configured and none is given (historical default). +_FALLBACK_ENV = "PRODUCTION" + + +def resolve_config_env(config_env: str | None, config_file: Path) -> str | None: + """Resolve the target environment name. + + Explicit *config_env* wins; non-interactive falls back to the current default (else + ``"PRODUCTION"``); interactive picks an existing name or a typed new one, prefilled with that + fallback. ``None`` if cancelled. + """ + if config_env is not None: + return config_env + config_path = Path(config_file).expanduser() + if config_path.is_file(): + loaded = ConfigFile.model_validate(yaml.safe_load(config_path.read_text())) + names, current = list(loaded.environments), loaded.general.default_config + else: + names, current = [], None + fallback = current or _FALLBACK_ENV + if not is_interactive(): + return fallback + return select_or_input("Environment name", names, default=fallback) + + +def _scope_menu_label(choice: str) -> str: + """Menu label for a scope preset (description + the scopes it maps to) or the Custom entry.""" + width = max(len(preset.description) for preset in SCOPE_PRESETS) + if choice == _CUSTOM: + return f"{'Custom…'.ljust(width)} (enter scopes manually)" + preset = SCOPE_PRESETS_BY_NAME[choice] + return f"{preset.description.ljust(width)} {preset.scope}" + + +def resolve_scope(scope: str | None) -> str | None: + """Resolve the OAuth scope string. + + A given *scope* expands preset names and passes anything else through as a raw scope string. + Non-interactive returns ``None`` (no default scope, so a headless login must pass ``--scope``; + the caller aborts). Interactive picks a preset (least-privilege preselected) or Custom + free-text. ``None`` if cancelled. + """ + if scope is not None: + preset = SCOPE_PRESETS_BY_NAME.get(scope) + return preset.scope if preset is not None else scope + if not is_interactive(): + return None + picked = select_choice( + "Select OAuth scope set", + [preset.name for preset in SCOPE_PRESETS] + [_CUSTOM], + default=SCOPE_PRESETS[0].name, + describe=_scope_menu_label, + ) + if picked is None: + return None + if picked == _CUSTOM: + return text_input("Enter OAuth scopes (space-separated)") + return SCOPE_PRESETS_BY_NAME[picked].scope + + +def resolve_set_default(set_default: bool | None, config_env: str) -> bool | None: + """Resolve whether the freshly-authenticated environment becomes the config default. + + Explicit *set_default* wins; non-interactive defaults to ``True``; interactive prompts (yes + preselected). ``None`` if cancelled (the caller aborts). + """ + if set_default is not None: + return set_default + if not is_interactive(): + return True + return confirm(f"Set '{config_env}' as the default environment?", default=True) diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/_constants.py b/bfabric_scripts/src/bfabric_scripts/cli/login/_constants.py index 3cce90fa..ffd61ede 100644 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/_constants.py +++ b/bfabric_scripts/src/bfabric_scripts/cli/login/_constants.py @@ -1,9 +1,8 @@ """Default OAuth client ID and scope policy for the CLI login commands. -These are CLI-level policy: ``"CLI"`` is the client identity the CLI registers -as, and the scopes below are the permission sets the CLI requests. The core -``bfabric`` library deliberately does not bake in these defaults — its OAuth -API requires callers to state ``client_id`` and ``scope`` explicitly. +CLI-level policy: ``"CLI"`` is the client identity the CLI registers as, and the scopes are the +permission sets it requests. The core ``bfabric`` library bakes in neither — its OAuth API +requires explicit ``client_id`` and ``scope``. """ from __future__ import annotations @@ -12,10 +11,9 @@ DEFAULT_CLIENT_ID = "CLI" -# Scope requested when *registering* an OAuth client/webapp (``auth register`` / -# ``auth register-webapp``). Broad and OIDC-inclusive on purpose: a registered webapp -# needs ``openid profile email groups`` for the user-identity claims it reads from the -# URL-token exchange, which the login presets below intentionally omit. +# Scope requested when *registering* an OAuth client/webapp. Broad and OIDC-inclusive on purpose: +# a registered webapp needs ``openid profile email groups`` for the user-identity claims it reads +# from the URL-token exchange, which the login presets below omit. DEFAULT_REGISTRATION_SCOPE = "api:read api:write openid profile email groups" @@ -27,11 +25,12 @@ class ScopePreset(NamedTuple): description: str -# Scope presets for interactive *login* (``auth login`` / ``auth device-code``), ordered -# by increasing capability. ``api:write`` implies ``api:read`` server-side, so the -# read-write set lists only ``api:write``; ``tus`` adds file-upload permission on top. +# Scope presets for interactive *login*, ordered by increasing capability. ``api:write`` implies +# ``api:read`` server-side, so read-write lists only ``api:write``; ``tus`` adds file upload. SCOPE_PRESETS: tuple[ScopePreset, ...] = ( ScopePreset("read-only", "api:read", "Read from the API"), ScopePreset("read-write", "api:write", "Write to the API (includes read from the API)"), ScopePreset("upload", "api:write tus", "Upload files to the API (includes read and write API)"), ) + +SCOPE_PRESETS_BY_NAME: dict[str, ScopePreset] = {preset.name: preset for preset in SCOPE_PRESETS} diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/default_config.py b/bfabric_scripts/src/bfabric_scripts/cli/login/default_config.py deleted file mode 100644 index af8170a4..00000000 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/default_config.py +++ /dev/null @@ -1,105 +0,0 @@ -"""Commands to show and set the default configuration environment.""" - -from __future__ import annotations - -from pathlib import Path -from typing import Annotated - -import cyclopts -import yaml -from rich.console import Console -from rich.prompt import Prompt -from rich.text import Text - -from bfabric.config import DEFAULT_CONFIG_FILE -from bfabric.config.config_file import ConfigFile -from bfabric.config.config_writer import set_default_config - -cmd_auth_default = cyclopts.App(help="Show or set the default configuration environment.") - - -def _environment_line(name: str, *, index: int | None, is_default: bool) -> Text: - """Render one environment row. Text (not markup) keeps names with "[" literal. - - The current default is prefixed with a bold-green arrow so it stands out in a long list. - """ - label = f"{index}. {name}" if index is not None else name - if is_default: - # Chained appends (each returns the Text) keep the whole row in one expression. - return Text("→ ", style="bold green").append(label, style="bold green").append(" (default)", style="green") - return Text(f" {label}") - - -def _prompt_for_environment(console: Console, names: list[str], default: str | None) -> str: - """Show a numbered menu of *names* and return the environment the user picks.""" - console.print("Configuration environments:") - for index, name in enumerate(names, start=1): - console.print(_environment_line(name, index=index, is_default=name == default)) - choices = [str(i) for i in range(1, len(names) + 1)] - # show_choices=False: the numbered menu above already lists the options, so the prompt stays - # short ("Select environment (3):") instead of repeating every name inline. Only offer a - # default when there is a current default to pre-select; otherwise the pick is required. - if default in names: - answer = Prompt.ask( - "Select environment", choices=choices, default=str(names.index(default) + 1), show_choices=False - ) - else: - answer = Prompt.ask("Select environment", choices=choices, show_choices=False) - return names[int(answer) - 1] - - -@cmd_auth_default.default -def cmd_auth_default_show( - *, - config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, -) -> None: - """List configuration environments and show the current default.""" - config_path = Path(config_file).expanduser() - if not config_path.is_file(): - print(f"Config file not found: {config_path}") - return - - config_file_obj = ConfigFile.model_validate(yaml.safe_load(config_path.read_text())) - names = list(config_file_obj.environments) - if not names: - print("No environments configured.") - return - - default = config_file_obj.general.default_config - console = Console() - console.print("Configuration environments:") - for name in names: - console.print(_environment_line(name, index=None, is_default=name == default)) - if default is None: - console.print("\nNo default configured.") - - -@cmd_auth_default.command(name="set") -def cmd_auth_default_set( - config_env: Annotated[ - str | None, cyclopts.Parameter(help="Environment to set as default (prompted if omitted).") - ] = None, - *, - config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, -) -> None: - """Set the default configuration environment.""" - config_path = Path(config_file).expanduser() - if not config_path.is_file(): - print(f"Config file not found: {config_path}") - return - - config_file_obj = ConfigFile.model_validate(yaml.safe_load(config_path.read_text())) - names = list(config_file_obj.environments) - if not names: - print("No environments configured.") - return - - if config_env is None: - config_env = _prompt_for_environment(Console(), names, config_file_obj.general.default_config) - - if config_env not in config_file_obj.environments: - print(f"Environment '{config_env}' not found. Available environments: {', '.join(names)}") - return - - set_default_config(config_path, config_env) - print(f"Default environment set to '{config_env}'.") diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/device_code.py b/bfabric_scripts/src/bfabric_scripts/cli/login/device_code.py deleted file mode 100644 index a72e5fde..00000000 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/device_code.py +++ /dev/null @@ -1,63 +0,0 @@ -"""Device-code login command.""" - -from __future__ import annotations - -from pathlib import Path -from typing import Annotated - -import cyclopts - -from bfabric._oauth.credential_provider import OAuthCredentialProvider -from bfabric._oauth.device_code import device_code_login -from bfabric._oauth.token_cache import compute_token_cache_path -from bfabric.config import DEFAULT_CONFIG_FILE -from bfabric.config.config_writer import write_environment_to_config -from bfabric_scripts.cli.login._constants import DEFAULT_CLIENT_ID - - -def cmd_login_device_code( - base_url: Annotated[str, cyclopts.Parameter(help="B-Fabric instance URL.")], - *, - scope: Annotated[str, cyclopts.Parameter(help="OAuth scope.")], - client_id: Annotated[str, cyclopts.Parameter(help="OAuth client ID.")] = DEFAULT_CLIENT_ID, - config_env: Annotated[str, cyclopts.Parameter(help="Environment name in the config file.")] = "PRODUCTION", - config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, - timeout: Annotated[float, cyclopts.Parameter(help="Seconds to wait for authorization.")] = 600.0, - set_default: Annotated[ - bool, cyclopts.Parameter(help="Set this environment as the default in the config file.") - ] = True, -) -> None: - """Authenticate via device code flow (for headless environments).""" - import sys - - base_url = base_url.rstrip("/") - try: - token = device_code_login( - base_url, - client_id=client_id, - scope=scope, - timeout=timeout, - ) - except RuntimeError as e: - print(f"Error: {e}", file=sys.stderr) - raise SystemExit(1) from None - cache_path = compute_token_cache_path(base_url, client_id, config_env).expanduser() - token_url = f"{base_url}/rest/oauth/token" - # Constructed for its side effect: persists the fresh token to the disk cache. - _ = OAuthCredentialProvider( - client_id=client_id, - client_secret="", - token_url=token_url, - token=token, - grant_type="refresh_token", - scope=scope, - token_cache_path=cache_path, - ) - env_data = { - "base_url": base_url, - "auth_method": "oauth", - "client_id": client_id, - } - write_environment_to_config(config_file, config_env, env_data, set_default=set_default) - print(f"Authenticated successfully.") - print(f"Config saved to environment '{config_env}' in {config_file}") diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/logout.py b/bfabric_scripts/src/bfabric_scripts/cli/login/logout.py deleted file mode 100644 index d55a3876..00000000 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/logout.py +++ /dev/null @@ -1,46 +0,0 @@ -"""Logout command — clears token cache for an environment.""" - -from __future__ import annotations - -import os -from pathlib import Path -from typing import Annotated - -import cyclopts -import yaml - -from bfabric.config import DEFAULT_CONFIG_FILE -from bfabric.config.config_file import ConfigFile -from bfabric._oauth.token_cache import TokenCache, compute_token_cache_path -from bfabric_scripts.cli.login._constants import DEFAULT_CLIENT_ID - - -def cmd_login_logout( - *, - config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, - config_env: Annotated[str | None, cyclopts.Parameter(help="Environment name (default: auto-detect).")] = None, -) -> None: - """Clear cached OAuth tokens for an environment.""" - config_path = Path(config_file).expanduser() - if not config_path.is_file(): - print(f"Config file not found: {config_path}") - return - - config_file_obj = ConfigFile.model_validate(yaml.safe_load(config_path.read_text())) - resolved_env = config_env or os.environ.get("BFABRICPY_CONFIG_ENV") or config_file_obj.general.default_config - if resolved_env is None: - print("No environment specified and no default configured.") - return - - if resolved_env not in config_file_obj.environments: - print(f"Environment '{resolved_env}' not found in config.") - return - - env = config_file_obj.environments[resolved_env] - if env.auth_method == "oauth": - client_id = env.client_id or DEFAULT_CLIENT_ID - cache_path = compute_token_cache_path(env.config.base_url.rstrip("/"), client_id, resolved_env).expanduser() - TokenCache(cache_path).clear() - print(f"Token cache cleared: {cache_path}") - else: - print(f"Environment '{resolved_env}' does not use OAuth; nothing to clear.") diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/manage.py b/bfabric_scripts/src/bfabric_scripts/cli/login/manage.py new file mode 100644 index 00000000..fda93eee --- /dev/null +++ b/bfabric_scripts/src/bfabric_scripts/cli/login/manage.py @@ -0,0 +1,267 @@ +"""Auth commands that inspect or manage existing config environments: list, default, status, logout. + +Their shared config-load, environment picker, and host/auth/scope rendering helpers live here too. +""" + +from __future__ import annotations + +import os +import sys +import time +from pathlib import Path +from typing import Annotated +from urllib.parse import urlsplit + +import cyclopts +import yaml + +from bfabric._oauth.token_cache import TokenCache, compute_token_cache_path +from bfabric.config import DEFAULT_CONFIG_FILE +from bfabric.config.config_file import ConfigFile, EnvironmentConfig +from bfabric.config.config_writer import remove_environment_from_config, set_default_config +from bfabric_scripts.cli.interactive import confirm, is_interactive, select_choice +from bfabric_scripts.cli.login._constants import DEFAULT_CLIENT_ID, SCOPE_PRESETS + + +def _load_config(config_file: Path) -> ConfigFile | None: + """Load and validate the config file, or print a "not found" notice and return ``None``.""" + config_path = Path(config_file).expanduser() + if not config_path.is_file(): + print(f"Config file not found: {config_path}") + return None + return ConfigFile.model_validate(yaml.safe_load(config_path.read_text())) + + +def _require_environments(config_file: Path) -> ConfigFile | None: + """Like :func:`_load_config`, but also prints a notice and returns ``None`` if no environments exist.""" + config = _load_config(config_file) + if config is None: + return None + if not config.environments: + print("No environments configured.") + return None + return config + + +def _auth_method(env: EnvironmentConfig) -> str: + """Effective auth method; falls back to ``auth``'s presence for legacy envs without ``auth_method``.""" + if env.auth_method in ("oauth", "pat"): + return env.auth_method + return "password" if env.auth is not None else "none" + + +def _oauth_cache_path(env: EnvironmentConfig, env_name: str) -> Path: + """Disk path of *env_name*'s cached OAuth token (keyed by base URL + client ID).""" + client_id = env.client_id or DEFAULT_CLIENT_ID + return compute_token_cache_path(env.config.base_url.rstrip("/"), client_id, env_name).expanduser() + + +def environment_summary(env: EnvironmentConfig) -> str: + """A compact "host · auth-method" descriptor shown next to each environment name.""" + base_url = str(env.config.base_url) + host = urlsplit(base_url).netloc or base_url + return f"{host} · {_auth_method(env)}" + + +def print_environments(environments: dict[str, EnvironmentConfig], default: str | None) -> None: + """List the configured environments with their host/auth summary, marking the default.""" + print("Configuration environments:") + width = max((len(name) for name in environments), default=0) + for name, env in environments.items(): + marker, tag = ("→", " (default)") if name == default else (" ", "") + print(f"{marker} {name.ljust(width)} {environment_summary(env)}{tag}") + + +def _select_environment(message: str, config: ConfigFile) -> str | None: + """Interactive picker over the configured environments, each labelled with its host/auth summary.""" + names = list(config.environments) + width = max(len(name) for name in names) + default = config.general.default_config + return select_choice( + message, + names, + default=default if default in names else None, + describe=lambda name: f"{name.ljust(width)} {environment_summary(config.environments[name])}", + search=True, + ) + + +def _normalize_scope(scope: str) -> str: + """Order-insensitive form of a scope string for comparing against the presets.""" + return " ".join(sorted(scope.split())) + + +def describe_scope(scope: object) -> str: + """Render a granted scope, appending ``[]`` on match; ``"(not recorded)"`` if missing/non-string.""" + if not isinstance(scope, str) or not scope.strip(): + return "(not recorded)" + normalized = _normalize_scope(scope) + for preset in SCOPE_PRESETS: + if _normalize_scope(preset.scope) == normalized: + return f"{scope} [{preset.name}]" + return scope + + +def describe_token_cache(cached: dict[str, object] | None, *, now: float) -> str: + """Summarize a cached token's freshness: ``missing`` / ``present`` (+ expiry when ``expires_at`` is set).""" + if cached is None: + return "missing" + expires_at = cached.get("expires_at") + if not isinstance(expires_at, (int, float)): + return "present" + remaining = float(expires_at) - now + if remaining <= 0: + return "present, expired" + minutes = int(remaining // 60) + label = f"~{minutes // 60}h" if minutes >= 60 else f"~{minutes}m" + return f"present, expires in {label}" + + +def cmd_auth_list( + *, + config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, +) -> None: + """List the configured environments, marking the default and showing each host / auth method.""" + config = _require_environments(config_file) + if config is None: + return + print_environments(config.environments, config.general.default_config) + + +def cmd_auth_default( + config_env: Annotated[ + str | None, + cyclopts.Parameter(help="Environment to set as default (interactive picker if omitted)."), + ] = None, + *, + config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, +) -> None: + """Set the default configuration environment. + + With no *config_env*, opens an interactive picker in a terminal, or lists the environments + non-interactively. + """ + config = _require_environments(config_file) + if config is None: + return + names = list(config.environments) + + if config_env is None and is_interactive(): + config_env = _select_environment("Select the default environment", config) + if config_env is None: + # Cancelled picker, or no TTY to prompt on. + if is_interactive(): + print("No changes made.") + else: + print("No environment specified. Pass an environment name or run in an interactive terminal.") + return + + if config_env not in config.environments: + print(f"Environment '{config_env}' not found. Available environments: {', '.join(names)}") + return + + set_default_config(config_file, config_env) + print(f"Default environment set to '{config_env}'.") + + +def cmd_login_status( + *, + config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, + config_env: Annotated[str | None, cyclopts.Parameter(help="Environment name (default: auto-detect).")] = None, +) -> None: + """Show current authentication status.""" + config = _load_config(config_file) + if config is None: + return + resolved_env = config_env or os.environ.get("BFABRICPY_CONFIG_ENV") or config.general.default_config + if resolved_env is None: + print("No environment specified and no default configured.") + return + if resolved_env not in config.environments: + print(f"Environment '{resolved_env}' not found in config.") + return + + env = config.environments[resolved_env] + print(f"Environment: {resolved_env}") + print(f"Base URL: {env.config.base_url}") + print(f"Auth method: {_auth_method(env)}") + + if env.auth_method == "oauth": + client_id = env.client_id or DEFAULT_CLIENT_ID + cache_path = _oauth_cache_path(env, resolved_env) + cached = TokenCache(cache_path).load() + print(f"Client ID: {client_id}") + print(f"Token cache: {cache_path} ({describe_token_cache(cached, now=time.time())})") + print(f"Scope: {describe_scope(cached.get('scope') if cached else None)}") + elif env.auth_method == "pat": + print("Token: stored in config file") + elif env.auth is not None: + print(f"Login: {env.auth.login}") + + +def cmd_login_logout( + config_env: Annotated[ + str | None, + cyclopts.Parameter(help="Environment to remove (interactive picker if omitted)."), + ] = None, + *, + config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, + no_confirm: Annotated[ + bool, cyclopts.Parameter(help="Skip the confirmation prompt (required to remove non-interactively).") + ] = False, +) -> None: + """Remove an environment: delete its config entry and clear any cached OAuth tokens. + + With no *config_env*, opens an interactive picker. A non-interactive run must name the + environment and pass ``--no-confirm`` (it cannot prompt for the destructive confirmation). + """ + config = _require_environments(config_file) + if config is None: + return + environments = config.environments + names = list(environments) + + if config_env is None: + if not is_interactive(): + print("Specify --config-env to choose an environment to remove non-interactively.", file=sys.stderr) + return + config_env = _select_environment("Select the environment to remove", config) + if config_env is None: + print("No changes made.") + return + + if config_env not in environments: + print(f"Environment '{config_env}' not found. Available environments: {', '.join(names)}") + return + + env = environments[config_env] + # Removing the current default clears it (a dangling default makes the config unloadable). Only + # worth flagging when other environments remain to default to. + leaves_no_default = config_env == config.general.default_config and len(names) > 1 + + if not no_confirm: + if not is_interactive(): + print( + f"Refusing to remove '{config_env}' without confirmation; pass --no-confirm to proceed.", + file=sys.stderr, + ) + return + prompt = ( + f"Remove environment '{config_env}' ({environment_summary(env)})? " + "This deletes its config entry and any cached OAuth tokens." + ) + if leaves_no_default: + prompt += " It is the current default; afterwards no default will be set." + if not confirm(prompt): + print("No changes made.") + return + + # Remove the config entry first: if that write fails, the cached token is left intact so the + # environment stays usable, rather than half-removed. + remove_environment_from_config(config_file, config_env) + if env.auth_method == "oauth": + TokenCache(_oauth_cache_path(env, config_env)).clear() + + print(f"Removed environment '{config_env}'.") + if leaves_no_default: + print("It was the default environment; set a new default with 'bfabric-cli auth default '.") diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/oauth_login.py b/bfabric_scripts/src/bfabric_scripts/cli/login/oauth_login.py new file mode 100644 index 00000000..d3b83ba1 --- /dev/null +++ b/bfabric_scripts/src/bfabric_scripts/cli/login/oauth_login.py @@ -0,0 +1,119 @@ +"""Interactive OAuth login commands: browser (PKCE) and device-code flows. + +They share param resolution and token persistence (``_resolve_params`` / ``_persist``); only the +token-acquisition step differs. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Annotated + +import cyclopts + +from bfabric._oauth.credential_provider import OAuthCredentialProvider +from bfabric._oauth.device_code import device_code_login +from bfabric._oauth.pkce import pkce_login +from bfabric._oauth.token_cache import compute_token_cache_path +from bfabric.config import DEFAULT_CONFIG_FILE +from bfabric.config.config_writer import write_environment_to_config +from bfabric_scripts.cli.login._common import resolve_config_env, resolve_scope, resolve_set_default +from bfabric_scripts.cli.login._constants import DEFAULT_CLIENT_ID + +_SCOPE_HELP = "Scope preset (read-only|read-write|upload) or a raw scope string (interactive picker if omitted)." +_CONFIG_ENV_HELP = "Environment name (interactive picker: existing or new, if omitted)." +_SET_DEFAULT_HELP = "Set this environment as the default in the config file (prompted if omitted)." + + +def _resolve_params( + config_env: str | None, scope: str | None, set_default: bool | None, config_file: Path +) -> tuple[str, str, bool] | None: + """Resolve env/scope/set-default, or print "Login aborted." and return ``None`` on cancel.""" + config_env = resolve_config_env(config_env, config_file) + scope = resolve_scope(scope) + # Only prompt for set-default once env and scope are settled, so a cancel there doesn't prompt. + set_default = resolve_set_default(set_default, config_env) if config_env and scope else None + if config_env is None or scope is None or set_default is None: + print("Login aborted.", file=sys.stderr) + return None + return config_env, scope, set_default + + +def _persist( + base_url: str, + client_id: str, + scope: str, + token: dict[str, object], + config_env: str, + config_file: Path, + set_default: bool, +) -> None: + """Cache the fresh OAuth *token* and record the environment in the config.""" + # OAuthCredentialProvider is constructed for its side effect: writing the token to the disk cache. + _ = OAuthCredentialProvider( + client_id=client_id, + client_secret="", + token_url=f"{base_url}/rest/oauth/token", + token=token, + grant_type="refresh_token", + scope=scope, + token_cache_path=compute_token_cache_path(base_url, client_id, config_env).expanduser(), + ) + env_data = {"base_url": base_url, "auth_method": "oauth", "client_id": client_id} + write_environment_to_config(config_file, config_env, env_data, set_default=set_default) + print("Authenticated successfully.") + print(f"Config saved to environment '{config_env}' in {config_file}") + + +def cmd_auth_login( + base_url: Annotated[str, cyclopts.Parameter(help="B-Fabric instance URL.")], + *, + client_id: Annotated[str, cyclopts.Parameter(help="OAuth client ID.")] = DEFAULT_CLIENT_ID, + config_env: Annotated[str | None, cyclopts.Parameter(help=_CONFIG_ENV_HELP)] = None, + config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, + scope: Annotated[str | None, cyclopts.Parameter(help=_SCOPE_HELP)] = None, + port: Annotated[int, cyclopts.Parameter(help="Local port for callback (0 = auto).")] = 0, + timeout: Annotated[float, cyclopts.Parameter(help="Seconds to wait for login.")] = 120.0, + set_default: Annotated[bool | None, cyclopts.Parameter(help=_SET_DEFAULT_HELP)] = None, +) -> None: + """Authenticate via browser-based login (OAuth PKCE flow).""" + resolved = _resolve_params(config_env, scope, set_default, config_file) + if resolved is None: + return + config_env, scope, set_default = resolved + + base_url = base_url.rstrip("/") + print("Opening browser for authentication...", file=sys.stderr) + print("Waiting for login to complete...", file=sys.stderr) + try: + token = pkce_login(base_url, client_id=client_id, scope=scope, port=port, timeout=timeout) + except RuntimeError as e: + print(f"Error: {e}", file=sys.stderr) + raise SystemExit(1) from None + _persist(base_url, client_id, scope, token, config_env, config_file, set_default) + + +def cmd_login_device_code( + base_url: Annotated[str, cyclopts.Parameter(help="B-Fabric instance URL.")], + *, + client_id: Annotated[str, cyclopts.Parameter(help="OAuth client ID.")] = DEFAULT_CLIENT_ID, + config_env: Annotated[str | None, cyclopts.Parameter(help=_CONFIG_ENV_HELP)] = None, + config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, + scope: Annotated[str | None, cyclopts.Parameter(help=_SCOPE_HELP)] = None, + timeout: Annotated[float, cyclopts.Parameter(help="Seconds to wait for authorization.")] = 600.0, + set_default: Annotated[bool | None, cyclopts.Parameter(help=_SET_DEFAULT_HELP)] = None, +) -> None: + """Authenticate via device code flow (for headless environments).""" + resolved = _resolve_params(config_env, scope, set_default, config_file) + if resolved is None: + return + config_env, scope, set_default = resolved + + base_url = base_url.rstrip("/") + try: + token = device_code_login(base_url, client_id=client_id, scope=scope, timeout=timeout) + except RuntimeError as e: + print(f"Error: {e}", file=sys.stderr) + raise SystemExit(1) from None + _persist(base_url, client_id, scope, token, config_env, config_file, set_default) diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/pat.py b/bfabric_scripts/src/bfabric_scripts/cli/login/pat.py index 91e2329e..7a74cee0 100644 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/pat.py +++ b/bfabric_scripts/src/bfabric_scripts/cli/login/pat.py @@ -11,26 +11,39 @@ from bfabric.config import DEFAULT_CONFIG_FILE from bfabric.config.config_writer import write_environment_to_config +from bfabric_scripts.cli.login._common import resolve_config_env, resolve_set_default def cmd_login_pat( base_url: Annotated[str, cyclopts.Parameter(help="B-Fabric instance URL.")], *, pat: Annotated[str | None, cyclopts.Parameter(help="Personal Access Token (prompted if omitted).")] = None, - config_env: Annotated[str, cyclopts.Parameter(help="Environment name in the config file.")] = "PRODUCTION", + config_env: Annotated[ + str | None, cyclopts.Parameter(help="Environment name (interactive picker: existing or new, if omitted).") + ] = None, config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, set_default: Annotated[ - bool, cyclopts.Parameter(help="Set this environment as the default in the config file.") - ] = True, + bool | None, + cyclopts.Parameter(help="Set this environment as the default in the config file (prompted if omitted)."), + ] = None, ) -> None: """Authenticate with a Personal Access Token (PAT).""" + config_env = resolve_config_env(config_env, config_file) + if config_env is None: + print("Login aborted.", file=sys.stderr) + return + set_default = resolve_set_default(set_default, config_env) + if set_default is None: + print("Login aborted.", file=sys.stderr) + return + if pat is None: pat = getpass.getpass("Personal Access Token: ") else: print("Warning: passing secrets via CLI flags is insecure (visible in ps, shell history).", file=sys.stderr) - # Store the PAT under ``pat`` (not ``login``/``password``): a PAT is not 32 characters, and an - # unmodified <=1.19.0 client validates every environment eagerly and would reject a short - # password — poisoning the whole shared config file. Under ``pat`` old clients ignore it. + # Store under ``pat``, not ``login``/``password``: a PAT isn't 32 chars, so an old (<=1.19.0) + # client eagerly validating every environment would reject a short password and poison the + # shared config; old clients ignore an unknown ``pat`` key. env_data = { "base_url": base_url.rstrip("/"), "auth_method": "pat", diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/pkce.py b/bfabric_scripts/src/bfabric_scripts/cli/login/pkce.py deleted file mode 100644 index ce33a68c..00000000 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/pkce.py +++ /dev/null @@ -1,67 +0,0 @@ -"""Browser-based login command (OAuth PKCE flow).""" - -from __future__ import annotations - -from pathlib import Path -from typing import Annotated - -import cyclopts - -from bfabric._oauth.credential_provider import OAuthCredentialProvider -from bfabric._oauth.pkce import pkce_login -from bfabric._oauth.token_cache import compute_token_cache_path -from bfabric.config import DEFAULT_CONFIG_FILE -from bfabric.config.config_writer import write_environment_to_config -from bfabric_scripts.cli.login._constants import DEFAULT_CLIENT_ID - - -def cmd_auth_login( - base_url: Annotated[str, cyclopts.Parameter(help="B-Fabric instance URL.")], - *, - scope: Annotated[str, cyclopts.Parameter(help="OAuth scope.")], - client_id: Annotated[str, cyclopts.Parameter(help="OAuth client ID.")] = DEFAULT_CLIENT_ID, - config_env: Annotated[str, cyclopts.Parameter(help="Environment name in the config file.")] = "PRODUCTION", - config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, - port: Annotated[int, cyclopts.Parameter(help="Local port for callback (0 = auto).")] = 0, - timeout: Annotated[float, cyclopts.Parameter(help="Seconds to wait for login.")] = 120.0, - set_default: Annotated[ - bool, cyclopts.Parameter(help="Set this environment as the default in the config file.") - ] = True, -) -> None: - """Authenticate via browser-based login (OAuth PKCE flow).""" - import sys - - base_url = base_url.rstrip("/") - print("Opening browser for authentication...", file=sys.stderr) - print("Waiting for login to complete...", file=sys.stderr) - try: - token = pkce_login( - base_url, - client_id=client_id, - scope=scope, - port=port, - timeout=timeout, - ) - except RuntimeError as e: - print(f"Error: {e}", file=sys.stderr) - raise SystemExit(1) from None - cache_path = compute_token_cache_path(base_url, client_id, config_env).expanduser() - token_url = f"{base_url}/rest/oauth/token" - # Constructed for its side effect: persists the fresh token to the disk cache. - _ = OAuthCredentialProvider( - client_id=client_id, - client_secret="", - token_url=token_url, - token=token, - grant_type="refresh_token", - scope=scope, - token_cache_path=cache_path, - ) - env_data = { - "base_url": base_url, - "auth_method": "oauth", - "client_id": client_id, - } - write_environment_to_config(config_file, config_env, env_data, set_default=set_default) - print(f"Authenticated successfully.") - print(f"Config saved to environment '{config_env}' in {config_file}") diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/register.py b/bfabric_scripts/src/bfabric_scripts/cli/login/register.py index 95e1b8cf..d053066e 100644 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/register.py +++ b/bfabric_scripts/src/bfabric_scripts/cli/login/register.py @@ -16,9 +16,8 @@ def _resolve_token_from_config(config_env: str, config_file: Path) -> tuple[str, str]: - """Load a cached OAuth access token and base_url for the given environment. + """Load the cached OAuth access token and base_url for *config_env* as ``(token, base_url)``. - Returns ``(access_token, base_url)``. Raises :class:`SystemExit` on failure. """ import yaml diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/register_webapp.py b/bfabric_scripts/src/bfabric_scripts/cli/login/register_webapp.py index 0d623764..6935f559 100644 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/register_webapp.py +++ b/bfabric_scripts/src/bfabric_scripts/cli/login/register_webapp.py @@ -29,10 +29,9 @@ def cmd_login_register_webapp( technology_id: Annotated[int | None, cyclopts.Parameter(help="Technology ID for the application.")] = None, description: Annotated[str | None, cyclopts.Parameter(help="Application description.")] = None, ) -> None: - """Register a new OAuth webapp: create OAuth client and B-Fabric application. + """Register a new OAuth webapp: create the OAuth client and the B-Fabric application. - Uses the current config environment's credentials for both the OAuth - registration endpoint (Bearer token) and the SOAP application save. + Uses the config environment's credentials for both the registration endpoint and the SOAP save. """ from bfabric import Bfabric from bfabric._oauth.registration import register_webapp diff --git a/bfabric_scripts/src/bfabric_scripts/cli/login/status.py b/bfabric_scripts/src/bfabric_scripts/cli/login/status.py deleted file mode 100644 index 955b5cf5..00000000 --- a/bfabric_scripts/src/bfabric_scripts/cli/login/status.py +++ /dev/null @@ -1,60 +0,0 @@ -"""Login status command.""" - -from __future__ import annotations - -import os -from pathlib import Path -from typing import Annotated - -import cyclopts -import yaml - -from bfabric.config import DEFAULT_CONFIG_FILE -from bfabric.config.config_file import ConfigFile -from bfabric._oauth.token_cache import TokenCache, compute_token_cache_path -from bfabric_scripts.cli.login._constants import DEFAULT_CLIENT_ID - - -def cmd_login_status( - *, - config_file: Annotated[Path, cyclopts.Parameter(help="Path to the config file.")] = DEFAULT_CONFIG_FILE, - config_env: Annotated[str | None, cyclopts.Parameter(help="Environment name (default: auto-detect).")] = None, -) -> None: - """Show current authentication status.""" - config_path = Path(config_file).expanduser() - if not config_path.is_file(): - print(f"Config file not found: {config_path}") - return - - config_file_obj = ConfigFile.model_validate(yaml.safe_load(config_path.read_text())) - resolved_env = config_env or os.environ.get("BFABRICPY_CONFIG_ENV") or config_file_obj.general.default_config - if resolved_env is None: - print("No environment specified and no default configured.") - return - - if resolved_env not in config_file_obj.environments: - print(f"Environment '{resolved_env}' not found in config.") - return - - env = config_file_obj.environments[resolved_env] - print(f"Environment: {resolved_env}") - print(f"Base URL: {env.config.base_url}") - - if env.auth_method == "oauth": - client_id = env.client_id or DEFAULT_CLIENT_ID - print(f"Auth method: oauth") - print(f"Client ID: {client_id}") - cache_path = compute_token_cache_path(env.config.base_url.rstrip("/"), client_id, resolved_env).expanduser() - cached = TokenCache(cache_path).load() - if cached: - print(f"Token cache: {cache_path} (present)") - else: - print(f"Token cache: {cache_path} (missing)") - elif env.auth_method == "pat": - print("Auth method: pat") - print("Token: stored in config file") - elif env.auth is not None: - print(f"Auth method: password") - print(f"Login: {env.auth.login}") - else: - print("Auth method: none") diff --git a/tests/bfabric/config/test_config_writer.py b/tests/bfabric/config/test_config_writer.py index 1139017f..62ce1736 100644 --- a/tests/bfabric/config/test_config_writer.py +++ b/tests/bfabric/config/test_config_writer.py @@ -8,7 +8,11 @@ from bfabric.config.bfabric_auth import OAUTH_LOGIN from bfabric.config.config_file import ConfigFile -from bfabric.config.config_writer import set_default_config, write_environment_to_config +from bfabric.config.config_writer import ( + remove_environment_from_config, + set_default_config, + write_environment_to_config, +) class TestWriteEnvironmentToConfig: @@ -193,3 +197,61 @@ def test_tightens_permissions(self, tmp_path): set_default_config(config_path, "TEST") mode = stat.S_IMODE(os.stat(config_path).st_mode) assert mode == 0o600 + + +class TestRemoveEnvironmentFromConfig: + @staticmethod + def _write_two_env_config(config_path, default="PROD"): + config_path.write_text( + yaml.dump( + { + "GENERAL": {"default_config": default}, + "PROD": {"base_url": "https://prod.example.com"}, + "TEST": {"base_url": "https://test.example.com"}, + } + ) + ) + + def test_removes_env_and_preserves_others(self, tmp_path): + config_path = tmp_path / "config.yml" + self._write_two_env_config(config_path, default="PROD") + remove_environment_from_config(config_path, "TEST") + data = yaml.safe_load(config_path.read_text()) + assert "TEST" not in data + assert data["PROD"]["base_url"] == "https://prod.example.com" + # Removing a non-default env leaves the default untouched. + assert data["GENERAL"]["default_config"] == "PROD" + + def test_removing_default_env_clears_default(self, tmp_path): + # A dangling default_config would make ConfigFile refuse to load the file, so removing the + # environment that is the default must also clear the default. + config_path = tmp_path / "config.yml" + self._write_two_env_config(config_path, default="PROD") + remove_environment_from_config(config_path, "PROD") + data = yaml.safe_load(config_path.read_text()) + assert "PROD" not in data + assert data.get("GENERAL", {}).get("default_config") is None + # The result must still parse back through the reader. + config_file = ConfigFile.model_validate(data) + assert set(config_file.environments) == {"TEST"} + + def test_raises_on_unknown_env_and_leaves_file_unchanged(self, tmp_path): + config_path = tmp_path / "config.yml" + self._write_two_env_config(config_path) + before = config_path.read_text() + with pytest.raises(ValueError): + remove_environment_from_config(config_path, "NOPE") + assert config_path.read_text() == before + + def test_raises_on_missing_file(self, tmp_path): + config_path = tmp_path / "nonexistent.yml" + with pytest.raises(FileNotFoundError): + remove_environment_from_config(config_path, "PROD") + + def test_tightens_permissions(self, tmp_path): + config_path = tmp_path / "config.yml" + self._write_two_env_config(config_path) + config_path.chmod(0o644) + remove_environment_from_config(config_path, "TEST") + mode = stat.S_IMODE(os.stat(config_path).st_mode) + assert mode == 0o600 diff --git a/tests/bfabric_scripts/cli/login/conftest.py b/tests/bfabric_scripts/cli/login/conftest.py new file mode 100644 index 00000000..dc77916c --- /dev/null +++ b/tests/bfabric_scripts/cli/login/conftest.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import time + +import pytest + + +@pytest.fixture(autouse=True) +def _clear_config_env(monkeypatch): + """Drop the global ``__MOCK`` env so commands resolve the environment from the temp config.""" + monkeypatch.delenv("BFABRICPY_CONFIG_ENV", raising=False) + + +@pytest.fixture +def oauth_token(): + """A representative token-endpoint response, valid for another hour.""" + return { + "access_token": "jwt123", + "refresh_token": "rt456", + "token_type": "Bearer", + "expires_at": time.time() + 3600, + } + + +@pytest.fixture +def oauth_session(mocker): + """Patch ``OAuth2Session`` so the credential provider caches without a real network call.""" + session = mocker.MagicMock() + session.token = None + session.metadata = {"token_endpoint": "https://example.com/bfabric/rest/oauth/token"} + mocker.patch("bfabric._oauth.credential_provider.OAuth2Session", return_value=session) + return session diff --git a/tests/bfabric_scripts/cli/login/test_cmd_auth_default.py b/tests/bfabric_scripts/cli/login/test_cmd_auth_default.py index 6005f7e7..ed54aa43 100644 --- a/tests/bfabric_scripts/cli/login/test_cmd_auth_default.py +++ b/tests/bfabric_scripts/cli/login/test_cmd_auth_default.py @@ -1,14 +1,8 @@ from __future__ import annotations -import pytest import yaml -from bfabric_scripts.cli.login.default_config import cmd_auth_default_set, cmd_auth_default_show - - -@pytest.fixture(autouse=True) -def _clear_config_env(monkeypatch): - monkeypatch.delenv("BFABRICPY_CONFIG_ENV", raising=False) +from bfabric_scripts.cli.login.manage import cmd_auth_default def _write_config(config_file, default="PROD"): @@ -17,63 +11,74 @@ def _write_config(config_file, default="PROD"): yaml.dump( { "GENERAL": general, - "PROD": {"base_url": "https://prod.example.com"}, - "TEST": {"base_url": "https://test.example.com"}, + "PROD": {"base_url": "https://prod.example.com", "auth_method": "oauth"}, + "TEST": {"base_url": "https://test.example.com", "auth_method": "pat", "pat": "tok123"}, } ) ) -class TestCmdAuthDefaultShow: - def test_lists_environments_and_marks_default(self, tmp_path, capsys): +class TestSetNonInteractive: + def test_positional_value_sets_default(self, tmp_path): config_file = tmp_path / "config.yml" _write_config(config_file, default="PROD") - cmd_auth_default_show(config_file=config_file) + cmd_auth_default("TEST", config_file=config_file) + data = yaml.safe_load(config_file.read_text()) + assert data["GENERAL"]["default_config"] == "TEST" + + def test_unknown_env_errors_and_leaves_file_unchanged(self, tmp_path, capsys): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + before = config_file.read_text() + cmd_auth_default("NOPE", config_file=config_file) output = capsys.readouterr().out - assert "PROD" in output - assert "TEST" in output - assert "(default)" in output - # The default env is marked prominently with an arrow. - assert "→ PROD" in output + assert "NOPE" in output + assert config_file.read_text() == before def test_missing_config_file(self, tmp_path, capsys): config_file = tmp_path / "nonexistent.yml" - cmd_auth_default_show(config_file=config_file) + cmd_auth_default("PROD", config_file=config_file) assert "not found" in capsys.readouterr().out -class TestCmdAuthDefaultSet: - def test_sets_default_non_interactively(self, tmp_path): +class TestInteractivePicker: + def test_selects_default_via_picker(self, tmp_path, mocker): config_file = tmp_path / "config.yml" _write_config(config_file, default="PROD") - cmd_auth_default_set("TEST", config_file=config_file) + # The picker runs only when no value is passed; it returns the chosen environment. + mocker.patch("bfabric_scripts.cli.login.manage.is_interactive", return_value=True) + picker = mocker.patch("bfabric_scripts.cli.login.manage.select_choice", return_value="TEST") + cmd_auth_default(config_file=config_file) data = yaml.safe_load(config_file.read_text()) assert data["GENERAL"]["default_config"] == "TEST" + # The current default is pre-selected in the picker. + assert picker.call_args.kwargs["default"] == "PROD" - def test_prompts_when_env_omitted(self, tmp_path, mocker, capsys): + def test_cancelled_picker_leaves_file_unchanged(self, tmp_path, mocker, capsys): config_file = tmp_path / "config.yml" _write_config(config_file, default="PROD") - # The picker shows a numbered menu and asks for a number; env names are sorted - # (PROD=1, TEST=2), so choosing "2" selects TEST. - mocker.patch("bfabric_scripts.cli.login.default_config.Prompt.ask", return_value="2") - cmd_auth_default_set(config_file=config_file) - data = yaml.safe_load(config_file.read_text()) - assert data["GENERAL"]["default_config"] == "TEST" - # The menu is printed with each environment on its own numbered line. - menu = capsys.readouterr().out - assert "1. PROD" in menu - assert "2. TEST" in menu + before = config_file.read_text() + mocker.patch("bfabric_scripts.cli.login.manage.select_choice", return_value=None) + mocker.patch("bfabric_scripts.cli.login.manage.is_interactive", return_value=True) + cmd_auth_default(config_file=config_file) + assert "No changes made" in capsys.readouterr().out + assert config_file.read_text() == before - def test_unknown_env_errors_and_leaves_file_unchanged(self, tmp_path, capsys): + +class TestNonInteractiveNoValue: + def test_reports_no_value_and_leaves_file_unchanged(self, tmp_path, mocker, capsys): config_file = tmp_path / "config.yml" _write_config(config_file, default="PROD") before = config_file.read_text() - cmd_auth_default_set("NOPE", config_file=config_file) + mocker.patch("bfabric_scripts.cli.login.manage.is_interactive", return_value=False) + cmd_auth_default(config_file=config_file) output = capsys.readouterr().out - assert "NOPE" in output + # Without a TTY to prompt on, it can't pick for the user: report and change nothing. + assert "No environment specified" in output assert config_file.read_text() == before - def test_missing_config_file(self, tmp_path, capsys): - config_file = tmp_path / "nonexistent.yml" - cmd_auth_default_set("PROD", config_file=config_file) - assert "not found" in capsys.readouterr().out + def test_empty_config_reports_no_environments(self, tmp_path, capsys): + config_file = tmp_path / "config.yml" + config_file.write_text(yaml.dump({"GENERAL": {}})) + cmd_auth_default(config_file=config_file) + assert "No environments configured" in capsys.readouterr().out diff --git a/tests/bfabric_scripts/cli/login/test_cmd_auth_list.py b/tests/bfabric_scripts/cli/login/test_cmd_auth_list.py new file mode 100644 index 00000000..ef4c340c --- /dev/null +++ b/tests/bfabric_scripts/cli/login/test_cmd_auth_list.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +import yaml + +from bfabric_scripts.cli.login.manage import cmd_auth_list + + +class TestCmdAuthList: + def test_lists_environments_and_marks_default(self, tmp_path, capsys): + config_file = tmp_path / "config.yml" + config_file.write_text( + yaml.dump( + { + "GENERAL": {"default_config": "TEST"}, + "PROD": {"base_url": "https://prod.example.com", "auth_method": "oauth"}, + "TEST": {"base_url": "https://test.example.com", "auth_method": "pat", "pat": "tok"}, + } + ) + ) + cmd_auth_list(config_file=config_file) + output = capsys.readouterr().out + assert "PROD" in output + assert "TEST" in output + # The default is flagged and each row carries the host + auth method. + assert "(default)" in output + assert "prod.example.com" in output + assert "oauth" in output + assert "pat" in output + + def test_missing_config_file(self, tmp_path, capsys): + config_file = tmp_path / "nonexistent.yml" + cmd_auth_list(config_file=config_file) + output = capsys.readouterr().out + assert "not found" in output + + def test_no_environments(self, tmp_path, capsys): + config_file = tmp_path / "config.yml" + config_file.write_text(yaml.dump({"GENERAL": {}})) + cmd_auth_list(config_file=config_file) + output = capsys.readouterr().out + assert "No environments configured" in output diff --git a/tests/bfabric_scripts/cli/login/test_cmd_auth_login.py b/tests/bfabric_scripts/cli/login/test_cmd_auth_login.py index 9a8c708b..505a4cd1 100644 --- a/tests/bfabric_scripts/cli/login/test_cmd_auth_login.py +++ b/tests/bfabric_scripts/cli/login/test_cmd_auth_login.py @@ -1,28 +1,15 @@ from __future__ import annotations -import time - import yaml -from bfabric_scripts.cli.login.pkce import cmd_auth_login +from bfabric_scripts.cli.login._constants import SCOPE_PRESETS_BY_NAME +from bfabric_scripts.cli.login.oauth_login import cmd_auth_login class TestCmdAuthLogin: - def test_writes_config_and_caches_token(self, tmp_path, mocker): + def test_writes_config_and_caches_token(self, tmp_path, mocker, oauth_token, oauth_session): config_file = tmp_path / "config.yml" - token = { - "access_token": "jwt123", - "refresh_token": "rt456", - "token_type": "Bearer", - "expires_at": time.time() + 3600, - } - - mock_session = mocker.MagicMock() - mock_session.token = None - mock_session.metadata = {"token_endpoint": "https://example.com/bfabric/rest/oauth/token"} - - mock_pkce = mocker.patch("bfabric_scripts.cli.login.pkce.pkce_login", return_value=token) - mocker.patch("bfabric._oauth.credential_provider.OAuth2Session", return_value=mock_session) + mock_pkce = mocker.patch("bfabric_scripts.cli.login.oauth_login.pkce_login", return_value=oauth_token) cmd_auth_login( base_url="https://example.com/bfabric", scope="api:read", @@ -37,20 +24,9 @@ def test_writes_config_and_caches_token(self, tmp_path, mocker): assert data["PROD"]["client_id"] == "test-client" assert data["PROD"]["base_url"] == "https://example.com/bfabric" - def test_set_default_false_does_not_set_default(self, tmp_path, mocker): + def test_set_default_false_does_not_set_default(self, tmp_path, mocker, oauth_token, oauth_session): config_file = tmp_path / "config.yml" - token = { - "access_token": "jwt123", - "refresh_token": "rt456", - "token_type": "Bearer", - "expires_at": time.time() + 3600, - } - mock_session = mocker.MagicMock() - mock_session.token = None - mock_session.metadata = {"token_endpoint": "https://example.com/bfabric/rest/oauth/token"} - - mocker.patch("bfabric_scripts.cli.login.pkce.pkce_login", return_value=token) - mocker.patch("bfabric._oauth.credential_provider.OAuth2Session", return_value=mock_session) + mocker.patch("bfabric_scripts.cli.login.oauth_login.pkce_login", return_value=oauth_token) cmd_auth_login( base_url="https://example.com/bfabric", scope="api:read", @@ -63,3 +39,71 @@ def test_set_default_false_does_not_set_default(self, tmp_path, mocker): data = yaml.safe_load(config_file.read_text()) assert "default_config" not in data["GENERAL"] assert data["PROD"]["auth_method"] == "oauth" + + def test_prompts_for_default_when_omitted(self, tmp_path, mocker, oauth_token, oauth_session): + config_file = tmp_path / "config.yml" + mocker.patch("bfabric_scripts.cli.login.oauth_login.pkce_login", return_value=oauth_token) + # No --set-default given: in a terminal the user is asked; here they decline. + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + confirm = mocker.patch("bfabric_scripts.cli.login._common.confirm", return_value=False) + cmd_auth_login( + base_url="https://example.com/bfabric", + client_id="test-client", + config_env="PROD", + config_file=config_file, + scope="read-write", + ) + confirm.assert_called_once() + data = yaml.safe_load(config_file.read_text()) + # Declining the prompt means the environment is not made the default. + assert "default_config" not in data["GENERAL"] + assert data["PROD"]["auth_method"] == "oauth" + + def test_cancel_at_set_default_aborts(self, tmp_path, mocker, capsys): + config_file = tmp_path / "config.yml" + mock_pkce = mocker.patch("bfabric_scripts.cli.login.oauth_login.pkce_login") + # No --set-default given: the user reaches the confirm prompt and cancels it (Ctrl-C -> None). + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login._common.confirm", return_value=None) + cmd_auth_login( + base_url="https://example.com/bfabric", + client_id="test-client", + config_env="PROD", + config_file=config_file, + scope="read-write", + ) + # Cancelling aborts the whole login: no browser flow, no config written. + mock_pkce.assert_not_called() + assert not config_file.exists() + assert "Login aborted." in capsys.readouterr().err + + def test_scope_preset_is_expanded(self, tmp_path, mocker, oauth_token, oauth_session): + config_file = tmp_path / "config.yml" + mock_pkce = mocker.patch("bfabric_scripts.cli.login.oauth_login.pkce_login", return_value=oauth_token) + cmd_auth_login( + base_url="https://example.com/bfabric", + client_id="test-client", + config_env="PROD", + config_file=config_file, + scope="upload", + ) + # The preset name expands to the real scope string requested from the OAuth flow. + assert mock_pkce.call_args.kwargs["scope"] == SCOPE_PRESETS_BY_NAME["upload"].scope + + def test_config_env_omitted_falls_back_to_current_default(self, tmp_path, mocker, oauth_token, oauth_session): + config_file = tmp_path / "config.yml" + config_file.write_text( + yaml.dump( + { + "GENERAL": {"default_config": "EXISTING"}, + "EXISTING": {"base_url": "https://old.example.com", "auth_method": "oauth"}, + } + ) + ) + mocker.patch("bfabric_scripts.cli.login.oauth_login.pkce_login", return_value=oauth_token) + # No TTY (pytest) and no --config-env => reuse the current default env, not PRODUCTION. + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=False) + cmd_auth_login(base_url="https://example.com/bfabric", client_id="c", config_file=config_file, scope="api:read") + data = yaml.safe_load(config_file.read_text()) + assert data["EXISTING"]["base_url"] == "https://example.com/bfabric" + assert "PRODUCTION" not in data diff --git a/tests/bfabric_scripts/cli/login/test_cmd_login_device_code.py b/tests/bfabric_scripts/cli/login/test_cmd_login_device_code.py index 62520d0d..54866d19 100644 --- a/tests/bfabric_scripts/cli/login/test_cmd_login_device_code.py +++ b/tests/bfabric_scripts/cli/login/test_cmd_login_device_code.py @@ -1,28 +1,15 @@ from __future__ import annotations -import time - import yaml -from bfabric_scripts.cli.login.device_code import cmd_login_device_code +from bfabric_scripts.cli.login._constants import SCOPE_PRESETS_BY_NAME +from bfabric_scripts.cli.login.oauth_login import cmd_login_device_code class TestCmdLoginDeviceCode: - def test_writes_config_and_caches_token(self, tmp_path, mocker): + def test_writes_config_and_caches_token(self, tmp_path, mocker, oauth_token, oauth_session): config_file = tmp_path / "config.yml" - token = { - "access_token": "jwt123", - "refresh_token": "rt456", - "token_type": "Bearer", - "expires_at": time.time() + 3600, - } - - mock_session = mocker.MagicMock() - mock_session.token = None - mock_session.metadata = {"token_endpoint": "https://example.com/bfabric/rest/oauth/token"} - - mock_dc = mocker.patch("bfabric_scripts.cli.login.device_code.device_code_login", return_value=token) - mocker.patch("bfabric._oauth.credential_provider.OAuth2Session", return_value=mock_session) + mock_dc = mocker.patch("bfabric_scripts.cli.login.oauth_login.device_code_login", return_value=oauth_token) cmd_login_device_code( base_url="https://example.com/bfabric", scope="api:read", @@ -37,20 +24,9 @@ def test_writes_config_and_caches_token(self, tmp_path, mocker): assert data["PROD"]["client_id"] == "test-client" assert data["PROD"]["base_url"] == "https://example.com/bfabric" - def test_set_default_false_does_not_set_default(self, tmp_path, mocker): + def test_set_default_false_does_not_set_default(self, tmp_path, mocker, oauth_token, oauth_session): config_file = tmp_path / "config.yml" - token = { - "access_token": "jwt123", - "refresh_token": "rt456", - "token_type": "Bearer", - "expires_at": time.time() + 3600, - } - mock_session = mocker.MagicMock() - mock_session.token = None - mock_session.metadata = {"token_endpoint": "https://example.com/bfabric/rest/oauth/token"} - - mocker.patch("bfabric_scripts.cli.login.device_code.device_code_login", return_value=token) - mocker.patch("bfabric._oauth.credential_provider.OAuth2Session", return_value=mock_session) + mocker.patch("bfabric_scripts.cli.login.oauth_login.device_code_login", return_value=oauth_token) cmd_login_device_code( base_url="https://example.com/bfabric", scope="api:read", @@ -63,3 +39,33 @@ def test_set_default_false_does_not_set_default(self, tmp_path, mocker): data = yaml.safe_load(config_file.read_text()) assert "default_config" not in data["GENERAL"] assert data["PROD"]["auth_method"] == "oauth" + + def test_cancel_at_set_default_aborts(self, tmp_path, mocker, capsys): + config_file = tmp_path / "config.yml" + mock_dc = mocker.patch("bfabric_scripts.cli.login.oauth_login.device_code_login") + # No --set-default given: the user reaches the confirm prompt and cancels it (Ctrl-C -> None). + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login._common.confirm", return_value=None) + cmd_login_device_code( + base_url="https://example.com/bfabric", + client_id="test-client", + config_env="PROD", + config_file=config_file, + scope="read-write", + ) + # Cancelling aborts the whole login: no device-code flow, no config written. + mock_dc.assert_not_called() + assert not config_file.exists() + assert "Login aborted." in capsys.readouterr().err + + def test_scope_preset_is_expanded(self, tmp_path, mocker, oauth_token, oauth_session): + config_file = tmp_path / "config.yml" + mock_dc = mocker.patch("bfabric_scripts.cli.login.oauth_login.device_code_login", return_value=oauth_token) + cmd_login_device_code( + base_url="https://example.com/bfabric", + client_id="test-client", + config_env="PROD", + config_file=config_file, + scope="upload", + ) + assert mock_dc.call_args.kwargs["scope"] == SCOPE_PRESETS_BY_NAME["upload"].scope diff --git a/tests/bfabric_scripts/cli/login/test_cmd_login_logout.py b/tests/bfabric_scripts/cli/login/test_cmd_login_logout.py index a35a5742..a7655b09 100644 --- a/tests/bfabric_scripts/cli/login/test_cmd_login_logout.py +++ b/tests/bfabric_scripts/cli/login/test_cmd_login_logout.py @@ -2,54 +2,140 @@ import json -import pytest import yaml -from bfabric._oauth.token_cache import compute_token_cache_path -from bfabric_scripts.cli.login.logout import cmd_login_logout +from bfabric_scripts.cli.login.manage import cmd_login_logout -@pytest.fixture(autouse=True) -def _clear_config_env(monkeypatch): - monkeypatch.delenv("BFABRICPY_CONFIG_ENV", raising=False) +def _write_config(config_file, *, default="PROD", extra_prod=None): + prod = {"base_url": "https://example.com/bfabric", "auth_method": "oauth", "client_id": "my-app"} + prod.update(extra_prod or {}) + config_file.write_text( + yaml.dump( + { + "GENERAL": {"default_config": default}, + "PROD": prod, + "TEST": {"base_url": "https://test.example.com/bfabric", "login": "user", "password": "x" * 32}, + } + ) + ) + + +def _environments(config_file): + data = yaml.safe_load(config_file.read_text()) + return {k for k in data if k != "GENERAL"} class TestCmdLoginLogout: - def test_clears_oauth_token_cache(self, tmp_path, capsys): - config_file = tmp_path / "config.yml" - base_url = "https://example.com/bfabric" - client_id = "my-app" - config_file.write_text( - yaml.dump({ - "GENERAL": {"default_config": "PROD"}, - "PROD": { - "base_url": base_url, - "auth_method": "oauth", - "client_id": client_id, - }, - }) - ) - cache_path = compute_token_cache_path(base_url, client_id, "PROD").expanduser() - cache_path.parent.mkdir(parents=True, exist_ok=True) + def test_removes_oauth_env_and_clears_cache(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + cache_path = tmp_path / "tok.json" cache_path.write_text(json.dumps({"access_token": "tok"})) + mocker.patch("bfabric_scripts.cli.login.manage.compute_token_cache_path", return_value=cache_path) + + cmd_login_logout("PROD", config_file=config_file, no_confirm=True) - cmd_login_logout(config_file=config_file) output = capsys.readouterr().out - assert "cleared" in output + assert "Removed" in output assert not cache_path.exists() + # The environment is gone and, since it was the default, the default is cleared. + assert _environments(config_file) == {"TEST"} + data = yaml.safe_load(config_file.read_text()) + assert data.get("GENERAL", {}).get("default_config") is None + + def test_removing_default_env_points_to_auth_default(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + mocker.patch("bfabric_scripts.cli.login.manage.compute_token_cache_path", return_value=tmp_path / "tok.json") + + cmd_login_logout("PROD", config_file=config_file, no_confirm=True) + + output = capsys.readouterr().out + # Removing the default while TEST remains must tell the user no default is set and how to fix it. + assert "auth default" in output + data = yaml.safe_load(config_file.read_text()) + assert data.get("GENERAL", {}).get("default_config") is None + + def test_removes_non_oauth_env_without_touching_cache(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + clear = mocker.patch("bfabric_scripts.cli.login.manage.TokenCache") + + cmd_login_logout("TEST", config_file=config_file, no_confirm=True) + + assert _environments(config_file) == {"PROD"} + # A password environment has no OAuth token cache to clear. + clear.assert_not_called() + + def test_confirmation_declined_keeps_env(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + mocker.patch("bfabric_scripts.cli.login.manage.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login.manage.confirm", return_value=False) + + cmd_login_logout("PROD", config_file=config_file) + + output = capsys.readouterr().out + assert "No changes made" in output + assert _environments(config_file) == {"PROD", "TEST"} + + def test_interactive_picker_selects_env(self, tmp_path, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + mocker.patch("bfabric_scripts.cli.login.manage.is_interactive", return_value=True) + pick = mocker.patch("bfabric_scripts.cli.login.manage.select_choice", return_value="TEST") + mocker.patch("bfabric_scripts.cli.login.manage.confirm", return_value=True) - def test_non_oauth_env(self, tmp_path, capsys): - config_file = tmp_path / "config.yml" - config_file.write_text( - yaml.dump({ - "GENERAL": {"default_config": "PROD"}, - "PROD": { - "base_url": "https://example.com/bfabric", - "login": "user", - "password": "x" * 32, - }, - }) - ) cmd_login_logout(config_file=config_file) + + pick.assert_called_once() + # The picker preselects the current default environment. + assert pick.call_args.kwargs["default"] == "PROD" + assert _environments(config_file) == {"PROD"} + + def test_interactive_picker_cancel_keeps_all(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + mocker.patch("bfabric_scripts.cli.login.manage.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login.manage.select_choice", return_value=None) + + cmd_login_logout(config_file=config_file) + + assert _environments(config_file) == {"PROD", "TEST"} + + def test_non_interactive_without_env_aborts(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + mocker.patch("bfabric_scripts.cli.login.manage.is_interactive", return_value=False) + + cmd_login_logout(config_file=config_file) + + err = capsys.readouterr().err + assert "--config-env" in err + assert _environments(config_file) == {"PROD", "TEST"} + + def test_non_interactive_without_no_confirm_refuses(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + mocker.patch("bfabric_scripts.cli.login.manage.is_interactive", return_value=False) + + cmd_login_logout("PROD", config_file=config_file) + + err = capsys.readouterr().err + assert "confirm" in err.lower() + assert _environments(config_file) == {"PROD", "TEST"} + + def test_unknown_env(self, tmp_path, capsys): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + cmd_login_logout("NOPE", config_file=config_file, no_confirm=True) + output = capsys.readouterr().out + assert "not found" in output + assert _environments(config_file) == {"PROD", "TEST"} + + def test_missing_config_file(self, tmp_path, capsys): + config_file = tmp_path / "nonexistent.yml" + cmd_login_logout("PROD", config_file=config_file, no_confirm=True) output = capsys.readouterr().out - assert "does not use OAuth" in output + assert "not found" in output diff --git a/tests/bfabric_scripts/cli/login/test_cmd_login_pat.py b/tests/bfabric_scripts/cli/login/test_cmd_login_pat.py index 8f138114..fb7d9d70 100644 --- a/tests/bfabric_scripts/cli/login/test_cmd_login_pat.py +++ b/tests/bfabric_scripts/cli/login/test_cmd_login_pat.py @@ -50,6 +50,23 @@ def test_prompts_when_pat_omitted(self, tmp_path, mocker): data = yaml.safe_load(config_file.read_text()) assert data["PROD"]["pat"] == "prompted-token" + def test_cancel_at_set_default_aborts(self, tmp_path, mocker, capsys): + config_file = tmp_path / "config.yml" + getpass = mocker.patch("bfabric_scripts.cli.login.pat.getpass.getpass") + # No --set-default given: the user reaches the confirm prompt and cancels it (Ctrl-C -> None). + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login._common.confirm", return_value=None) + cmd_login_pat( + base_url="https://example.com/bfabric", + pat="tok", + config_env="PROD", + config_file=config_file, + ) + # Cancelling aborts before any secret prompt and writes no config. + getpass.assert_not_called() + assert not config_file.exists() + assert "Login aborted." in capsys.readouterr().err + def test_set_default_false_does_not_set_default(self, tmp_path): config_file = tmp_path / "config.yml" cmd_login_pat( diff --git a/tests/bfabric_scripts/cli/login/test_cmd_login_status.py b/tests/bfabric_scripts/cli/login/test_cmd_login_status.py index 721412c1..3a179d26 100644 --- a/tests/bfabric_scripts/cli/login/test_cmd_login_status.py +++ b/tests/bfabric_scripts/cli/login/test_cmd_login_status.py @@ -1,14 +1,12 @@ from __future__ import annotations -import pytest -import yaml - -from bfabric_scripts.cli.login.status import cmd_login_status +import json +import time +import yaml -@pytest.fixture(autouse=True) -def _clear_config_env(monkeypatch): - monkeypatch.delenv("BFABRICPY_CONFIG_ENV", raising=False) +from bfabric_scripts.cli.login._constants import SCOPE_PRESETS_BY_NAME +from bfabric_scripts.cli.login.manage import cmd_login_status, describe_scope, describe_token_cache class TestCmdLoginStatus: @@ -70,8 +68,95 @@ def test_shows_pat_auth(self, tmp_path, capsys): # The secret itself must never be printed. assert "short-pat-token" not in output + def test_oauth_reports_missing_token_and_unrecorded_scope(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + config_file.write_text( + yaml.dump( + { + "GENERAL": {"default_config": "PROD"}, + "PROD": { + "base_url": "https://example.com/bfabric", + "auth_method": "oauth", + "client_id": "my-app", + }, + } + ) + ) + # No cache file on disk -> missing token, no scope to report. + mocker.patch("bfabric_scripts.cli.login.manage.compute_token_cache_path", return_value=tmp_path / "absent.json") + cmd_login_status(config_file=config_file) + output = capsys.readouterr().out + assert "missing" in output + assert "(not recorded)" in output + + def test_oauth_shows_matched_scope_and_expiry_when_cached(self, tmp_path, capsys, mocker): + config_file = tmp_path / "config.yml" + config_file.write_text( + yaml.dump( + { + "GENERAL": {"default_config": "PROD"}, + "PROD": { + "base_url": "https://example.com/bfabric", + "auth_method": "oauth", + "client_id": "my-app", + }, + } + ) + ) + cache_path = tmp_path / "tok.json" + cache_path.write_text( + json.dumps( + { + "access_token": "x", + "scope": SCOPE_PRESETS_BY_NAME["upload"].scope, + "expires_at": time.time() + 9000, + } + ) + ) + mocker.patch("bfabric_scripts.cli.login.manage.compute_token_cache_path", return_value=cache_path) + cmd_login_status(config_file=config_file) + output = capsys.readouterr().out + # The granted scope matches the upload preset and the token is still valid. + assert "upload" in output + assert "expires in" in output + def test_missing_config_file(self, tmp_path, capsys): config_file = tmp_path / "nonexistent.yml" cmd_login_status(config_file=config_file) output = capsys.readouterr().out assert "not found" in output + + +class TestDescribeScope: + def test_matched_preset_is_annotated(self): + # A granted scope equal to a preset (order-insensitive) is annotated with the preset name. + scope = "tus api:write" # reordered to prove match is order-insensitive + described = describe_scope(scope) + assert "upload" in described + assert "tus" in described + + def test_unmatched_scope_shown_raw(self): + assert describe_scope("api:read custom:thing") == "api:read custom:thing" + + def test_absent_scope_is_not_recorded(self): + assert describe_scope(None) == "(not recorded)" + assert describe_scope("") == "(not recorded)" + # A non-string (unexpected cache shape) must not blow up. + assert describe_scope(123) == "(not recorded)" + + +class TestDescribeTokenCache: + def test_missing_cache(self): + assert describe_token_cache(None, now=1000.0) == "missing" + + def test_present_without_expiry(self): + assert describe_token_cache({"access_token": "x"}, now=1000.0) == "present" + + def test_expired(self): + assert "expired" in describe_token_cache({"access_token": "x", "expires_at": 900}, now=1000.0) + + def test_valid_reports_remaining(self): + # 2h30m from now. + described = describe_token_cache({"access_token": "x", "expires_at": 1000 + 9000}, now=1000.0) + assert "present" in described + assert "2h" in described diff --git a/tests/bfabric_scripts/cli/login/test_login_common.py b/tests/bfabric_scripts/cli/login/test_login_common.py new file mode 100644 index 00000000..8ba87823 --- /dev/null +++ b/tests/bfabric_scripts/cli/login/test_login_common.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import yaml + +from bfabric_scripts.cli.login._common import resolve_config_env, resolve_scope, resolve_set_default + + +def _write_config(config_file, default="PROD"): + general = {"default_config": default} if default is not None else {} + config_file.write_text( + yaml.dump( + { + "GENERAL": general, + "PROD": {"base_url": "https://prod.example.com", "auth_method": "oauth"}, + "TEST": {"base_url": "https://test.example.com", "auth_method": "oauth"}, + } + ) + ) + + +class TestResolveConfigEnv: + def test_explicit_value_returned_as_is(self, tmp_path, mocker): + # An explicit value short-circuits: no file read, no prompt. + prompt = mocker.patch("bfabric_scripts.cli.login._common.select_or_input") + assert resolve_config_env("STAGE", tmp_path / "missing.yml") == "STAGE" + prompt.assert_not_called() + + def test_non_interactive_uses_current_default(self, tmp_path, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="TEST") + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=False) + assert resolve_config_env(None, config_file) == "TEST" + + def test_non_interactive_without_default_falls_back_to_production(self, tmp_path, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default=None) + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=False) + assert resolve_config_env(None, config_file) == "PRODUCTION" + + def test_non_interactive_missing_file_falls_back_to_production(self, tmp_path, mocker): + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=False) + assert resolve_config_env(None, tmp_path / "missing.yml") == "PRODUCTION" + + def test_interactive_offers_existing_and_allows_new(self, tmp_path, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + prompt = mocker.patch("bfabric_scripts.cli.login._common.select_or_input", return_value="NEWENV") + assert resolve_config_env(None, config_file) == "NEWENV" + args = prompt.call_args.args + # select_or_input offers the existing names as suggestions but lets the user type a new one. + assert set(args[1]) == {"PROD", "TEST"} + # The current default is prefilled. + assert prompt.call_args.kwargs["default"] == "PROD" + + def test_interactive_cancel_returns_none(self, tmp_path, mocker): + config_file = tmp_path / "config.yml" + _write_config(config_file, default="PROD") + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login._common.select_or_input", return_value=None) + assert resolve_config_env(None, config_file) is None + + +class TestResolveScope: + def test_preset_slug_expands(self): + assert resolve_scope("read-only") == "api:read" + assert resolve_scope("read-write") == "api:write" + assert resolve_scope("upload") == "api:write tus" + + def test_raw_string_passthrough(self): + assert resolve_scope("api:read custom:thing") == "api:read custom:thing" + + def test_non_interactive_without_scope_returns_none(self, mocker): + # No baked-in default: a headless login must pass --scope explicitly. + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=False) + assert resolve_scope(None) is None + + def test_interactive_preset_pick_expands(self, mocker): + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login._common.select_choice", return_value="upload") + assert resolve_scope(None) == "api:write tus" + + def test_interactive_custom_prompts_for_raw_scopes(self, mocker): + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login._common.select_choice", return_value="custom") + text = mocker.patch("bfabric_scripts.cli.login._common.text_input", return_value="api:read containers") + assert resolve_scope(None) == "api:read containers" + text.assert_called_once() + + def test_interactive_cancel_returns_none(self, mocker): + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login._common.select_choice", return_value=None) + assert resolve_scope(None) is None + + +class TestResolveSetDefault: + def test_explicit_true_honored_without_prompt(self, mocker): + confirm = mocker.patch("bfabric_scripts.cli.login._common.confirm") + assert resolve_set_default(True, "PROD") is True + confirm.assert_not_called() + + def test_explicit_false_honored_without_prompt(self, mocker): + confirm = mocker.patch("bfabric_scripts.cli.login._common.confirm") + assert resolve_set_default(False, "PROD") is False + confirm.assert_not_called() + + def test_non_interactive_defaults_to_true(self, mocker): + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=False) + confirm = mocker.patch("bfabric_scripts.cli.login._common.confirm") + assert resolve_set_default(None, "PROD") is True + confirm.assert_not_called() + + def test_interactive_prompts_preselected_yes(self, mocker): + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + confirm = mocker.patch("bfabric_scripts.cli.login._common.confirm", return_value=False) + assert resolve_set_default(None, "PROD") is False + # The prompt is preselected to "yes". + assert confirm.call_args.kwargs["default"] is True + + def test_interactive_cancel_returns_none(self, mocker): + # A cancelled prompt (confirm -> None) is propagated so the caller aborts the login. + mocker.patch("bfabric_scripts.cli.login._common.is_interactive", return_value=True) + mocker.patch("bfabric_scripts.cli.login._common.confirm", return_value=None) + assert resolve_set_default(None, "PROD") is None diff --git a/tests/bfabric_scripts/cli/test_interactive.py b/tests/bfabric_scripts/cli/test_interactive.py new file mode 100644 index 00000000..6a783d15 --- /dev/null +++ b/tests/bfabric_scripts/cli/test_interactive.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +import questionary + +from bfabric_scripts.cli.interactive import confirm, select_choice, select_or_input, text_input + + +class TestSelectChoice: + def test_delegates_to_questionary_select(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = "TEST" + select = mocker.patch("bfabric_scripts.cli.interactive.questionary.select", return_value=question) + assert select_choice("Pick", ["PROD", "TEST"], default="TEST") == "TEST" + # j/k are always disabled so ↑/↓ are the sole navigation keys (and never clash with the + # search filter, which questionary would otherwise reject). + select.assert_called_once_with( + "Pick", choices=["PROD", "TEST"], default="TEST", use_search_filter=False, use_jk_keys=False + ) + + def test_describe_builds_labelled_choices_but_returns_plain_value(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = "PROD" + select = mocker.patch("bfabric_scripts.cli.interactive.questionary.select", return_value=question) + assert select_choice("Pick", ["PROD", "TEST"], describe=lambda c: f"{c} info") == "PROD" + passed = select.call_args.kwargs["choices"] + assert all(isinstance(choice, questionary.Choice) for choice in passed) + assert [choice.title for choice in passed] == ["PROD info", "TEST info"] + assert [choice.value for choice in passed] == ["PROD", "TEST"] + + def test_search_flag_enables_live_filter(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = "PROD" + select = mocker.patch("bfabric_scripts.cli.interactive.questionary.select", return_value=question) + select_choice("Pick", ["PROD", "TEST"], search=True) + assert select.call_args.kwargs["use_search_filter"] is True + # j/k stay off (as always) so they don't clash with typing into the search filter. + assert select.call_args.kwargs["use_jk_keys"] is False + + +class TestSelectOrInput: + def test_maps_empty_answer_to_none(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = "" + mocker.patch("bfabric_scripts.cli.interactive.questionary.autocomplete", return_value=question) + assert select_or_input("Pick", ["PROD"]) is None + + def test_returns_typed_value_and_hints_tab_when_choices_present(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = "NEW" + autocomplete = mocker.patch("bfabric_scripts.cli.interactive.questionary.autocomplete", return_value=question) + assert select_or_input("Pick", ["PROD"], default="X") == "NEW" + # The Tab-autocomplete hint is appended when there are suggestions to complete. + autocomplete.assert_called_once_with("Pick (Tab to autocomplete)", choices=["PROD"], default="X") + + def test_no_hint_when_no_choices(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = "NEW" + autocomplete = mocker.patch("bfabric_scripts.cli.interactive.questionary.autocomplete", return_value=question) + assert select_or_input("Pick", []) == "NEW" + autocomplete.assert_called_once_with("Pick", choices=[], default="") + + +class TestConfirm: + def test_returns_true_when_confirmed(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = True + confirm_prompt = mocker.patch("bfabric_scripts.cli.interactive.questionary.confirm", return_value=question) + assert confirm("Delete?") is True + confirm_prompt.assert_called_once_with("Delete?", default=False) + + def test_returns_false_when_declined(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = False + mocker.patch("bfabric_scripts.cli.interactive.questionary.confirm", return_value=question) + assert confirm("Delete?") is False + + def test_cancel_returns_none(self, mocker): + # Ctrl-C / Esc yields None from questionary; confirm surfaces it so callers can tell an + # aborted prompt apart from an explicit "no". + question = mocker.MagicMock() + question.ask.return_value = None + mocker.patch("bfabric_scripts.cli.interactive.questionary.confirm", return_value=question) + assert confirm("Delete?") is None + + +class TestTextInput: + def test_returns_entered_value(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = "api:read foo" + text = mocker.patch("bfabric_scripts.cli.interactive.questionary.text", return_value=question) + assert text_input("Scopes", default="api:read") == "api:read foo" + text.assert_called_once_with("Scopes", default="api:read") + + def test_maps_empty_answer_to_none(self, mocker): + question = mocker.MagicMock() + question.ask.return_value = "" + mocker.patch("bfabric_scripts.cli.interactive.questionary.text", return_value=question) + assert text_input("Scopes") is None