diff --git a/cwmscli/__main__.py b/cwmscli/__main__.py index e97e6b9..99811ad 100644 --- a/cwmscli/__main__.py +++ b/cwmscli/__main__.py @@ -7,6 +7,7 @@ from click.core import ParameterSource from cwmscli.commands import commands_cwms +from cwmscli.commands.env import env_group from cwmscli.load import __main__ as load from cwmscli.usgs import usgs_group from cwmscli.utils.click_help import add_version_to_help_tree @@ -88,6 +89,7 @@ def cli( cli.add_command(commands_cwms.blob_group) cli.add_command(commands_cwms.clob_group) cli.add_command(commands_cwms.users_group) +cli.add_command(env_group) cli.add_command(load.load_group) add_version_to_help_tree(cli) diff --git a/cwmscli/commands/env.py b/cwmscli/commands/env.py new file mode 100644 index 0000000..4135da7 --- /dev/null +++ b/cwmscli/commands/env.py @@ -0,0 +1,508 @@ +import os +import shutil +import subprocess +import sys +import time +from typing import Dict, Optional + +import click + +from cwmscli.utils.env_store import ( + ENV_DEFAULTS, + EnvStoreError, + delete_env, + env_exists_on_disk, + list_envs, + load_env, + save_env, +) + +SENSITIVE_KEYS = {"CDA_API_KEY"} + + +def _stdout_is_tty() -> bool: + """Indirection so tests can override TTY detection.""" + return sys.stdout.isatty() + + +def _check_env(env_config: Dict[str, str]) -> Dict: + import requests + + api_root = env_config.get("CDA_API_ROOT", "").rstrip("/") + if not api_root: + return { + "reachable": False, + "latency_ms": None, + "auth": "skipped", + "error": "no API root", + } + + url = f"{api_root}/offices" + try: + t0 = time.monotonic() + resp = requests.get(url, timeout=5) + latency_ms = int((time.monotonic() - t0) * 1000) + except requests.RequestException as e: + return { + "reachable": False, + "latency_ms": None, + "auth": "skipped", + "error": str(e), + } + + if resp.status_code >= 400: + return { + "reachable": False, + "latency_ms": latency_ms, + "auth": "skipped", + "error": f"HTTP {resp.status_code}", + } + + api_key = env_config.get("CDA_API_KEY") + if not api_key: + return { + "reachable": True, + "latency_ms": latency_ms, + "auth": "skipped", + "error": None, + } + + try: + auth_resp = requests.get(url, headers={"Authorization": api_key}, timeout=5) + except requests.RequestException: + return { + "reachable": True, + "latency_ms": latency_ms, + "auth": "failed", + "error": None, + } + + if auth_resp.status_code == 401: + return { + "reachable": True, + "latency_ms": latency_ms, + "auth": "failed", + "error": None, + } + + return {"reachable": True, "latency_ms": latency_ms, "auth": "ok", "error": None} + + +@click.group("env", help="Manage CDA environments and API keys") +def env_group(): + """Environment management commands for cwms-cli.""" + pass + + +@env_group.command("setup", help="Create or update an environment configuration") +@click.argument("env_name") +@click.option( + "--api-root", + help="CDA API root URL (e.g., https://example.mil/cwms-data)", +) +@click.option( + "--api-key", + help="API key for authentication", +) +@click.option( + "--office", + help="Default office code (e.g., SWT)", +) +def setup_cmd( + env_name: str, + api_root: Optional[str], + api_key: Optional[str], + office: Optional[str], +): + """ + Create or update environment configuration. + + ENV_NAME can be: cwbi-dev, cwbi-test, cwbi-prod, onsite, localhost, or custom + """ + existing = load_env(env_name) or {} + env_vars = dict(existing) + env_vars["ENVIRONMENT"] = env_name + + if api_root: + env_vars["CDA_API_ROOT"] = api_root + elif "CDA_API_ROOT" not in env_vars and env_name in ENV_DEFAULTS: + env_vars["CDA_API_ROOT"] = ENV_DEFAULTS[env_name]["CDA_API_ROOT"] + + if api_key: + env_vars["CDA_API_KEY"] = api_key + + if office: + env_vars["OFFICE"] = office.upper() + + if "CDA_API_ROOT" not in env_vars: + click.echo( + f"Error: --api-root is required for '{env_name}' (not a default environment)", + err=True, + ) + click.echo(f"Available defaults: {', '.join(ENV_DEFAULTS.keys())}", err=True) + sys.exit(1) + + try: + path = save_env(env_name, env_vars) + except EnvStoreError as e: + click.echo(f"Error: {e}", err=True) + sys.exit(1) + + click.echo(f"Environment '{env_name}' saved to {path}") + + +@env_group.command("show", help="Show current environment and available configurations") +@click.option( + "--check", + is_flag=True, + default=False, + help="Test connectivity and authentication for each environment.", +) +def show_cmd(check: bool): + """ + Display current environment and list all configured environments. + + Lists all environments with API root, office, and key status. + Use --check to test connectivity and API key validity (requires network). + """ + current_env = os.environ.get("ENVIRONMENT") + + if current_env: + click.echo( + f"Current environment: {click.style(current_env, fg='green', bold=True)}\n" + ) + else: + click.echo("No environment currently active\n") + + names = list_envs() + if not names: + click.echo("No environments configured") + click.echo("Run 'cwms-cli env setup ' to create one") + return + + click.echo("Available environments:") + for env_name in names: + env_config = load_env(env_name) + if not env_config: + continue + marker = "* " if env_name == current_env else " " + builtin = " (built-in)" if not env_exists_on_disk(env_name) else "" + api_root = env_config.get("CDA_API_ROOT", "not set") + office = env_config.get("OFFICE", "not set") + has_key = "has API key" if env_config.get("CDA_API_KEY") else "no API key" + + click.echo(f"{marker}{env_name}{builtin}") + click.echo(f" API Root: {api_root}") + click.echo(f" Office: {office}") + click.echo(f" Status: {has_key}") + + if check: + result = _check_env(env_config) + if result["reachable"]: + latency = f" ({result['latency_ms']}ms)" + reach_str = click.style("reachable", fg="green") + latency + else: + err = f" — {result['error']}" if result["error"] else "" + reach_str = click.style("unreachable", fg="red") + err + + auth_str = "" + if result["auth"] == "ok": + auth_str = click.style("authenticated", fg="green") + elif result["auth"] == "failed": + auth_str = click.style("auth failed", fg="red") + + click.echo(f" Connect: {reach_str}") + if auth_str: + click.echo(f" Auth: {auth_str}") + + +@env_group.command("delete", help="Delete an environment configuration") +@click.argument("env_name") +@click.option("--yes", "-y", is_flag=True, help="Skip confirmation prompt") +def delete_cmd(env_name: str, yes: bool): + """ + Delete an environment configuration. + + Examples: + cwms-cli env delete myenv + cwms-cli env delete myenv --yes + """ + if not yes and not click.confirm(f"Delete environment '{env_name}'?"): + click.echo("Cancelled") + return + + try: + existed = delete_env(env_name) + except EnvStoreError as e: + click.echo(f"Error: {e}", err=True) + sys.exit(1) + + if existed: + click.echo(f"Environment '{env_name}' deleted") + else: + click.echo(f"Environment '{env_name}' not found", err=True) + sys.exit(1) + + +def _detect_shell() -> str: + """Best-effort detection of the user's interactive shell.""" + if sys.platform == "win32": + # PowerShell sets PSModulePath; prefer pwsh/powershell when present. + if os.environ.get("PSModulePath"): + for candidate in ("pwsh", "powershell"): + found = shutil.which(candidate) + if found: + return found + return os.environ.get("COMSPEC", "cmd.exe") + + return os.environ.get("SHELL", "/bin/bash") + + +def _detect_shell_kind() -> str: + """Map the detected shell path to a known kind, defaulting to 'bash'.""" + path = _detect_shell().lower() + base = os.path.basename(path) + if "pwsh" in base or "powershell" in base: + return "powershell" + if "cmd" in base: + return "cmd" + if "fish" in base: + return "fish" + if "zsh" in base: + return "zsh" + if "bash" in base or "sh" in base: + return "bash" + return "bash" + + +def _export_help_lines(env_name: str) -> str: + """Per-shell instructions for loading an env into the current shell.""" + recipes = { + "bash": f'eval "$(cwms-cli env export {env_name} --format bash)"', + "zsh": f'eval "$(cwms-cli env export {env_name} --format bash)"', + "powershell": ( + f"cwms-cli env export {env_name} --format powershell " + "| Out-String | Invoke-Expression" + ), + "cmd": ( + f"cwms-cli env export {env_name} --format cmd " + f"--output %TEMP%\\cwms-env.cmd && call %TEMP%\\cwms-env.cmd" + ), + "fish": f"cwms-cli env export {env_name} --format fish | source", + } + detected = _detect_shell_kind() + primary = recipes.get(detected, recipes["bash"]) + + lines = [ + f"To load '{env_name}' into your current shell ({detected} detected):", + f" {primary}", + "", + "For other shells:", + ] + label_width = max(len(k) for k in recipes) + for kind, recipe in recipes.items(): + if kind == detected: + continue + lines.append(f" {kind.ljust(label_width)} {recipe}") + lines.extend( + [ + "", + f"Write a .env file: cwms-cli env export {env_name} --output .env", + "Print to terminal anyway: --show-key", + ] + ) + return "\n".join(lines) + + +def spawn_shell_with_env(env_vars: Dict[str, str], env_name: str): + """Spawn a new shell with environment variables set.""" + user_shell = _detect_shell() + new_env = os.environ.copy() + new_env.update(env_vars) + + click.echo( + f"Activating environment: {click.style(env_name, fg='green', bold=True)}", + err=True, + ) + click.echo(f"Shell: {user_shell}", err=True) + click.echo( + "Type 'exit' or press Ctrl+D to return to your original environment\n", + err=True, + ) + + try: + result = subprocess.run([user_shell], env=new_env) + sys.exit(result.returncode) + except OSError as e: + click.echo(f"Error spawning shell: {e}", err=True) + sys.exit(1) + + +@env_group.command("activate", help="Activate an environment in a new shell") +@click.argument("env_name") +def activate_cmd(env_name: str): + """ + Activate an environment in a new shell session. + + The environment variables will be set in the new shell and persist + until you exit the shell. Type 'exit' to return to your original environment. + + Note: This spawns a child shell. Your parent shell, and any IDE + already open, will not see these variables. To populate the current + shell, use: eval "$(cwms-cli env export --format bash)" + + Examples: + cwms-cli env activate cwbi-prod + cwms-cli env activate localhost + """ + env_vars = load_env(env_name) + if not env_vars: + click.echo(f"Error: Environment '{env_name}' not found", err=True) + click.echo(f"Run 'cwms-cli env setup {env_name}' to create it", err=True) + sys.exit(1) + + spawn_shell_with_env(env_vars, env_name) + + +def _quote_dotenv(value: str) -> str: + escaped = value.replace("\\", "\\\\").replace('"', '\\"') + return f'"{escaped}"' + + +def _quote_bash(value: str) -> str: + return "'" + value.replace("'", "'\\''") + "'" + + +def _quote_powershell(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def _quote_cmd(value: str) -> str: + # set "K=V" handles spaces and most special chars. Escape embedded " and %. + return value.replace("%", "%%").replace('"', '""') + + +def _quote_fish(value: str) -> str: + # Fish single-quoted strings: backslash escapes \ and '. + return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'" + + +def _format_env(env_vars: Dict[str, str], fmt: str) -> str: + items = sorted(env_vars.items()) + lines = [] + for key, value in items: + value = str(value) + if fmt == "dotenv": + lines.append(f"{key}={_quote_dotenv(value)}") + elif fmt == "bash": + lines.append(f"export {key}={_quote_bash(value)}") + elif fmt == "powershell": + lines.append(f"$env:{key} = {_quote_powershell(value)}") + elif fmt == "cmd": + # @ prefix suppresses cmd's default echoing of each line when run via `call`. + lines.append(f'@set "{key}={_quote_cmd(value)}"') + elif fmt == "fish": + lines.append(f"set -gx {key} {_quote_fish(value)}") + else: + raise ValueError(f"Unknown format: {fmt}") + return "\n".join(lines) + + +@env_group.command( + "export", + help="Export an environment's variables to your current shell or a .env file", +) +@click.argument("env_name") +@click.option( + "--format", + "fmt", + type=click.Choice(["dotenv", "bash", "powershell", "cmd", "fish"]), + default="dotenv", + show_default=True, + help="Output syntax. Match this to your shell, or use 'dotenv' for a .env file.", +) +@click.option( + "--output", + "-o", + type=click.Path(dir_okay=False, writable=True, resolve_path=True), + default=None, + help="Write to FILE (mode 0600) instead of standard output.", +) +@click.option( + "--no-key", + is_flag=True, + default=False, + help="Omit CDA_API_KEY (useful for sharing templates).", +) +@click.option( + "--show-key", + is_flag=True, + default=False, + help="Allow the API key to be displayed in your terminal.", +) +def export_cmd( + env_name: str, + fmt: str, + output: Optional[str], + no_key: bool, + show_key: bool, +): + """ + Export a stored environment so your current shell, IDE, or another tool + can use its variables. + + A child process cannot directly modify its parent shell, so this command + emits values that you (or your shell) load. Three common ways: + + \b + # Load into the current bash/zsh shell + eval "$(cwms-cli env export cwbi-prod --format bash)" + + # Load into the current PowerShell session + cwms-cli env export cwbi-prod --format powershell | Out-String | Invoke-Expression + + # Write a .env file for an IDE, docker-compose, or direnv to read + cwms-cli env export cwbi-prod --output .env + + Run with no flags in an interactive terminal to see the right recipe + for your detected shell. The API key is never displayed in a terminal + unless you pass --show-key. + """ + env_vars = load_env(env_name) + if env_vars is None: + click.echo(f"Error: Environment '{env_name}' not found", err=True) + sys.exit(1) + + if no_key: + env_vars = {k: v for k, v in env_vars.items() if k not in SENSITIVE_KEYS} + + has_key = any(k in env_vars for k in SENSITIVE_KEYS) + writing_to_file = output is not None + + if has_key and not writing_to_file and _stdout_is_tty() and not show_key: + click.echo(_export_help_lines(env_name), err=True) + sys.exit(1) + + rendered = _format_env(env_vars, fmt) + + if writing_to_file: + path = output + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + if sys.platform == "win32": + fd = os.open(path, flags) + else: + fd = os.open(path, flags, 0o600) + try: + with os.fdopen(fd, "w") as f: + f.write(rendered + "\n") + if sys.platform != "win32": + os.chmod(path, 0o600) + except OSError as e: + click.echo(f"Error writing {path}: {e}", err=True) + sys.exit(1) + click.echo(f"Wrote {path} (0600)", err=True) + if path.endswith(".env") or os.path.basename(path).startswith(".env"): + click.echo("Reminder: add this file to .gitignore.", err=True) + return + + click.echo(rendered) diff --git a/cwmscli/load/root.py b/cwmscli/load/root.py index 1366ad3..f2ba1b9 100644 --- a/cwmscli/load/root.py +++ b/cwmscli/load/root.py @@ -10,6 +10,7 @@ from cwmscli import requirements as reqs from cwmscli.utils.deps import requires +from cwmscli.utils.env_store import load_env logger = logging.getLogger(__name__) @@ -43,6 +44,51 @@ def _norm_office(o: Optional[str]) -> str: def validate_cda_targets(func): @functools.wraps(func) def wrapper(*args, **kwargs): + source_env = kwargs.pop("source_env", None) + target_env = kwargs.pop("target_env", None) + + if source_env is not None: + if _param_was_explicit("source_cda"): + raise click.ClickException( + "--source-env and --source-cda are mutually exclusive." + ) + env_data = load_env(source_env) + if env_data is None: + raise click.ClickException( + f"Environment '{source_env}' not found. " + "Run 'cwms-cli env show' to list available environments." + ) + if "CDA_API_ROOT" not in env_data: + raise click.ClickException( + f"Environment '{source_env}' has no CDA_API_ROOT configured. " + f"Run 'cwms-cli env setup {source_env} --api-root ' to fix." + ) + kwargs["source_cda"] = env_data["CDA_API_ROOT"] + if not _param_was_explicit("source_office") and env_data.get("OFFICE"): + kwargs["source_office"] = env_data["OFFICE"] + + if target_env is not None: + if _param_was_explicit("target_cda"): + raise click.ClickException( + "--target-env and --target-cda are mutually exclusive." + ) + env_data = load_env(target_env) + if env_data is None: + raise click.ClickException( + f"Environment '{target_env}' not found. " + "Run 'cwms-cli env show' to list available environments." + ) + if "CDA_API_ROOT" not in env_data: + raise click.ClickException( + f"Environment '{target_env}' has no CDA_API_ROOT configured. " + f"Run 'cwms-cli env setup {target_env} --api-root ' to fix." + ) + kwargs["target_cda"] = env_data["CDA_API_ROOT"] + if not _param_was_explicit("target_api_key") and env_data.get( + "CDA_API_KEY" + ): + kwargs["target_api_key"] = env_data["CDA_API_KEY"] + source_csv = kwargs.get("source_csv") target_csv = kwargs.get("target_csv") @@ -53,16 +99,20 @@ def wrapper(*args, **kwargs): ) if source_csv: - if kwargs.get("source_cda") and _param_was_explicit("source_cda"): + if kwargs.get("source_cda") and ( + _param_was_explicit("source_cda") or source_env is not None + ): raise click.ClickException( - "--source-csv and --source-cda are mutually exclusive." + "--source-csv and --source-cda/--source-env are mutually exclusive." ) kwargs["source_cda"] = None if target_csv: - if kwargs.get("target_cda") and _param_was_explicit("target_cda"): + if kwargs.get("target_cda") and ( + _param_was_explicit("target_cda") or target_env is not None + ): raise click.ClickException( - "--target-csv and --target-cda are mutually exclusive." + "--target-csv and --target-cda/--target-env are mutually exclusive." ) kwargs["target_cda"] = None @@ -134,6 +184,24 @@ def shared_source_target_options(f): envvar="CDA_API_KEY", help="Target API key used when no saved cwms-cli login token is available.", )(f) + f = click.option( + "--source-env", + default=None, + help=( + "Named CDA environment for the source " + "(resolves api-root and office from ~/.config/cwms-cli/envs/). " + "Mutually exclusive with --source-cda." + ), + )(f) + f = click.option( + "--target-env", + default=None, + help=( + "Named CDA environment for the target " + "(resolves api-root and api-key from ~/.config/cwms-cli/envs/). " + "Mutually exclusive with --target-cda." + ), + )(f) f = click.option( "--dry-run/--no-dry-run", is_flag=True, diff --git a/cwmscli/utils/auth.py b/cwmscli/utils/auth.py index d6f5a9c..67ac887 100644 --- a/cwmscli/utils/auth.py +++ b/cwmscli/utils/auth.py @@ -15,6 +15,8 @@ from pathlib import Path from typing import Any, Dict, Optional +from cwmscli.utils.paths import config_dir + DEFAULT_CLIENT_ID = "cwms" DEFAULT_CDA_API_ROOT = "https://cwms-data.usace.army.mil/cwms-data" DEFAULT_OIDC_BASE_URL = ( @@ -109,12 +111,7 @@ def log_message(self, format: str, *args: Any) -> None: def default_token_file(provider: str) -> Path: - config_root = os.getenv("XDG_CONFIG_HOME") - if config_root: - base_dir = Path(config_root) - else: - base_dir = Path.home() / ".config" - return base_dir / "cwms-cli" / "auth" / f"{provider}.json" + return config_dir("auth") / f"{provider}.json" def _oidc_cache_file() -> Path: diff --git a/cwmscli/utils/env_store.py b/cwmscli/utils/env_store.py new file mode 100644 index 0000000..6500250 --- /dev/null +++ b/cwmscli/utils/env_store.py @@ -0,0 +1,133 @@ +"""File-based storage for named CDA environments. + +One JSON file per environment under the user's config dir, mode 0600 on +POSIX (current-user-only ACL on Windows). Same threat model as +~/.aws/credentials, ~/.config/gh, ~/.kube/config: the user account is +the security boundary. +""" + +import json +import os +import sys +from pathlib import Path +from typing import Dict, List, Optional + +from cwmscli.utils.paths import config_dir + +ENV_DEFAULTS = { + "cwbi-prod": { + "ENVIRONMENT": "cwbi-prod", + "CDA_API_ROOT": "https://cwms-data.usace.army.mil/cwms-data", + }, +} + + +class EnvStoreError(Exception): + """Raised when an env file cannot be read, written, or deleted.""" + + +def envs_dir() -> Path: + """Return the directory where env files live, creating it if needed.""" + return config_dir("envs", create=True) + + +def _env_path(name: str) -> Path: + if not name or "/" in name or "\\" in name or name in (".", ".."): + raise EnvStoreError(f"Invalid environment name: {name!r}") + return envs_dir() / f"{name}.json" + + +def _restrict_windows_acl(path: Path) -> None: + """Restrict a file to the current user on Windows via icacls. + + No-op on non-Windows. Failures are swallowed: the file already exists + in the user's APPDATA, and surfacing icacls errors hurts more than + it helps. + """ + if sys.platform != "win32": + return + import subprocess + + user = os.environ.get("USERNAME") + if not user: + return + try: + subprocess.run( + ["icacls", str(path), "/inheritance:r", "/grant:r", f"{user}:F"], + check=False, + capture_output=True, + ) + except (OSError, FileNotFoundError): + pass + + +def save_env(name: str, config: Dict[str, str]) -> Path: + """Write a config dict to /.json with 0600 perms. + + Uses os.open with O_CREAT|O_TRUNC|O_WRONLY and explicit mode so the + file is never briefly world-readable between create and chmod. + """ + path = _env_path(name) + payload = json.dumps(config, indent=2, sort_keys=True) + "\n" + + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + if sys.platform == "win32": + # Windows ignores the mode arg; ACL is set after the write. + fd = os.open(path, flags) + else: + fd = os.open(path, flags, 0o600) + + try: + with os.fdopen(fd, "w") as f: + f.write(payload) + except OSError as e: + raise EnvStoreError(f"Failed to write env file {path}: {e}") from e + + if sys.platform != "win32": + # Re-chmod in case the file already existed with different perms. + try: + os.chmod(path, 0o600) + except OSError: + pass + else: + _restrict_windows_acl(path) + + return path + + +def env_exists_on_disk(name: str) -> bool: + """Return True if a user-created env file exists for *name*.""" + return _env_path(name).exists() + + +def load_env(name: str) -> Optional[Dict[str, str]]: + """Read and parse an env file, falling back to built-in defaults.""" + path = _env_path(name) + if not path.exists(): + return dict(ENV_DEFAULTS[name]) if name in ENV_DEFAULTS else None + try: + with open(path, "r") as f: + data = json.load(f) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(data, dict): + return None + return data + + +def delete_env(name: str) -> bool: + """Remove an env file. Returns True if it existed.""" + path = _env_path(name) + try: + path.unlink() + return True + except FileNotFoundError: + return False + except OSError as e: + raise EnvStoreError(f"Failed to delete env file {path}: {e}") from e + + +def list_envs() -> List[str]: + """Return sorted env names, including built-in defaults.""" + on_disk = {p.stem for p in envs_dir().glob("*.json")} + return sorted(on_disk | ENV_DEFAULTS.keys()) diff --git a/cwmscli/utils/logging/__init__.py b/cwmscli/utils/logging/__init__.py index 6abd5eb..513245d 100644 --- a/cwmscli/utils/logging/__init__.py +++ b/cwmscli/utils/logging/__init__.py @@ -98,7 +98,7 @@ def setup_logging(cfg: LoggingConfig) -> None: base_fmt = "%(asctime)s;%(levelname)s;%(message)s" date_fmt = "%Y-%m-%d %H:%M:%S" - stream_handler = logging.StreamHandler(sys.stdout) + stream_handler = logging.StreamHandler(sys.stderr) stream_handler.setFormatter(ColorLevelFormatter(base_fmt, date_fmt, color_enabled)) root.addHandler(stream_handler) diff --git a/cwmscli/utils/paths.py b/cwmscli/utils/paths.py new file mode 100644 index 0000000..873d1bb --- /dev/null +++ b/cwmscli/utils/paths.py @@ -0,0 +1,23 @@ +"""Per-user config directory resolution for cwms-cli. + +Single source of truth so saved logins, named environments, and any future +per-user state all live under one root on every platform. +""" + +import os +from pathlib import Path + + +def config_dir(*parts: str, create: bool = False) -> Path: + """Return cwms-cli's per-user config dir, joined with *parts. + + Uses ``$XDG_CONFIG_HOME/cwms-cli`` when set, else ``~/.config/cwms-cli``, + on all platforms so a user's tokens and environments share one root. + Pass ``create=True`` to ensure the directory exists. + """ + xdg = os.environ.get("XDG_CONFIG_HOME") + base = Path(xdg).expanduser() if xdg else Path("~/.config").expanduser() + path = base.joinpath("cwms-cli", *parts) + if create: + path.mkdir(parents=True, exist_ok=True) + return path diff --git a/docs/cli/env.rst b/docs/cli/env.rst new file mode 100644 index 0000000..531290a --- /dev/null +++ b/docs/cli/env.rst @@ -0,0 +1,303 @@ +Environment Manager +=================== + +Manage named CDA environments with ``cwms-cli env``. Each environment stores +a CDA API root URL, office code, and optional API key in a JSON file under +``~/.config/cwms-cli/envs/`` (or ``$XDG_CONFIG_HOME/cwms-cli/envs/`` when that +variable is set), on all platforms. Files are created with mode ``0600`` +(owner-only read/write) so only your user account can read them. + +This keeps API keys out of project directories, shell history, and command +lines, and lets you reference environments by name instead of juggling +URLs and credentials. + + +Built-in Environments +--------------------- + +``cwbi-prod`` ships preconfigured with the production CDA URL. It is +available immediately — no ``env setup`` required — and appears in +``env show`` as ``(built-in)``. + +Because the built-in has no office or API key, you still need to pass +``--source-office`` when using it as a source: + +.. code-block:: bash + + cwms-cli load location ids-all \ + --source-env cwbi-prod --source-office SWT \ + --target-env localhost + +To avoid repeating ``--source-office`` every time, run ``env setup`` once +to attach an office (and optionally an API key): + +.. code-block:: bash + + cwms-cli env setup cwbi-prod --office SWT --api-key YOUR_KEY + + +Quick Start +----------- + +**1. Create environments:** + +.. code-block:: bash + + # Production — customize the built-in with your office and key + cwms-cli env setup cwbi-prod --office SWT --api-key YOUR_KEY + + # Development (needs --api-root) + cwms-cli env setup cwbi-dev \ + --api-root https://cwms-data-dev.example.mil/cwms-data \ + --office SWT --api-key YOUR_KEY + + # Test (needs --api-root) + cwms-cli env setup cwbi-test \ + --api-root https://cwms-data-test.example.mil/cwms-data \ + --office SWT --api-key YOUR_KEY + + # Local development server + cwms-cli env setup localhost \ + --api-root http://localhost:8082/cwms-data --office DEV + +**2. Use environments with load commands:** + +.. code-block:: bash + + cwms-cli load location ids-all \ + --source-env cwbi-prod --target-env localhost + + cwms-cli load timeseries data \ + --source-env cwbi-prod --target-env localhost \ + --ts-id "Black Butte.Flow.Inst.1Hour.0.raw-cda" + +**3. View and manage environments:** + +.. code-block:: bash + + cwms-cli env show + cwms-cli env delete old-env --yes + + +Using Environments with ``load`` +--------------------------------- + +The ``--source-env`` and ``--target-env`` options resolve a named +environment into the corresponding source/target options: + +- ``--source-env`` sets ``--source-cda`` and ``--source-office`` +- ``--target-env`` sets ``--target-cda`` and ``--target-api-key`` + +These two invocations are equivalent: + +.. code-block:: bash + + # Explicit flags + cwms-cli load location ids-all \ + --source-cda https://cwms-data.usace.army.mil/cwms-data/ \ + --source-office SWT \ + --target-cda http://localhost:8082/cwms-data/ \ + --target-api-key "apikey 0123456789abcdef" + + # Named environments + cwms-cli load location ids-all \ + --source-env cwbi-prod --target-env localhost + +**Rules:** + +- ``--source-env`` and ``--source-cda`` are mutually exclusive. +- ``--target-env`` and ``--target-cda`` are mutually exclusive. +- Explicit ``--source-office`` or ``--target-api-key`` flags override the + values from the environment file. + + +Commands +-------- + +cwms-cli env setup +~~~~~~~~~~~~~~~~~~~~~~~~~ + +Create or update an environment configuration. + +.. code-block:: bash + + # Setup with all options + cwms-cli env setup myenv \ + --api-root https://cwms-data-dev.example.mil/cwms-data \ + --api-key YOUR_KEY --office SWT + + # Update just the API key (other fields preserved) + cwms-cli env setup myenv --api-key NEW_KEY + + # Update just the office + cwms-cli env setup myenv --office LRD + +``cwbi-prod`` is built-in and already has the production URL. Running +``env setup cwbi-prod`` creates a user file that overrides the built-in, +letting you attach an office and API key. All other environment names +require ``--api-root``. + + +cwms-cli env show +~~~~~~~~~~~~~~~~~ + +List all configured environments with their API root, office, and key status. +The API key is always redacted — only ``has API key`` or ``no API key`` is shown. + +.. code-block:: bash + + cwms-cli env show + +**Example output:** + +.. code-block:: text + + Current environment: cwbi-prod + + Available environments: + * cwbi-prod + API Root: https://cwms-data.usace.army.mil/cwms-data + Office: SWT + Status: has API key + cwbi-dev + API Root: https://cwms-data-dev.example.mil/cwms-data + Office: SWT + Status: no API key + +On a fresh install (before any ``env setup``), ``cwbi-prod`` appears with +``(built-in)`` and shows ``Office: not set``. + +The ``*`` marks the currently active environment (from the ``ENVIRONMENT`` +variable). + +**Options:** + +- ``--check`` — test connectivity and API key validity for each environment + (requires network access). Adds ``Connect`` and ``Auth`` lines to the output. + +.. code-block:: bash + + cwms-cli env show --check + +.. code-block:: text + + Available environments: + * cwbi-prod + API Root: https://cwms-data.usace.army.mil/cwms-data + Office: SWT + Status: has API key + Connect: reachable (284ms) + Auth: authenticated + cwbi-dev + API Root: https://cwms-data-dev.example.mil/cwms-data + Office: SWT + Status: no API key + Connect: unreachable — Connection refused + + +cwms-cli env export +~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Export an environment's variables to your current shell or a file. + +.. code-block:: bash + + # Load into the current bash/zsh shell + eval "$(cwms-cli env export cwbi-prod --format bash)" + + # Load into PowerShell + cwms-cli env export cwbi-prod --format powershell | Out-String | Invoke-Expression + + # Write a .env file for an IDE or docker-compose + cwms-cli env export cwbi-prod --output .env + +**Formats:** ``dotenv`` (default), ``bash``, ``powershell``, ``cmd``, ``fish``. + +**Safety:** The API key is never printed to a terminal by default. If stdout +is a TTY and the environment has an API key, ``export`` shows shell-specific +recipes instead. Use ``--show-key`` to override, or ``--output FILE`` to write +directly to disk (recommended — guarantees ``0600`` permissions and no +scrollback exposure). + +**Options:** + +- ``--output FILE`` — write to a file with ``0600`` permissions instead of stdout. +- ``--no-key`` — omit ``CDA_API_KEY`` (useful for templates or sharing). +- ``--show-key`` — allow the API key to be displayed in the terminal. + + +cwms-cli env activate +~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Activate an environment in a new shell session. + +.. code-block:: bash + + cwms-cli env activate cwbi-prod + +This spawns a child shell with the environment variables set. Type ``exit`` +or press ``Ctrl+D`` to return to your original shell. + +.. note:: + + The parent shell and any already-open IDE will **not** see these variables. + For IDE integration, use ``cwms-cli env export --output .env`` + instead. + + +cwms-cli env delete +~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Delete an environment configuration. + +.. code-block:: bash + + # Delete with confirmation prompt + cwms-cli env delete myenv + + # Delete without confirmation + cwms-cli env delete myenv --yes + + +Storage and Security +-------------------- + +**File locations:** + +- All platforms: ``~/.config/cwms-cli/envs/.json`` + (respects ``XDG_CONFIG_HOME`` when set) + +**File permissions:** ``0600`` on POSIX (owner-only read/write). On Windows, +an ACL restricts access to the current user. + +**Security model:** The user account is the security boundary, matching +``aws``, ``gcloud``, ``kubectl``, and ``gh``. This feature defends against: + +- Accidental ``git add`` of a key — files live in ``~/.config/``, not the repo +- Key pasted into an LLM — users share ``env show`` output (always redacted) +- Key visible in ``ps`` or shell history — users reference the env name, not values +- Key in terminal scrollback — ``export`` refuses TTY output by default + +This feature does **not** defend against root access or same-user process +reads. For encrypted-at-rest storage, use a vault (1Password CLI, HashiCorp +Vault, AWS Secrets Manager) and feed values in via environment variables. + + +Headless and CI Usage +--------------------- + +For headless or CI environments where ``cwms-cli env`` is not practical, +set environment variables directly: + +.. code-block:: bash + + export CDA_API_ROOT="https://cwms-data.usace.army.mil/cwms-data" + export CDA_API_KEY="your_key" + export OFFICE="SWT" + + cwms-cli blob list + + +.. click:: cwmscli.commands.env:env_group + :prog: cwms-cli env + :nested: full diff --git a/docs/cli/load_location_ids_all.rst b/docs/cli/load_location_ids_all.rst index 1c2189d..e92178e 100644 --- a/docs/cli/load_location_ids_all.rst +++ b/docs/cli/load_location_ids_all.rst @@ -17,6 +17,22 @@ to the source CDA catalog, so both options use CDA regular expression behavior. The CLI does not apply extra exact-match filtering. For CDA regex syntax, see the :doc:`CWMS Data API regular expression guide `. +Using Named Environments +------------------------ + +If you have configured environments with ``cwms-cli env setup``, use +``--source-env`` and ``--target-env`` instead of spelling out URLs and keys: + +.. code-block:: shell + + cwms-cli load location ids-all \ + --source-env cwbi-prod \ + --target-env localhost \ + --like "^Black Butte$" \ + --location-kind-like PROJECT + +See :doc:`env` for environment setup. + CDA to CDA Examples ------------------- diff --git a/docs/index.rst b/docs/index.rst index 200cc81..5b811bf 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -60,6 +60,7 @@ Contents cli/csv2cwms cli/blob + cli/env cli/login cli/clob cli/users diff --git a/tests/commands/test_env.py b/tests/commands/test_env.py new file mode 100644 index 0000000..9238c5f --- /dev/null +++ b/tests/commands/test_env.py @@ -0,0 +1,625 @@ +"""Tests for `cwms-cli env` commands and the file-based env store.""" + +import json +import os +import stat +import subprocess +import sys + +import pytest +from click.testing import CliRunner + +from cwmscli.commands.env import ( + _format_env, + _quote_bash, + _quote_cmd, + _quote_dotenv, + _quote_fish, + _quote_powershell, + env_group, +) +from cwmscli.utils.env_store import ( + EnvStoreError, + delete_env, + envs_dir, + list_envs, + load_env, + save_env, +) + + +@pytest.fixture +def isolated_envs(monkeypatch, tmp_path): + """Redirect env storage to a tmp dir for each test.""" + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + monkeypatch.setenv("APPDATA", str(tmp_path)) + monkeypatch.delenv("ENVIRONMENT", raising=False) + return tmp_path + + +# ---------- env_store ---------- + + +def test_save_load_roundtrip(isolated_envs): + config = { + "ENVIRONMENT": "demo", + "CDA_API_ROOT": "https://example.mil/cwms-data", + "CDA_API_KEY": "abc123", + "OFFICE": "SWT", + } + save_env("demo", config) + assert load_env("demo") == config + + +def test_save_env_writes_0600(isolated_envs): + if sys.platform == "win32": + pytest.skip("POSIX permission semantics") + path = save_env("demo", {"CDA_API_KEY": "k"}) + mode = stat.S_IMODE(os.stat(path).st_mode) + assert mode == 0o600 + + +def test_save_env_overwrites_with_0600(isolated_envs): + if sys.platform == "win32": + pytest.skip("POSIX permission semantics") + path = save_env("demo", {"a": "1"}) + os.chmod(path, 0o644) + save_env("demo", {"a": "2"}) + assert stat.S_IMODE(os.stat(path).st_mode) == 0o600 + assert load_env("demo") == {"a": "2"} + + +def test_load_env_missing_returns_none(isolated_envs): + assert load_env("nope") is None + + +def test_load_env_malformed_returns_none(isolated_envs): + path = envs_dir() / "bad.json" + path.write_text("{not valid json") + assert load_env("bad") is None + + +def test_load_env_non_dict_returns_none(isolated_envs): + path = envs_dir() / "list.json" + path.write_text("[1, 2, 3]") + assert load_env("list") is None + + +def test_delete_env_existing(isolated_envs): + save_env("demo", {"a": "1"}) + assert delete_env("demo") is True + assert load_env("demo") is None + + +def test_delete_env_missing(isolated_envs): + assert delete_env("nope") is False + + +def test_list_envs(isolated_envs): + assert list_envs() == ["cwbi-prod"] + save_env("alpha", {"a": "1"}) + save_env("beta", {"a": "1"}) + save_env("gamma", {"a": "1"}) + assert list_envs() == ["alpha", "beta", "cwbi-prod", "gamma"] + + +def test_list_envs_deduplicates_defaults(isolated_envs): + save_env("cwbi-prod", {"CDA_API_ROOT": "https://custom"}) + assert list_envs().count("cwbi-prod") == 1 + + +@pytest.mark.parametrize("bad", ["", ".", "..", "a/b", "a\\b"]) +def test_invalid_env_names_rejected(isolated_envs, bad): + with pytest.raises(EnvStoreError): + save_env(bad, {"a": "1"}) + + +# ---------- built-in defaults ---------- + + +def test_load_env_returns_builtin_default(isolated_envs): + data = load_env("cwbi-prod") + assert data is not None + assert data["CDA_API_ROOT"] == "https://cwms-data.usace.army.mil/cwms-data" + assert data["ENVIRONMENT"] == "cwbi-prod" + + +def test_on_disk_env_overrides_builtin(isolated_envs): + save_env("cwbi-prod", {"CDA_API_ROOT": "https://custom", "CDA_API_KEY": "k"}) + data = load_env("cwbi-prod") + assert data["CDA_API_ROOT"] == "https://custom" + assert data["CDA_API_KEY"] == "k" + + +def test_show_labels_builtin(isolated_envs): + runner = CliRunner() + result = runner.invoke(env_group, ["show"]) + assert result.exit_code == 0 + assert "cwbi-prod (built-in)" in result.output + + +def test_show_no_builtin_label_when_customized(isolated_envs): + save_env( + "cwbi-prod", + {"CDA_API_ROOT": "https://cwms-data.usace.army.mil/cwms-data", "OFFICE": "SWT"}, + ) + runner = CliRunner() + result = runner.invoke(env_group, ["show"]) + assert result.exit_code == 0 + assert "cwbi-prod" in result.output + assert "(built-in)" not in result.output + + +# ---------- env setup ---------- + + +def test_setup_creates_file_with_0600(isolated_envs): + runner = CliRunner() + result = runner.invoke( + env_group, + ["setup", "myenv", "--api-root", "https://x.mil/cwms-data", "--api-key", "k"], + ) + assert result.exit_code == 0, result.output + path = envs_dir() / "myenv.json" + assert path.exists() + if sys.platform != "win32": + assert stat.S_IMODE(os.stat(path).st_mode) == 0o600 + stored = json.loads(path.read_text()) + assert stored["CDA_API_ROOT"] == "https://x.mil/cwms-data" + assert stored["CDA_API_KEY"] == "k" + assert stored["ENVIRONMENT"] == "myenv" + + +def test_setup_uses_default_for_known_name(isolated_envs): + runner = CliRunner() + result = runner.invoke(env_group, ["setup", "cwbi-prod", "--api-key", "k"]) + assert result.exit_code == 0, result.output + stored = load_env("cwbi-prod") + assert stored["CDA_API_ROOT"] == "https://cwms-data.usace.army.mil/cwms-data" + + +def test_setup_requires_api_root_for_unknown_name(isolated_envs): + runner = CliRunner() + result = runner.invoke(env_group, ["setup", "newenv", "--api-key", "k"]) + assert result.exit_code != 0 + assert "--api-root is required" in result.output + + +def test_setup_partial_update_preserves_fields(isolated_envs): + runner = CliRunner() + runner.invoke( + env_group, + [ + "setup", + "myenv", + "--api-root", + "https://x.mil/cwms-data", + "--api-key", + "old", + "--office", + "SWT", + ], + ) + result = runner.invoke(env_group, ["setup", "myenv", "--api-key", "new"]) + assert result.exit_code == 0, result.output + stored = load_env("myenv") + assert stored["CDA_API_KEY"] == "new" + assert stored["CDA_API_ROOT"] == "https://x.mil/cwms-data" + assert stored["OFFICE"] == "SWT" + + +def test_setup_uppercases_office(isolated_envs): + runner = CliRunner() + runner.invoke( + env_group, + ["setup", "myenv", "--api-root", "https://x", "--office", "swt"], + ) + assert load_env("myenv")["OFFICE"] == "SWT" + + +# ---------- env show ---------- + + +def test_show_empty_still_shows_builtins(isolated_envs): + runner = CliRunner() + result = runner.invoke(env_group, ["show"]) + assert result.exit_code == 0 + assert "cwbi-prod (built-in)" in result.output + + +def test_show_lists_envs_and_redacts_key(isolated_envs): + save_env( + "withkey", + {"CDA_API_ROOT": "https://x", "CDA_API_KEY": "supersecret", "OFFICE": "SWT"}, + ) + save_env("nokey", {"CDA_API_ROOT": "https://y"}) + runner = CliRunner() + result = runner.invoke(env_group, ["show"]) + assert result.exit_code == 0 + assert "withkey" in result.output + assert "nokey" in result.output + assert "has API key" in result.output + assert "no API key" in result.output + assert "supersecret" not in result.output + + +def test_show_marks_current_env(isolated_envs, monkeypatch): + save_env("active", {"CDA_API_ROOT": "https://x"}) + monkeypatch.setenv("ENVIRONMENT", "active") + runner = CliRunner() + result = runner.invoke(env_group, ["show"]) + assert "Current environment:" in result.output + assert "* active" in result.output + + +# ---------- env show --check ---------- + +from cwmscli.commands.env import _check_env + + +def _fake_check(result_map): + """Return a _check_env replacement that returns canned results by API root.""" + default = { + "reachable": True, + "latency_ms": 42, + "auth": "skipped", + "error": None, + } + + def _check(env_config): + root = env_config.get("CDA_API_ROOT", "") + return result_map.get(root, default) + + return _check + + +def test_show_check_reachable(isolated_envs, monkeypatch): + save_env("demo", {"CDA_API_ROOT": "https://x.mil/cwms-data"}) + monkeypatch.setattr( + "cwmscli.commands.env._check_env", + _fake_check({}), + ) + runner = CliRunner() + result = runner.invoke(env_group, ["show", "--check"]) + assert result.exit_code == 0 + assert "reachable" in result.output + assert "42ms)" in result.output + + +def test_show_check_unreachable(isolated_envs, monkeypatch): + save_env("demo", {"CDA_API_ROOT": "https://x.mil/cwms-data"}) + monkeypatch.setattr( + "cwmscli.commands.env._check_env", + _fake_check( + { + "https://x.mil/cwms-data": { + "reachable": False, + "latency_ms": None, + "auth": "skipped", + "error": "Connection refused", + }, + } + ), + ) + runner = CliRunner() + result = runner.invoke(env_group, ["show", "--check"]) + assert result.exit_code == 0 + assert "unreachable" in result.output + assert "Connection refused" in result.output + + +def test_show_check_auth_ok(isolated_envs, monkeypatch): + save_env( + "demo", + {"CDA_API_ROOT": "https://x.mil/cwms-data", "CDA_API_KEY": "apikey mykey"}, + ) + monkeypatch.setattr( + "cwmscli.commands.env._check_env", + _fake_check( + { + "https://x.mil/cwms-data": { + "reachable": True, + "latency_ms": 100, + "auth": "ok", + "error": None, + }, + } + ), + ) + runner = CliRunner() + result = runner.invoke(env_group, ["show", "--check"]) + assert result.exit_code == 0 + assert "authenticated" in result.output + + +def test_show_check_auth_failed(isolated_envs, monkeypatch): + save_env( + "demo", + {"CDA_API_ROOT": "https://x.mil/cwms-data", "CDA_API_KEY": "apikey bad"}, + ) + monkeypatch.setattr( + "cwmscli.commands.env._check_env", + _fake_check( + { + "https://x.mil/cwms-data": { + "reachable": True, + "latency_ms": 100, + "auth": "failed", + "error": None, + }, + } + ), + ) + runner = CliRunner() + result = runner.invoke(env_group, ["show", "--check"]) + assert result.exit_code == 0 + assert "auth failed" in result.output + + +def test_show_check_no_key_skips_auth(isolated_envs, monkeypatch): + save_env("demo", {"CDA_API_ROOT": "https://x.mil/cwms-data"}) + monkeypatch.setattr( + "cwmscli.commands.env._check_env", + _fake_check({}), + ) + runner = CliRunner() + result = runner.invoke(env_group, ["show", "--check"]) + assert result.exit_code == 0 + assert "authenticated" not in result.output + assert "auth failed" not in result.output + + +def test_show_without_check_flag_unchanged(isolated_envs): + save_env("demo", {"CDA_API_ROOT": "https://x.mil/cwms-data"}) + runner = CliRunner() + result = runner.invoke(env_group, ["show"]) + assert result.exit_code == 0 + assert "Connect:" not in result.output + assert "Auth:" not in result.output + + +# ---------- env delete ---------- + + +def test_delete_with_yes_flag(isolated_envs): + save_env("doomed", {"a": "1"}) + runner = CliRunner() + result = runner.invoke(env_group, ["delete", "doomed", "--yes"]) + assert result.exit_code == 0 + assert load_env("doomed") is None + + +def test_delete_confirmation_cancel(isolated_envs): + save_env("safe", {"a": "1"}) + runner = CliRunner() + result = runner.invoke(env_group, ["delete", "safe"], input="n\n") + assert result.exit_code == 0 + assert "Cancelled" in result.output + assert load_env("safe") == {"a": "1"} + + +def test_delete_missing_env_errors(isolated_envs): + runner = CliRunner() + result = runner.invoke(env_group, ["delete", "ghost", "--yes"]) + assert result.exit_code != 0 + assert "not found" in result.output + + +# ---------- quoting helpers ---------- + + +@pytest.mark.parametrize( + "value", + [ + "plain", + "with spaces", + "with'apostrophe", + 'with"quote', + "with\\backslash", + "with$dollar", + "with`backtick", + "with'multi'apos", + "a'b\"c\\d$e", + ], +) +def test_bash_quoting_roundtrip(value): + """Eval the bash output and confirm the value comes back unchanged.""" + line = f"export X={_quote_bash(value)}" + out = subprocess.check_output(["bash", "-c", f'{line}; printf %s "$X"']) + assert out.decode() == value + + +@pytest.mark.parametrize( + "value", + [ + "plain", + "with spaces", + 'with"quote', + "with\\backslash", + "with$dollar", + ], +) +def test_dotenv_quoting_roundtrip(value): + """Parse dotenv with python-dotenv-style rules and confirm round-trip.""" + line = f"X={_quote_dotenv(value)}" + # Strip the wrapping double quotes, then unescape \\ and \" + raw = line.split("=", 1)[1] + assert raw.startswith('"') and raw.endswith('"') + inner = raw[1:-1] + decoded = inner.replace('\\"', '"').replace("\\\\", "\\") + assert decoded == value + + +def test_powershell_quote_doubles_apostrophes(): + assert _quote_powershell("a'b") == "'a''b'" + assert _quote_powershell("plain") == "'plain'" + + +def test_cmd_quote_doubles_percent_and_quote(): + assert _quote_cmd("100%") == "100%%" + assert _quote_cmd('a"b') == 'a""b' + + +def test_fish_quote_escapes_apostrophe_and_backslash(): + assert _quote_fish("a'b") == "'a\\'b'" + assert _quote_fish("a\\b") == "'a\\\\b'" + assert _quote_fish("plain") == "'plain'" + + +# ---------- env export ---------- + + +@pytest.fixture +def env_with_key(isolated_envs): + save_env( + "demo", + { + "ENVIRONMENT": "demo", + "CDA_API_ROOT": "https://x.mil/cwms-data", + "CDA_API_KEY": "secret", + "OFFICE": "SWT", + }, + ) + return "demo" + + +def test_export_dotenv_format(env_with_key): + runner = CliRunner() + result = runner.invoke(env_group, ["export", env_with_key, "--show-key"]) + assert result.exit_code == 0 + out = result.output + assert 'CDA_API_ROOT="https://x.mil/cwms-data"' in out + assert 'CDA_API_KEY="secret"' in out + assert 'OFFICE="SWT"' in out + + +def test_export_bash_format_evaluates(env_with_key): + runner = CliRunner() + result = runner.invoke( + env_group, ["export", env_with_key, "--format", "bash", "--show-key"] + ) + assert result.exit_code == 0 + script = ( + result.output + '\nprintf "%s|%s|%s" "$CDA_API_ROOT" "$CDA_API_KEY" "$OFFICE"' + ) + out = subprocess.check_output(["bash", "-c", script]).decode() + assert out == "https://x.mil/cwms-data|secret|SWT" + + +def test_export_powershell_format_structure(env_with_key): + runner = CliRunner() + result = runner.invoke( + env_group, ["export", env_with_key, "--format", "powershell", "--show-key"] + ) + assert result.exit_code == 0 + assert "$env:CDA_API_KEY = 'secret'" in result.output + + +def test_export_cmd_format_structure(env_with_key): + runner = CliRunner() + result = runner.invoke( + env_group, ["export", env_with_key, "--format", "cmd", "--show-key"] + ) + assert result.exit_code == 0 + # @ prefix prevents cmd from echoing the line (which would leak the key) + assert '@set "CDA_API_KEY=secret"' in result.output + + +def test_export_cmd_format_all_lines_silenced(env_with_key): + runner = CliRunner() + result = runner.invoke( + env_group, ["export", env_with_key, "--format", "cmd", "--show-key"] + ) + assert result.exit_code == 0 + for line in result.output.strip().splitlines(): + if line.startswith("set "): + pytest.fail(f"unsuppressed set line would echo the value: {line!r}") + + +def test_export_fish_format_structure(env_with_key): + runner = CliRunner() + result = runner.invoke( + env_group, ["export", env_with_key, "--format", "fish", "--show-key"] + ) + assert result.exit_code == 0 + assert "set -gx CDA_API_KEY 'secret'" in result.output + + +def test_export_tty_refusal(env_with_key, monkeypatch): + monkeypatch.setattr("cwmscli.commands.env._stdout_is_tty", lambda: True) + runner = CliRunner() + result = runner.invoke(env_group, ["export", env_with_key]) + assert result.exit_code != 0 + assert "into your current shell" in result.output + assert "--format bash" in result.output + assert "Out-String | Invoke-Expression" in result.output + assert "secret" not in result.output + + +def test_export_help_uses_detected_shell(env_with_key, monkeypatch): + monkeypatch.setattr("cwmscli.commands.env._stdout_is_tty", lambda: True) + monkeypatch.setattr("cwmscli.commands.env._detect_shell_kind", lambda: "powershell") + runner = CliRunner() + result = runner.invoke(env_group, ["export", env_with_key]) + assert result.exit_code != 0 + # Powershell recipe is the primary; bash etc. listed as "other shells" + first_line = result.output.splitlines()[0] + assert "powershell detected" in first_line + assert "For other shells:" in result.output + + +def test_export_show_key_overrides_tty_refusal(env_with_key, monkeypatch): + monkeypatch.setattr("cwmscli.commands.env._stdout_is_tty", lambda: True) + runner = CliRunner() + result = runner.invoke(env_group, ["export", env_with_key, "--show-key"]) + assert result.exit_code == 0 + assert "secret" in result.output + + +def test_export_no_key_omits_key(env_with_key): + runner = CliRunner() + result = runner.invoke(env_group, ["export", env_with_key, "--no-key"]) + assert result.exit_code == 0 + assert "CDA_API_KEY" not in result.output + assert "CDA_API_ROOT" in result.output + + +def test_export_no_key_allows_tty_when_no_key(isolated_envs, monkeypatch): + save_env("nokey", {"CDA_API_ROOT": "https://x", "OFFICE": "SWT"}) + monkeypatch.setattr("cwmscli.commands.env._stdout_is_tty", lambda: True) + runner = CliRunner() + result = runner.invoke(env_group, ["export", "nokey"]) + assert result.exit_code == 0 + assert "https://x" in result.output + + +def test_export_output_writes_0600(env_with_key, tmp_path): + out = tmp_path / "out.env" + runner = CliRunner() + result = runner.invoke(env_group, ["export", env_with_key, "--output", str(out)]) + assert result.exit_code == 0, result.output + assert out.exists() + if sys.platform != "win32": + assert stat.S_IMODE(os.stat(out).st_mode) == 0o600 + contents = out.read_text() + assert 'CDA_API_KEY="secret"' in contents + + +def test_export_output_dotenv_prints_gitignore_reminder(env_with_key, tmp_path): + out = tmp_path / ".env" + runner = CliRunner() + result = runner.invoke(env_group, ["export", env_with_key, "--output", str(out)]) + assert result.exit_code == 0 + assert "gitignore" in result.output + + +def test_export_missing_env_errors(isolated_envs): + runner = CliRunner() + result = runner.invoke(env_group, ["export", "ghost"]) + assert result.exit_code != 0 + assert "not found" in result.output + + +def test_format_env_sorts_keys(): + out = _format_env({"B": "2", "A": "1"}, "dotenv") + assert out.splitlines() == ['A="1"', 'B="2"'] diff --git a/tests/load/test_source_target_env.py b/tests/load/test_source_target_env.py new file mode 100644 index 0000000..f6b52e7 --- /dev/null +++ b/tests/load/test_source_target_env.py @@ -0,0 +1,255 @@ +"""Tests for --source-env / --target-env resolution in load commands.""" + +import pytest +from click.testing import CliRunner + +from cwmscli.load.location.location import location as location_group +from cwmscli.utils.env_store import save_env + + +@pytest.fixture +def isolated_envs(monkeypatch, tmp_path): + """Redirect env storage to a tmp dir for each test.""" + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + monkeypatch.setenv("APPDATA", str(tmp_path)) + monkeypatch.delenv("ENVIRONMENT", raising=False) + monkeypatch.delenv("CDA_SOURCE_URL", raising=False) + monkeypatch.delenv("CDA_TARGET_URL", raising=False) + monkeypatch.delenv("CDA_SOURCE_OFFICE", raising=False) + monkeypatch.delenv("CDA_API_KEY", raising=False) + return tmp_path + + +@pytest.fixture +def capture_load(monkeypatch): + """Monkeypatch the inner load function to capture kwargs instead of calling CDA.""" + captured = {} + + def fake_load(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr("cwmscli.load.location.location_ids.load_locations", fake_load) + monkeypatch.setattr("cwmscli.utils.get_saved_login_token", lambda *a, **kw: None) + return captured + + +def _invoke(args): + return CliRunner().invoke(location_group, ["ids-all"] + args) + + +# ---------- resolution ---------- + + +def test_source_env_resolves_cda_and_office(isolated_envs, capture_load): + save_env( + "prod", + { + "CDA_API_ROOT": "https://prod.mil/cwms-data", + "OFFICE": "SWT", + }, + ) + result = _invoke( + [ + "--source-env", + "prod", + "--target-cda", + "http://localhost:8082/cwms-data", + ] + ) + assert result.exit_code == 0, result.output + assert capture_load["source_cda"] == "https://prod.mil/cwms-data" + assert capture_load["source_office"] == "SWT" + + +def test_target_env_resolves_cda_and_api_key(isolated_envs, capture_load): + save_env( + "local", + { + "CDA_API_ROOT": "http://localhost:8082/cwms-data", + "CDA_API_KEY": "apikey abc123", + }, + ) + result = _invoke( + [ + "--source-cda", + "https://prod.mil/cwms-data", + "--source-office", + "SWT", + "--target-env", + "local", + ] + ) + assert result.exit_code == 0, result.output + assert capture_load["target_cda"] == "http://localhost:8082/cwms-data" + assert capture_load["target_api_key"] == "apikey abc123" + + +def test_both_envs_resolve(isolated_envs, capture_load): + save_env( + "prod", + { + "CDA_API_ROOT": "https://prod.mil/cwms-data", + "OFFICE": "SWT", + }, + ) + save_env( + "local", + { + "CDA_API_ROOT": "http://localhost:8082/cwms-data", + "CDA_API_KEY": "apikey abc123", + }, + ) + result = _invoke(["--source-env", "prod", "--target-env", "local"]) + assert result.exit_code == 0, result.output + assert capture_load["source_cda"] == "https://prod.mil/cwms-data" + assert capture_load["source_office"] == "SWT" + assert capture_load["target_cda"] == "http://localhost:8082/cwms-data" + assert capture_load["target_api_key"] == "apikey abc123" + + +# ---------- explicit override ---------- + + +def test_explicit_source_office_overrides_env(isolated_envs, capture_load): + save_env( + "prod", + { + "CDA_API_ROOT": "https://prod.mil/cwms-data", + "OFFICE": "SWT", + }, + ) + result = _invoke( + [ + "--source-env", + "prod", + "--source-office", + "LRD", + "--target-cda", + "http://localhost:8082/cwms-data", + ] + ) + assert result.exit_code == 0, result.output + assert capture_load["source_office"] == "LRD" + + +# ---------- mutual exclusivity ---------- + + +def test_source_env_and_source_cda_mutually_exclusive(isolated_envs): + save_env("prod", {"CDA_API_ROOT": "https://prod.mil/cwms-data", "OFFICE": "SWT"}) + result = _invoke( + [ + "--source-env", + "prod", + "--source-cda", + "https://other.mil/cwms-data", + "--source-office", + "SWT", + ] + ) + assert result.exit_code != 0 + assert "mutually exclusive" in result.output + + +def test_target_env_and_target_cda_mutually_exclusive(isolated_envs): + save_env("local", {"CDA_API_ROOT": "http://localhost/cwms-data"}) + result = _invoke( + [ + "--source-cda", + "https://prod.mil/cwms-data", + "--source-office", + "SWT", + "--target-env", + "local", + "--target-cda", + "http://other:8082/cwms-data", + ] + ) + assert result.exit_code != 0 + assert "mutually exclusive" in result.output + + +def test_source_env_and_source_csv_mutually_exclusive(isolated_envs, tmp_path): + save_env("prod", {"CDA_API_ROOT": "https://prod.mil/cwms-data", "OFFICE": "SWT"}) + csv = tmp_path / "in.csv" + csv.write_text("name,office-id,active\nLOC_A,SWT,True\n") + result = _invoke( + [ + "--source-env", + "prod", + "--source-csv", + str(csv), + "--target-cda", + "http://localhost:8082/cwms-data", + ] + ) + assert result.exit_code != 0 + assert "mutually exclusive" in result.output + + +# ---------- error paths ---------- + + +def test_source_env_not_found(isolated_envs): + result = _invoke( + [ + "--source-env", + "nonexistent", + "--target-cda", + "http://localhost:8082/cwms-data", + ] + ) + assert result.exit_code != 0 + assert "not found" in result.output + + +def test_target_env_not_found(isolated_envs): + result = _invoke( + [ + "--source-cda", + "https://prod.mil/cwms-data", + "--source-office", + "SWT", + "--target-env", + "nonexistent", + ] + ) + assert result.exit_code != 0 + assert "not found" in result.output + + +def test_source_env_missing_api_root(isolated_envs): + save_env("broken", {"OFFICE": "SWT"}) + result = _invoke( + [ + "--source-env", + "broken", + "--target-cda", + "http://localhost:8082/cwms-data", + ] + ) + assert result.exit_code != 0 + assert "CDA_API_ROOT" in result.output + + +# ---------- kwargs don't leak ---------- + + +def test_env_kwargs_not_passed_to_command(isolated_envs, capture_load): + save_env( + "prod", + { + "CDA_API_ROOT": "https://prod.mil/cwms-data", + "OFFICE": "SWT", + }, + ) + save_env( + "local", + { + "CDA_API_ROOT": "http://localhost:8082/cwms-data", + }, + ) + result = _invoke(["--source-env", "prod", "--target-env", "local"]) + assert result.exit_code == 0, result.output + assert "source_env" not in capture_load + assert "target_env" not in capture_load diff --git a/tests/utils/test_paths.py b/tests/utils/test_paths.py new file mode 100644 index 0000000..a072d98 --- /dev/null +++ b/tests/utils/test_paths.py @@ -0,0 +1,33 @@ +"""Tests for the shared per-user config-dir helper.""" + +from pathlib import Path + +from cwmscli.utils.auth import default_token_file +from cwmscli.utils.env_store import envs_dir +from cwmscli.utils.paths import config_dir + + +def test_config_dir_uses_xdg_when_set(monkeypatch, tmp_path): + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + assert config_dir() == tmp_path / "cwms-cli" + assert config_dir("envs") == tmp_path / "cwms-cli" / "envs" + + +def test_config_dir_falls_back_to_home_config(monkeypatch): + monkeypatch.delenv("XDG_CONFIG_HOME", raising=False) + assert config_dir("auth") == Path("~/.config").expanduser() / "cwms-cli" / "auth" + + +def test_config_dir_create_makes_directory(monkeypatch, tmp_path): + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + path = config_dir("envs", create=True) + assert path.is_dir() + + +def test_auth_and_env_share_one_root(monkeypatch, tmp_path): + """Saved logins and named environments must live under the same root.""" + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + token = default_token_file("federation-eams") + envs = envs_dir() + assert token.parent.parent == envs.parent + assert token.parent.parent == tmp_path / "cwms-cli"