feat(cli): add docuelevate CLI tool for power users
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
@@ -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()
|
||||
Reference in New Issue
Block a user