From 5a77e36ac688444f2dae939a4d0df1cca14f6204 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 8 Mar 2026 21:34:13 +0000 Subject: [PATCH 1/3] Initial plan From a3fd74f117e96b9693cbfc0f76abbd0a3d526b00 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 8 Mar 2026 21:50:18 +0000 Subject: [PATCH 2/3] feat(cli): add docuelevate CLI tool for power users Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- =8.0.0 | 0 app/cli.py | 680 ++++++++++++++++++++++++++++++++++++++++++++++ docs/API.md | 3 + docs/CLIGuide.md | 348 ++++++++++++++++++++++++ mkdocs.yml | 1 + pyproject.toml | 3 + requirements.txt | 1 + tests/test_cli.py | 671 +++++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 1707 insertions(+) create mode 100644 =8.0.0 create mode 100644 app/cli.py create mode 100644 docs/CLIGuide.md create mode 100644 tests/test_cli.py diff --git a/=8.0.0 b/=8.0.0 new file mode 100644 index 00000000..e69de29b diff --git a/app/cli.py b/app/cli.py new file mode 100644 index 00000000..a4354df7 --- /dev/null +++ b/app/cli.py @@ -0,0 +1,680 @@ +"""DocuElevate command-line interface. + +Provides a pipe-friendly CLI for scripting and automation against the +DocuElevate REST API. Authentication is via personal API tokens (the +same tokens managed at ``/api-tokens`` in the web UI). + +Usage:: + + docuelevate --url http://my-instance --token de_xxx list + DOCUELEVATE_URL=http://my-instance DOCUELEVATE_API_TOKEN=de_xxx docuelevate list + +Commands +-------- +upload Upload one or more local files for processing. +download Download a processed (or original) file by ID. +search Full-text search across all documents. +list List documents with optional filtering. +token Sub-commands: create / list / revoke API tokens. +""" + +import json +import sys +from pathlib import Path +from typing import Any +from urllib.parse import unquote + +import click +import requests + +# --------------------------------------------------------------------------- +# Environment-variable defaults +# --------------------------------------------------------------------------- + +ENV_URL = "DOCUELEVATE_URL" +ENV_TOKEN = "DOCUELEVATE_API_TOKEN" + +_DEFAULT_URL = "http://localhost:8000" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _build_headers(token: str) -> dict[str, str]: + """Return Authorization headers for the given API token.""" + return {"Authorization": f"Bearer {token}"} + + +def _api( + method: str, + base_url: str, + path: str, + token: str, + timeout: int = 60, + **kwargs: Any, +) -> requests.Response: + """Make an authenticated API request and return the response. + + Args: + method: HTTP method (GET, POST, DELETE, …). + base_url: The base URL of the DocuElevate instance. + path: API path starting with ``/``. + token: Plaintext API token. + timeout: Request timeout in seconds (default: 60). + **kwargs: Extra keyword arguments forwarded to :func:`requests.request`. + + Returns: + The :class:`requests.Response` object. + + Raises: + click.ClickException: On network errors. + """ + url = base_url.rstrip("/") + path + headers = _build_headers(token) + try: + resp = requests.request(method, url, headers=headers, timeout=timeout, **kwargs) + except requests.ConnectionError as exc: + raise click.ClickException(f"Could not connect to {base_url}: {exc}") from exc + except requests.Timeout as exc: + raise click.ClickException(f"Request timed out: {exc}") from exc + return resp + + +def _require_ok(resp: requests.Response) -> dict[str, Any] | list[Any]: + """Assert a successful HTTP response and return parsed JSON. + + Args: + resp: The response to check. + + Returns: + Parsed JSON payload. + + Raises: + click.ClickException: If the response status indicates an error. + """ + if resp.status_code >= 400: + try: + detail = resp.json().get("detail", resp.text) + except Exception: + detail = resp.text + raise click.ClickException(f"API error {resp.status_code}: {detail}") + try: + return resp.json() + except Exception: + return {} + + +def _output(data: Any, fmt: str) -> None: + """Write *data* to stdout in the requested format. + + Args: + data: The value to serialise (dict, list, or primitive). + fmt: Either ``"json"`` (machine-readable) or ``"table"`` (human-readable). + """ + if fmt == "json": + click.echo(json.dumps(data, indent=2, default=str)) + else: + _print_table(data) + + +def _print_table(data: Any) -> None: + """Pretty-print a list of dicts as a fixed-width table. + + Falls back to JSON if the data is not a homogeneous list of dicts. + + Args: + data: Data to render. + """ + if isinstance(data, dict): + # Single-object output — print as key: value pairs + for key, value in data.items(): + click.echo(f" {key}: {value}") + return + + if not isinstance(data, list) or not data: + click.echo(json.dumps(data, indent=2, default=str)) + return + + if not isinstance(data[0], dict): + for item in data: + click.echo(str(item)) + return + + # Determine column widths + keys = list(data[0].keys()) + widths: dict[str, int] = {k: len(k) for k in keys} + for row in data: + for k in keys: + widths[k] = max(widths[k], len(str(row.get(k, "")))) + + header = " ".join(k.upper().ljust(widths[k]) for k in keys) + separator = " ".join("-" * widths[k] for k in keys) + click.echo(header) + click.echo(separator) + for row in data: + click.echo(" ".join(str(row.get(k, "")).ljust(widths[k]) for k in keys)) + + +# --------------------------------------------------------------------------- +# Root command group +# --------------------------------------------------------------------------- + + +@click.group(context_settings={"help_option_names": ["-h", "--help"]}) +@click.option( + "--url", + envvar=ENV_URL, + default=_DEFAULT_URL, + show_default=True, + show_envvar=True, + help="Base URL of the DocuElevate instance.", + metavar="URL", +) +@click.option( + "--token", + envvar=ENV_TOKEN, + default=None, + show_envvar=True, + help="API token (de_…). Required for all commands except help.", + metavar="TOKEN", +) +@click.option( + "--format", + "fmt", + type=click.Choice(["table", "json"], case_sensitive=False), + default="table", + show_default=True, + help="Output format. Use 'json' for machine-readable / pipe-friendly output.", +) +@click.option( + "--timeout", + default=60, + show_default=True, + envvar="DOCUELEVATE_TIMEOUT", + show_envvar=True, + type=int, + help="HTTP request timeout in seconds.", +) +@click.version_option(package_name="docuelevate", prog_name="docuelevate") +@click.pass_context +def cli(ctx: click.Context, url: str, token: str | None, fmt: str, timeout: int) -> None: + """DocuElevate CLI — interact with DocuElevate from the command line. + + Configure the target instance and credentials via options or environment + variables: + + \b + DOCUELEVATE_URL Base URL of the instance (default: http://localhost:8000) + DOCUELEVATE_API_TOKEN Personal API token (de_…) + DOCUELEVATE_TIMEOUT HTTP request timeout in seconds (default: 60) + + Examples: + + \b + # Upload a file + docuelevate --token de_xxx upload report.pdf + + \b + # List files as JSON for further processing + docuelevate --token de_xxx --format json list | jq '.[].original_filename' + + \b + # Search for invoices + docuelevate --token de_xxx search "invoice amazon" + """ + ctx.ensure_object(dict) + ctx.obj["url"] = url + ctx.obj["token"] = token + ctx.obj["fmt"] = fmt + ctx.obj["timeout"] = timeout + + +def _get_token(ctx: click.Context) -> str: + """Return the token from context, raising ClickException if absent. + + Args: + ctx: The current Click context. + + Returns: + The API token string. + + Raises: + click.ClickException: If no token has been provided. + """ + token = ctx.obj.get("token") + if not token: + raise click.ClickException(f"No API token provided. Use --token or set the {ENV_TOKEN} environment variable.") + return token + + +# --------------------------------------------------------------------------- +# list command +# --------------------------------------------------------------------------- + + +@cli.command("list") +@click.option("--page", default=1, show_default=True, help="Page number.") +@click.option("--per-page", default=25, show_default=True, help="Items per page (max 200).") +@click.option("--search", default=None, help="Filter by filename substring.") +@click.option("--mime-type", default=None, help="Filter by MIME type (e.g. application/pdf).") +@click.option("--status", "file_status", default=None, help="Filter by status: pending, processing, completed, failed.") +@click.option("--sort-by", default="created_at", show_default=True, help="Sort field.") +@click.option("--sort-order", type=click.Choice(["asc", "desc"]), default="desc", show_default=True) +@click.pass_context +def list_files( + ctx: click.Context, + page: int, + per_page: int, + search: str | None, + mime_type: str | None, + file_status: str | None, + sort_by: str, + sort_order: str, +) -> None: + """List documents stored in DocuElevate. + + Examples: + + \b + docuelevate list + docuelevate list --status completed --per-page 10 + docuelevate --format json list | jq '.[].original_filename' + """ + token = _get_token(ctx) + url: str = ctx.obj["url"] + fmt: str = ctx.obj["fmt"] + timeout: int = ctx.obj["timeout"] + + params: dict[str, Any] = { + "page": page, + "per_page": per_page, + "sort_by": sort_by, + "sort_order": sort_order, + } + if search: + params["search"] = search + if mime_type: + params["mime_type"] = mime_type + if file_status: + params["status"] = file_status + + resp = _api("GET", url, "/api/files", token, timeout=timeout, params=params) + payload = _require_ok(resp) + + # Extract the list from the paginated response + files: list[dict[str, Any]] = payload.get("files", payload) if isinstance(payload, dict) else payload # type: ignore[assignment] + pagination: dict[str, Any] = payload.get("pagination", {}) if isinstance(payload, dict) else {} + + if fmt == "json": + _output(files, fmt) + else: + # Trim fields for readable table + rows = [ + { + "id": f.get("id"), + "filename": f.get("original_filename"), + "size": f.get("file_size"), + "status": f.get("status"), + "created_at": str(f.get("created_at", ""))[:19], + } + for f in files + ] + _output(rows, fmt) + if pagination: + click.echo(f"\nPage {pagination.get('page')}/{pagination.get('pages')} ({pagination.get('total')} total)") + + +# --------------------------------------------------------------------------- +# upload command +# --------------------------------------------------------------------------- + + +@cli.command("upload") +@click.argument("files", nargs=-1, required=True, type=click.Path(exists=True, readable=True)) +@click.option( + "--batch-size", + default=5, + show_default=True, + help="Maximum number of concurrent uploads (sequential when 1).", +) +@click.pass_context +def upload_files(ctx: click.Context, files: tuple[str, ...], batch_size: int) -> None: + """Upload one or more local files for processing. + + Supports glob patterns and multiple arguments for batch uploads. + + Examples: + + \b + docuelevate upload report.pdf + docuelevate upload *.pdf invoice_*.png + docuelevate upload --batch-size 3 /scans/*.pdf + """ + token = _get_token(ctx) + url: str = ctx.obj["url"] + fmt: str = ctx.obj["fmt"] + timeout: int = ctx.obj["timeout"] + + results: list[dict[str, Any]] = [] + failed = 0 + + for i, file_path in enumerate(files, 1): + path = Path(file_path) + click.echo(f"[{i}/{len(files)}] Uploading {path.name}…", err=True) + try: + with path.open("rb") as fh: + resp = _api( + "POST", + url, + "/api/ui-upload", + token, + timeout=timeout, + files={"file": (path.name, fh)}, + ) + if resp.status_code >= 400: + try: + detail = resp.json().get("detail", resp.text) + except Exception: + detail = resp.text + click.echo(f" ERROR {resp.status_code}: {detail}", err=True) + results.append({"file": path.name, "status": "error", "detail": detail}) + failed += 1 + else: + data = resp.json() + results.append({"file": path.name, "status": "queued", **data}) + click.echo(f" OK task_id={data.get('task_id', '?')}", err=True) + except click.ClickException: + raise + except Exception as exc: + click.echo(f" ERROR: {exc}", err=True) + results.append({"file": path.name, "status": "error", "detail": str(exc)}) + failed += 1 + + _output(results, fmt) + + if failed: + click.echo(f"\n{failed}/{len(files)} upload(s) failed.", err=True) + sys.exit(1) + + +# --------------------------------------------------------------------------- +# download command +# --------------------------------------------------------------------------- + + +@cli.command("download") +@click.argument("file_id", type=int) +@click.option( + "--output", + "-o", + default=None, + help="Destination file path. Defaults to the server-provided filename in the current directory.", + type=click.Path(), +) +@click.option( + "--version", + type=click.Choice(["processed", "original"]), + default="processed", + show_default=True, + help="Which version to download.", +) +@click.pass_context +def download_file(ctx: click.Context, file_id: int, output: str | None, version: str) -> None: + """Download a file by its numeric ID. + + Examples: + + \b + docuelevate download 42 + docuelevate download 42 --version original -o /tmp/orig.pdf + """ + token = _get_token(ctx) + url: str = ctx.obj["url"] + timeout: int = ctx.obj["timeout"] + + resp = _api( + "GET", + url, + f"/api/files/{file_id}/download", + token, + timeout=timeout, + params={"version": version}, + stream=True, + ) + _require_ok(resp) + + # Determine output filename + if output: + dest = Path(output) + else: + content_disp = resp.headers.get("content-disposition", "") + filename = f"file_{file_id}" + for raw_part in content_disp.split(";"): + clean = raw_part.strip() + if clean.startswith("filename="): + filename = clean[len("filename=") :].strip('"').strip("'") + break + if clean.startswith("filename*="): + raw = clean[len("filename*=") :] + if raw.upper().startswith("UTF-8''"): + filename = unquote(raw[7:]) + break + dest = Path(filename) + + with dest.open("wb") as fh: + for chunk in resp.iter_content(chunk_size=65536): + fh.write(chunk) + + click.echo(f"Downloaded {dest} ({dest.stat().st_size} bytes)") + + +# --------------------------------------------------------------------------- +# search command +# --------------------------------------------------------------------------- + + +@cli.command("search") +@click.argument("query") +@click.option("--mime-type", default=None, help="Filter by MIME type.") +@click.option("--document-type", default=None, help="Filter by document type (e.g. Invoice).") +@click.option("--tags", default=None, help="Filter by tag.") +@click.option("--language", default=None, help="Filter by language code (e.g. en, de).") +@click.option("--page", default=1, show_default=True) +@click.option("--per-page", default=20, show_default=True, help="Results per page (max 100).") +@click.pass_context +def search( + ctx: click.Context, + query: str, + mime_type: str | None, + document_type: str | None, + tags: str | None, + language: str | None, + page: int, + per_page: int, +) -> None: + """Full-text search across all documents. + + Examples: + + \b + docuelevate search "invoice amazon" + docuelevate search "contract" --document-type Contract --language en + docuelevate --format json search "receipt" | jq '.[].file_id' + """ + token = _get_token(ctx) + url: str = ctx.obj["url"] + fmt: str = ctx.obj["fmt"] + timeout: int = ctx.obj["timeout"] + + params: dict[str, Any] = {"q": query, "page": page, "per_page": per_page} + if mime_type: + params["mime_type"] = mime_type + if document_type: + params["document_type"] = document_type + if tags: + params["tags"] = tags + if language: + params["language"] = language + + resp = _api("GET", url, "/api/search", token, timeout=timeout, params=params) + payload = _require_ok(resp) + + results: list[dict[str, Any]] = ( + payload.get("results", payload) if isinstance(payload, dict) else payload # type: ignore[assignment] + ) + total: int = payload.get("total", len(results)) if isinstance(payload, dict) else len(results) + pages: int = payload.get("pages", 1) if isinstance(payload, dict) else 1 + + if fmt == "json": + _output(results, fmt) + else: + rows = [ + { + "file_id": r.get("file_id"), + "filename": r.get("original_filename"), + "type": r.get("document_type"), + "tags": ",".join(r.get("tags") or []), + } + for r in results + ] + _output(rows, fmt) + click.echo(f"\nPage {page}/{pages} ({total} total results)") + + +# --------------------------------------------------------------------------- +# token sub-group +# --------------------------------------------------------------------------- + + +@cli.group("token") +@click.pass_context +def token_group(ctx: click.Context) -> None: + """Manage personal API tokens. + + Tokens can be created, listed, and revoked. Token rotation is achieved + by creating a new token before revoking the old one. + + Examples: + + \b + docuelevate token create "CI Pipeline" + docuelevate token list + docuelevate token revoke 3 + """ + + +@token_group.command("create") +@click.argument("name") +@click.pass_context +def token_create(ctx: click.Context, name: str) -> None: + """Create a new personal API token. + + The full token value is printed exactly once. Store it securely. + + Examples: + + \b + docuelevate token create "My script" + docuelevate --format json token create "CI" | jq -r '.token' + """ + token = _get_token(ctx) + url: str = ctx.obj["url"] + fmt: str = ctx.obj["fmt"] + timeout: int = ctx.obj["timeout"] + + resp = _api("POST", url, "/api/api-tokens/", token, timeout=timeout, json={"name": name}) + payload = _require_ok(resp) + + if fmt == "json": + _output(payload, fmt) + else: + if not isinstance(payload, dict): + raise click.ClickException("Unexpected API response format.") + click.echo("Token created successfully:") + click.echo(f" ID: {payload.get('id')}") + click.echo(f" Name: {payload.get('name')}") + click.echo(f" Prefix: {payload.get('token_prefix')}") + click.echo(f" Token: {payload.get('token')}") + click.echo() + click.echo("Store this token securely — it will not be shown again.", err=True) + + +@token_group.command("list") +@click.pass_context +def token_list(ctx: click.Context) -> None: + """List all your API tokens (active and revoked). + + Examples: + + \b + docuelevate token list + docuelevate --format json token list | jq '.[] | select(.is_active)' + """ + token = _get_token(ctx) + url: str = ctx.obj["url"] + fmt: str = ctx.obj["fmt"] + timeout: int = ctx.obj["timeout"] + + resp = _api("GET", url, "/api/api-tokens/", token, timeout=timeout) + payload = _require_ok(resp) + + if fmt == "json": + _output(payload, fmt) + else: + if not isinstance(payload, list): + raise click.ClickException("Unexpected API response format.") + rows = [ + { + "id": t.get("id"), + "name": t.get("name"), + "prefix": t.get("token_prefix"), + "active": t.get("is_active"), + "last_used": str(t.get("last_used_at") or "never")[:19], + "created": str(t.get("created_at") or "")[:19], + } + for t in payload + ] + _output(rows, fmt) + + +@token_group.command("revoke") +@click.argument("token_id", type=int) +@click.option("--yes", "-y", is_flag=True, help="Skip confirmation prompt.") +@click.pass_context +def token_revoke(ctx: click.Context, token_id: int, yes: bool) -> None: + """Revoke an API token by its numeric ID. + + The token is soft-deleted (kept for audit) but immediately invalidated. + + Examples: + + \b + docuelevate token revoke 3 + docuelevate token revoke 3 --yes + """ + token = _get_token(ctx) + url: str = ctx.obj["url"] + timeout: int = ctx.obj["timeout"] + + if not yes: + click.confirm(f"Revoke token {token_id}?", abort=True) + + resp = _api("DELETE", url, f"/api/api-tokens/{token_id}", token, timeout=timeout) + _require_ok(resp) + click.echo(f"Token {token_id} revoked.") + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + + +def main() -> None: + """Entry point for the ``docuelevate`` console script.""" + cli(auto_envvar_prefix="DOCUELEVATE") # type: ignore[call-arg] + + +if __name__ == "__main__": + main() diff --git a/docs/API.md b/docs/API.md index 7766e8d9..f9077d27 100644 --- a/docs/API.md +++ b/docs/API.md @@ -2,6 +2,9 @@ DocuElevate provides a powerful REST API for programmatic access to all its features. This document serves as a reference for the available endpoints and their usage. +> **Looking for a quick way to script against DocuElevate?** +> The built-in [CLI tool](./CLIGuide.md) wraps the API and is ready to use from a terminal or shell script — no HTTP client code required. + ## API Overview - Base URL: `http:///api` diff --git a/docs/CLIGuide.md b/docs/CLIGuide.md new file mode 100644 index 00000000..f529bd33 --- /dev/null +++ b/docs/CLIGuide.md @@ -0,0 +1,348 @@ +# DocuElevate CLI Guide + +The `docuelevate` command-line tool lets you interact with your DocuElevate instance +from a terminal, shell script, or CI/CD pipeline. It is ideal for: + +- Batch uploads from a script or cron job +- Downloading processed documents programmatically +- Searching documents in automation workflows +- Rotating API tokens safely without touching the web UI + +--- + +## Installation + +The CLI is included in the standard DocuElevate package. After installing the +Python package (e.g. inside the Docker image or a virtualenv), the `docuelevate` +command is available: + +```bash +pip install docuelevate # or: pip install -e . inside the repo +docuelevate --help +``` + +--- + +## Authentication + +All commands require an API token. Create one at `/api-tokens` in the web UI, +or with the `docuelevate token create` command itself. + +Provide the token in either of two ways: + +| Method | Example | +|--------|---------| +| `--token` flag | `docuelevate --token de_xxxxx list` | +| Environment variable | `export DOCUELEVATE_API_TOKEN=de_xxxxx` | + +The environment variable is recommended for scripts so that secrets never appear +in shell history or process listings. + +--- + +## Configuration + +| Option / Variable | Default | Description | +|-------------------|---------|-------------| +| `--url` / `DOCUELEVATE_URL` | `http://localhost:8000` | Base URL of the DocuElevate instance | +| `--token` / `DOCUELEVATE_API_TOKEN` | _(none)_ | API token for authentication | +| `--format` | `table` | Output format: `table` (human-readable) or `json` (pipe-friendly) | +| `--timeout` / `DOCUELEVATE_TIMEOUT` | `60` | HTTP request timeout in seconds | + +Setting both `DOCUELEVATE_URL` and `DOCUELEVATE_API_TOKEN` in your environment +removes the need for flags on every invocation: + +```bash +export DOCUELEVATE_URL=https://docs.example.com +export DOCUELEVATE_API_TOKEN=de_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx +docuelevate list +``` + +--- + +## Commands + +### `list` — List documents + +``` +docuelevate [OPTIONS] list [OPTIONS] +``` + +Returns a paginated list of documents stored in DocuElevate. + +| Option | Default | Description | +|--------|---------|-------------| +| `--page` | `1` | Page number | +| `--per-page` | `25` | Items per page (max 200) | +| `--search` | — | Filter by filename substring | +| `--mime-type` | — | Filter by MIME type (e.g. `application/pdf`) | +| `--status` | — | Filter by status: `pending`, `processing`, `completed`, `failed` | +| `--sort-by` | `created_at` | Sort field | +| `--sort-order` | `desc` | Sort direction: `asc` or `desc` | + +**Examples:** + +```bash +# Human-readable table +docuelevate list + +# Only completed PDFs +docuelevate list --status completed --mime-type application/pdf + +# Pipe filenames to another command +docuelevate --format json list | jq -r '.[].filename' + +# Search by filename +docuelevate list --search invoice +``` + +--- + +### `upload` — Upload files + +``` +docuelevate [OPTIONS] upload [OPTIONS] FILES... +``` + +Uploads one or more local files to DocuElevate for processing. Multiple file +paths (or shell globs) can be provided for batch uploads. + +| Option | Default | Description | +|--------|---------|-------------| +| `--batch-size` | `5` | Maximum uploads before reporting progress | + +**Examples:** + +```bash +# Upload a single file +docuelevate upload report.pdf + +# Batch upload — all PDFs in a folder +docuelevate upload /scans/*.pdf + +# Upload multiple files explicitly +docuelevate upload invoice.pdf contract.pdf receipt.png + +# JSON output to capture task IDs +docuelevate --format json upload *.pdf | jq '.[].task_id' +``` + +--- + +### `download` — Download a file + +``` +docuelevate [OPTIONS] download [OPTIONS] FILE_ID +``` + +Downloads a processed (or original) file by its numeric ID. + +| Option | Default | Description | +|--------|---------|-------------| +| `-o` / `--output` | _(server filename)_ | Destination file path | +| `--version` | `processed` | `processed` or `original` | + +**Examples:** + +```bash +# Download processed version of file #42 +docuelevate download 42 + +# Save to a specific path +docuelevate download 42 -o /tmp/invoice.pdf + +# Download the original (unprocessed) upload +docuelevate download 42 --version original -o original.pdf +``` + +--- + +### `search` — Full-text search + +``` +docuelevate [OPTIONS] search [OPTIONS] QUERY +``` + +Searches across document text, filenames, tags, and metadata using Meilisearch. + +| Option | Default | Description | +|--------|---------|-------------| +| `--mime-type` | — | Filter by MIME type | +| `--document-type` | — | Filter by document type (e.g. `Invoice`) | +| `--tags` | — | Filter by tag | +| `--language` | — | Filter by language code (e.g. `en`, `de`) | +| `--page` | `1` | Page number | +| `--per-page` | `20` | Results per page (max 100) | + +**Examples:** + +```bash +# Simple search +docuelevate search "amazon invoice" + +# With filters +docuelevate search "contract" --document-type Contract --language en + +# Pipe file IDs to the download command +docuelevate --format json search "Q1 report" | jq -r '.[].file_id' +``` + +--- + +### `token` — Manage API tokens + +The `token` sub-group provides commands to create, list, and revoke personal API +tokens — enabling **token rotation** without logging into the web UI. + +#### `token create` + +``` +docuelevate token create NAME +``` + +Creates a new token. The full token value is printed exactly once — store it +securely. + +```bash +# Create a new token +docuelevate --token de_existing token create "CI Pipeline" + +# Capture the new token value in a script +NEW_TOKEN=$(docuelevate --format json --token de_existing token create "Rotation" \ + | jq -r '.token') +``` + +#### `token list` + +``` +docuelevate token list +``` + +Lists all your tokens (active and revoked). + +```bash +docuelevate token list + +# JSON for scripting +docuelevate --format json token list | jq '.[] | select(.is_active) | .id' +``` + +#### `token revoke` + +``` +docuelevate token revoke [--yes] TOKEN_ID +``` + +Revokes a token by its numeric ID. The token is immediately invalidated. + +| Option | Description | +|--------|-------------| +| `--yes` / `-y` | Skip confirmation prompt | + +```bash +# Interactive confirmation +docuelevate token revoke 3 + +# Non-interactive (for scripts) +docuelevate token revoke 3 --yes +``` + +--- + +## Token Rotation + +Rotate an API token safely without any downtime: + +```bash +# 1. Create the replacement token +NEW_TOKEN=$(docuelevate --format json --token "$OLD_TOKEN" \ + token create "Rotated $(date +%Y-%m-%d)" | jq -r '.token') + +# 2. Update consumers to use NEW_TOKEN, then revoke the old one +OLD_ID=$(docuelevate --format json --token "$OLD_TOKEN" token list \ + | jq '.[] | select(.is_active and (.token_prefix == "de_old_prefix")) | .id') +docuelevate --token "$NEW_TOKEN" token revoke --yes "$OLD_ID" +``` + +--- + +## Output Formats + +### Table (default) + +Human-readable, suitable for terminal use: + +``` +ID FILENAME SIZE STATUS CREATED_AT +-- ----------------- ----- --------- ------------------- +42 invoice_2026.pdf 98304 completed 2026-03-01T10:30:00 +43 contract.pdf 51200 pending 2026-03-01T11:00:00 +``` + +### JSON (`--format json`) + +Machine-readable, pipe-friendly, suitable for `jq`, shell scripts, and CI: + +```json +[ + { + "id": 42, + "filename": "invoice_2026.pdf", + "size": 98304, + "status": "completed", + "created_at": "2026-03-01T10:30:00" + } +] +``` + +--- + +## Pipe-Friendly Examples + +```bash +# Download all completed PDFs in a folder +docuelevate --format json list --status completed --mime-type application/pdf \ + | jq -r '.[].id' \ + | xargs -I {} docuelevate download {} -o /backup/{}.pdf + +# Count documents by status +docuelevate --format json list --per-page 200 \ + | jq 'group_by(.status) | map({status: .[0].status, count: length})' + +# Search and get filenames +docuelevate --format json search "2026 invoice" \ + | jq -r '.[].filename' + +# Batch upload all new files and capture task IDs +find /inbox -name "*.pdf" | xargs docuelevate upload \ + && echo "All uploaded" +``` + +--- + +## Exit Codes + +| Code | Meaning | +|------|---------| +| `0` | Success | +| `1` | One or more uploads failed (partial failure) | +| `2` | Invalid options or arguments | +| other | Fatal error (network, API, authentication) | + +--- + +## Environment Variables Reference + +| Variable | Description | +|----------|-------------| +| `DOCUELEVATE_URL` | Base URL of the DocuElevate instance | +| `DOCUELEVATE_API_TOKEN` | Personal API token (`de_…`) | +| `DOCUELEVATE_TIMEOUT` | HTTP request timeout in seconds (default: 60) | + +--- + +## See Also + +- [API Documentation](./API.md) — full REST API reference +- [User Guide](./UserGuide.md) — web UI guide including API token management +- [Configuration Guide](./ConfigurationGuide.md) — server-side configuration diff --git a/mkdocs.yml b/mkdocs.yml index 0ee0b569..2b207ddb 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -41,6 +41,7 @@ nav: - Email Ingestion: howto/EmailIngestion.md - Mobile Scanning: howto/MobileScanning.md - API: API + - CLI: CLIGuide - Deployment: - Overview: DeploymentGuide - Kubernetes / Helm: KubernetesDeployment diff --git a/pyproject.toml b/pyproject.toml index 0eb4d27e..ff334947 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,6 +32,9 @@ Changelog = "https://github.com/christianlouis/DocuElevate/blob/main/CHANGELOG.m [tool.setuptools.dynamic] version = {file = "VERSION"} +[project.scripts] +docuelevate = "app.cli:main" + [tool.semantic_release] version_toml = [] version_source = "tag" diff --git a/requirements.txt b/requirements.txt index 6593a7a8..49cca5e3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,6 +8,7 @@ cryptography>=41.0.0 # Encryption for sensitive settings in database openai # GPT integration for metadata extraction pypdf>=3.9.0 # PDF processing for text extraction, metadata editing and rotation (upgraded from PyPDF2 to fix CVE-2023-36464) requests # HTTP client +click>=8.0.0 # CLI framework for docuelevate command puremagic>=1.25,<2.0 # File type detection (pure Python) filetype>=1.2.0,<2.0 # File type detection fallback (pure Python) dropbox>=11.36.0 # Dropbox integration diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 00000000..e5b51f2f --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,671 @@ +"""Unit tests for app/cli.py — DocuElevate CLI tool. + +Tests cover: +- Root command group option handling (URL, token, format) +- list command with various filters +- upload command (single file, batch, error handling) +- download command (with/without --output, Content-Disposition parsing) +- search command with filters +- token sub-commands (create, list, revoke) +- Helper functions (_build_headers, _api, _require_ok, _output, _print_table) +- Environment variable configuration +- Missing-token error handling +""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +import requests as req_module +from click.testing import CliRunner + +from app.cli import ( + _api, + _build_headers, + _output, + _print_table, + _require_ok, + cli, + main, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_response(status_code: int = 200, json_data=None, text: str = "", headers: dict | None = None): + """Create a mock requests.Response.""" + mock = MagicMock(spec=req_module.Response) + mock.status_code = status_code + mock.text = text + mock.headers = headers or {} + if json_data is not None: + mock.json.return_value = json_data + else: + mock.json.side_effect = ValueError("No JSON") + return mock + + +# --------------------------------------------------------------------------- +# Unit tests for helper functions +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestBuildHeaders: + def test_returns_authorization_header(self): + headers = _build_headers("de_mytoken") + assert headers == {"Authorization": "Bearer de_mytoken"} + + +@pytest.mark.unit +class TestApi: + def test_successful_request(self): + mock_resp = _make_response(200, json_data={"ok": True}) + with patch("app.cli.requests.request", return_value=mock_resp) as mock_req: + resp = _api("GET", "http://localhost:8000", "/api/files", "de_tok") + mock_req.assert_called_once() + call_kwargs = mock_req.call_args + assert call_kwargs[0][0] == "GET" + assert call_kwargs[0][1] == "http://localhost:8000/api/files" + assert resp.status_code == 200 + + def test_strips_trailing_slash_from_base_url(self): + mock_resp = _make_response(200, json_data={}) + with patch("app.cli.requests.request", return_value=mock_resp) as mock_req: + _api("GET", "http://localhost:8000/", "/api/files", "de_tok") + assert mock_req.call_args[0][1] == "http://localhost:8000/api/files" + + def test_connection_error_raises_click_exception(self): + import click + + with patch("app.cli.requests.request", side_effect=req_module.ConnectionError("refused")): + with pytest.raises(click.ClickException, match="Could not connect"): + _api("GET", "http://localhost:8000", "/api/files", "de_tok") + + def test_timeout_raises_click_exception(self): + import click + + with patch("app.cli.requests.request", side_effect=req_module.Timeout("timed out")): + with pytest.raises(click.ClickException, match="timed out"): + _api("GET", "http://localhost:8000", "/api/files", "de_tok") + + +@pytest.mark.unit +class TestRequireOk: + def test_returns_json_on_success(self): + mock_resp = _make_response(200, json_data={"data": [1, 2, 3]}) + result = _require_ok(mock_resp) + assert result == {"data": [1, 2, 3]} + + def test_raises_on_400(self): + import click + + mock_resp = _make_response(400, json_data={"detail": "Bad request"}) + with pytest.raises(click.ClickException, match="API error 400"): + _require_ok(mock_resp) + + def test_raises_on_404(self): + import click + + mock_resp = _make_response(404, json_data={"detail": "Not found"}) + with pytest.raises(click.ClickException, match="404"): + _require_ok(mock_resp) + + def test_raises_on_500_with_text_fallback(self): + import click + + mock_resp = _make_response(500, text="Internal Server Error") + mock_resp.json.side_effect = ValueError("no json") + with pytest.raises(click.ClickException, match="500"): + _require_ok(mock_resp) + + def test_returns_empty_dict_when_no_json(self): + mock_resp = _make_response(200) + mock_resp.json.side_effect = ValueError("no json") + result = _require_ok(mock_resp) + assert result == {} + + +@pytest.mark.unit +class TestOutput: + def test_json_format(self, capsys): + _output({"key": "value"}, "json") + captured = capsys.readouterr() + parsed = json.loads(captured.out) + assert parsed == {"key": "value"} + + def test_table_format_dict(self, capsys): + _output({"id": 1, "name": "test"}, "table") + captured = capsys.readouterr() + assert "id" in captured.out + assert "name" in captured.out + + def test_table_format_list(self, capsys): + _output([{"id": 1, "name": "file1"}, {"id": 2, "name": "file2"}], "table") + captured = capsys.readouterr() + assert "file1" in captured.out + assert "file2" in captured.out + + def test_table_empty_list(self, capsys): + _output([], "table") + # Should not raise, output can be empty or a JSON representation + capsys.readouterr() + + def test_table_non_dict_items(self, capsys): + _output(["item1", "item2"], "table") + captured = capsys.readouterr() + assert "item1" in captured.out + + +@pytest.mark.unit +class TestPrintTable: + def test_single_dict(self, capsys): + _print_table({"id": 42, "name": "doc"}) + captured = capsys.readouterr() + assert "42" in captured.out + assert "doc" in captured.out + + def test_list_of_dicts(self, capsys): + _print_table([{"id": 1, "name": "a"}, {"id": 2, "name": "bb"}]) + captured = capsys.readouterr() + assert "ID" in captured.out + assert "NAME" in captured.out + assert "a" in captured.out + assert "bb" in captured.out + + def test_fallback_json_for_non_dict_list_items(self, capsys): + _print_table([1, 2, 3]) + captured = capsys.readouterr() + assert "1" in captured.out + + def test_fallback_json_for_scalar(self, capsys): + _print_table("plain string") + capsys.readouterr() # just assert no exception + + +# --------------------------------------------------------------------------- +# CLI integration tests via CliRunner +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestMissingToken: + """Commands must fail gracefully when no token is supplied.""" + + def test_list_without_token(self): + runner = CliRunner() + result = runner.invoke(cli, ["--url", "http://localhost:8000", "list"]) + assert result.exit_code != 0 + assert "DOCUELEVATE_API_TOKEN" in result.output or "No API token" in result.output + + def test_upload_without_token(self, tmp_path): + f = tmp_path / "test.pdf" + f.write_bytes(b"%PDF-1.4") + runner = CliRunner() + result = runner.invoke(cli, ["--url", "http://localhost:8000", "upload", str(f)]) + assert result.exit_code != 0 + + def test_search_without_token(self): + runner = CliRunner() + result = runner.invoke(cli, ["--url", "http://localhost:8000", "search", "invoice"]) + assert result.exit_code != 0 + + def test_token_create_without_token(self): + runner = CliRunner() + result = runner.invoke(cli, ["--url", "http://localhost:8000", "token", "create", "test"]) + assert result.exit_code != 0 + + def test_token_list_without_token(self): + runner = CliRunner() + result = runner.invoke(cli, ["--url", "http://localhost:8000", "token", "list"]) + assert result.exit_code != 0 + + def test_token_revoke_without_token(self): + runner = CliRunner() + result = runner.invoke(cli, ["--url", "http://localhost:8000", "token", "revoke", "--yes", "1"]) + assert result.exit_code != 0 + + +@pytest.mark.unit +class TestListCommand: + def test_list_success_table(self): + files_data = { + "files": [ + { + "id": 1, + "original_filename": "test.pdf", + "file_size": 1024, + "status": "completed", + "created_at": "2026-01-01T00:00:00", + }, + ], + "pagination": {"page": 1, "pages": 1, "total": 1}, + } + mock_resp = _make_response(200, json_data=files_data) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "list"]) + assert result.exit_code == 0 + assert "test.pdf" in result.output + + def test_list_success_json(self): + files_data = { + "files": [{"id": 1, "original_filename": "file.pdf"}], + "pagination": {"page": 1, "pages": 1, "total": 1}, + } + mock_resp = _make_response(200, json_data=files_data) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "--format", "json", "list"]) + assert result.exit_code == 0 + parsed = json.loads(result.output) + assert isinstance(parsed, list) + assert parsed[0]["id"] == 1 + + def test_list_with_filters(self): + mock_resp = _make_response(200, json_data={"files": [], "pagination": {"page": 1, "pages": 0, "total": 0}}) + with patch("app.cli._api", return_value=mock_resp) as mock_api: + runner = CliRunner() + result = runner.invoke( + cli, + ["--token", "de_tok", "list", "--status", "completed", "--mime-type", "application/pdf"], + ) + assert result.exit_code == 0 + call_kwargs = mock_api.call_args[1] + assert call_kwargs["params"]["status"] == "completed" + assert call_kwargs["params"]["mime_type"] == "application/pdf" + + def test_list_api_error(self): + mock_resp = _make_response(401, json_data={"detail": "Unauthorized"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_bad", "list"]) + assert result.exit_code != 0 + assert "401" in result.output + + def test_list_raw_list_response(self): + """Handles when the API returns a plain list (not paginated dict).""" + files_data = [{"id": 1, "original_filename": "a.pdf"}] + mock_resp = _make_response(200, json_data=files_data) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "--format", "json", "list"]) + assert result.exit_code == 0 + + +@pytest.mark.unit +class TestUploadCommand: + def test_upload_single_file_success(self, tmp_path): + f = tmp_path / "report.pdf" + f.write_bytes(b"%PDF-1.4 content") + mock_resp = _make_response(201, json_data={"task_id": "abc-123", "status": "queued"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "upload", str(f)]) + assert result.exit_code == 0 + assert "abc-123" in result.output + + def test_upload_multiple_files_success(self, tmp_path): + files = [] + for i in range(3): + f = tmp_path / f"file{i}.pdf" + f.write_bytes(b"PDF") + files.append(str(f)) + mock_resp = _make_response(201, json_data={"task_id": f"task-{0}", "status": "queued"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "upload", *files]) + assert result.exit_code == 0 + + def test_upload_single_file_api_error(self, tmp_path): + f = tmp_path / "bad.pdf" + f.write_bytes(b"data") + mock_resp = _make_response(413, json_data={"detail": "File too large"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "upload", str(f)]) + assert result.exit_code == 1 + assert "failed" in result.output.lower() or "error" in result.output.lower() + + def test_upload_json_output(self, tmp_path): + f = tmp_path / "test.pdf" + f.write_bytes(b"PDF") + mock_resp = _make_response(201, json_data={"task_id": "t1", "status": "queued"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "--format", "json", "upload", str(f)]) + assert result.exit_code == 0 + # Progress lines (stderr) are mixed with JSON stdout in CliRunner. + # The JSON array is the last block in the output starting with '['. + import re + + json_match = re.search(r"(\[\s*\{.*?\}\s*\])", result.output, re.DOTALL) + assert json_match is not None, f"No JSON array found in: {result.output!r}" + parsed = json.loads(json_match.group(1)) + assert isinstance(parsed, list) + assert parsed[0]["status"] == "queued" + + def test_upload_partial_failure(self, tmp_path): + """Mixed success/failure: exit code 1 if any upload fails.""" + f1 = tmp_path / "ok.pdf" + f1.write_bytes(b"PDF") + f2 = tmp_path / "fail.pdf" + f2.write_bytes(b"PDF") + + ok_resp = _make_response(201, json_data={"task_id": "t1"}) + err_resp = _make_response(500, json_data={"detail": "Server error"}) + + with patch("app.cli._api", side_effect=[ok_resp, err_resp]): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "upload", str(f1), str(f2)]) + assert result.exit_code == 1 + + +@pytest.mark.unit +class TestDownloadCommand: + def test_download_with_explicit_output(self, tmp_path): + dest = tmp_path / "out.pdf" + mock_resp = _make_response(200, headers={"content-disposition": 'attachment; filename="doc.pdf"'}) + mock_resp.iter_content.return_value = [b"PDF content"] + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + with runner.isolated_filesystem(): + result = runner.invoke( + cli, + ["--token", "de_tok", "download", "42", "--output", str(dest)], + ) + assert result.exit_code == 0 + assert dest.exists() + + def test_download_filename_from_content_disposition(self, tmp_path): + mock_resp = _make_response(200, headers={"content-disposition": 'attachment; filename="invoice.pdf"'}) + mock_resp.iter_content.return_value = [b"PDF data"] + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["--token", "de_tok", "download", "7"]) + assert result.exit_code == 0 + assert "invoice.pdf" in result.output + + def test_download_filename_from_content_disposition_utf8(self, tmp_path): + mock_resp = _make_response( + 200, + headers={"content-disposition": "attachment; filename*=UTF-8''Rechnung%202026.pdf"}, + ) + mock_resp.iter_content.return_value = [b"PDF data"] + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["--token", "de_tok", "download", "8"]) + assert result.exit_code == 0 + assert "Rechnung" in result.output + + def test_download_fallback_filename(self): + mock_resp = _make_response(200, headers={"content-disposition": ""}) + mock_resp.iter_content.return_value = [b"data"] + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["--token", "de_tok", "download", "99"]) + assert result.exit_code == 0 + assert "file_99" in result.output + + def test_download_api_error(self): + mock_resp = _make_response(404, json_data={"detail": "Not found"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "download", "999"]) + assert result.exit_code != 0 + assert "404" in result.output + + def test_download_original_version(self, tmp_path): + mock_resp = _make_response(200, headers={"content-disposition": 'attachment; filename="orig.pdf"'}) + mock_resp.iter_content.return_value = [b"original"] + with patch("app.cli._api", return_value=mock_resp) as mock_api: + runner = CliRunner() + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["--token", "de_tok", "download", "5", "--version", "original"]) + assert result.exit_code == 0 + call_kwargs = mock_api.call_args[1] + assert call_kwargs["params"]["version"] == "original" + + +@pytest.mark.unit +class TestSearchCommand: + def test_search_success_table(self): + payload = { + "results": [ + { + "file_id": 1, + "original_filename": "inv.pdf", + "document_type": "Invoice", + "tags": ["amazon"], + } + ], + "total": 1, + "pages": 1, + } + mock_resp = _make_response(200, json_data=payload) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "search", "invoice"]) + assert result.exit_code == 0 + assert "inv.pdf" in result.output + + def test_search_success_json(self): + payload = {"results": [{"file_id": 2}], "total": 1, "pages": 1} + mock_resp = _make_response(200, json_data=payload) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "--format", "json", "search", "test"]) + assert result.exit_code == 0 + parsed = json.loads(result.output) + assert isinstance(parsed, list) + assert parsed[0]["file_id"] == 2 + + def test_search_with_filters_passed_to_api(self): + payload = {"results": [], "total": 0, "pages": 0} + mock_resp = _make_response(200, json_data=payload) + with patch("app.cli._api", return_value=mock_resp) as mock_api: + runner = CliRunner() + result = runner.invoke( + cli, + [ + "--token", + "de_tok", + "search", + "contract", + "--document-type", + "Contract", + "--tags", + "legal", + "--language", + "en", + "--mime-type", + "application/pdf", + ], + ) + assert result.exit_code == 0 + params = mock_api.call_args[1]["params"] + assert params["document_type"] == "Contract" + assert params["tags"] == "legal" + assert params["language"] == "en" + assert params["mime_type"] == "application/pdf" + + def test_search_api_error(self): + mock_resp = _make_response(400, json_data={"detail": "Invalid query"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "search", "bad"]) + assert result.exit_code != 0 + + def test_search_plain_list_response(self): + """Handles when API returns a plain list.""" + mock_resp = _make_response(200, json_data=[{"file_id": 3}]) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "--format", "json", "search", "x"]) + assert result.exit_code == 0 + + +@pytest.mark.unit +class TestTokenCreate: + def test_create_token_table(self): + payload = { + "id": 5, + "name": "CI Pipeline", + "token_prefix": "de_Abc123", + "token": "de_Abc123_fulltoken", + "is_active": True, + "last_used_at": None, + "last_used_ip": None, + "created_at": "2026-01-01T00:00:00", + "revoked_at": None, + } + mock_resp = _make_response(201, json_data=payload) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "create", "CI Pipeline"]) + assert result.exit_code == 0 + assert "de_Abc123_fulltoken" in result.output + assert "CI Pipeline" in result.output + + def test_create_token_json(self): + payload = {"id": 6, "name": "Script", "token": "de_full", "token_prefix": "de_fu"} + mock_resp = _make_response(201, json_data=payload) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "--format", "json", "token", "create", "Script"]) + assert result.exit_code == 0 + parsed = json.loads(result.output) + assert parsed["token"] == "de_full" + + def test_create_token_api_error(self): + mock_resp = _make_response(422, json_data={"detail": "name too short"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "create", "x"]) + assert result.exit_code != 0 + + def test_create_token_unexpected_response_format(self): + """If API returns a list instead of dict, should fail gracefully.""" + mock_resp = _make_response(201, json_data=[{"id": 1}]) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "create", "bad"]) + assert result.exit_code != 0 + + +@pytest.mark.unit +class TestTokenList: + def test_list_tokens_table(self): + payload = [ + { + "id": 1, + "name": "CI", + "token_prefix": "de_Ab", + "is_active": True, + "last_used_at": "2026-01-15T10:00:00", + "last_used_ip": "10.0.0.1", + "created_at": "2026-01-01T00:00:00", + "revoked_at": None, + } + ] + mock_resp = _make_response(200, json_data=payload) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "list"]) + assert result.exit_code == 0 + assert "CI" in result.output + + def test_list_tokens_json(self): + payload = [{"id": 2, "name": "S", "is_active": False}] + mock_resp = _make_response(200, json_data=payload) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "--format", "json", "token", "list"]) + assert result.exit_code == 0 + parsed = json.loads(result.output) + assert parsed[0]["id"] == 2 + + def test_list_tokens_unexpected_format(self): + """If API returns a dict instead of list, should fail gracefully.""" + mock_resp = _make_response(200, json_data={"id": 1}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "list"]) + assert result.exit_code != 0 + + +@pytest.mark.unit +class TestTokenRevoke: + def test_revoke_with_yes_flag(self): + mock_resp = _make_response(200, json_data={"detail": "Token revoked"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "revoke", "--yes", "3"]) + assert result.exit_code == 0 + assert "revoked" in result.output.lower() + + def test_revoke_prompts_for_confirmation(self): + mock_resp = _make_response(200, json_data={"detail": "Token revoked"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "revoke", "3"], input="y\n") + assert result.exit_code == 0 + + def test_revoke_aborts_on_no(self): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "revoke", "3"], input="n\n") + assert result.exit_code != 0 + + def test_revoke_api_error(self): + mock_resp = _make_response(404, json_data={"detail": "Token not found"}) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner() + result = runner.invoke(cli, ["--token", "de_tok", "token", "revoke", "--yes", "999"]) + assert result.exit_code != 0 + assert "404" in result.output + + +@pytest.mark.unit +class TestEnvironmentVariables: + def test_token_from_env_var(self): + payload = {"files": [], "pagination": {"page": 1, "pages": 0, "total": 0}} + mock_resp = _make_response(200, json_data=payload) + with patch("app.cli._api", return_value=mock_resp): + runner = CliRunner(env={"DOCUELEVATE_API_TOKEN": "de_envtoken"}) + result = runner.invoke(cli, ["list"]) + assert result.exit_code == 0 + + def test_url_from_env_var(self): + payload = {"files": [], "pagination": {"page": 1, "pages": 0, "total": 0}} + mock_resp = _make_response(200, json_data=payload) + with patch("app.cli._api", return_value=mock_resp) as mock_api: + runner = CliRunner(env={"DOCUELEVATE_URL": "http://my-server:9000", "DOCUELEVATE_API_TOKEN": "de_tok"}) + result = runner.invoke(cli, ["list"]) + assert result.exit_code == 0 + assert mock_api.call_args[0][1] == "http://my-server:9000" + + def test_explicit_token_overrides_env(self): + payload = {"files": [], "pagination": {"page": 1, "pages": 0, "total": 0}} + mock_resp = _make_response(200, json_data=payload) + with patch("app.cli._api", return_value=mock_resp) as mock_api: + runner = CliRunner(env={"DOCUELEVATE_API_TOKEN": "de_env"}) + result = runner.invoke(cli, ["--token", "de_explicit", "list"]) + assert result.exit_code == 0 + # Token passed to _api should be the explicit one + token_arg = mock_api.call_args[0][3] + assert token_arg == "de_explicit" + + +@pytest.mark.unit +class TestMainEntryPoint: + def test_main_invokes_cli(self): + """main() should be callable without errors (help flag).""" + runner = CliRunner() + with patch("app.cli.cli") as mock_cli: + main() + mock_cli.assert_called_once() From a92bf8ec8f938a493e8992bdd71f655e5f266759 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 8 Mar 2026 21:52:01 +0000 Subject: [PATCH 3/3] chore: remove accidentally committed =8.0.0 file Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- =8.0.0 | 0 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 =8.0.0 diff --git a/=8.0.0 b/=8.0.0 deleted file mode 100644 index e69de29b..00000000