shawn
9c3d3ef7ed
- cantaloupe/ package with click CLI (python -m cantaloupe export) - export command: --output, --headless/--visible, --scheduled, --email, --password - configure command: saves credentials to ~/.cantaloupe.env (600 perms) - cantaloupe/config.py: credential loading from env vars + env file - Updated src/auth.py _get_credentials() to also check ~/.cantaloupe.env - File size, row count (openpyxl), and elapsed time reporting - 39 new CLI/config tests + 4 updated auth tests; 86/86 pass
549 lines
21 KiB
Python
549 lines
21 KiB
Python
"""
|
|
Tests for cantaloupe CLI and config modules.
|
|
|
|
Covers:
|
|
- config module: load_credentials, save_credentials, prompt_credentials
|
|
- cli module: export command, configure command, --scheduled, --headless/--visible
|
|
- auth.py: _get_credentials env-file fallback (integration with config)
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from cantaloupe.cli import cli, _count_excel_rows, _format_size, _resolve_output
|
|
from cantaloupe.config import load_credentials, save_credentials
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────
|
|
# config module
|
|
# ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_load_credentials_from_env():
|
|
"""load_credentials prefers env vars over env file."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "env@test.com", "CANTALOUPE_PASSWORD": "envpass"},
|
|
):
|
|
email, password = load_credentials()
|
|
assert email == "env@test.com"
|
|
assert password == "envpass"
|
|
|
|
|
|
def test_load_credentials_from_file(tmp_path):
|
|
"""load_credentials falls back to env file when env vars are absent."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
env_file.write_text(
|
|
"CANTALOUPE_EMAIL=file@test.com\nCANTALOUPE_PASSWORD=filepass\n"
|
|
)
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch("cantaloupe.config.ENV_FILE", env_file):
|
|
email, password = load_credentials()
|
|
assert email == "file@test.com"
|
|
assert password == "filepass"
|
|
|
|
|
|
def test_load_credentials_missing_both():
|
|
"""load_credentials returns empty strings when nothing is set."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch.object(Path, "exists", return_value=False):
|
|
email, password = load_credentials()
|
|
assert email == ""
|
|
assert password == ""
|
|
|
|
|
|
def test_load_credentials_env_overrides_file(tmp_path):
|
|
"""env vars take priority over the env file."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
env_file.write_text(
|
|
"CANTALOUPE_EMAIL=file@test.com\nCANTALOUPE_PASSWORD=filepass\n"
|
|
)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "env@test.com"},
|
|
):
|
|
with patch("cantaloupe.config.ENV_FILE", env_file):
|
|
email, password = load_credentials()
|
|
assert email == "env@test.com" # env overrides file
|
|
assert password == "filepass" # file fallback for password
|
|
|
|
|
|
def test_load_credentials_file_with_comments_and_quotes(tmp_path):
|
|
"""load_credentials handles comments and quoted values."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
env_file.write_text(
|
|
"# This is a comment\n"
|
|
'CANTALOUPE_EMAIL="quoted@test.com"\n'
|
|
"CANTALOUPE_PASSWORD='singlequoted'\n"
|
|
"# Another comment"
|
|
)
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch("cantaloupe.config.ENV_FILE", env_file):
|
|
email, password = load_credentials()
|
|
assert email == "quoted@test.com"
|
|
assert password == "singlequoted"
|
|
|
|
|
|
def test_save_credentials_creates_file(tmp_path):
|
|
"""save_credentials writes the env file with correct permissions."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
|
|
with patch("cantaloupe.config.ENV_FILE", env_file):
|
|
result = save_credentials("new@test.com", "newpass")
|
|
assert result == env_file
|
|
assert env_file.exists()
|
|
|
|
content = env_file.read_text()
|
|
assert "CANTALOUPE_EMAIL=new@test.com" in content
|
|
assert "CANTALOUPE_PASSWORD=newpass" in content
|
|
|
|
|
|
def test_save_credentials_preserves_other_lines(tmp_path):
|
|
"""save_credentials updates email/password lines, preserves everything else."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
env_file.write_text(
|
|
"# Config file\n"
|
|
"SOME_OTHER_KEY=value\n"
|
|
"CANTALOUPE_EMAIL=old@test.com\n"
|
|
"CANTALOUPE_PASSWORD=oldpass\n"
|
|
)
|
|
|
|
with patch("cantaloupe.config.ENV_FILE", env_file):
|
|
save_credentials("updated@test.com", "updpw")
|
|
content = env_file.read_text()
|
|
assert "CANTALOUPE_EMAIL=updated@test.com" in content
|
|
assert "CANTALOUPE_PASSWORD=updpw" in content
|
|
assert "CANTALOUPE_EMAIL=old@test.com" not in content
|
|
assert "CANTALOUPE_PASSWORD=oldpass" not in content
|
|
assert "SOME_OTHER_KEY=value" in content # preserved
|
|
|
|
|
|
def test_save_credentials_empty_password():
|
|
"""save_credentials handles empty password (allowed)."""
|
|
env_file = Path(tempfile.mkdtemp()) / ".cantaloupe.env"
|
|
|
|
with patch("cantaloupe.config.ENV_FILE", env_file):
|
|
save_credentials("e@test.com", "")
|
|
content = env_file.read_text()
|
|
assert "CANTALOUPE_EMAIL=e@test.com" in content
|
|
assert "CANTALOUPE_PASSWORD=" in content
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────
|
|
# _format_size
|
|
# ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_format_size_bytes():
|
|
assert _format_size(0) == "0.0 B"
|
|
assert _format_size(500) == "500.0 B"
|
|
assert _format_size(1023) == "1023.0 B"
|
|
|
|
|
|
def test_format_size_kb():
|
|
assert _format_size(1024) == "1.0 KB"
|
|
assert _format_size(1536) == "1.5 KB"
|
|
|
|
|
|
def test_format_size_mb():
|
|
assert _format_size(1024 * 1024) == "1.0 MB"
|
|
assert _format_size(int(2.5 * 1024 * 1024)) == "2.5 MB"
|
|
|
|
|
|
def test_format_size_gb():
|
|
assert _format_size(1024 * 1024 * 1024) == "1.0 GB"
|
|
|
|
|
|
def test_format_size_large():
|
|
"""1 TB = 1024 GB."""
|
|
one_tb = 1024 * 1024 * 1024 * 1024
|
|
assert _format_size(one_tb) == "1.0 TB"
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────
|
|
# _count_excel_rows
|
|
# ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_count_excel_rows_no_file():
|
|
"""Returns 0 when file doesn't exist."""
|
|
with patch("openpyxl.load_workbook", side_effect=FileNotFoundError):
|
|
result = _count_excel_rows(Path("/nonexistent.xlsx"))
|
|
assert result == 0
|
|
|
|
|
|
def test_count_excel_rows_empty_sheet():
|
|
"""Returns 0 when sheet has only header row."""
|
|
mock_wb = MagicMock()
|
|
mock_ws = MagicMock()
|
|
mock_ws.max_row = 1 # just header
|
|
mock_wb.active = mock_ws
|
|
mock_wb.close = MagicMock()
|
|
|
|
with patch("openpyxl.load_workbook", return_value=mock_wb):
|
|
result = _count_excel_rows(Path("/fake.xlsx"))
|
|
assert result == 0
|
|
|
|
|
|
def test_count_excel_rows_with_data():
|
|
"""Subtracts header row from total rows."""
|
|
mock_wb = MagicMock()
|
|
mock_ws = MagicMock()
|
|
mock_ws.max_row = 51 # 1 header + 50 data rows
|
|
mock_wb.active = mock_ws
|
|
mock_wb.close = MagicMock()
|
|
|
|
with patch("openpyxl.load_workbook", return_value=mock_wb):
|
|
result = _count_excel_rows(Path("/fake.xlsx"))
|
|
assert result == 50
|
|
|
|
|
|
def test_count_excel_rows_none_active():
|
|
"""Returns 0 when workbook.active is None."""
|
|
mock_wb = MagicMock()
|
|
mock_wb.active = None
|
|
mock_wb.close = MagicMock()
|
|
|
|
with patch("openpyxl.load_workbook", return_value=mock_wb):
|
|
result = _count_excel_rows(Path("/fake.xlsx"))
|
|
assert result == 0
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────
|
|
# _resolve_output
|
|
# ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_resolve_output_interactive_default(tmp_path, monkeypatch):
|
|
"""Interactive mode defaults to cwd."""
|
|
monkeypatch.chdir(tmp_path)
|
|
result = _resolve_output(None, scheduled=False)
|
|
assert result == tmp_path
|
|
|
|
|
|
def test_resolve_output_scheduled_default(tmp_path):
|
|
"""Scheduled mode defaults to ~/cantaloupe-exports."""
|
|
home = tmp_path / "home" / "testuser"
|
|
home.mkdir(parents=True)
|
|
with patch("cantaloupe.cli.Path.home", return_value=home):
|
|
result = _resolve_output(None, scheduled=True)
|
|
assert result == home / "cantaloupe-exports"
|
|
|
|
|
|
def test_resolve_output_explicit_path(tmp_path):
|
|
"""Explicit output path is used as-is."""
|
|
out = tmp_path / "my-exports"
|
|
result = _resolve_output(str(out), scheduled=False)
|
|
assert result == out
|
|
assert out.exists() # created
|
|
|
|
|
|
def test_resolve_output_file_path(tmp_path):
|
|
"""File path with extension — parent is created."""
|
|
out = tmp_path / "deep" / "nested" / "file.xlsx"
|
|
result = _resolve_output(str(out), scheduled=True)
|
|
assert result == out
|
|
assert out.parent.exists()
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────
|
|
# CLI via Click test runner
|
|
# ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def runner():
|
|
return CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_env_file(tmp_path):
|
|
"""Create a temporary env file with test credentials."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
env_file.write_text(
|
|
"CANTALOUPE_EMAIL=cli@test.com\nCANTALOUPE_PASSWORD=clipw\n"
|
|
)
|
|
return env_file
|
|
|
|
|
|
# ── configure command ──
|
|
|
|
|
|
def test_configure_saves_credentials(runner, tmp_path):
|
|
"""configure command writes to ~/.cantaloupe.env."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
with patch("cantaloupe.config.ENV_FILE", env_file):
|
|
result = runner.invoke(
|
|
cli,
|
|
["configure"],
|
|
input="user@example.com\nsecret123\n",
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "Credentials saved" in result.output
|
|
content = env_file.read_text()
|
|
assert "CANTALOUPE_EMAIL=user@example.com" in content
|
|
assert "CANTALOUPE_PASSWORD=secret123" in content
|
|
|
|
|
|
# ── export command ──
|
|
|
|
|
|
def test_export_no_credentials_interactive(runner, tmp_path):
|
|
"""export prompts for credentials when none found."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch.object(Path, "exists", return_value=False):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "test.xlsx"
|
|
mock_path.write_bytes(b"fake xlsx data")
|
|
mock_download.return_value = mock_path
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
["export", "-o", str(tmp_path)],
|
|
input="user@example.com\nsecret123\ny\n",
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert "Cantaloupe credentials not found" in result.output
|
|
|
|
|
|
def test_export_no_credentials_scheduled(runner):
|
|
"""export in scheduled mode exits with error when no credentials."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch.object(Path, "exists", return_value=False):
|
|
result = runner.invoke(cli, ["export", "--scheduled"])
|
|
assert result.exit_code == 1
|
|
assert "ERROR" in result.output or "credentials" in result.output.lower()
|
|
|
|
|
|
def test_export_with_env_vars(runner, tmp_path):
|
|
"""export picks up credentials from environment variables."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "env@test.com", "CANTALOUPE_PASSWORD": "envpw"},
|
|
):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "export.xlsx"
|
|
mock_path.write_bytes(b"xlsx content")
|
|
mock_download.return_value = mock_path
|
|
|
|
result = runner.invoke(cli, ["export", "-o", str(tmp_path)])
|
|
assert result.exit_code == 0
|
|
assert "Export complete" in result.output
|
|
|
|
|
|
def test_export_with_env_file(runner, mock_env_file, tmp_path):
|
|
"""export picks up credentials from ~/.cantaloupe.env."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch("cantaloupe.config.ENV_FILE", mock_env_file):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "export.xlsx"
|
|
mock_path.write_bytes(b"data")
|
|
mock_download.return_value = mock_path
|
|
|
|
result = runner.invoke(cli, ["export", "-o", str(tmp_path)])
|
|
assert result.exit_code == 0
|
|
assert "Export complete" in result.output
|
|
|
|
|
|
def test_export_with_cli_flags(runner, tmp_path):
|
|
"""--email and --password CLI flags override everything else."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch.object(Path, "exists", return_value=False):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "export.xlsx"
|
|
mock_path.write_bytes(b"data")
|
|
mock_download.return_value = mock_path
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"export",
|
|
"--email", "flag@test.com",
|
|
"--password", "flagpw",
|
|
"-o", str(tmp_path),
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "Export complete" in result.output
|
|
|
|
|
|
def test_export_scheduled_flag(runner, tmp_path):
|
|
"""export --scheduled runs in non-interactive mode."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "cron@test.com", "CANTALOUPE_PASSWORD": "cronpw"},
|
|
):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "export.xlsx"
|
|
mock_path.write_bytes(b"data")
|
|
mock_download.return_value = mock_path
|
|
|
|
result = runner.invoke(cli, ["export", "--scheduled", "-o", str(tmp_path)])
|
|
assert result.exit_code == 0
|
|
assert "[scheduled]" in result.output
|
|
assert "Export complete" in result.output
|
|
|
|
|
|
def test_export_shows_file_stats(runner, tmp_path):
|
|
"""export reports file size and row count on completion."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"},
|
|
):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "export.xlsx"
|
|
mock_path.write_bytes(b"x" * 2048) # exactly 2 KB
|
|
mock_download.return_value = mock_path
|
|
|
|
with patch("cantaloupe.cli._count_excel_rows", return_value=42):
|
|
result = runner.invoke(cli, ["export", "-o", str(tmp_path)])
|
|
assert result.exit_code == 0
|
|
assert "2.0 KB" in result.output
|
|
assert "42" in result.output
|
|
assert "Export complete" in result.output
|
|
|
|
|
|
def test_export_download_error(runner):
|
|
"""export handles ExportError gracefully."""
|
|
from src.download import ExportError
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"},
|
|
):
|
|
with patch("src.download.download", side_effect=ExportError("simulated failure")):
|
|
result = runner.invoke(cli, ["export", "-o", "/tmp"])
|
|
assert result.exit_code == 1
|
|
assert "ERROR" in result.output or "simulated failure" in result.output
|
|
|
|
|
|
def test_export_headless_flag(runner, tmp_path):
|
|
"""--headless and --visible flags are accepted (for Playwright fallback)."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"},
|
|
):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "test.xlsx"
|
|
mock_path.write_bytes(b"fake")
|
|
mock_download.return_value = mock_path
|
|
result = runner.invoke(cli, ["export", "--headless", "-o", str(tmp_path)])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
def test_export_visible_flag(runner, tmp_path):
|
|
"""--visible flag is accepted."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"},
|
|
):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "test.xlsx"
|
|
mock_path.write_bytes(b"fake")
|
|
mock_download.return_value = mock_path
|
|
result = runner.invoke(cli, ["export", "--visible", "-o", str(tmp_path)])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
def test_export_verbose_flag(runner, tmp_path):
|
|
"""-v / --verbose enables debug logging."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"},
|
|
):
|
|
with patch("src.download.download") as mock_download:
|
|
mock_path = tmp_path / "export.xlsx"
|
|
mock_path.write_bytes(b"data")
|
|
mock_download.return_value = mock_path
|
|
|
|
result = runner.invoke(cli, ["-v", "export", "-o", str(tmp_path)])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
def test_export_keyboard_interrupt(runner, tmp_path):
|
|
"""export exits cleanly on Ctrl+C."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"},
|
|
):
|
|
with patch("src.download.download", side_effect=KeyboardInterrupt):
|
|
result = runner.invoke(cli, ["export", "-o", str(tmp_path)])
|
|
assert result.exit_code == 130
|
|
assert "Cancelled" in result.output
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────
|
|
# auth.py: _get_credentials env-file fallback
|
|
# ─────────────────────────────────────────────────────────────────
|
|
|
|
from src.auth import _get_credentials, AuthenticationError
|
|
|
|
|
|
def test_get_credentials_from_env_file(tmp_path):
|
|
"""_get_credentials reads from ~/.cantaloupe.env when env vars are absent."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
env_file.write_text(
|
|
"CANTALOUPE_EMAIL=file@auth.com\nCANTALOUPE_PASSWORD=filepw\n"
|
|
)
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch("src.auth.os.path.expanduser", return_value=str(env_file)):
|
|
email, password = _get_credentials()
|
|
assert email == "file@auth.com"
|
|
assert password == "filepw"
|
|
|
|
|
|
def test_get_credentials_env_overrides_file(tmp_path):
|
|
"""env vars take priority over the env file in auth.py."""
|
|
env_file = tmp_path / ".cantaloupe.env"
|
|
env_file.write_text(
|
|
"CANTALOUPE_EMAIL=file@auth.com\nCANTALOUPE_PASSWORD=filepw\n"
|
|
)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CANTALOUPE_EMAIL": "env@auth.com", "CANTALOUPE_PASSWORD": "envpw"},
|
|
):
|
|
with patch("src.auth.os.path.expanduser", return_value=str(env_file)):
|
|
email, password = _get_credentials()
|
|
assert email == "env@auth.com"
|
|
assert password == "envpw"
|
|
|
|
|
|
def test_get_credentials_missing_both_still_raises():
|
|
"""_get_credentials raises AuthenticationError when nothing is set."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch("src.auth.os.path.expanduser", return_value="/nonexistent/.cantaloupe.env"):
|
|
with pytest.raises(AuthenticationError, match="Credentials not found"):
|
|
_get_credentials()
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────
|
|
# CLI version
|
|
# ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_cli_version(runner):
|
|
"""--version prints the version string."""
|
|
result = runner.invoke(cli, ["--version"])
|
|
assert result.exit_code == 0
|
|
assert "0.1.0" in result.output
|
|
|
|
|
|
def test_cli_help(runner):
|
|
"""--help prints usage info."""
|
|
result = runner.invoke(cli, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "Cantaloupe Downloader" in result.output
|
|
assert "export" in result.output
|
|
assert "configure" in result.output
|