Chris Coutinho
2e7774654b
Split the monolithic management.py (1988 lines) into 4 focused modules: - management.py: Server status, user sessions, shared helpers (~520 lines) - passwords.py: App password provisioning for BasicAuth mode (~300 lines) - webhooks.py: Webhook registration management (~290 lines) - visualization.py: Search and PDF preview endpoints (~810 lines) Backward compatibility maintained via __init__.py re-exports. Updated test imports to use new module paths. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
717 lines
25 KiB
Python
717 lines
25 KiB
Python
"""
|
|
Unit tests for Management API PDF preview endpoint.
|
|
|
|
Tests the /api/v1/pdf-preview endpoint focusing on:
|
|
- Parameter validation (file_path, page, scale)
|
|
- OAuth token validation
|
|
- PDF rendering with PyMuPDF
|
|
- Error handling (file not found, invalid page, etc.)
|
|
"""
|
|
|
|
import base64
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from starlette.applications import Starlette
|
|
from starlette.routing import Route
|
|
from starlette.testclient import TestClient
|
|
|
|
from nextcloud_mcp_server.api.visualization import get_pdf_preview
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
def create_test_app():
|
|
"""Create a test Starlette app with the PDF preview endpoint."""
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/api/v1/pdf-preview", get_pdf_preview, methods=["GET"]),
|
|
]
|
|
)
|
|
# Set up OAuth context (required by endpoint)
|
|
app.state.oauth_context = {"config": {"nextcloud_host": "http://localhost:8080"}}
|
|
return app
|
|
|
|
|
|
def create_mock_pdf_bytes():
|
|
"""Create a minimal valid PDF for testing."""
|
|
# Minimal PDF structure that PyMuPDF can parse
|
|
# This is a 1-page PDF with a blank page
|
|
pdf_content = b"""%PDF-1.4
|
|
1 0 obj
|
|
<< /Type /Catalog /Pages 2 0 R >>
|
|
endobj
|
|
2 0 obj
|
|
<< /Type /Pages /Kids [3 0 R] /Count 1 >>
|
|
endobj
|
|
3 0 obj
|
|
<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] >>
|
|
endobj
|
|
xref
|
|
0 4
|
|
0000000000 65535 f
|
|
0000000009 00000 n
|
|
0000000058 00000 n
|
|
0000000115 00000 n
|
|
trailer
|
|
<< /Size 4 /Root 1 0 R >>
|
|
startxref
|
|
196
|
|
%%EOF"""
|
|
return pdf_content
|
|
|
|
|
|
class TestPdfPreviewParameterValidation:
|
|
"""Tests for parameter validation in PDF preview endpoint."""
|
|
|
|
def test_missing_file_path_returns_400(self):
|
|
"""Test that missing file_path parameter returns 400."""
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "file_path" in data["error"].lower()
|
|
|
|
def test_invalid_page_number_returns_400(self):
|
|
"""Test that invalid page number (0 or negative) returns 400."""
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
|
|
# Test page=0
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&page=0",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "page" in data["error"].lower()
|
|
|
|
# Test negative page
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&page=-1",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_invalid_scale_returns_400(self):
|
|
"""Test that scale outside valid range returns 400."""
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
|
|
# Test scale too small
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&scale=0.1",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "scale" in data["error"].lower()
|
|
|
|
# Test scale too large
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&scale=10.0",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_non_numeric_page_returns_400(self):
|
|
"""Test that non-numeric page parameter returns 400."""
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&page=abc",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
|
|
|
|
class TestPdfPreviewAuthentication:
|
|
"""Tests for authentication in PDF preview endpoint."""
|
|
|
|
def test_unauthorized_without_token_returns_401(self):
|
|
"""Test that request without token returns 401."""
|
|
with patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
side_effect=Exception("Invalid token"),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get("/api/v1/pdf-preview?file_path=/test.pdf")
|
|
|
|
assert response.status_code == 401
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
|
|
def test_unauthorized_with_invalid_token_returns_401(self):
|
|
"""Test that request with invalid token returns 401."""
|
|
with patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
side_effect=Exception("Token expired"),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf",
|
|
headers={"Authorization": "Bearer invalid-token"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
|
|
|
|
class TestPdfPreviewRendering:
|
|
"""Tests for PDF rendering functionality."""
|
|
|
|
def test_successful_pdf_render(self):
|
|
"""Test successful PDF page rendering."""
|
|
pdf_bytes = create_mock_pdf_bytes()
|
|
|
|
# Mock the WebDAV client
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(return_value=(pdf_bytes, "application/pdf"))
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&page=1&scale=1.0",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert "image" in data
|
|
assert data["page_number"] == 1
|
|
assert data["total_pages"] == 1
|
|
|
|
# Verify image is valid base64
|
|
try:
|
|
decoded = base64.b64decode(data["image"])
|
|
# PNG magic bytes
|
|
assert decoded[:8] == b"\x89PNG\r\n\x1a\n"
|
|
except Exception as e:
|
|
pytest.fail(f"Image is not valid base64-encoded PNG: {e}")
|
|
|
|
def test_page_out_of_range_returns_400(self):
|
|
"""Test that requesting page beyond total pages returns 400."""
|
|
pdf_bytes = create_mock_pdf_bytes()
|
|
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(return_value=(pdf_bytes, "application/pdf"))
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&page=999",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "page" in data["error"].lower()
|
|
assert "999" in data["error"]
|
|
|
|
def test_file_not_found_returns_404(self):
|
|
"""Test that non-existent file returns 404."""
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(
|
|
side_effect=FileNotFoundError("File not found")
|
|
)
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/nonexistent.pdf",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "not found" in data["error"].lower()
|
|
|
|
def test_default_parameters(self):
|
|
"""Test that default parameters (page=1, scale=2.0) are used."""
|
|
pdf_bytes = create_mock_pdf_bytes()
|
|
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(return_value=(pdf_bytes, "application/pdf"))
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
# Only file_path, no page or scale
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["page_number"] == 1 # Default page
|
|
|
|
|
|
class TestPdfPreviewEdgeCases:
|
|
"""Tests for edge cases in PDF preview endpoint."""
|
|
|
|
def test_url_encoded_file_path(self):
|
|
"""Test that URL-encoded file paths are handled correctly."""
|
|
pdf_bytes = create_mock_pdf_bytes()
|
|
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(return_value=(pdf_bytes, "application/pdf"))
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
# URL-encoded path with spaces
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/Documents/My%20File.pdf",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
# Verify the path was passed correctly to WebDAV
|
|
mock_webdav.read_file.assert_called_once()
|
|
call_args = mock_webdav.read_file.call_args[0]
|
|
assert "My File.pdf" in call_args[0]
|
|
|
|
def test_missing_nextcloud_host_config(self):
|
|
"""Test handling when Nextcloud host is not configured."""
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
# Override with empty config
|
|
app.state.oauth_context = {"config": {"nextcloud_host": ""}}
|
|
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 500
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
|
|
def test_corrupted_pdf_returns_400(self):
|
|
"""Test that corrupted PDF data returns 400 with specific error."""
|
|
mock_webdav = AsyncMock()
|
|
# Return invalid PDF bytes
|
|
mock_webdav.read_file = AsyncMock(
|
|
return_value=(b"not a valid pdf", "application/pdf")
|
|
)
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/corrupted.pdf",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert (
|
|
"corrupted" in data["error"].lower()
|
|
or "invalid" in data["error"].lower()
|
|
)
|
|
|
|
def test_boundary_scale_values(self):
|
|
"""Test boundary scale values (min and max)."""
|
|
pdf_bytes = create_mock_pdf_bytes()
|
|
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(return_value=(pdf_bytes, "application/pdf"))
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
|
|
# Test minimum valid scale (0.5)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&scale=0.5",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Test maximum valid scale (5.0)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/test.pdf&scale=5.0",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
class TestPdfPreviewSecurityValidation:
|
|
"""Tests for security validations in PDF preview endpoint."""
|
|
|
|
def test_path_traversal_returns_400(self):
|
|
"""Test that path traversal attempts are blocked with 400."""
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
|
|
# Test various path traversal patterns
|
|
traversal_paths = [
|
|
"/Documents/../../../etc/passwd",
|
|
"/../secret.pdf",
|
|
"/folder/..%2F..%2Fetc/passwd", # URL-encoded
|
|
"/test/../secret.pdf",
|
|
]
|
|
|
|
for path in traversal_paths:
|
|
response = client.get(
|
|
f"/api/v1/pdf-preview?file_path={path}",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400, (
|
|
f"Path traversal not blocked: {path}"
|
|
)
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "invalid file path" in data["error"].lower()
|
|
|
|
def test_file_size_limit_exceeded_returns_413(self):
|
|
"""Test that files exceeding 50MB limit return 413."""
|
|
# Create bytes larger than 50MB limit
|
|
large_pdf_bytes = b"x" * (51 * 1024 * 1024) # 51 MB
|
|
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(
|
|
return_value=(large_pdf_bytes, "application/pdf")
|
|
)
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/large.pdf",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 413
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "size limit" in data["error"].lower()
|
|
|
|
def test_corrupted_pdf_returns_400(self):
|
|
"""Test that corrupted PDF returns 400 with specific error message."""
|
|
# Invalid PDF content that PyMuPDF cannot parse
|
|
corrupted_pdf_bytes = b"not a valid PDF file content"
|
|
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(
|
|
return_value=(corrupted_pdf_bytes, "application/pdf")
|
|
)
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/corrupted.pdf",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert (
|
|
"corrupted" in data["error"].lower()
|
|
or "invalid" in data["error"].lower()
|
|
)
|
|
|
|
def test_empty_pdf_returns_400(self):
|
|
"""Test that empty PDF file returns 400."""
|
|
empty_pdf_bytes = b""
|
|
|
|
mock_webdav = AsyncMock()
|
|
mock_webdav.read_file = AsyncMock(
|
|
return_value=(empty_pdf_bytes, "application/pdf")
|
|
)
|
|
|
|
mock_nc_client = MagicMock()
|
|
mock_nc_client.webdav = mock_webdav
|
|
mock_nc_client.__aenter__ = AsyncMock(return_value=mock_nc_client)
|
|
mock_nc_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.extract_bearer_token",
|
|
return_value="test-token",
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.client.NextcloudClient.from_token",
|
|
return_value=mock_nc_client,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/pdf-preview?file_path=/empty.pdf",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|