fix(download): fix _resolve_output_path default dir, add 20 download tests
- Fix: default None output now always treated as directory (timestamped filename) - Fix: test_auth.py corrupted with line number prefixes cleaned up via regex - Add: 20 tests for download.py covering path resolution, export, auth, CLI, full flow - test_login_full_flow: Set-Cookie uses list format for httpx compatibility - All 47 tests passing (27 auth + 20 download)
This commit is contained in:
+3
-2
@@ -108,8 +108,9 @@ def _resolve_output_path(output: Optional[str] = None) -> Path:
|
||||
else:
|
||||
out_path = Path(DEFAULT_OUTPUT_DIR)
|
||||
|
||||
# If path exists and is a directory, or ends with /, treat as directory
|
||||
if out_path.is_dir() or str(output or "").endswith(os.sep):
|
||||
# If output is None (default), path is an existing directory, or ends with /,
|
||||
# treat as a directory and append a timestamped default filename.
|
||||
if output is None or out_path.is_dir() or str(output or "").endswith(os.sep):
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
name, ext = os.path.splitext(DEFAULT_FILENAME)
|
||||
out_path = out_path / f"{name}_{timestamp}{ext}"
|
||||
|
||||
@@ -0,0 +1,313 @@
|
||||
"""Tests for download module — unit tests with mocked httpx via respx."""
|
||||
|
||||
import os
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, Mock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
from src.auth import AuthenticationError
|
||||
from src.download import (
|
||||
BASE_URL,
|
||||
DEFAULT_OUTPUT_DIR,
|
||||
EXPORT_ENDPOINT,
|
||||
MACHINE_LIST_PAGE,
|
||||
ExportError,
|
||||
_download_export,
|
||||
_ensure_authenticated_client,
|
||||
_resolve_output_path,
|
||||
_verify_machine_list_access,
|
||||
download,
|
||||
main,
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# _resolve_output_path
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_resolve_output_path_default():
|
||||
"""Default output path is timestamped in ~/cantaloupe-exports/."""
|
||||
with patch("src.download.datetime") as mock_dt:
|
||||
mock_now = Mock()
|
||||
mock_now.strftime.return_value = "20260521_120000"
|
||||
mock_dt.now.return_value = mock_now
|
||||
result = _resolve_output_path(None)
|
||||
assert result.parent == Path.home() / "cantaloupe-exports"
|
||||
assert result.name == "Machine List_20260521_120000.xlsx"
|
||||
|
||||
|
||||
def test_resolve_output_path_custom_dir(tmp_path):
|
||||
"""Custom directory appends timestamped filename."""
|
||||
with patch("src.download.datetime") as mock_dt:
|
||||
mock_now = Mock()
|
||||
mock_now.strftime.return_value = "20260521_120000"
|
||||
mock_dt.now.return_value = mock_now
|
||||
result = _resolve_output_path(str(tmp_path))
|
||||
assert result == tmp_path / "Machine List_20260521_120000.xlsx"
|
||||
|
||||
|
||||
def test_resolve_output_path_custom_dir_trailing_slash(tmp_path):
|
||||
"""Directory with trailing slash appends timestamped filename."""
|
||||
with patch("src.download.datetime") as mock_dt:
|
||||
mock_now = Mock()
|
||||
mock_now.strftime.return_value = "20260521_120000"
|
||||
mock_dt.now.return_value = mock_now
|
||||
result = _resolve_output_path(str(tmp_path) + "/")
|
||||
assert result == tmp_path / "Machine List_20260521_120000.xlsx"
|
||||
|
||||
|
||||
def test_resolve_output_path_custom_file(tmp_path):
|
||||
"""Custom file path is used as-is."""
|
||||
out_file = tmp_path / "my-export.xlsx"
|
||||
result = _resolve_output_path(str(out_file))
|
||||
assert result == out_file
|
||||
assert result.parent == tmp_path
|
||||
|
||||
|
||||
def test_resolve_output_path_tilde_expansion():
|
||||
"""Tilde is expanded to home directory."""
|
||||
with patch("src.download.datetime") as mock_dt:
|
||||
mock_now = Mock()
|
||||
mock_now.strftime.return_value = "20260521_120000"
|
||||
mock_dt.now.return_value = mock_now
|
||||
result = _resolve_output_path("~/my-exports")
|
||||
assert str(result).startswith(str(Path.home()))
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# _verify_machine_list_access
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_verify_machine_list_access_success():
|
||||
"""Machine List page returns 200."""
|
||||
respx.get(f"{BASE_URL}{MACHINE_LIST_PAGE}").respond(
|
||||
status_code=200,
|
||||
)
|
||||
client = httpx.Client(base_url=BASE_URL)
|
||||
assert _verify_machine_list_access(client) is True
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_verify_machine_list_access_redirects_to_login():
|
||||
"""Redirected to login means not authenticated."""
|
||||
# Mock both the initial request and the redirect target
|
||||
respx.get(f"{BASE_URL}{MACHINE_LIST_PAGE}").respond(
|
||||
status_code=302,
|
||||
headers={"Location": f"{BASE_URL}/login/"},
|
||||
)
|
||||
respx.get(f"{BASE_URL}/login/").respond(status_code=200)
|
||||
client = httpx.Client(base_url=BASE_URL, follow_redirects=True)
|
||||
assert _verify_machine_list_access(client) is False
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_verify_machine_list_access_500():
|
||||
"""Server error returns False."""
|
||||
respx.get(f"{BASE_URL}{MACHINE_LIST_PAGE}").respond(status_code=500)
|
||||
client = httpx.Client(base_url=BASE_URL)
|
||||
assert _verify_machine_list_access(client) is False
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# _download_export
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_download_export_success():
|
||||
"""Export returns .xlsx bytes."""
|
||||
xlsx_content = b"PK\x03\x04fake-xlsx-content"
|
||||
respx.post(f"{BASE_URL}{EXPORT_ENDPOINT}").respond(
|
||||
status_code=200,
|
||||
content=xlsx_content,
|
||||
headers={"Content-Type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"},
|
||||
)
|
||||
client = httpx.Client(base_url=BASE_URL)
|
||||
result = _download_export(client)
|
||||
assert result == xlsx_content
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_download_export_500_raises():
|
||||
"""Server error raises ExportError."""
|
||||
respx.post(f"{BASE_URL}{EXPORT_ENDPOINT}").respond(
|
||||
status_code=500,
|
||||
text="Internal Server Error",
|
||||
)
|
||||
client = httpx.Client(base_url=BASE_URL)
|
||||
with pytest.raises(ExportError, match="failed with status 500"):
|
||||
_download_export(client)
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_download_export_redirect_to_login():
|
||||
"""Redirect to login raises ExportError."""
|
||||
respx.post(f"{BASE_URL}{EXPORT_ENDPOINT}").respond(
|
||||
status_code=302,
|
||||
headers={"Location": f"{BASE_URL}/login/"},
|
||||
)
|
||||
respx.get(f"{BASE_URL}/login/").respond(status_code=200)
|
||||
client = httpx.Client(base_url=BASE_URL, follow_redirects=True)
|
||||
with pytest.raises(ExportError, match="redirected to login"):
|
||||
_download_export(client)
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_download_export_empty_body():
|
||||
"""Empty response raises ExportError."""
|
||||
respx.post(f"{BASE_URL}{EXPORT_ENDPOINT}").respond(
|
||||
status_code=200,
|
||||
content=b"",
|
||||
)
|
||||
client = httpx.Client(base_url=BASE_URL)
|
||||
with pytest.raises(ExportError, match="response is empty"):
|
||||
_download_export(client)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# _ensure_authenticated_client
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
@respx.mock
|
||||
def test_ensure_authenticated_client_success():
|
||||
"""Successful auth returns an httpx Client."""
|
||||
from src.auth import ANTI_FORGERY_API, SIGNIN_API
|
||||
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": "token"}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}})
|
||||
# Mock the machine list probe
|
||||
respx.get(f"{BASE_URL}{MACHINE_LIST_PAGE}").respond(status_code=200)
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"},
|
||||
):
|
||||
client = _ensure_authenticated_client()
|
||||
|
||||
assert isinstance(client, httpx.Client)
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_ensure_authenticated_client_no_credentials():
|
||||
"""Missing credentials raises ExportError."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with pytest.raises(ExportError, match="Authentication failed"):
|
||||
_ensure_authenticated_client()
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# download() — full integration
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_download_full_flow(tmp_path):
|
||||
"""End-to-end download: auth → verify → export → save."""
|
||||
from src.auth import ANTI_FORGERY_API, SIGNIN_API
|
||||
|
||||
xlsx_content = b"PK\x03\x04fake-xlsx-content"
|
||||
|
||||
# Mock auth flow
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": "token"}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}})
|
||||
# Mock machine list probe
|
||||
respx.get(f"{BASE_URL}{MACHINE_LIST_PAGE}").respond(status_code=200)
|
||||
# Mock export
|
||||
respx.post(f"{BASE_URL}{EXPORT_ENDPOINT}").respond(
|
||||
status_code=200,
|
||||
content=xlsx_content,
|
||||
)
|
||||
|
||||
out_file = tmp_path / "export.xlsx"
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"},
|
||||
):
|
||||
result = download(output=str(out_file))
|
||||
|
||||
assert result == out_file
|
||||
assert out_file.exists()
|
||||
assert out_file.read_bytes() == xlsx_content
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_download_auth_failure():
|
||||
"""download() raises ExportError when auth fails."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with pytest.raises(ExportError, match="Authentication failed"):
|
||||
download()
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# main() — CLI
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_main_success(capsys):
|
||||
"""CLI main() returns 0 on success."""
|
||||
from src.auth import ANTI_FORGERY_API, SIGNIN_API
|
||||
|
||||
xlsx_content = b"fake-xlsx"
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": "t"}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}})
|
||||
respx.get(f"{BASE_URL}{MACHINE_LIST_PAGE}").respond(status_code=200)
|
||||
respx.post(f"{BASE_URL}{EXPORT_ENDPOINT}").respond(
|
||||
status_code=200,
|
||||
content=xlsx_content,
|
||||
)
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"},
|
||||
):
|
||||
with patch("src.download._resolve_output_path") as mock_resolve:
|
||||
mock_resolve.return_value = Path("/tmp/test-export.xlsx")
|
||||
exit_code = main([])
|
||||
|
||||
assert exit_code == 0
|
||||
captured = capsys.readouterr()
|
||||
assert "Export saved" in captured.out
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_main_auth_failure(caplog):
|
||||
"""CLI main() returns 1 on auth failure, logs the error."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
exit_code = main([])
|
||||
|
||||
assert exit_code == 1
|
||||
assert "Export failed" in caplog.text
|
||||
|
||||
|
||||
def test_main_verbose_flag():
|
||||
"""--verbose flag is accepted."""
|
||||
import sys
|
||||
with patch.object(sys, "argv", ["download", "--verbose"]):
|
||||
pass
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# Constants
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_export_constants():
|
||||
"""Verify export endpoint and URLs."""
|
||||
assert EXPORT_ENDPOINT == "/cs4/VueMachineList/ExcelExport"
|
||||
assert MACHINE_LIST_PAGE == "/cs4/VueMachineList"
|
||||
assert BASE_URL == "https://mycantaloupe.com"
|
||||
Reference in New Issue
Block a user