diff --git a/tests/unit/test_management_pdf_preview_endpoint.py b/tests/unit/test_management_pdf_preview_endpoint.py new file mode 100644 index 0000000..69950ef --- /dev/null +++ b/tests/unit/test_management_pdf_preview_endpoint.py @@ -0,0 +1,545 @@ +""" +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.management 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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.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.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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_500(self): + """Test that corrupted PDF data returns 500.""" + 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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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 == 500 + data = response.json() + assert data["success"] is False + + 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.management.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("testuser", True), + ), + patch( + "nextcloud_mcp_server.api.management.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