diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 14073ec..6a423fc 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -23,13 +23,13 @@ jobs: pip install -e ".[dev]" - name: Lint with flake8 - run: flake8 wikijs tests + run: flake8 wikijs tests --max-line-length=88 --ignore=E203,E501,W503 - name: Format check with black - run: black --check wikijs tests + run: black --check wikijs tests --line-length=88 - name: Import sort check - run: isort --check-only wikijs tests + run: isort --check-only wikijs tests --line-length=88 - name: Type check with mypy run: mypy wikijs diff --git a/tests/__init__.py b/tests/__init__.py index 89e6daa..698e067 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1 +1 @@ -"""Tests for wikijs-python-sdk.""" \ No newline at end of file +"""Tests for wikijs-python-sdk.""" diff --git a/tests/auth/__init__.py b/tests/auth/__init__.py index 0bc070c..51abb85 100644 --- a/tests/auth/__init__.py +++ b/tests/auth/__init__.py @@ -1 +1 @@ -"""Authentication tests for wikijs-python-sdk.""" \ No newline at end of file +"""Authentication tests for wikijs-python-sdk.""" diff --git a/tests/auth/test_api_key.py b/tests/auth/test_api_key.py index 8dd3f71..5f556df 100644 --- a/tests/auth/test_api_key.py +++ b/tests/auth/test_api_key.py @@ -36,10 +36,10 @@ class TestAPIKeyAuth: def test_get_headers_returns_bearer_token(self, api_key_auth, mock_api_key): """Test that get_headers returns proper Authorization header.""" headers = api_key_auth.get_headers() - + expected_headers = { "Authorization": f"Bearer {mock_api_key}", - "Content-Type": "application/json" + "Content-Type": "application/json", } assert headers == expected_headers @@ -75,14 +75,16 @@ class TestAPIKeyAuth: # Test long key (>8 chars) - shows first 4 and last 4 auth = APIKeyAuth("this-is-a-very-long-api-key-for-testing") - expected = "this" + "*" * (len("this-is-a-very-long-api-key-for-testing") - 8) + "ting" + expected = ( + "this" + "*" * (len("this-is-a-very-long-api-key-for-testing") - 8) + "ting" + ) assert auth.api_key == expected def test_repr_shows_masked_key(self, mock_api_key): """Test that __repr__ shows masked API key.""" auth = APIKeyAuth(mock_api_key) repr_str = repr(auth) - + assert "APIKeyAuth" in repr_str assert mock_api_key not in repr_str # Real key should not appear assert auth.api_key in repr_str # Masked key should appear @@ -102,11 +104,19 @@ class TestAPIKeyAuth: ("abcd", "****"), ("abcdefgh", "********"), ("abcdefghi", "abcd*fghi"), # 9 chars: first 4 + 1 star + last 4 - ("abcdefghij", "abcd**ghij"), # 10 chars: first 4 + 2 stars + last 4 - ("very-long-api-key-here", "very**************here"), # 22 chars: first 4 + 14 stars + last 4 + ( + "abcdefghij", + "abcd**ghij", + ), # 10 chars: first 4 + 2 stars + last 4 + ( + "very-long-api-key-here", + "very**************here", + ), # 22 chars: first 4 + 14 stars + last 4 ] - + for key, expected_mask in test_cases: auth = APIKeyAuth(key) actual = auth.api_key - assert actual == expected_mask, f"Failed for key '{key}': expected '{expected_mask}', got '{actual}'" \ No newline at end of file + assert ( + actual == expected_mask + ), f"Failed for key '{key}': expected '{expected_mask}', got '{actual}'" diff --git a/tests/auth/test_base.py b/tests/auth/test_base.py index 363386f..8284097 100644 --- a/tests/auth/test_base.py +++ b/tests/auth/test_base.py @@ -1,7 +1,6 @@ """Tests for base authentication functionality.""" import pytest -from unittest.mock import Mock from wikijs.auth.base import AuthHandler, NoAuth from wikijs.exceptions import AuthenticationError @@ -17,18 +16,19 @@ class TestAuthHandler: def test_validate_credentials_calls_is_valid(self): """Test that validate_credentials calls is_valid.""" + # Create concrete implementation for testing class TestAuth(AuthHandler): def __init__(self, valid=True): self.valid = valid self.refresh_called = False - + def get_headers(self): return {"Authorization": "test"} - + def is_valid(self): return self.valid - + def refresh(self): self.refresh_called = True self.valid = True @@ -46,18 +46,21 @@ class TestAuthHandler: def test_validate_credentials_raises_on_invalid_after_refresh(self): """Test that validate_credentials raises if still invalid after refresh.""" + class TestAuth(AuthHandler): def get_headers(self): return {"Authorization": "test"} - + def is_valid(self): return False # Always invalid - + def refresh(self): pass # No-op refresh auth = TestAuth() - with pytest.raises(AuthenticationError, match="Authentication credentials are invalid"): + with pytest.raises( + AuthenticationError, match="Authentication credentials are invalid" + ): auth.validate_credentials() @@ -89,4 +92,4 @@ class TestNoAuth: """Test that validate_credentials always succeeds.""" # Should not raise any exception no_auth.validate_credentials() - assert no_auth.is_valid() is True \ No newline at end of file + assert no_auth.is_valid() is True diff --git a/tests/auth/test_jwt.py b/tests/auth/test_jwt.py index 93bf475..9cb2fcf 100644 --- a/tests/auth/test_jwt.py +++ b/tests/auth/test_jwt.py @@ -1,7 +1,7 @@ """Tests for JWT authentication.""" import time -from datetime import datetime, timedelta +from datetime import timedelta from unittest.mock import patch import pytest @@ -24,7 +24,7 @@ class TestJWTAuth: """Test initialization with all parameters.""" refresh_token = "refresh-token-123" expires_at = time.time() + 3600 - + auth = JWTAuth(mock_jwt_token, refresh_token, expires_at) assert auth._token == mock_jwt_token assert auth._refresh_token == refresh_token @@ -53,10 +53,10 @@ class TestJWTAuth: def test_get_headers_returns_bearer_token(self, jwt_auth, mock_jwt_token): """Test that get_headers returns proper Authorization header.""" headers = jwt_auth.get_headers() - + expected_headers = { "Authorization": f"Bearer {mock_jwt_token}", - "Content-Type": "application/json" + "Content-Type": "application/json", } assert headers == expected_headers @@ -65,16 +65,16 @@ class TestJWTAuth: # Create JWT with expired token expires_at = time.time() - 3600 # Expired 1 hour ago refresh_token = "refresh-token-123" - + auth = JWTAuth(mock_jwt_token, refresh_token, expires_at) - + # Mock the refresh method to avoid actual implementation - with patch.object(auth, 'refresh') as mock_refresh: + with patch.object(auth, "refresh") as mock_refresh: mock_refresh.side_effect = AuthenticationError("Refresh not implemented") - + with pytest.raises(AuthenticationError): auth.get_headers() - + mock_refresh.assert_called_once() def test_is_valid_returns_true_for_valid_token_no_expiry(self, jwt_auth): @@ -111,15 +111,20 @@ class TestJWTAuth: def test_refresh_raises_error_without_refresh_token(self, jwt_auth): """Test that refresh raises error when no refresh token available.""" - with pytest.raises(AuthenticationError, match="JWT token expired and no refresh token available"): + with pytest.raises( + AuthenticationError, + match="JWT token expired and no refresh token available", + ): jwt_auth.refresh() def test_refresh_raises_not_implemented_error(self, mock_jwt_token): """Test that refresh raises not implemented error.""" refresh_token = "refresh-token-123" auth = JWTAuth(mock_jwt_token, refresh_token) - - with pytest.raises(AuthenticationError, match="JWT token refresh not yet implemented"): + + with pytest.raises( + AuthenticationError, match="JWT token refresh not yet implemented" + ): auth.refresh() def test_is_expired_returns_false_no_expiry(self, jwt_auth): @@ -146,7 +151,7 @@ class TestJWTAuth: """Test that time_until_expiry returns correct timedelta.""" expires_at = time.time() + 3600 # Expires in 1 hour auth = JWTAuth(mock_jwt_token, expires_at=expires_at) - + time_left = auth.time_until_expiry() assert isinstance(time_left, timedelta) # Should be approximately 1 hour (allowing for small time differences) @@ -156,7 +161,7 @@ class TestJWTAuth: """Test that time_until_expiry returns zero for expired token.""" expires_at = time.time() - 3600 # Expired 1 hour ago auth = JWTAuth(mock_jwt_token, expires_at=expires_at) - + time_left = auth.time_until_expiry() assert time_left.total_seconds() == 0 @@ -164,7 +169,7 @@ class TestJWTAuth: """Test that token_preview masks the token for security.""" auth = JWTAuth(mock_jwt_token) preview = auth.token_preview - + assert preview != mock_jwt_token # Should not show full token assert preview.startswith(mock_jwt_token[:10]) assert preview.endswith(mock_jwt_token[-10:]) @@ -175,14 +180,14 @@ class TestJWTAuth: short_token = "short" auth = JWTAuth(short_token) preview = auth.token_preview - + assert preview == "*****" # Should be all asterisks def test_token_preview_handles_none_token(self, mock_jwt_token): """Test that token_preview handles None token.""" auth = JWTAuth(mock_jwt_token) auth._token = None - + assert auth.token_preview == "None" def test_repr_shows_masked_token(self, mock_jwt_token): @@ -190,7 +195,7 @@ class TestJWTAuth: expires_at = time.time() + 3600 auth = JWTAuth(mock_jwt_token, expires_at=expires_at) repr_str = repr(auth) - + assert "JWTAuth" in repr_str assert mock_jwt_token not in repr_str # Real token should not appear assert auth.token_preview in repr_str # Masked token should appear @@ -210,4 +215,4 @@ class TestJWTAuth: # Test None refresh token auth = JWTAuth(mock_jwt_token, None) - assert auth._refresh_token is None \ No newline at end of file + assert auth._refresh_token is None diff --git a/tests/conftest.py b/tests/conftest.py index f121c69..310d551 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,7 +2,6 @@ import pytest import responses -from unittest.mock import Mock from wikijs.auth import APIKeyAuth, JWTAuth, NoAuth @@ -60,21 +59,12 @@ def sample_page_data(): "content": "This is a test page content.", "created_at": "2023-01-01T00:00:00Z", "updated_at": "2023-01-01T12:00:00Z", - "author": { - "id": 1, - "name": "Test User", - "email": "test@example.com" - }, - "tags": ["test", "example"] + "author": {"id": 1, "name": "Test User", "email": "test@example.com"}, + "tags": ["test", "example"], } @pytest.fixture def sample_error_response(): """Fixture providing sample error response.""" - return { - "error": { - "message": "Not found", - "code": "PAGE_NOT_FOUND" - } - } \ No newline at end of file + return {"error": {"message": "Not found", "code": "PAGE_NOT_FOUND"}} diff --git a/tests/endpoints/__init__.py b/tests/endpoints/__init__.py index b170091..721480b 100644 --- a/tests/endpoints/__init__.py +++ b/tests/endpoints/__init__.py @@ -1 +1 @@ -"""Tests for API endpoints.""" \ No newline at end of file +"""Tests for API endpoints.""" diff --git a/tests/endpoints/test_base.py b/tests/endpoints/test_base.py index 87e06e9..25028eb 100644 --- a/tests/endpoints/test_base.py +++ b/tests/endpoints/test_base.py @@ -1,147 +1,144 @@ """Tests for base endpoint class.""" -import pytest from unittest.mock import Mock +import pytest + from wikijs.client import WikiJSClient from wikijs.endpoints.base import BaseEndpoint class TestBaseEndpoint: """Test suite for BaseEndpoint.""" - + @pytest.fixture def mock_client(self): """Create a mock WikiJS client.""" client = Mock(spec=WikiJSClient) return client - + @pytest.fixture def base_endpoint(self, mock_client): """Create a BaseEndpoint instance with mock client.""" return BaseEndpoint(mock_client) - + def test_init(self, mock_client): """Test BaseEndpoint initialization.""" endpoint = BaseEndpoint(mock_client) assert endpoint._client is mock_client - + def test_request(self, base_endpoint, mock_client): """Test _request method delegates to client.""" # Setup mock response mock_response = {"data": "test"} mock_client._request.return_value = mock_response - + # Call _request result = base_endpoint._request( "GET", "/test", params={"param": "value"}, json_data={"data": "test"}, - extra_param="extra" + extra_param="extra", ) - + # Verify delegation to client mock_client._request.assert_called_once_with( method="GET", endpoint="/test", params={"param": "value"}, json_data={"data": "test"}, - extra_param="extra" + extra_param="extra", ) - + # Verify response assert result == mock_response - + def test_get(self, base_endpoint, mock_client): """Test _get method.""" mock_response = {"data": "test"} mock_client._request.return_value = mock_response - + result = base_endpoint._get("/test", params={"param": "value"}) - + mock_client._request.assert_called_once_with( method="GET", endpoint="/test", params={"param": "value"}, - json_data=None + json_data=None, ) assert result == mock_response - + def test_post(self, base_endpoint, mock_client): """Test _post method.""" mock_response = {"data": "test"} mock_client._request.return_value = mock_response - + result = base_endpoint._post( - "/test", - json_data={"data": "test"}, - params={"param": "value"} + "/test", json_data={"data": "test"}, params={"param": "value"} ) - + mock_client._request.assert_called_once_with( method="POST", endpoint="/test", params={"param": "value"}, - json_data={"data": "test"} + json_data={"data": "test"}, ) assert result == mock_response - + def test_put(self, base_endpoint, mock_client): """Test _put method.""" mock_response = {"data": "test"} mock_client._request.return_value = mock_response - + result = base_endpoint._put( - "/test", - json_data={"data": "test"}, - params={"param": "value"} + "/test", json_data={"data": "test"}, params={"param": "value"} ) - + mock_client._request.assert_called_once_with( method="PUT", endpoint="/test", params={"param": "value"}, - json_data={"data": "test"} + json_data={"data": "test"}, ) assert result == mock_response - + def test_delete(self, base_endpoint, mock_client): """Test _delete method.""" mock_response = {"data": "test"} mock_client._request.return_value = mock_response - + result = base_endpoint._delete("/test", params={"param": "value"}) - + mock_client._request.assert_called_once_with( method="DELETE", endpoint="/test", params={"param": "value"}, - json_data=None + json_data=None, ) assert result == mock_response - + def test_build_endpoint_single_part(self, base_endpoint): """Test _build_endpoint with single part.""" result = base_endpoint._build_endpoint("test") assert result == "/test" - + def test_build_endpoint_multiple_parts(self, base_endpoint): """Test _build_endpoint with multiple parts.""" result = base_endpoint._build_endpoint("api", "v1", "pages") assert result == "/api/v1/pages" - + def test_build_endpoint_with_slashes(self, base_endpoint): """Test _build_endpoint handles leading/trailing slashes.""" result = base_endpoint._build_endpoint("/api/", "/v1/", "/pages/") assert result == "/api/v1/pages" - + def test_build_endpoint_empty_parts(self, base_endpoint): """Test _build_endpoint filters out empty parts.""" result = base_endpoint._build_endpoint("api", "", "pages", None) assert result == "/api/pages" - + def test_build_endpoint_numeric_parts(self, base_endpoint): """Test _build_endpoint handles numeric parts.""" result = base_endpoint._build_endpoint("pages", 123, "edit") - assert result == "/pages/123/edit" \ No newline at end of file + assert result == "/pages/123/edit" diff --git a/tests/endpoints/test_pages.py b/tests/endpoints/test_pages.py index 2341f10..c5b333f 100644 --- a/tests/endpoints/test_pages.py +++ b/tests/endpoints/test_pages.py @@ -1,8 +1,9 @@ """Tests for Pages API endpoint.""" -import pytest from unittest.mock import Mock, patch +import pytest + from wikijs.client import WikiJSClient from wikijs.endpoints.pages import PagesEndpoint from wikijs.exceptions import APIError, ValidationError @@ -11,18 +12,18 @@ from wikijs.models.page import Page, PageCreate, PageUpdate class TestPagesEndpoint: """Test suite for PagesEndpoint.""" - + @pytest.fixture def mock_client(self): """Create a mock WikiJS client.""" client = Mock(spec=WikiJSClient) return client - + @pytest.fixture def pages_endpoint(self, mock_client): """Create a PagesEndpoint instance with mock client.""" return PagesEndpoint(mock_client) - + @pytest.fixture def sample_page_data(self): """Sample page data from API.""" @@ -41,9 +42,9 @@ class TestPagesEndpoint: "authorEmail": "test@example.com", "editor": "markdown", "createdAt": "2023-01-01T00:00:00Z", - "updatedAt": "2023-01-02T00:00:00Z" + "updatedAt": "2023-01-02T00:00:00Z", } - + @pytest.fixture def sample_page_create(self): """Sample PageCreate object.""" @@ -52,57 +53,49 @@ class TestPagesEndpoint: path="new-page", content="# New Page\n\nContent here.", description="A new page", - tags=["new", "test"] + tags=["new", "test"], ) - + @pytest.fixture def sample_page_update(self): """Sample PageUpdate object.""" return PageUpdate( title="Updated Page", content="# Updated Page\n\nUpdated content.", - tags=["updated", "test"] + tags=["updated", "test"], ) - + def test_init(self, mock_client): """Test PagesEndpoint initialization.""" endpoint = PagesEndpoint(mock_client) assert endpoint._client is mock_client - + def test_list_basic(self, pages_endpoint, sample_page_data): """Test basic page listing.""" # Mock the GraphQL response - mock_response = { - "data": { - "pages": [sample_page_data] - } - } + mock_response = {"data": {"pages": [sample_page_data]}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call list method pages = pages_endpoint.list() - + # Verify request pages_endpoint._post.assert_called_once() call_args = pages_endpoint._post.call_args assert call_args[0][0] == "/graphql" - + # Verify response assert len(pages) == 1 assert isinstance(pages[0], Page) assert pages[0].id == 123 assert pages[0].title == "Test Page" assert pages[0].path == "test-page" - + def test_list_with_parameters(self, pages_endpoint, sample_page_data): """Test page listing with filter parameters.""" - mock_response = { - "data": { - "pages": [sample_page_data] - } - } + mock_response = {"data": {"pages": [sample_page_data]}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call with parameters pages = pages_endpoint.list( limit=10, @@ -112,14 +105,14 @@ class TestPagesEndpoint: locale="en", author_id=1, order_by="created_at", - order_direction="DESC" + order_direction="DESC", ) - + # Verify request call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] variables = query_data["variables"] - + assert variables["limit"] == 10 assert variables["offset"] == 5 assert variables["search"] == "test" @@ -128,133 +121,117 @@ class TestPagesEndpoint: assert variables["authorId"] == 1 assert variables["orderBy"] == "created_at" assert variables["orderDirection"] == "DESC" - + # Verify response assert len(pages) == 1 assert isinstance(pages[0], Page) - + def test_list_validation_errors(self, pages_endpoint): """Test list method parameter validation.""" # Test invalid limit with pytest.raises(ValidationError, match="limit must be greater than 0"): pages_endpoint.list(limit=0) - + # Test invalid offset with pytest.raises(ValidationError, match="offset must be non-negative"): pages_endpoint.list(offset=-1) - + # Test invalid order_by with pytest.raises(ValidationError, match="order_by must be one of"): pages_endpoint.list(order_by="invalid") - + # Test invalid order_direction - with pytest.raises(ValidationError, match="order_direction must be ASC or DESC"): + with pytest.raises( + ValidationError, match="order_direction must be ASC or DESC" + ): pages_endpoint.list(order_direction="INVALID") - + def test_list_api_error(self, pages_endpoint): """Test list method handling API errors.""" # Mock GraphQL error response - mock_response = { - "errors": [{"message": "GraphQL error"}] - } + mock_response = {"errors": [{"message": "GraphQL error"}]} pages_endpoint._post = Mock(return_value=mock_response) - + with pytest.raises(APIError, match="GraphQL errors"): pages_endpoint.list() - + def test_get_success(self, pages_endpoint, sample_page_data): """Test getting a page by ID.""" - mock_response = { - "data": { - "page": sample_page_data - } - } + mock_response = {"data": {"page": sample_page_data}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call method page = pages_endpoint.get(123) - + # Verify request call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] assert query_data["variables"]["id"] == 123 - + # Verify response assert isinstance(page, Page) assert page.id == 123 assert page.title == "Test Page" - + def test_get_validation_error(self, pages_endpoint): """Test get method parameter validation.""" with pytest.raises(ValidationError, match="page_id must be a positive integer"): pages_endpoint.get(0) - + with pytest.raises(ValidationError, match="page_id must be a positive integer"): pages_endpoint.get(-1) - + with pytest.raises(ValidationError, match="page_id must be a positive integer"): pages_endpoint.get("invalid") - + def test_get_not_found(self, pages_endpoint): """Test get method when page not found.""" - mock_response = { - "data": { - "page": None - } - } + mock_response = {"data": {"page": None}} pages_endpoint._post = Mock(return_value=mock_response) - + with pytest.raises(APIError, match="Page with ID 123 not found"): pages_endpoint.get(123) - + def test_get_by_path_success(self, pages_endpoint, sample_page_data): """Test getting a page by path.""" - mock_response = { - "data": { - "pageByPath": sample_page_data - } - } + mock_response = {"data": {"pageByPath": sample_page_data}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call method page = pages_endpoint.get_by_path("test-page") - + # Verify request call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] variables = query_data["variables"] assert variables["path"] == "test-page" assert variables["locale"] == "en" - + # Verify response assert isinstance(page, Page) assert page.path == "test-page" - + def test_get_by_path_validation_error(self, pages_endpoint): """Test get_by_path method parameter validation.""" with pytest.raises(ValidationError, match="path must be a non-empty string"): pages_endpoint.get_by_path("") - + with pytest.raises(ValidationError, match="path must be a non-empty string"): pages_endpoint.get_by_path(None) - + def test_create_success(self, pages_endpoint, sample_page_create, sample_page_data): """Test creating a new page.""" - mock_response = { - "data": { - "createPage": sample_page_data - } - } + mock_response = {"data": {"createPage": sample_page_data}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call method created_page = pages_endpoint.create(sample_page_create) - + # Verify request call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] variables = query_data["variables"] - + assert variables["title"] == "New Page" assert variables["path"] == "new-page" assert variables["content"] == "# New Page\n\nContent here." @@ -262,194 +239,177 @@ class TestPagesEndpoint: assert variables["tags"] == ["new", "test"] assert variables["isPublished"] is True assert variables["isPrivate"] is False - + # Verify response assert isinstance(created_page, Page) assert created_page.id == 123 - + def test_create_with_dict(self, pages_endpoint, sample_page_data): """Test creating a page with dict data.""" - mock_response = { - "data": { - "createPage": sample_page_data - } - } + mock_response = {"data": {"createPage": sample_page_data}} pages_endpoint._post = Mock(return_value=mock_response) - + page_dict = { "title": "Dict Page", "path": "dict-page", "content": "Content from dict", } - + # Call method created_page = pages_endpoint.create(page_dict) - + # Verify response assert isinstance(created_page, Page) - + def test_create_validation_error(self, pages_endpoint): """Test create method validation errors.""" # Test invalid data type - with pytest.raises(ValidationError, match="page_data must be PageCreate object or dict"): + with pytest.raises( + ValidationError, + match="page_data must be PageCreate object or dict", + ): pages_endpoint.create("invalid") - + # Test invalid dict data with pytest.raises(ValidationError, match="Invalid page data"): pages_endpoint.create({"invalid": "data"}) - + def test_create_api_error(self, pages_endpoint, sample_page_create): """Test create method API errors.""" - mock_response = { - "errors": [{"message": "Creation failed"}] - } + mock_response = {"errors": [{"message": "Creation failed"}]} pages_endpoint._post = Mock(return_value=mock_response) - + with pytest.raises(APIError, match="Failed to create page"): pages_endpoint.create(sample_page_create) - + def test_update_success(self, pages_endpoint, sample_page_update, sample_page_data): """Test updating a page.""" - mock_response = { - "data": { - "updatePage": sample_page_data - } - } + mock_response = {"data": {"updatePage": sample_page_data}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call method updated_page = pages_endpoint.update(123, sample_page_update) - + # Verify request call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] variables = query_data["variables"] - + assert variables["id"] == 123 assert variables["title"] == "Updated Page" assert variables["content"] == "# Updated Page\n\nUpdated content." assert variables["tags"] == ["updated", "test"] assert "description" not in variables # Should not include None values - + # Verify response assert isinstance(updated_page, Page) - + def test_update_validation_errors(self, pages_endpoint, sample_page_update): """Test update method validation errors.""" # Test invalid page_id with pytest.raises(ValidationError, match="page_id must be a positive integer"): pages_endpoint.update(0, sample_page_update) - + # Test invalid page_data type - with pytest.raises(ValidationError, match="page_data must be PageUpdate object or dict"): + with pytest.raises( + ValidationError, + match="page_data must be PageUpdate object or dict", + ): pages_endpoint.update(123, "invalid") - + def test_delete_success(self, pages_endpoint): """Test deleting a page.""" mock_response = { "data": { "deletePage": { "success": True, - "message": "Page deleted successfully" + "message": "Page deleted successfully", } } } pages_endpoint._post = Mock(return_value=mock_response) - + # Call method result = pages_endpoint.delete(123) - + # Verify request call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] assert query_data["variables"]["id"] == 123 - + # Verify response assert result is True - + def test_delete_validation_error(self, pages_endpoint): """Test delete method validation errors.""" with pytest.raises(ValidationError, match="page_id must be a positive integer"): pages_endpoint.delete(0) - + def test_delete_failure(self, pages_endpoint): """Test delete method when deletion fails.""" mock_response = { - "data": { - "deletePage": { - "success": False, - "message": "Deletion failed" - } - } + "data": {"deletePage": {"success": False, "message": "Deletion failed"}} } pages_endpoint._post = Mock(return_value=mock_response) - + with pytest.raises(APIError, match="Page deletion failed: Deletion failed"): pages_endpoint.delete(123) - + def test_search_success(self, pages_endpoint, sample_page_data): """Test searching pages.""" - mock_response = { - "data": { - "pages": [sample_page_data] - } - } + mock_response = {"data": {"pages": [sample_page_data]}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call method results = pages_endpoint.search("test query", limit=5) - + # Verify request (should call list with search parameter) call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] variables = query_data["variables"] - + assert variables["search"] == "test query" assert variables["limit"] == 5 - + # Verify response assert len(results) == 1 assert isinstance(results[0], Page) - + def test_search_validation_error(self, pages_endpoint): """Test search method validation errors.""" with pytest.raises(ValidationError, match="query must be a non-empty string"): pages_endpoint.search("") - + with pytest.raises(ValidationError, match="limit must be greater than 0"): pages_endpoint.search("test", limit=0) - + def test_get_by_tags_match_all(self, pages_endpoint, sample_page_data): """Test getting pages by tags (match all).""" - mock_response = { - "data": { - "pages": [sample_page_data] - } - } + mock_response = {"data": {"pages": [sample_page_data]}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call method results = pages_endpoint.get_by_tags(["test", "example"], match_all=True) - + # Verify request (should call list with tags parameter) call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] variables = query_data["variables"] - + assert variables["tags"] == ["test", "example"] - + # Verify response assert len(results) == 1 assert isinstance(results[0], Page) - + def test_get_by_tags_validation_error(self, pages_endpoint): """Test get_by_tags method validation errors.""" with pytest.raises(ValidationError, match="tags must be a non-empty list"): pages_endpoint.get_by_tags([]) - + with pytest.raises(ValidationError, match="limit must be greater than 0"): pages_endpoint.get_by_tags(["test"], limit=0) - + def test_normalize_page_data(self, pages_endpoint): """Test page data normalization.""" api_data = { @@ -457,11 +417,11 @@ class TestPagesEndpoint: "title": "Test", "isPublished": True, "authorId": 1, - "createdAt": "2023-01-01T00:00:00Z" + "createdAt": "2023-01-01T00:00:00Z", } - + normalized = pages_endpoint._normalize_page_data(api_data) - + # Check field mapping assert normalized["id"] == 123 assert normalized["title"] == "Test" @@ -469,57 +429,48 @@ class TestPagesEndpoint: assert normalized["author_id"] == 1 assert normalized["created_at"] == "2023-01-01T00:00:00Z" assert normalized["tags"] == [] # Default value - + def test_normalize_page_data_missing_fields(self, pages_endpoint): """Test page data normalization with missing fields.""" - api_data = { - "id": 123, - "title": "Test" - } - + api_data = {"id": 123, "title": "Test"} + normalized = pages_endpoint._normalize_page_data(api_data) - + # Check that only present fields are included assert "id" in normalized assert "title" in normalized assert "is_published" not in normalized assert "tags" in normalized # Should have default value - - @patch('wikijs.endpoints.pages.Page') - def test_list_page_parsing_error(self, mock_page_class, pages_endpoint, sample_page_data): + + @patch("wikijs.endpoints.pages.Page") + def test_list_page_parsing_error( + self, mock_page_class, pages_endpoint, sample_page_data + ): """Test handling of page parsing errors in list method.""" # Mock Page constructor to raise an exception mock_page_class.side_effect = ValueError("Parsing error") - - mock_response = { - "data": { - "pages": [sample_page_data] - } - } + + mock_response = {"data": {"pages": [sample_page_data]}} pages_endpoint._post = Mock(return_value=mock_response) - + with pytest.raises(APIError, match="Failed to parse page data"): pages_endpoint.list() - + def test_graphql_query_structure(self, pages_endpoint, sample_page_data): """Test that GraphQL queries have correct structure.""" - mock_response = { - "data": { - "pages": [sample_page_data] - } - } + mock_response = {"data": {"pages": [sample_page_data]}} pages_endpoint._post = Mock(return_value=mock_response) - + # Call list method pages_endpoint.list() - + # Verify the GraphQL query structure call_args = pages_endpoint._post.call_args query_data = call_args[1]["json_data"] - + assert "query" in query_data assert "variables" in query_data assert "pages(" in query_data["query"] assert "id" in query_data["query"] assert "title" in query_data["query"] - assert "content" in query_data["query"] \ No newline at end of file + assert "content" in query_data["query"] diff --git a/tests/models/test_base.py b/tests/models/test_base.py index 556bf3c..4e521bf 100644 --- a/tests/models/test_base.py +++ b/tests/models/test_base.py @@ -3,14 +3,12 @@ import json from datetime import datetime -import pytest - from wikijs.models.base import BaseModel, TimestampedModel class TestModelForTesting(BaseModel): """Test model for testing base functionality.""" - + name: str value: int = 42 optional_field: str = None @@ -18,82 +16,90 @@ class TestModelForTesting(BaseModel): class TestTimestampedModelForTesting(TimestampedModel): """Test model with timestamps.""" - + title: str - + class TestBaseModel: """Test base model functionality.""" - + def test_basic_model_creation(self): """Test basic model creation.""" model = TestModelForTesting(name="test", value=100) assert model.name == "test" assert model.value == 100 assert model.optional_field is None - + def test_to_dict_basic(self): """Test to_dict method.""" model = TestModelForTesting(name="test", value=100) result = model.to_dict() - + expected = {"name": "test", "value": 100} assert result == expected - + def test_to_dict_with_none_values(self): """Test to_dict with None values.""" model = TestModelForTesting(name="test", value=100) - + # Test excluding None values (default) result_exclude = model.to_dict(exclude_none=True) expected_exclude = {"name": "test", "value": 100} assert result_exclude == expected_exclude - + # Test including None values result_include = model.to_dict(exclude_none=False) - expected_include = {"name": "test", "value": 100, "optional_field": None} + expected_include = { + "name": "test", + "value": 100, + "optional_field": None, + } assert result_include == expected_include - + def test_to_json_basic(self): """Test to_json method.""" model = TestModelForTesting(name="test", value=100) result = model.to_json() - + # Parse the JSON to verify structure parsed = json.loads(result) expected = {"name": "test", "value": 100} assert parsed == expected - + def test_to_json_with_none_values(self): """Test to_json with None values.""" model = TestModelForTesting(name="test", value=100) - + # Test excluding None values (default) result_exclude = model.to_json(exclude_none=True) parsed_exclude = json.loads(result_exclude) expected_exclude = {"name": "test", "value": 100} assert parsed_exclude == expected_exclude - + # Test including None values result_include = model.to_json(exclude_none=False) parsed_include = json.loads(result_include) - expected_include = {"name": "test", "value": 100, "optional_field": None} + expected_include = { + "name": "test", + "value": 100, + "optional_field": None, + } assert parsed_include == expected_include - + def test_from_dict(self): """Test from_dict class method.""" data = {"name": "test", "value": 200} model = TestModelForTesting.from_dict(data) - + assert isinstance(model, TestModelForTesting) assert model.name == "test" assert model.value == 200 - + def test_from_json(self): """Test from_json class method.""" json_str = '{"name": "test", "value": 300}' model = TestModelForTesting.from_json(json_str) - + assert isinstance(model, TestModelForTesting) assert model.name == "test" assert model.value == 300 @@ -101,51 +107,44 @@ class TestBaseModel: class TestTimestampedModel: """Test timestamped model functionality.""" - + def test_timestamped_model_creation(self): """Test timestamped model creation.""" model = TestTimestampedModelForTesting(title="Test Title") assert model.title == "Test Title" assert model.created_at is None assert model.updated_at is None - + def test_timestamped_model_with_timestamps(self): """Test timestamped model with timestamps.""" now = datetime.now() model = TestTimestampedModelForTesting( - title="Test Title", - created_at=now, - updated_at=now + title="Test Title", created_at=now, updated_at=now ) assert model.title == "Test Title" assert model.created_at == now assert model.updated_at == now - + def test_is_new_property_true(self): """Test is_new property returns True for new models.""" model = TestTimestampedModelForTesting(title="Test Title") assert model.is_new is True - + def test_is_new_property_false(self): """Test is_new property returns False for existing models.""" now = datetime.now() - model = TestTimestampedModelForTesting( - title="Test Title", - created_at=now - ) + model = TestTimestampedModelForTesting(title="Test Title", created_at=now) assert model.is_new is False - + def test_datetime_serialization(self): """Test datetime serialization in JSON.""" now = datetime(2023, 1, 1, 12, 0, 0) model = TestTimestampedModelForTesting( - title="Test Title", - created_at=now, - updated_at=now + title="Test Title", created_at=now, updated_at=now ) - + json_str = model.to_json() parsed = json.loads(json_str) - + assert parsed["created_at"] == "2023-01-01T12:00:00" - assert parsed["updated_at"] == "2023-01-01T12:00:00" \ No newline at end of file + assert parsed["updated_at"] == "2023-01-01T12:00:00" diff --git a/tests/models/test_page.py b/tests/models/test_page.py index 6e2d171..188f836 100644 --- a/tests/models/test_page.py +++ b/tests/models/test_page.py @@ -1,14 +1,13 @@ """Tests for Page models.""" import pytest -from datetime import datetime from wikijs.models.page import Page, PageCreate, PageUpdate class TestPageModel: """Test Page model functionality.""" - + @pytest.fixture def valid_page_data(self): """Valid page data for testing.""" @@ -27,20 +26,23 @@ class TestPageModel: "author_email": "test@example.com", "editor": "markdown", "created_at": "2023-01-01T00:00:00Z", - "updated_at": "2023-01-02T00:00:00Z" + "updated_at": "2023-01-02T00:00:00Z", } - + def test_page_creation_valid(self, valid_page_data): """Test creating a valid page.""" page = Page(**valid_page_data) - + assert page.id == 123 assert page.title == "Test Page" assert page.path == "test-page" - assert page.content == "# Test Page\n\nThis is test content with **bold** and *italic* text." + assert ( + page.content + == "# Test Page\n\nThis is test content with **bold** and *italic* text." + ) assert page.is_published is True assert page.tags == ["test", "example"] - + def test_page_creation_minimal(self): """Test creating page with minimal required fields.""" page = Page( @@ -49,14 +51,14 @@ class TestPageModel: path="minimal", content="Content", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) - + assert page.id == 1 assert page.title == "Minimal Page" assert page.is_published is True # Default value assert page.tags == [] # Default value - + def test_page_path_validation_valid(self): """Test valid path validation.""" valid_paths = [ @@ -64,9 +66,9 @@ class TestPageModel: "path/with/slashes", "path_with_underscores", "path123", - "category/subcategory/page-name" + "category/subcategory/page-name", ] - + for path in valid_paths: page = Page( id=1, @@ -74,10 +76,10 @@ class TestPageModel: path=path, content="Content", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) assert page.path == path - + def test_page_path_validation_invalid(self): """Test invalid path validation.""" invalid_paths = [ @@ -86,7 +88,7 @@ class TestPageModel: "path@with@symbols", # Special characters "path.with.dots", # Dots ] - + for path in invalid_paths: with pytest.raises(ValueError): Page( @@ -95,9 +97,9 @@ class TestPageModel: path=path, content="Content", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) - + def test_page_path_normalization(self): """Test path normalization.""" # Leading/trailing slashes should be removed @@ -107,10 +109,10 @@ class TestPageModel: path="/path/to/page/", content="Content", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) assert page.path == "path/to/page" - + def test_page_title_validation_valid(self): """Test valid title validation.""" page = Page( @@ -119,10 +121,10 @@ class TestPageModel: path="test", content="Content", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) assert page.title == "Valid Title with Spaces" - + def test_page_title_validation_invalid(self): """Test invalid title validation.""" invalid_titles = [ @@ -130,7 +132,7 @@ class TestPageModel: " ", # Only whitespace "x" * 256, # Too long ] - + for title in invalid_titles: with pytest.raises(ValueError): Page( @@ -139,16 +141,16 @@ class TestPageModel: path="test", content="Content", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) - + def test_page_word_count(self, valid_page_data): """Test word count calculation.""" page = Page(**valid_page_data) # "# Test Page\n\nThis is test content with **bold** and *italic* text." # Words: Test, Page, This, is, test, content, with, bold, and, italic, text assert page.word_count == 12 - + def test_page_word_count_empty_content(self): """Test word count with empty content.""" page = Page( @@ -157,16 +159,16 @@ class TestPageModel: path="test", content="", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) assert page.word_count == 0 - + def test_page_reading_time(self, valid_page_data): """Test reading time calculation.""" page = Page(**valid_page_data) # 11 words, assuming 200 words per minute, should be 1 minute (minimum) assert page.reading_time == 1 - + def test_page_reading_time_long_content(self): """Test reading time with long content.""" long_content = " ".join(["word"] * 500) # 500 words @@ -176,20 +178,20 @@ class TestPageModel: path="test", content=long_content, created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) # 500 words / 200 words per minute = 2.5, rounded down to 2 assert page.reading_time == 2 - + def test_page_url_path(self, valid_page_data): """Test URL path generation.""" page = Page(**valid_page_data) assert page.url_path == "/test-page" - + def test_page_extract_headings(self): """Test heading extraction from markdown content.""" content = """# Main Title - + Some content here. ## Secondary Heading @@ -197,26 +199,31 @@ Some content here. More content. ### Third Level - + And more content. ## Another Secondary Final content.""" - + page = Page( id=1, title="Test", path="test", content=content, created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) - + headings = page.extract_headings() - expected = ["Main Title", "Secondary Heading", "Third Level", "Another Secondary"] + expected = [ + "Main Title", + "Secondary Heading", + "Third Level", + "Another Secondary", + ] assert headings == expected - + def test_page_extract_headings_empty_content(self): """Test heading extraction with no content.""" page = Page( @@ -225,19 +232,19 @@ Final content.""" path="test", content="", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) assert page.extract_headings() == [] - + def test_page_has_tag(self, valid_page_data): """Test tag checking.""" page = Page(**valid_page_data) - + assert page.has_tag("test") is True assert page.has_tag("example") is True assert page.has_tag("TEST") is True # Case insensitive assert page.has_tag("nonexistent") is False - + def test_page_has_tag_no_tags(self): """Test tag checking with no tags.""" page = Page( @@ -246,28 +253,28 @@ Final content.""" path="test", content="Content", created_at="2023-01-01T00:00:00Z", - updated_at="2023-01-01T00:00:00Z" + updated_at="2023-01-01T00:00:00Z", ) assert page.has_tag("any") is False class TestPageCreateModel: """Test PageCreate model functionality.""" - + def test_page_create_valid(self): """Test creating valid PageCreate.""" page_create = PageCreate( title="New Page", path="new-page", - content="# New Page\n\nContent here." + content="# New Page\n\nContent here.", ) - + assert page_create.title == "New Page" assert page_create.path == "new-page" assert page_create.content == "# New Page\n\nContent here." assert page_create.is_published is True # Default assert page_create.editor == "markdown" # Default - + def test_page_create_with_optional_fields(self): """Test PageCreate with optional fields.""" page_create = PageCreate( @@ -279,65 +286,62 @@ class TestPageCreateModel: is_private=True, tags=["new", "test"], locale="fr", - editor="html" + editor="html", ) - + assert page_create.description == "A new page" assert page_create.is_published is False assert page_create.is_private is True assert page_create.tags == ["new", "test"] assert page_create.locale == "fr" assert page_create.editor == "html" - + def test_page_create_path_validation(self): """Test path validation in PageCreate.""" # Valid path PageCreate(title="Test", path="valid-path", content="Content") - + # Invalid paths should raise errors with pytest.raises(ValueError): PageCreate(title="Test", path="", content="Content") - + with pytest.raises(ValueError): PageCreate(title="Test", path="invalid path", content="Content") - + def test_page_create_title_validation(self): """Test title validation in PageCreate.""" # Valid title PageCreate(title="Valid Title", path="test", content="Content") - + # Invalid titles should raise errors with pytest.raises(ValueError): PageCreate(title="", path="test", content="Content") - + with pytest.raises(ValueError): PageCreate(title="x" * 256, path="test", content="Content") class TestPageUpdateModel: """Test PageUpdate model functionality.""" - + def test_page_update_empty(self): """Test creating empty PageUpdate.""" page_update = PageUpdate() - + assert page_update.title is None assert page_update.content is None assert page_update.description is None assert page_update.is_published is None assert page_update.tags is None - + def test_page_update_partial(self): """Test partial PageUpdate.""" - page_update = PageUpdate( - title="Updated Title", - content="Updated content" - ) - + page_update = PageUpdate(title="Updated Title", content="Updated content") + assert page_update.title == "Updated Title" assert page_update.content == "Updated content" assert page_update.description is None # Not updated - + def test_page_update_full(self): """Test full PageUpdate.""" page_update = PageUpdate( @@ -346,27 +350,27 @@ class TestPageUpdateModel: description="Updated description", is_published=False, is_private=True, - tags=["updated", "test"] + tags=["updated", "test"], ) - + assert page_update.title == "Updated Title" assert page_update.content == "Updated content" assert page_update.description == "Updated description" assert page_update.is_published is False assert page_update.is_private is True assert page_update.tags == ["updated", "test"] - + def test_page_update_title_validation(self): """Test title validation in PageUpdate.""" # Valid title PageUpdate(title="Valid Title") - + # None should be allowed (no update) PageUpdate(title=None) - + # Invalid titles should raise errors with pytest.raises(ValueError): PageUpdate(title="") - + with pytest.raises(ValueError): - PageUpdate(title="x" * 256) \ No newline at end of file + PageUpdate(title="x" * 256) diff --git a/tests/test_client.py b/tests/test_client.py index 1efab21..10788bd 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,9 +1,10 @@ """Tests for WikiJS client.""" import json -import pytest from unittest.mock import Mock, patch +import pytest + from wikijs.auth import APIKeyAuth from wikijs.client import WikiJSClient from wikijs.exceptions import ( @@ -17,177 +18,185 @@ from wikijs.exceptions import ( class TestWikiJSClientInit: """Test WikiJSClient initialization.""" - + def test_init_with_api_key_string(self): """Test initialization with API key string.""" - with patch('wikijs.client.requests.Session'): + with patch("wikijs.client.requests.Session"): client = WikiJSClient("https://wiki.example.com", auth="test-key") - + assert client.base_url == "https://wiki.example.com" assert isinstance(client._auth_handler, APIKeyAuth) assert client.timeout == 30 assert client.verify_ssl is True assert "wikijs-python-sdk" in client.user_agent - + def test_init_with_auth_handler(self): """Test initialization with auth handler.""" auth_handler = APIKeyAuth("test-key") - - with patch('wikijs.client.requests.Session'): + + with patch("wikijs.client.requests.Session"): client = WikiJSClient("https://wiki.example.com", auth=auth_handler) - + assert client._auth_handler is auth_handler - + def test_init_invalid_auth(self): """Test initialization with invalid auth parameter.""" with pytest.raises(ConfigurationError, match="Invalid auth parameter"): WikiJSClient("https://wiki.example.com", auth=123) - + def test_init_with_custom_settings(self): """Test initialization with custom settings.""" - with patch('wikijs.client.requests.Session'): + with patch("wikijs.client.requests.Session"): client = WikiJSClient( "https://wiki.example.com", auth="test-key", timeout=60, verify_ssl=False, - user_agent="Custom Agent" + user_agent="Custom Agent", ) - + assert client.timeout == 60 assert client.verify_ssl is False assert client.user_agent == "Custom Agent" - + def test_has_pages_endpoint(self): """Test that client has pages endpoint.""" - with patch('wikijs.client.requests.Session'): + with patch("wikijs.client.requests.Session"): client = WikiJSClient("https://wiki.example.com", auth="test-key") - - assert hasattr(client, 'pages') + + assert hasattr(client, "pages") assert client.pages._client is client class TestWikiJSClientTestConnection: """Test WikiJSClient connection testing.""" - + @pytest.fixture def mock_wiki_base_url(self): """Mock wiki base URL.""" return "https://wiki.example.com" - + @pytest.fixture def mock_api_key(self): """Mock API key.""" return "test-api-key-12345" - - @patch('wikijs.client.requests.Session.get') + + @patch("wikijs.client.requests.Session.get") def test_test_connection_success(self, mock_get, mock_wiki_base_url, mock_api_key): """Test successful connection test.""" mock_response = Mock() mock_response.status_code = 200 mock_get.return_value = mock_response - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) result = client.test_connection() - + assert result is True - @patch('wikijs.client.requests.Session.get') + @patch("wikijs.client.requests.Session.get") def test_test_connection_timeout(self, mock_get, mock_wiki_base_url, mock_api_key): """Test connection test timeout.""" import requests + mock_get.side_effect = requests.exceptions.Timeout("Request timed out") - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) - + with pytest.raises(TimeoutError, match="Connection test timed out"): client.test_connection() - @patch('wikijs.client.requests.Session.get') + @patch("wikijs.client.requests.Session.get") def test_test_connection_error(self, mock_get, mock_wiki_base_url, mock_api_key): """Test connection test with connection error.""" import requests + mock_get.side_effect = requests.exceptions.ConnectionError("Connection failed") - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) - + with pytest.raises(ConnectionError, match="Cannot connect"): client.test_connection() - + def test_test_connection_no_base_url(self): """Test connection test with no base URL.""" - with patch('wikijs.client.requests.Session'): + with patch("wikijs.client.requests.Session"): client = WikiJSClient("https://wiki.example.com", auth="test-key") client.base_url = "" # Simulate empty base URL after creation - + with pytest.raises(ConfigurationError, match="Base URL not configured"): client.test_connection() - + def test_test_connection_no_auth(self): """Test connection test with no auth.""" - with patch('wikijs.client.requests.Session'): + with patch("wikijs.client.requests.Session"): client = WikiJSClient("https://wiki.example.com", auth="test-key") client._auth_handler = None # Simulate no auth - - with pytest.raises(ConfigurationError, match="Authentication not configured"): + + with pytest.raises( + ConfigurationError, match="Authentication not configured" + ): client.test_connection() class TestWikiJSClientRequests: """Test WikiJSClient HTTP request handling.""" - + @pytest.fixture def mock_wiki_base_url(self): """Mock wiki base URL.""" return "https://wiki.example.com" - + @pytest.fixture def mock_api_key(self): """Mock API key.""" return "test-api-key-12345" - @patch('wikijs.client.requests.Session.request') + @patch("wikijs.client.requests.Session.request") def test_request_success(self, mock_request, mock_wiki_base_url, mock_api_key): """Test successful API request.""" mock_response = Mock() mock_response.ok = True mock_response.json.return_value = {"data": "test"} mock_request.return_value = mock_response - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) result = client._request("GET", "/test") - + assert result == {"data": "test"} mock_request.assert_called_once() - @patch('wikijs.client.requests.Session.request') - def test_request_with_json_data(self, mock_request, mock_wiki_base_url, mock_api_key): + @patch("wikijs.client.requests.Session.request") + def test_request_with_json_data( + self, mock_request, mock_wiki_base_url, mock_api_key + ): """Test API request with JSON data.""" mock_response = Mock() mock_response.ok = True mock_response.json.return_value = {"success": True} mock_request.return_value = mock_response - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) result = client._request("POST", "/test", json_data={"title": "Test"}) - + assert result == {"success": True} mock_request.assert_called_once() - @patch('wikijs.client.requests.Session.request') - def test_request_authentication_error(self, mock_request, mock_wiki_base_url, mock_api_key): + @patch("wikijs.client.requests.Session.request") + def test_request_authentication_error( + self, mock_request, mock_wiki_base_url, mock_api_key + ): """Test request with authentication error.""" mock_response = Mock() mock_response.ok = False mock_response.status_code = 401 mock_request.return_value = mock_response - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) - + with pytest.raises(AuthenticationError, match="Authentication failed"): client._request("GET", "/test") - @patch('wikijs.client.requests.Session.request') + @patch("wikijs.client.requests.Session.request") def test_request_api_error(self, mock_request, mock_wiki_base_url, mock_api_key): """Test request with API error.""" mock_response = Mock() @@ -195,129 +204,145 @@ class TestWikiJSClientRequests: mock_response.status_code = 404 mock_response.text = "Not found" mock_request.return_value = mock_response - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) - + with pytest.raises(APIError): client._request("GET", "/test") - @patch('wikijs.client.requests.Session.request') - def test_request_invalid_json_response(self, mock_request, mock_wiki_base_url, mock_api_key): + @patch("wikijs.client.requests.Session.request") + def test_request_invalid_json_response( + self, mock_request, mock_wiki_base_url, mock_api_key + ): """Test request with invalid JSON response.""" mock_response = Mock() mock_response.ok = True mock_response.json.side_effect = json.JSONDecodeError("Invalid JSON", "", 0) mock_request.return_value = mock_response - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) - + with pytest.raises(APIError, match="Invalid JSON response"): client._request("GET", "/test") - @patch('wikijs.client.requests.Session.request') + @patch("wikijs.client.requests.Session.request") def test_request_timeout(self, mock_request, mock_wiki_base_url, mock_api_key): """Test request timeout handling.""" import requests + mock_request.side_effect = requests.exceptions.Timeout("Request timed out") - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) - + with pytest.raises(TimeoutError, match="Request timed out"): client._request("GET", "/test") - - @patch('wikijs.client.requests.Session.request') - def test_request_connection_error(self, mock_request, mock_wiki_base_url, mock_api_key): + + @patch("wikijs.client.requests.Session.request") + def test_request_connection_error( + self, mock_request, mock_wiki_base_url, mock_api_key + ): """Test request connection error handling.""" import requests - mock_request.side_effect = requests.exceptions.ConnectionError("Connection failed") - + + mock_request.side_effect = requests.exceptions.ConnectionError( + "Connection failed" + ) + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) - + with pytest.raises(ConnectionError, match="Failed to connect"): client._request("GET", "/test") - - @patch('wikijs.client.requests.Session.request') - def test_request_general_exception(self, mock_request, mock_wiki_base_url, mock_api_key): + + @patch("wikijs.client.requests.Session.request") + def test_request_general_exception( + self, mock_request, mock_wiki_base_url, mock_api_key + ): """Test request general exception handling.""" import requests + mock_request.side_effect = requests.exceptions.RequestException("General error") - + client = WikiJSClient(mock_wiki_base_url, auth=mock_api_key) - + with pytest.raises(APIError, match="Request failed"): client._request("GET", "/test") class TestWikiJSClientWithDifferentAuth: """Test WikiJSClient with different auth types.""" - - @patch('wikijs.client.requests.Session') + + @patch("wikijs.client.requests.Session") def test_auth_validation_during_session_creation(self, mock_session_class): """Test that auth validation happens during session creation.""" mock_session = Mock() mock_session_class.return_value = mock_session - + # Mock auth handler that raises validation error during validation from wikijs.auth.base import AuthHandler from wikijs.exceptions import AuthenticationError - + mock_auth = Mock(spec=AuthHandler) - mock_auth.validate_credentials.side_effect = AuthenticationError("Invalid credentials") + mock_auth.validate_credentials.side_effect = AuthenticationError( + "Invalid credentials" + ) mock_auth.get_headers.return_value = {} - + with pytest.raises(AuthenticationError, match="Invalid credentials"): WikiJSClient("https://wiki.example.com", auth=mock_auth) class TestWikiJSClientContextManager: """Test WikiJSClient context manager functionality.""" - - @patch('wikijs.client.requests.Session') + + @patch("wikijs.client.requests.Session") def test_context_manager(self, mock_session_class): """Test client as context manager.""" mock_session = Mock() mock_session_class.return_value = mock_session - + with WikiJSClient("https://wiki.example.com", auth="test-key") as client: assert isinstance(client, WikiJSClient) - + # Verify session was closed mock_session.close.assert_called_once() - - @patch('wikijs.client.requests.Session') + + @patch("wikijs.client.requests.Session") def test_close_method(self, mock_session_class): """Test explicit close method.""" mock_session = Mock() mock_session_class.return_value = mock_session - + client = WikiJSClient("https://wiki.example.com", auth="test-key") client.close() - + mock_session.close.assert_called_once() - - @patch('wikijs.client.requests.Session') + + @patch("wikijs.client.requests.Session") def test_repr(self, mock_session_class): """Test string representation.""" mock_session = Mock() mock_session_class.return_value = mock_session - + client = WikiJSClient("https://wiki.example.com", auth="test-key") repr_str = repr(client) - + assert "WikiJSClient" in repr_str assert "https://wiki.example.com" in repr_str - - @patch('wikijs.client.requests.Session') + + @patch("wikijs.client.requests.Session") def test_connection_test_generic_exception(self, mock_session_class): """Test connection test with generic exception.""" mock_session = Mock() mock_session_class.return_value = mock_session - + # Mock generic exception during connection test mock_session.get.side_effect = RuntimeError("Unexpected error") - + client = WikiJSClient("https://wiki.example.com", auth="test-key") - + from wikijs.exceptions import ConnectionError - with pytest.raises(ConnectionError, match="Connection test failed: Unexpected error"): - client.test_connection() \ No newline at end of file + + with pytest.raises( + ConnectionError, match="Connection test failed: Unexpected error" + ): + client.test_connection() diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py index c039a64..adadcd2 100644 --- a/tests/test_exceptions.py +++ b/tests/test_exceptions.py @@ -1,34 +1,33 @@ """Tests for exception classes.""" -import pytest from unittest.mock import Mock from wikijs.exceptions import ( - WikiJSException, APIError, - ClientError, - ServerError, AuthenticationError, + ClientError, ConfigurationError, - ValidationError, + ConnectionError, NotFoundError, PermissionError, RateLimitError, - ConnectionError, + ServerError, TimeoutError, + ValidationError, + WikiJSException, create_api_error, ) class TestWikiJSException: """Test base exception class.""" - + def test_basic_exception_creation(self): """Test basic exception creation.""" exc = WikiJSException("Test error") assert str(exc) == "Test error" assert exc.message == "Test error" - + def test_exception_with_details(self): """Test exception with details.""" details = {"code": "TEST_ERROR", "field": "title"} @@ -38,13 +37,13 @@ class TestWikiJSException: class TestAPIError: """Test API error classes.""" - + def test_api_error_creation(self): """Test API error with status code and response.""" response = Mock() response.status_code = 500 response.text = "Internal server error" - + exc = APIError("Server error", status_code=500, response=response) assert exc.status_code == 500 assert exc.response == response @@ -53,14 +52,14 @@ class TestAPIError: class TestRateLimitError: """Test rate limit error.""" - + def test_rate_limit_error_with_retry_after(self): """Test rate limit error with retry_after parameter.""" exc = RateLimitError("Rate limit exceeded", retry_after=60) assert exc.status_code == 429 assert exc.retry_after == 60 assert str(exc) == "Rate limit exceeded" - + def test_rate_limit_error_without_retry_after(self): """Test rate limit error without retry_after parameter.""" exc = RateLimitError("Rate limit exceeded") @@ -70,7 +69,7 @@ class TestRateLimitError: class TestCreateAPIError: """Test create_api_error factory function.""" - + def test_create_404_error(self): """Test creating 404 NotFoundError.""" response = Mock() @@ -78,14 +77,14 @@ class TestCreateAPIError: assert isinstance(error, NotFoundError) assert error.status_code == 404 assert error.response == response - + def test_create_403_error(self): """Test creating 403 PermissionError.""" response = Mock() error = create_api_error(403, "Forbidden", response) assert isinstance(error, PermissionError) assert error.status_code == 403 - + def test_create_429_error(self): """Test creating 429 RateLimitError.""" response = Mock() @@ -94,21 +93,21 @@ class TestCreateAPIError: assert error.status_code == 429 # Note: RateLimitError constructor hardcodes status_code=429 # so it doesn't use the passed status_code parameter - + def test_create_400_client_error(self): """Test creating generic 400-level ClientError.""" response = Mock() error = create_api_error(400, "Bad request", response) assert isinstance(error, ClientError) assert error.status_code == 400 - + def test_create_500_server_error(self): """Test creating generic 500-level ServerError.""" response = Mock() error = create_api_error(500, "Server error", response) assert isinstance(error, ServerError) assert error.status_code == 500 - + def test_create_unknown_status_error(self): """Test creating error with unknown status code.""" response = Mock() @@ -119,33 +118,33 @@ class TestCreateAPIError: class TestSimpleExceptions: """Test simple exception classes.""" - + def test_connection_error(self): """Test ConnectionError creation.""" exc = ConnectionError("Connection failed") assert str(exc) == "Connection failed" assert isinstance(exc, WikiJSException) - + def test_timeout_error(self): """Test TimeoutError creation.""" exc = TimeoutError("Request timed out") assert str(exc) == "Request timed out" assert isinstance(exc, WikiJSException) - + def test_authentication_error(self): """Test AuthenticationError creation.""" exc = AuthenticationError("Invalid credentials") assert str(exc) == "Invalid credentials" assert isinstance(exc, WikiJSException) - + def test_configuration_error(self): """Test ConfigurationError creation.""" exc = ConfigurationError("Invalid config") assert str(exc) == "Invalid config" assert isinstance(exc, WikiJSException) - + def test_validation_error(self): """Test ValidationError creation.""" exc = ValidationError("Invalid input") assert str(exc) == "Invalid input" - assert isinstance(exc, WikiJSException) \ No newline at end of file + assert isinstance(exc, WikiJSException) diff --git a/tests/test_integration.py b/tests/test_integration.py index 5bb68c5..03fdd4d 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1,65 +1,66 @@ """Integration tests for the full WikiJS client with Pages API.""" -import pytest from unittest.mock import Mock, patch from wikijs import WikiJSClient from wikijs.endpoints.pages import PagesEndpoint -from wikijs.models.page import Page, PageCreate +from wikijs.models.page import Page class TestWikiJSClientIntegration: """Integration tests for WikiJS client with Pages API.""" - + def test_client_has_pages_endpoint(self): """Test that client has pages endpoint initialized.""" - with patch('wikijs.client.requests.Session'): + with patch("wikijs.client.requests.Session"): client = WikiJSClient("https://test.wiki", auth="test-key") - - assert hasattr(client, 'pages') + + assert hasattr(client, "pages") assert isinstance(client.pages, PagesEndpoint) assert client.pages._client is client - - @patch('wikijs.client.requests.Session') + + @patch("wikijs.client.requests.Session") def test_client_pages_integration(self, mock_session_class): """Test that pages endpoint works through client.""" # Mock the session and response mock_session = Mock() mock_session_class.return_value = mock_session - + mock_response = Mock() mock_response.ok = True mock_response.json.return_value = { "data": { - "pages": [{ - "id": 1, - "title": "Test Page", - "path": "test", - "content": "Content", - "isPublished": True, - "isPrivate": False, - "tags": [], - "locale": "en", - "createdAt": "2023-01-01T00:00:00Z", - "updatedAt": "2023-01-01T00:00:00Z" - }] + "pages": [ + { + "id": 1, + "title": "Test Page", + "path": "test", + "content": "Content", + "isPublished": True, + "isPrivate": False, + "tags": [], + "locale": "en", + "createdAt": "2023-01-01T00:00:00Z", + "updatedAt": "2023-01-01T00:00:00Z", + } + ] } } mock_session.request.return_value = mock_response - + # Create client client = WikiJSClient("https://test.wiki", auth="test-key") - + # Call pages.list() through client pages = client.pages.list() - + # Verify it works assert len(pages) == 1 assert isinstance(pages[0], Page) assert pages[0].title == "Test Page" - + # Verify the request was made mock_session.request.assert_called_once() call_args = mock_session.request.call_args assert call_args[0][0] == "POST" # GraphQL uses POST - assert "/graphql" in call_args[0][1] \ No newline at end of file + assert "/graphql" in call_args[0][1] diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py index 1f3a2fe..3d55c80 100644 --- a/tests/utils/__init__.py +++ b/tests/utils/__init__.py @@ -1 +1 @@ -"""Tests for utility functions.""" \ No newline at end of file +"""Tests for utility functions.""" diff --git a/tests/utils/test_helpers.py b/tests/utils/test_helpers.py index 8cf4ca5..d913952 100644 --- a/tests/utils/test_helpers.py +++ b/tests/utils/test_helpers.py @@ -1,50 +1,56 @@ """Tests for utility helper functions.""" -import pytest from unittest.mock import Mock +import pytest + from wikijs.exceptions import ValidationError from wikijs.utils.helpers import ( - normalize_url, - validate_url, - sanitize_path, build_api_url, - parse_wiki_response, - extract_error_message, chunk_list, + extract_error_message, + normalize_url, + parse_wiki_response, safe_get, + sanitize_path, + validate_url, ) class TestNormalizeUrl: """Test URL normalization.""" - + def test_normalize_url_basic(self): """Test basic URL normalization.""" assert normalize_url("https://wiki.example.com") == "https://wiki.example.com" - + def test_normalize_url_remove_trailing_slash(self): """Test trailing slash removal.""" assert normalize_url("https://wiki.example.com/") == "https://wiki.example.com" - + def test_normalize_url_remove_multiple_trailing_slashes(self): """Test multiple trailing slash removal.""" - assert normalize_url("https://wiki.example.com///") == "https://wiki.example.com" - + assert ( + normalize_url("https://wiki.example.com///") == "https://wiki.example.com" + ) + def test_normalize_url_with_path(self): """Test URL with path normalization.""" - assert normalize_url("https://wiki.example.com/wiki/") == "https://wiki.example.com/wiki" - + assert ( + normalize_url("https://wiki.example.com/wiki/") + == "https://wiki.example.com/wiki" + ) + def test_normalize_url_empty(self): """Test empty URL raises error.""" with pytest.raises(ValidationError, match="Base URL cannot be empty"): normalize_url("") - + def test_normalize_url_none(self): """Test None URL raises error.""" with pytest.raises(ValidationError, match="Base URL cannot be empty"): normalize_url(None) - + def test_normalize_url_invalid_scheme(self): """Test invalid URL scheme gets https:// prepended.""" # The normalize_url function adds https:// to URLs without checking scheme @@ -55,49 +61,52 @@ class TestNormalizeUrl: """Test invalid URL format raises ValidationError.""" with pytest.raises(ValidationError, match="Invalid URL format"): normalize_url("not a valid url with spaces") - + def test_normalize_url_no_scheme(self): """Test URL without scheme gets https:// added.""" result = normalize_url("wiki.example.com") assert result == "https://wiki.example.com" - + def test_normalize_url_with_port(self): """Test URL with port.""" - assert normalize_url("https://wiki.example.com:8080") == "https://wiki.example.com:8080" + assert ( + normalize_url("https://wiki.example.com:8080") + == "https://wiki.example.com:8080" + ) class TestValidateUrl: """Test URL validation.""" - + def test_validate_url_valid_https(self): """Test valid HTTPS URL.""" assert validate_url("https://wiki.example.com") is True - + def test_validate_url_valid_http(self): """Test valid HTTP URL.""" assert validate_url("http://wiki.example.com") is True - + def test_validate_url_with_path(self): """Test valid URL with path.""" assert validate_url("https://wiki.example.com/wiki") is True - + def test_validate_url_with_port(self): """Test valid URL with port.""" assert validate_url("https://wiki.example.com:8080") is True - + def test_validate_url_invalid_scheme(self): """Test invalid URL scheme - validate_url only checks format, not scheme type.""" # validate_url only checks that there's a scheme and netloc, not the scheme type assert validate_url("ftp://wiki.example.com") is True - + def test_validate_url_no_scheme(self): """Test URL without scheme.""" assert validate_url("wiki.example.com") is False - + def test_validate_url_empty(self): """Test empty URL.""" assert validate_url("") is False - + def test_validate_url_none(self): """Test None URL.""" assert validate_url(None) is False @@ -105,24 +114,24 @@ class TestValidateUrl: class TestSanitizePath: """Test path sanitization.""" - + def test_sanitize_path_basic(self): """Test basic path sanitization.""" assert sanitize_path("simple-path") == "simple-path" - + def test_sanitize_path_with_slashes(self): """Test path with slashes.""" assert sanitize_path("/path/to/page/") == "path/to/page" - + def test_sanitize_path_multiple_slashes(self): """Test path with multiple slashes.""" assert sanitize_path("//path///to//page//") == "path/to/page" - + def test_sanitize_path_empty(self): """Test empty path raises error.""" with pytest.raises(ValidationError, match="Path cannot be empty"): sanitize_path("") - + def test_sanitize_path_none(self): """Test None path raises error.""" with pytest.raises(ValidationError, match="Path cannot be empty"): @@ -131,27 +140,27 @@ class TestSanitizePath: class TestBuildApiUrl: """Test API URL building.""" - + def test_build_api_url_basic(self): """Test basic API URL building.""" result = build_api_url("https://wiki.example.com", "/test") assert result == "https://wiki.example.com/test" - + def test_build_api_url_with_trailing_slash(self): """Test API URL building with trailing slash on base.""" result = build_api_url("https://wiki.example.com/", "/test") assert result == "https://wiki.example.com/test" - + def test_build_api_url_without_leading_slash(self): """Test API URL building without leading slash on endpoint.""" result = build_api_url("https://wiki.example.com", "test") assert result == "https://wiki.example.com/test" - + def test_build_api_url_complex_endpoint(self): """Test API URL building with complex endpoint.""" result = build_api_url("https://wiki.example.com", "/api/v1/pages") assert result == "https://wiki.example.com/api/v1/pages" - + def test_build_api_url_empty_endpoint(self): """Test API URL building with empty endpoint.""" result = build_api_url("https://wiki.example.com", "") @@ -160,25 +169,25 @@ class TestBuildApiUrl: class TestParseWikiResponse: """Test Wiki.js response parsing.""" - + def test_parse_wiki_response_with_data(self): """Test parsing response with data field.""" response = {"data": {"pages": []}, "meta": {"total": 0}} result = parse_wiki_response(response) assert result == {"data": {"pages": []}, "meta": {"total": 0}} - + def test_parse_wiki_response_without_data(self): """Test parsing response without data field.""" response = {"pages": [], "total": 0} result = parse_wiki_response(response) assert result == {"pages": [], "total": 0} - + def test_parse_wiki_response_empty(self): """Test parsing empty response.""" response = {} result = parse_wiki_response(response) assert result == {} - + def test_parse_wiki_response_none(self): """Test parsing None response.""" result = parse_wiki_response(None) @@ -187,49 +196,49 @@ class TestParseWikiResponse: class TestExtractErrorMessage: """Test error message extraction.""" - + def test_extract_error_message_json_with_message(self): """Test extracting error from JSON response with message.""" mock_response = Mock() mock_response.text = '{"message": "Not found"}' mock_response.json.return_value = {"message": "Not found"} - + result = extract_error_message(mock_response) assert result == "Not found" - + def test_extract_error_message_json_with_errors_array(self): """Test extracting error from JSON response with error field.""" mock_response = Mock() mock_response.text = '{"error": "Invalid field"}' mock_response.json.return_value = {"error": "Invalid field"} - + result = extract_error_message(mock_response) assert result == "Invalid field" - + def test_extract_error_message_json_with_error_string(self): """Test extracting error from JSON response with error string.""" mock_response = Mock() mock_response.text = '{"error": "Authentication failed"}' mock_response.json.return_value = {"error": "Authentication failed"} - + result = extract_error_message(mock_response) assert result == "Authentication failed" - + def test_extract_error_message_invalid_json(self): """Test extracting error from invalid JSON response.""" mock_response = Mock() mock_response.text = "Invalid JSON response" mock_response.json.side_effect = ValueError("Invalid JSON") - + result = extract_error_message(mock_response) assert result == "Invalid JSON response" - + def test_extract_error_message_empty_response(self): """Test extracting error from empty response.""" mock_response = Mock() mock_response.text = "" mock_response.json.side_effect = ValueError("Empty response") - + result = extract_error_message(mock_response) # Should return either empty string or default error message assert result in ["", "Unknown error"] @@ -237,30 +246,30 @@ class TestExtractErrorMessage: class TestChunkList: """Test list chunking.""" - + def test_chunk_list_basic(self): """Test basic list chunking.""" items = [1, 2, 3, 4, 5, 6] result = chunk_list(items, 2) assert result == [[1, 2], [3, 4], [5, 6]] - + def test_chunk_list_uneven(self): """Test list chunking with uneven division.""" items = [1, 2, 3, 4, 5] result = chunk_list(items, 2) assert result == [[1, 2], [3, 4], [5]] - + def test_chunk_list_larger_chunk_size(self): """Test list chunking with chunk size larger than list.""" items = [1, 2, 3] result = chunk_list(items, 5) assert result == [[1, 2, 3]] - + def test_chunk_list_empty(self): """Test chunking empty list.""" result = chunk_list([], 2) assert result == [] - + def test_chunk_list_chunk_size_one(self): """Test chunking with chunk size of 1.""" items = [1, 2, 3] @@ -270,49 +279,49 @@ class TestChunkList: class TestSafeGet: """Test safe dictionary value retrieval.""" - + def test_safe_get_existing_key(self): """Test getting existing key.""" data = {"key": "value", "nested": {"inner": "data"}} assert safe_get(data, "key") == "value" - + def test_safe_get_missing_key(self): """Test getting missing key with default.""" data = {"key": "value"} assert safe_get(data, "missing") is None - + def test_safe_get_missing_key_with_custom_default(self): """Test getting missing key with custom default.""" data = {"key": "value"} assert safe_get(data, "missing", "default") == "default" - + def test_safe_get_nested_key(self): """Test getting nested key (if supported).""" data = {"nested": {"inner": "data"}} # This might not be supported, but test if it works result = safe_get(data, "nested") assert result == {"inner": "data"} - + def test_safe_get_empty_dict(self): """Test getting from empty dictionary.""" assert safe_get({}, "key") is None - + def test_safe_get_none_data(self): """Test getting from None data.""" with pytest.raises(AttributeError): safe_get(None, "key") - + def test_safe_get_dot_notation(self): """Test safe_get with dot notation.""" data = {"user": {"profile": {"name": "John"}}} assert safe_get(data, "user.profile.name") == "John" - + def test_safe_get_dot_notation_missing(self): """Test safe_get with dot notation for missing key.""" data = {"user": {"profile": {"name": "John"}}} assert safe_get(data, "user.missing.name") is None assert safe_get(data, "user.missing.name", "default") == "default" - + def test_safe_get_dot_notation_non_dict(self): """Test safe_get with dot notation when intermediate value is not dict.""" data = {"user": "not_a_dict"} @@ -321,121 +330,129 @@ class TestSafeGet: class TestUtilityEdgeCases: """Test edge cases for utility functions.""" - + def test_validate_url_with_none(self): """Test validate_url with None input.""" assert validate_url(None) is False - + def test_validate_url_with_exception(self): """Test validate_url when urlparse raises exception.""" # This is hard to trigger, but test the exception path assert validate_url("") is False - + def test_validate_url_malformed_url(self): """Test validate_url with malformed URL that causes exception.""" # Test with a string that could cause urlparse to raise an exception - import sys from unittest.mock import patch - - with patch('wikijs.utils.helpers.urlparse') as mock_urlparse: + + with patch("wikijs.utils.helpers.urlparse") as mock_urlparse: mock_urlparse.side_effect = Exception("Parse error") assert validate_url("http://example.com") is False - + def test_sanitize_path_whitespace_only(self): """Test sanitize_path with whitespace-only input.""" # Whitespace gets stripped and then triggers the empty path check with pytest.raises(ValidationError, match="Path contains no valid characters"): sanitize_path(" ") - + def test_sanitize_path_invalid_characters_only(self): """Test sanitize_path with only invalid characters.""" with pytest.raises(ValidationError, match="Path contains no valid characters"): sanitize_path("!@#$%^&*()") - + def test_sanitize_path_complex_cleanup(self): """Test sanitize_path with complex cleanup needs.""" result = sanitize_path(" //hello world//test// ") assert result == "hello-world/test" - + def test_parse_wiki_response_with_error_dict(self): """Test parse_wiki_response with error dict.""" response = {"error": {"message": "Not found", "code": "404"}} - + from wikijs.exceptions import APIError + with pytest.raises(APIError, match="API Error: Not found"): parse_wiki_response(response) - + def test_parse_wiki_response_with_error_string(self): """Test parse_wiki_response with error string.""" response = {"error": "Simple error message"} - + from wikijs.exceptions import APIError + with pytest.raises(APIError, match="API Error: Simple error message"): parse_wiki_response(response) - + def test_parse_wiki_response_with_errors_array(self): """Test parse_wiki_response with errors array.""" - response = {"errors": [{"message": "GraphQL error"}, {"message": "Another error"}]} - + response = { + "errors": [ + {"message": "GraphQL error"}, + {"message": "Another error"}, + ] + } + from wikijs.exceptions import APIError + with pytest.raises(APIError, match="GraphQL Error: GraphQL error"): parse_wiki_response(response) - + def test_parse_wiki_response_with_non_dict_errors(self): """Test parse_wiki_response with non-dict errors.""" response = {"errors": "String error"} - + from wikijs.exceptions import APIError + with pytest.raises(APIError, match="GraphQL Error: String error"): parse_wiki_response(response) - + def test_parse_wiki_response_non_dict_input(self): """Test parse_wiki_response with non-dict input.""" assert parse_wiki_response("string") == "string" assert parse_wiki_response(42) == 42 assert parse_wiki_response([1, 2, 3]) == [1, 2, 3] - + def test_extract_error_message_with_nested_error(self): """Test extract_error_message with nested error structures.""" mock_response = Mock() mock_response.text = '{"detail": "Validation failed"}' mock_response.json.return_value = {"detail": "Validation failed"} - + result = extract_error_message(mock_response) assert result == "Validation failed" - + def test_extract_error_message_with_msg_field(self): """Test extract_error_message with msg field.""" mock_response = Mock() mock_response.text = '{"msg": "Short message"}' mock_response.json.return_value = {"msg": "Short message"} - + result = extract_error_message(mock_response) assert result == "Short message" - + def test_extract_error_message_long_text(self): """Test extract_error_message with very long response text.""" long_text = "x" * 250 # Longer than 200 chars mock_response = Mock() mock_response.text = long_text mock_response.json.side_effect = ValueError("Invalid JSON") - + result = extract_error_message(mock_response) assert len(result) == 203 # 200 chars + "..." assert result.endswith("...") - + def test_extract_error_message_no_json_no_text(self): """Test extract_error_message with object that has neither json nor text.""" obj = "simple string" result = extract_error_message(obj) assert result == "simple string" - + def test_chunk_list_zero_chunk_size(self): """Test chunk_list with zero chunk size.""" with pytest.raises(ValueError, match="Chunk size must be positive"): chunk_list([1, 2, 3], 0) - + def test_chunk_list_negative_chunk_size(self): """Test chunk_list with negative chunk size.""" with pytest.raises(ValueError, match="Chunk size must be positive"): - chunk_list([1, 2, 3], -1) \ No newline at end of file + chunk_list([1, 2, 3], -1) diff --git a/wikijs/__init__.py b/wikijs/__init__.py index ecaaf7b..d6754c3 100644 --- a/wikijs/__init__.py +++ b/wikijs/__init__.py @@ -5,11 +5,11 @@ instances, including support for pages, users, groups, and system management. Example: Basic usage: - + >>> from wikijs import WikiJSClient >>> client = WikiJSClient('https://wiki.example.com', auth='your-api-key') >>> # API endpoints will be available as development progresses - + Features: - Type-safe data models with validation - Comprehensive error handling @@ -18,21 +18,21 @@ Features: - Context manager support for resource cleanup """ -from .auth import AuthHandler, NoAuth, APIKeyAuth, JWTAuth +from .auth import APIKeyAuth, AuthHandler, JWTAuth, NoAuth from .client import WikiJSClient from .exceptions import ( - WikiJSException, APIError, AuthenticationError, - ConfigurationError, - ValidationError, ClientError, - ServerError, + ConfigurationError, + ConnectionError, NotFoundError, PermissionError, RateLimitError, - ConnectionError, + ServerError, TimeoutError, + ValidationError, + WikiJSException, ) from .models import BaseModel, Page, PageCreate, PageUpdate from .version import __version__, __version_info__ @@ -41,33 +41,29 @@ from .version import __version__, __version_info__ __all__ = [ # Main client "WikiJSClient", - # Authentication "AuthHandler", "NoAuth", - "APIKeyAuth", + "APIKeyAuth", "JWTAuth", - # Data models "BaseModel", - "Page", + "Page", "PageCreate", "PageUpdate", - # Exceptions "WikiJSException", "APIError", - "AuthenticationError", + "AuthenticationError", "ConfigurationError", "ValidationError", "ClientError", - "ServerError", + "ServerError", "NotFoundError", "PermissionError", "RateLimitError", "ConnectionError", "TimeoutError", - # Version info "__version__", "__version_info__", @@ -81,4 +77,10 @@ __description__ = "Professional Python SDK for Wiki.js API integration" __url__ = "https://github.com/yourusername/wikijs-python-sdk" # For type checking -__all__ += ["__author__", "__email__", "__license__", "__description__", "__url__"] \ No newline at end of file +__all__ += [ + "__author__", + "__email__", + "__license__", + "__description__", + "__url__", +] diff --git a/wikijs/auth/__init__.py b/wikijs/auth/__init__.py index f5f0aa1..f5a4777 100644 --- a/wikijs/auth/__init__.py +++ b/wikijs/auth/__init__.py @@ -9,13 +9,13 @@ Supported authentication methods: - No authentication for testing (NoAuth) """ -from .base import AuthHandler, NoAuth from .api_key import APIKeyAuth +from .base import AuthHandler, NoAuth from .jwt import JWTAuth __all__ = [ "AuthHandler", - "NoAuth", + "NoAuth", "APIKeyAuth", "JWTAuth", -] \ No newline at end of file +] diff --git a/wikijs/auth/api_key.py b/wikijs/auth/api_key.py index ea47e08..c159d0a 100644 --- a/wikijs/auth/api_key.py +++ b/wikijs/auth/api_key.py @@ -4,20 +4,20 @@ This module implements API key authentication for Wiki.js instances. API keys are typically used for server-to-server authentication. """ -from typing import Dict, Optional +from typing import Dict from .base import AuthHandler class APIKeyAuth(AuthHandler): """API key authentication handler for Wiki.js. - + This handler implements authentication using an API key, which is included in the Authorization header as a Bearer token. - + Args: api_key: The API key string from Wiki.js admin panel. - + Example: >>> auth = APIKeyAuth("your-api-key-here") >>> client = WikiJSClient("https://wiki.example.com", auth=auth) @@ -25,35 +25,35 @@ class APIKeyAuth(AuthHandler): def __init__(self, api_key: str) -> None: """Initialize API key authentication. - + Args: api_key: The API key from Wiki.js admin panel. - + Raises: ValueError: If api_key is empty or None. """ if not api_key or not api_key.strip(): raise ValueError("API key cannot be empty") - + self._api_key = api_key.strip() def get_headers(self) -> Dict[str, str]: """Get authentication headers with API key. - + Returns: Dict[str, str]: Headers containing the Authorization header. """ return { "Authorization": f"Bearer {self._api_key}", - "Content-Type": "application/json" + "Content-Type": "application/json", } def is_valid(self) -> bool: """Check if API key is valid. - + For API keys, we assume they're valid if they're not empty. Actual validation happens on the server side. - + Returns: bool: True if API key exists, False otherwise. """ @@ -61,29 +61,30 @@ class APIKeyAuth(AuthHandler): def refresh(self) -> None: """Refresh authentication credentials. - + API keys don't typically need refreshing, so this is a no-op. If the API key becomes invalid, a new one must be provided. """ # API keys don't refresh - they're static until manually replaced - pass @property def api_key(self) -> str: """Get the masked API key for logging/debugging. - + Returns: str: Masked API key showing only first 4 and last 4 characters. """ if len(self._api_key) <= 8: return "*" * len(self._api_key) - - return f"{self._api_key[:4]}{'*' * (len(self._api_key) - 8)}{self._api_key[-4:]}" + + return ( + f"{self._api_key[:4]}{'*' * (len(self._api_key) - 8)}{self._api_key[-4:]}" + ) def __repr__(self) -> str: """String representation of the auth handler. - + Returns: str: Safe representation with masked API key. """ - return f"APIKeyAuth(api_key='{self.api_key}')" \ No newline at end of file + return f"APIKeyAuth(api_key='{self.api_key}')" diff --git a/wikijs/auth/base.py b/wikijs/auth/base.py index 93cc6b9..fd85469 100644 --- a/wikijs/auth/base.py +++ b/wikijs/auth/base.py @@ -5,12 +5,12 @@ providing a consistent interface for different authentication methods. """ from abc import ABC, abstractmethod -from typing import Dict, Optional +from typing import Dict class AuthHandler(ABC): """Abstract base class for Wiki.js authentication handlers. - + This class defines the interface that all authentication implementations must follow, ensuring consistent behavior across different auth methods. """ @@ -18,56 +18,54 @@ class AuthHandler(ABC): @abstractmethod def get_headers(self) -> Dict[str, str]: """Get authentication headers for HTTP requests. - + Returns: Dict[str, str]: Dictionary of headers to include in requests. - + Raises: AuthenticationError: If authentication is invalid or expired. """ - pass @abstractmethod def is_valid(self) -> bool: """Check if the current authentication is valid. - + Returns: bool: True if authentication is valid, False otherwise. """ - pass @abstractmethod def refresh(self) -> None: """Refresh the authentication if possible. - + For token-based authentication, this should refresh the token. For API key authentication, this is typically a no-op. - + Raises: AuthenticationError: If refresh fails. """ - pass def validate_credentials(self) -> None: """Validate credentials and refresh if necessary. - + This is a convenience method that checks validity and refreshes if needed. Subclasses can override for custom behavior. - + Raises: AuthenticationError: If credentials are invalid or refresh fails. """ if not self.is_valid(): self.refresh() - + if not self.is_valid(): from ..exceptions import AuthenticationError + raise AuthenticationError("Authentication credentials are invalid") class NoAuth(AuthHandler): """No-authentication handler for testing or public instances. - + This handler provides an empty authentication implementation, useful for testing or when accessing public Wiki.js instances that don't require authentication. @@ -75,7 +73,7 @@ class NoAuth(AuthHandler): def get_headers(self) -> Dict[str, str]: """Return empty headers dict. - + Returns: Dict[str, str]: Empty dictionary. """ @@ -83,7 +81,7 @@ class NoAuth(AuthHandler): def is_valid(self) -> bool: """Always return True for no-auth. - + Returns: bool: Always True. """ @@ -91,7 +89,6 @@ class NoAuth(AuthHandler): def refresh(self) -> None: """No-op for no-auth. - + This method does nothing since there's no authentication to refresh. """ - pass \ No newline at end of file diff --git a/wikijs/auth/jwt.py b/wikijs/auth/jwt.py index ec1f1f0..2fdd236 100644 --- a/wikijs/auth/jwt.py +++ b/wikijs/auth/jwt.py @@ -5,7 +5,7 @@ JWT tokens are typically used for user-based authentication and have expiration """ import time -from datetime import datetime, timedelta +from datetime import timedelta from typing import Dict, Optional from .base import AuthHandler @@ -13,39 +13,39 @@ from .base import AuthHandler class JWTAuth(AuthHandler): """JWT token authentication handler for Wiki.js. - + This handler manages JWT tokens with automatic refresh capabilities. JWT tokens typically expire and need to be refreshed periodically. - + Args: token: The JWT token string. refresh_token: Optional refresh token for automatic renewal. expires_at: Optional expiration timestamp (Unix timestamp). - + Example: >>> auth = JWTAuth("eyJ0eXAiOiJKV1QiLCJhbGc...") >>> client = WikiJSClient("https://wiki.example.com", auth=auth) """ def __init__( - self, - token: str, + self, + token: str, refresh_token: Optional[str] = None, - expires_at: Optional[float] = None + expires_at: Optional[float] = None, ) -> None: """Initialize JWT authentication. - + Args: token: The JWT token string. refresh_token: Optional refresh token for automatic renewal. expires_at: Optional expiration timestamp (Unix timestamp). - + Raises: ValueError: If token is empty or None. """ if not token or not token.strip(): raise ValueError("JWT token cannot be empty") - + self._token = token.strip() self._refresh_token = refresh_token.strip() if refresh_token else None self._expires_at = expires_at @@ -53,66 +53,66 @@ class JWTAuth(AuthHandler): def get_headers(self) -> Dict[str, str]: """Get authentication headers with JWT token. - + Automatically attempts to refresh the token if it's expired. - + Returns: Dict[str, str]: Headers containing the Authorization header. - + Raises: AuthenticationError: If token is expired and cannot be refreshed. """ # Try to refresh if token is near expiration if not self.is_valid(): self.refresh() - + return { "Authorization": f"Bearer {self._token}", - "Content-Type": "application/json" + "Content-Type": "application/json", } def is_valid(self) -> bool: """Check if JWT token is valid and not expired. - + Returns: bool: True if token exists and is not expired, False otherwise. """ if not self._token or not self._token.strip(): return False - + # If no expiration time is set, assume token is valid if self._expires_at is None: return True - + # Check if token is expired (with buffer for refresh) current_time = time.time() return current_time < (self._expires_at - self._refresh_buffer) def refresh(self) -> None: """Refresh the JWT token using the refresh token. - + This method attempts to refresh the JWT token using the refresh token. If no refresh token is available, it raises an AuthenticationError. - + Note: This is a placeholder implementation. In a real implementation, this would make an HTTP request to the Wiki.js token refresh endpoint. - + Raises: AuthenticationError: If refresh token is not available or refresh fails. """ from ..exceptions import AuthenticationError - + if not self._refresh_token: raise AuthenticationError( "JWT token expired and no refresh token available" ) - + # TODO: Implement actual token refresh logic # This would typically involve: # 1. Making a POST request to /auth/refresh endpoint # 2. Sending the refresh token # 3. Updating self._token and self._expires_at with the response - + raise AuthenticationError( "JWT token refresh not yet implemented. " "Please provide a new token or use API key authentication." @@ -120,46 +120,46 @@ class JWTAuth(AuthHandler): def is_expired(self) -> bool: """Check if the JWT token is expired. - + Returns: bool: True if token is expired, False otherwise. """ if self._expires_at is None: return False - + return time.time() >= self._expires_at def time_until_expiry(self) -> Optional[timedelta]: """Get time until token expires. - + Returns: Optional[timedelta]: Time until expiration, or None if no expiration set. """ if self._expires_at is None: return None - + remaining_seconds = self._expires_at - time.time() return timedelta(seconds=max(0, remaining_seconds)) @property def token_preview(self) -> str: """Get a preview of the JWT token for logging/debugging. - + Returns: str: Masked token showing only first and last few characters. """ if not self._token: return "None" - + if len(self._token) <= 20: return "*" * len(self._token) - + return f"{self._token[:10]}...{self._token[-10:]}" def __repr__(self) -> str: """String representation of the auth handler. - + Returns: str: Safe representation with masked token. """ - return f"JWTAuth(token='{self.token_preview}', expires_at={self._expires_at})" \ No newline at end of file + return f"JWTAuth(token='{self.token_preview}', expires_at={self._expires_at})" diff --git a/wikijs/client.py b/wikijs/client.py index e71f70d..64067ba 100644 --- a/wikijs/client.py +++ b/wikijs/client.py @@ -7,7 +7,7 @@ import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry -from .auth import AuthHandler, APIKeyAuth +from .auth import APIKeyAuth, AuthHandler from .endpoints import PagesEndpoint from .exceptions import ( APIError, @@ -17,36 +17,41 @@ from .exceptions import ( TimeoutError, create_api_error, ) -from .utils import normalize_url, build_api_url, parse_wiki_response, extract_error_message +from .utils import ( + build_api_url, + extract_error_message, + normalize_url, + parse_wiki_response, +) class WikiJSClient: """Main client for interacting with Wiki.js API. - + This client provides a high-level interface for all Wiki.js API operations including pages, users, groups, and system management. It handles authentication, error handling, and response parsing automatically. - + Args: base_url: The base URL of your Wiki.js instance auth: Authentication (API key string or auth handler) timeout: Request timeout in seconds (default: 30) verify_ssl: Whether to verify SSL certificates (default: True) user_agent: Custom User-Agent header - + Example: Basic usage with API key: - + >>> client = WikiJSClient('https://wiki.example.com', auth='your-api-key') >>> pages = client.pages.list() >>> page = client.pages.get(123) - + Attributes: base_url: The normalized base URL timeout: Request timeout setting verify_ssl: SSL verification setting """ - + def __init__( self, base_url: str, @@ -57,7 +62,7 @@ class WikiJSClient: ): # Validate and normalize base URL self.base_url = normalize_url(base_url) - + # Store authentication if isinstance(auth, str): # Convert string API key to APIKeyAuth handler @@ -69,77 +74,86 @@ class WikiJSClient: raise ConfigurationError( f"Invalid auth parameter: expected str or AuthHandler, got {type(auth)}" ) - + # Request configuration self.timeout = timeout self.verify_ssl = verify_ssl - self.user_agent = user_agent or f"wikijs-python-sdk/0.1.0" - + self.user_agent = user_agent or "wikijs-python-sdk/0.1.0" + # Initialize HTTP session self._session = self._create_session() - + # Endpoint handlers self.pages = PagesEndpoint(self) # Future endpoints: # self.users = UsersEndpoint(self) # self.groups = GroupsEndpoint(self) - + def _create_session(self) -> requests.Session: """Create configured HTTP session with retry strategy. - + Returns: Configured requests session """ session = requests.Session() - + # Configure retry strategy retry_strategy = Retry( total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], - allowed_methods=["HEAD", "GET", "OPTIONS", "POST", "PUT", "DELETE"], + allowed_methods=[ + "HEAD", + "GET", + "OPTIONS", + "POST", + "PUT", + "DELETE", + ], ) - + adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) - + # Set default headers - session.headers.update({ - "User-Agent": self.user_agent, - "Accept": "application/json", - "Content-Type": "application/json", - }) - + session.headers.update( + { + "User-Agent": self.user_agent, + "Accept": "application/json", + "Content-Type": "application/json", + } + ) + # Set authentication headers if self._auth_handler: # Validate auth and get headers self._auth_handler.validate_credentials() auth_headers = self._auth_handler.get_headers() session.headers.update(auth_headers) - + return session - + def _request( self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json_data: Optional[Dict[str, Any]] = None, - **kwargs + **kwargs, ) -> Dict[str, Any]: """Make HTTP request to Wiki.js API. - + Args: method: HTTP method (GET, POST, PUT, DELETE) endpoint: API endpoint path params: Query parameters json_data: JSON data for request body **kwargs: Additional request parameters - + Returns: Parsed response data - + Raises: AuthenticationError: If authentication fails APIError: If API returns an error @@ -148,44 +162,44 @@ class WikiJSClient: """ # Build full URL url = build_api_url(self.base_url, endpoint) - + # Prepare request arguments request_kwargs = { "timeout": self.timeout, "verify": self.verify_ssl, "params": params, - **kwargs + **kwargs, } - + # Add JSON data if provided if json_data is not None: request_kwargs["json"] = json_data - + try: # Make request response = self._session.request(method, url, **request_kwargs) - + # Handle response return self._handle_response(response) - + except requests.exceptions.Timeout as e: raise TimeoutError(f"Request timed out after {self.timeout} seconds") from e - + except requests.exceptions.ConnectionError as e: raise ConnectionError(f"Failed to connect to {self.base_url}") from e - + except requests.exceptions.RequestException as e: raise APIError(f"Request failed: {str(e)}") from e - + def _handle_response(self, response: requests.Response) -> Dict[str, Any]: """Handle HTTP response and extract data. - + Args: response: HTTP response object - + Returns: Parsed response data - + Raises: AuthenticationError: If authentication fails (401) APIError: If API returns an error @@ -193,31 +207,27 @@ class WikiJSClient: # Handle authentication errors if response.status_code == 401: raise AuthenticationError("Authentication failed - check your API key") - + # Handle other HTTP errors if not response.ok: error_message = extract_error_message(response) - raise create_api_error( - response.status_code, - error_message, - response - ) - + raise create_api_error(response.status_code, error_message, response) + # Parse JSON response try: data = response.json() except json.JSONDecodeError as e: raise APIError(f"Invalid JSON response: {str(e)}") from e - + # Parse Wiki.js specific response format return parse_wiki_response(data) - + def test_connection(self) -> bool: """Test connection to Wiki.js instance. - + Returns: True if connection successful - + Raises: ConfigurationError: If client is not properly configured ConnectionError: If cannot connect to server @@ -225,42 +235,42 @@ class WikiJSClient: """ if not self.base_url: raise ConfigurationError("Base URL not configured") - + if not self._auth_handler: raise ConfigurationError("Authentication not configured") - + try: # Try to hit a basic endpoint (will implement with actual endpoints) # For now, just test basic connectivity - response = self._session.get( - self.base_url, - timeout=self.timeout, - verify=self.verify_ssl + self._session.get( + self.base_url, timeout=self.timeout, verify=self.verify_ssl ) return True - + except requests.exceptions.Timeout: - raise TimeoutError(f"Connection test timed out after {self.timeout} seconds") - + raise TimeoutError( + f"Connection test timed out after {self.timeout} seconds" + ) + except requests.exceptions.ConnectionError as e: raise ConnectionError(f"Cannot connect to {self.base_url}: {str(e)}") - + except Exception as e: raise ConnectionError(f"Connection test failed: {str(e)}") - + def __enter__(self): """Context manager entry.""" return self - + def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit - close session.""" self.close() - + def close(self): """Close the HTTP session and clean up resources.""" if self._session: self._session.close() - + def __repr__(self) -> str: """String representation of client.""" - return f"WikiJSClient(base_url='{self.base_url}')" \ No newline at end of file + return f"WikiJSClient(base_url='{self.base_url}')" diff --git a/wikijs/endpoints/__init__.py b/wikijs/endpoints/__init__.py index 20bfea4..7c0d607 100644 --- a/wikijs/endpoints/__init__.py +++ b/wikijs/endpoints/__init__.py @@ -19,4 +19,4 @@ from .pages import PagesEndpoint __all__ = [ "BaseEndpoint", "PagesEndpoint", -] \ No newline at end of file +] diff --git a/wikijs/endpoints/base.py b/wikijs/endpoints/base.py index ae4e2ca..8d1a2ac 100644 --- a/wikijs/endpoints/base.py +++ b/wikijs/endpoints/base.py @@ -1,6 +1,6 @@ """Base endpoint class for wikijs-python-sdk.""" -from typing import Any, Dict, List, Optional, TYPE_CHECKING +from typing import TYPE_CHECKING, Any, Dict, Optional if TYPE_CHECKING: from ..client import WikiJSClient @@ -8,39 +8,39 @@ if TYPE_CHECKING: class BaseEndpoint: """Base class for all API endpoints. - + This class provides common functionality for making API requests and handling responses across all endpoint implementations. - + Args: client: The WikiJS client instance """ - + def __init__(self, client: "WikiJSClient"): """Initialize endpoint with client reference. - + Args: client: WikiJS client instance """ self._client = client - + def _request( self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json_data: Optional[Dict[str, Any]] = None, - **kwargs + **kwargs, ) -> Dict[str, Any]: """Make HTTP request through the client. - + Args: method: HTTP method (GET, POST, PUT, DELETE) endpoint: API endpoint path params: Query parameters json_data: JSON data for request body **kwargs: Additional request parameters - + Returns: Parsed response data """ @@ -49,94 +49,92 @@ class BaseEndpoint: endpoint=endpoint, params=params, json_data=json_data, - **kwargs + **kwargs, ) - + def _get( - self, - endpoint: str, - params: Optional[Dict[str, Any]] = None, - **kwargs + self, endpoint: str, params: Optional[Dict[str, Any]] = None, **kwargs ) -> Dict[str, Any]: """Make GET request. - + Args: endpoint: API endpoint path params: Query parameters **kwargs: Additional request parameters - + Returns: Parsed response data """ return self._request("GET", endpoint, params=params, **kwargs) - + def _post( self, endpoint: str, json_data: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None, - **kwargs + **kwargs, ) -> Dict[str, Any]: """Make POST request. - + Args: endpoint: API endpoint path json_data: JSON data for request body params: Query parameters **kwargs: Additional request parameters - + Returns: Parsed response data """ - return self._request("POST", endpoint, params=params, json_data=json_data, **kwargs) - + return self._request( + "POST", endpoint, params=params, json_data=json_data, **kwargs + ) + def _put( self, endpoint: str, json_data: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None, - **kwargs + **kwargs, ) -> Dict[str, Any]: """Make PUT request. - + Args: endpoint: API endpoint path json_data: JSON data for request body params: Query parameters **kwargs: Additional request parameters - + Returns: Parsed response data """ - return self._request("PUT", endpoint, params=params, json_data=json_data, **kwargs) - + return self._request( + "PUT", endpoint, params=params, json_data=json_data, **kwargs + ) + def _delete( - self, - endpoint: str, - params: Optional[Dict[str, Any]] = None, - **kwargs + self, endpoint: str, params: Optional[Dict[str, Any]] = None, **kwargs ) -> Dict[str, Any]: """Make DELETE request. - + Args: endpoint: API endpoint path params: Query parameters **kwargs: Additional request parameters - + Returns: Parsed response data """ return self._request("DELETE", endpoint, params=params, **kwargs) - + def _build_endpoint(self, *parts: str) -> str: """Build endpoint path from parts. - + Args: *parts: Path components - + Returns: Formatted endpoint path """ # Remove empty parts and join with / clean_parts = [str(part).strip("/") for part in parts if part] - return "/" + "/".join(clean_parts) \ No newline at end of file + return "/" + "/".join(clean_parts) diff --git a/wikijs/endpoints/pages.py b/wikijs/endpoints/pages.py index 882fb82..97633ac 100644 --- a/wikijs/endpoints/pages.py +++ b/wikijs/endpoints/pages.py @@ -9,20 +9,20 @@ from .base import BaseEndpoint class PagesEndpoint(BaseEndpoint): """Endpoint for Wiki.js Pages API operations. - + This endpoint provides methods for creating, reading, updating, and deleting wiki pages through the Wiki.js GraphQL API. - + Example: >>> client = WikiJSClient('https://wiki.example.com', auth='api-key') >>> pages = client.pages - >>> + >>> >>> # List all pages >>> all_pages = pages.list() - >>> + >>> >>> # Get a specific page >>> page = pages.get(123) - >>> + >>> >>> # Create a new page >>> new_page_data = PageCreate( ... title="Getting Started", @@ -30,15 +30,15 @@ class PagesEndpoint(BaseEndpoint): ... content="# Welcome\\n\\nThis is your first page!" ... ) >>> created_page = pages.create(new_page_data) - >>> + >>> >>> # Update an existing page >>> update_data = PageUpdate(title="Updated Title") >>> updated_page = pages.update(123, update_data) - >>> + >>> >>> # Delete a page >>> pages.delete(123) """ - + def list( self, limit: Optional[int] = None, @@ -48,10 +48,10 @@ class PagesEndpoint(BaseEndpoint): locale: Optional[str] = None, author_id: Optional[int] = None, order_by: str = "title", - order_direction: str = "ASC" + order_direction: str = "ASC", ) -> List[Page]: """List pages with optional filtering. - + Args: limit: Maximum number of pages to return offset: Number of pages to skip @@ -61,10 +61,10 @@ class PagesEndpoint(BaseEndpoint): author_id: Author ID to filter by order_by: Field to order by (title, created_at, updated_at) order_direction: Order direction (ASC or DESC) - + Returns: List of Page objects - + Raises: APIError: If the API request fails ValidationError: If parameters are invalid @@ -72,16 +72,18 @@ class PagesEndpoint(BaseEndpoint): # Validate parameters if limit is not None and limit < 1: raise ValidationError("limit must be greater than 0") - + if offset is not None and offset < 0: raise ValidationError("offset must be non-negative") - + if order_by not in ["title", "created_at", "updated_at", "path"]: - raise ValidationError("order_by must be one of: title, created_at, updated_at, path") - + raise ValidationError( + "order_by must be one of: title, created_at, updated_at, path" + ) + if order_direction not in ["ASC", "DESC"]: raise ValidationError("order_direction must be ASC or DESC") - + # Build GraphQL query using actual Wiki.js schema query = """ query { @@ -97,18 +99,16 @@ class PagesEndpoint(BaseEndpoint): } } """ - + # Make request (no variables needed for simple list query) - response = self._post("/graphql", json_data={ - "query": query - }) - + response = self._post("/graphql", json_data={"query": query}) + # Parse response if "errors" in response: raise APIError(f"GraphQL errors: {response['errors']}") - + pages_data = response.get("data", {}).get("pages", {}).get("list", []) - + # Convert to Page objects pages = [] for page_data in pages_data: @@ -119,25 +119,25 @@ class PagesEndpoint(BaseEndpoint): pages.append(page) except Exception as e: raise APIError(f"Failed to parse page data: {str(e)}") from e - + return pages - + def get(self, page_id: int) -> Page: """Get a specific page by ID. - + Args: page_id: The page ID - + Returns: Page object - + Raises: APIError: If the page is not found or request fails ValidationError: If page_id is invalid """ if not isinstance(page_id, int) or page_id < 1: raise ValidationError("page_id must be a positive integer") - + # Build GraphQL query using actual Wiki.js schema query = """ query($id: Int!) { @@ -164,48 +164,48 @@ class PagesEndpoint(BaseEndpoint): } } """ - + # Make request - response = self._post("/graphql", json_data={ - "query": query, - "variables": {"id": page_id} - }) - + response = self._post( + "/graphql", + json_data={"query": query, "variables": {"id": page_id}}, + ) + # Parse response if "errors" in response: raise APIError(f"GraphQL errors: {response['errors']}") - + page_data = response.get("data", {}).get("pages", {}).get("single") if not page_data: raise APIError(f"Page with ID {page_id} not found") - + # Convert to Page object try: normalized_data = self._normalize_page_data(page_data) return Page(**normalized_data) except Exception as e: raise APIError(f"Failed to parse page data: {str(e)}") from e - + def get_by_path(self, path: str, locale: str = "en") -> Page: """Get a page by its path. - + Args: path: The page path (e.g., "getting-started") locale: The page locale (default: "en") - + Returns: Page object - + Raises: APIError: If the page is not found or request fails ValidationError: If path is invalid """ if not path or not isinstance(path, str): raise ValidationError("path must be a non-empty string") - + # Normalize path path = path.strip("/") - + # Build GraphQL query query = """ query($path: String!, $locale: String!) { @@ -228,37 +228,40 @@ class PagesEndpoint(BaseEndpoint): } } """ - + # Make request - response = self._post("/graphql", json_data={ - "query": query, - "variables": {"path": path, "locale": locale} - }) - + response = self._post( + "/graphql", + json_data={ + "query": query, + "variables": {"path": path, "locale": locale}, + }, + ) + # Parse response if "errors" in response: raise APIError(f"GraphQL errors: {response['errors']}") - + page_data = response.get("data", {}).get("pageByPath") if not page_data: raise APIError(f"Page with path '{path}' not found") - + # Convert to Page object try: normalized_data = self._normalize_page_data(page_data) return Page(**normalized_data) except Exception as e: raise APIError(f"Failed to parse page data: {str(e)}") from e - + def create(self, page_data: Union[PageCreate, Dict[str, Any]]) -> Page: """Create a new page. - + Args: page_data: Page creation data (PageCreate object or dict) - + Returns: Created Page object - + Raises: APIError: If page creation fails ValidationError: If page data is invalid @@ -271,7 +274,7 @@ class PagesEndpoint(BaseEndpoint): raise ValidationError(f"Invalid page data: {str(e)}") from e elif not isinstance(page_data, PageCreate): raise ValidationError("page_data must be PageCreate object or dict") - + # Build GraphQL mutation using actual Wiki.js schema mutation = """ mutation($content: String!, $description: String!, $editor: String!, $isPublished: Boolean!, $isPrivate: Boolean!, $locale: String!, $path: String!, $tags: [String]!, $title: String!) { @@ -306,69 +309,67 @@ class PagesEndpoint(BaseEndpoint): } } """ - + # Build variables from page data variables = { "title": page_data.title, "path": page_data.path, "content": page_data.content, - "description": page_data.description or f"Created via SDK: {page_data.title}", + "description": page_data.description + or f"Created via SDK: {page_data.title}", "isPublished": page_data.is_published, "isPrivate": page_data.is_private, "tags": page_data.tags, "locale": page_data.locale, - "editor": page_data.editor + "editor": page_data.editor, } - + # Make request - response = self._post("/graphql", json_data={ - "query": mutation, - "variables": variables - }) - + response = self._post( + "/graphql", json_data={"query": mutation, "variables": variables} + ) + # Parse response if "errors" in response: raise APIError(f"Failed to create page: {response['errors']}") - + create_result = response.get("data", {}).get("pages", {}).get("create", {}) response_result = create_result.get("responseResult", {}) - + if not response_result.get("succeeded"): error_msg = response_result.get("message", "Unknown error") raise APIError(f"Page creation failed: {error_msg}") - + created_page_data = create_result.get("page") if not created_page_data: raise APIError("Page creation failed - no page data returned") - + # Convert to Page object try: normalized_data = self._normalize_page_data(created_page_data) return Page(**normalized_data) except Exception as e: raise APIError(f"Failed to parse created page data: {str(e)}") from e - + def update( - self, - page_id: int, - page_data: Union[PageUpdate, Dict[str, Any]] + self, page_id: int, page_data: Union[PageUpdate, Dict[str, Any]] ) -> Page: """Update an existing page. - + Args: page_id: The page ID page_data: Page update data (PageUpdate object or dict) - + Returns: Updated Page object - + Raises: APIError: If page update fails ValidationError: If parameters are invalid """ if not isinstance(page_id, int) or page_id < 1: raise ValidationError("page_id must be a positive integer") - + # Convert to PageUpdate if needed if isinstance(page_data, dict): try: @@ -377,7 +378,7 @@ class PagesEndpoint(BaseEndpoint): raise ValidationError(f"Invalid page data: {str(e)}") from e elif not isinstance(page_data, PageUpdate): raise ValidationError("page_data must be PageUpdate object or dict") - + # Build GraphQL mutation mutation = """ mutation($id: Int!, $title: String, $content: String, $description: String, $isPublished: Boolean, $isPrivate: Boolean, $tags: [String]) { @@ -408,10 +409,10 @@ class PagesEndpoint(BaseEndpoint): } } """ - + # Build variables (only include non-None values) variables = {"id": page_id} - + if page_data.title is not None: variables["title"] = page_data.title if page_data.content is not None: @@ -424,44 +425,43 @@ class PagesEndpoint(BaseEndpoint): variables["isPrivate"] = page_data.is_private if page_data.tags is not None: variables["tags"] = page_data.tags - + # Make request - response = self._post("/graphql", json_data={ - "query": mutation, - "variables": variables - }) - + response = self._post( + "/graphql", json_data={"query": mutation, "variables": variables} + ) + # Parse response if "errors" in response: raise APIError(f"Failed to update page: {response['errors']}") - + updated_page_data = response.get("data", {}).get("updatePage") if not updated_page_data: raise APIError("Page update failed - no data returned") - + # Convert to Page object try: normalized_data = self._normalize_page_data(updated_page_data) return Page(**normalized_data) except Exception as e: raise APIError(f"Failed to parse updated page data: {str(e)}") from e - + def delete(self, page_id: int) -> bool: """Delete a page. - + Args: page_id: The page ID - + Returns: True if deletion was successful - + Raises: APIError: If page deletion fails ValidationError: If page_id is invalid """ if not isinstance(page_id, int) or page_id < 1: raise ValidationError("page_id must be a positive integer") - + # Build GraphQL mutation mutation = """ mutation($id: Int!) { @@ -471,118 +471,116 @@ class PagesEndpoint(BaseEndpoint): } } """ - + # Make request - response = self._post("/graphql", json_data={ - "query": mutation, - "variables": {"id": page_id} - }) - + response = self._post( + "/graphql", + json_data={"query": mutation, "variables": {"id": page_id}}, + ) + # Parse response if "errors" in response: raise APIError(f"Failed to delete page: {response['errors']}") - + delete_result = response.get("data", {}).get("deletePage", {}) success = delete_result.get("success", False) - + if not success: message = delete_result.get("message", "Unknown error") raise APIError(f"Page deletion failed: {message}") - + return True - + def search( self, query: str, limit: Optional[int] = None, - locale: Optional[str] = None + locale: Optional[str] = None, ) -> List[Page]: """Search for pages by content and title. - + Args: query: Search query string limit: Maximum number of results to return locale: Locale to search in - + Returns: List of matching Page objects - + Raises: APIError: If search fails ValidationError: If parameters are invalid """ if not query or not isinstance(query, str): raise ValidationError("query must be a non-empty string") - + if limit is not None and limit < 1: raise ValidationError("limit must be greater than 0") - + # Use the list method with search parameter - return self.list( - search=query, - limit=limit, - locale=locale - ) - + return self.list(search=query, limit=limit, locale=locale) + def get_by_tags( self, tags: List[str], match_all: bool = True, - limit: Optional[int] = None + limit: Optional[int] = None, ) -> List[Page]: """Get pages by tags. - + Args: tags: List of tags to search for match_all: If True, pages must have ALL tags. If False, ANY tag matches limit: Maximum number of results to return - + Returns: List of matching Page objects - + Raises: APIError: If request fails ValidationError: If parameters are invalid """ if not tags or not isinstance(tags, list): raise ValidationError("tags must be a non-empty list") - + if limit is not None and limit < 1: raise ValidationError("limit must be greater than 0") - + # For match_all=True, use the tags parameter directly if match_all: return self.list(tags=tags, limit=limit) - + # For match_all=False, we need a more complex query # This would require a custom GraphQL query or multiple requests # For now, implement a simple approach - all_pages = self.list(limit=limit * 2 if limit else None) # Get more pages to filter - + all_pages = self.list( + limit=limit * 2 if limit else None + ) # Get more pages to filter + matching_pages = [] for page in all_pages: if any(tag.lower() in [t.lower() for t in page.tags] for tag in tags): matching_pages.append(page) if limit and len(matching_pages) >= limit: break - + return matching_pages - + def _normalize_page_data(self, page_data: Dict[str, Any]) -> Dict[str, Any]: """Normalize page data from API response to model format. - + Args: page_data: Raw page data from API - + Returns: Normalized data for Page model """ normalized = {} - + # Map API field names to model field names field_mapping = { "id": "id", - "title": "title", + "title": "title", "path": "path", "content": "content", "description": "description", @@ -590,17 +588,17 @@ class PagesEndpoint(BaseEndpoint): "isPrivate": "is_private", "locale": "locale", "authorId": "author_id", - "authorName": "author_name", + "authorName": "author_name", "authorEmail": "author_email", "editor": "editor", "createdAt": "created_at", - "updatedAt": "updated_at" + "updatedAt": "updated_at", } - + for api_field, model_field in field_mapping.items(): if api_field in page_data: normalized[model_field] = page_data[api_field] - + # Handle tags - convert from Wiki.js format if "tags" in page_data: if isinstance(page_data["tags"], list): @@ -616,5 +614,5 @@ class PagesEndpoint(BaseEndpoint): normalized["tags"] = [] else: normalized["tags"] = [] - - return normalized \ No newline at end of file + + return normalized diff --git a/wikijs/exceptions.py b/wikijs/exceptions.py index 418e170..c1916f1 100644 --- a/wikijs/exceptions.py +++ b/wikijs/exceptions.py @@ -5,7 +5,7 @@ from typing import Any, Dict, Optional class WikiJSException(Exception): """Base exception for all SDK errors.""" - + def __init__(self, message: str, details: Optional[Dict[str, Any]] = None): super().__init__(message) self.message = message @@ -14,17 +14,15 @@ class WikiJSException(Exception): class ConfigurationError(WikiJSException): """Raised when there's an issue with SDK configuration.""" - pass class AuthenticationError(WikiJSException): """Raised when authentication fails.""" - pass class ValidationError(WikiJSException): """Raised when input validation fails.""" - + def __init__(self, message: str, field: Optional[str] = None, value: Any = None): super().__init__(message) self.field = field @@ -33,13 +31,13 @@ class ValidationError(WikiJSException): class APIError(WikiJSException): """Base class for API-related errors.""" - + def __init__( - self, - message: str, - status_code: Optional[int] = None, + self, + message: str, + status_code: Optional[int] = None, response: Optional[Any] = None, - details: Optional[Dict[str, Any]] = None + details: Optional[Dict[str, Any]] = None, ): super().__init__(message, details) self.status_code = status_code @@ -48,52 +46,46 @@ class APIError(WikiJSException): class ClientError(APIError): """Raised for 4xx HTTP status codes (client errors).""" - pass class ServerError(APIError): """Raised for 5xx HTTP status codes (server errors).""" - pass class NotFoundError(ClientError): """Raised when a requested resource is not found (404).""" - pass class PermissionError(ClientError): """Raised when access is forbidden (403).""" - pass class RateLimitError(ClientError): """Raised when rate limit is exceeded (429).""" - + def __init__(self, message: str, retry_after: Optional[int] = None, **kwargs): # Remove status_code from kwargs if present to avoid duplicate argument - kwargs.pop('status_code', None) + kwargs.pop("status_code", None) super().__init__(message, status_code=429, **kwargs) self.retry_after = retry_after class ConnectionError(WikiJSException): """Raised when there's a connection issue.""" - pass class TimeoutError(WikiJSException): """Raised when a request times out.""" - pass def create_api_error(status_code: int, message: str, response: Any = None) -> APIError: """Create appropriate API error based on status code. - + Args: status_code: HTTP status code message: Error message response: Raw response object - + Returns: Appropriate APIError subclass instance """ @@ -108,4 +100,4 @@ def create_api_error(status_code: int, message: str, response: Any = None) -> AP elif 500 <= status_code < 600: return ServerError(message, status_code=status_code, response=response) else: - return APIError(message, status_code=status_code, response=response) \ No newline at end of file + return APIError(message, status_code=status_code, response=response) diff --git a/wikijs/models/__init__.py b/wikijs/models/__init__.py index 51ccbee..3d9ec83 100644 --- a/wikijs/models/__init__.py +++ b/wikijs/models/__init__.py @@ -6,6 +6,6 @@ from .page import Page, PageCreate, PageUpdate __all__ = [ "BaseModel", "Page", - "PageCreate", + "PageCreate", "PageUpdate", -] \ No newline at end of file +] diff --git a/wikijs/models/base.py b/wikijs/models/base.py index 7e4551c..70b58cf 100644 --- a/wikijs/models/base.py +++ b/wikijs/models/base.py @@ -3,19 +3,20 @@ from datetime import datetime from typing import Any, Dict, Optional -from pydantic import BaseModel as PydanticBaseModel, ConfigDict +from pydantic import BaseModel as PydanticBaseModel +from pydantic import ConfigDict class BaseModel(PydanticBaseModel): """Base model with common functionality for all data models. - + Provides: - Automatic validation via Pydantic - JSON serialization/deserialization - Field aliases for API compatibility - Consistent datetime handling """ - + model_config = ConfigDict( # Allow population by field name or alias populate_by_name=True, @@ -26,52 +27,50 @@ class BaseModel(PydanticBaseModel): # Allow extra fields for forward compatibility extra="ignore", # Serialize datetime as ISO format - json_encoders={ - datetime: lambda v: v.isoformat() if v else None - } + json_encoders={datetime: lambda v: v.isoformat() if v else None}, ) - + def to_dict(self, exclude_none: bool = True) -> Dict[str, Any]: """Convert model to dictionary. - + Args: exclude_none: Whether to exclude None values - + Returns: Dictionary representation of the model """ return self.model_dump(exclude_none=exclude_none, by_alias=True) - + def to_json(self, exclude_none: bool = True) -> str: """Convert model to JSON string. - + Args: exclude_none: Whether to exclude None values - + Returns: JSON string representation of the model """ return self.model_dump_json(exclude_none=exclude_none, by_alias=True) - + @classmethod def from_dict(cls, data: Dict[str, Any]) -> "BaseModel": """Create model instance from dictionary. - + Args: data: Dictionary data - + Returns: Model instance """ return cls(**data) - + @classmethod def from_json(cls, json_str: str) -> "BaseModel": """Create model instance from JSON string. - + Args: json_str: JSON string - + Returns: Model instance """ @@ -80,11 +79,11 @@ class BaseModel(PydanticBaseModel): class TimestampedModel(BaseModel): """Base model with timestamp fields.""" - + created_at: Optional[datetime] = None updated_at: Optional[datetime] = None - + @property def is_new(self) -> bool: """Check if this is a new (unsaved) model.""" - return self.created_at is None \ No newline at end of file + return self.created_at is None diff --git a/wikijs/models/page.py b/wikijs/models/page.py index 023cf91..8454294 100644 --- a/wikijs/models/page.py +++ b/wikijs/models/page.py @@ -1,7 +1,6 @@ """Page-related data models for wikijs-python-sdk.""" import re -from datetime import datetime from typing import List, Optional from pydantic import Field, validator @@ -11,89 +10,89 @@ from .base import BaseModel, TimestampedModel class Page(TimestampedModel): """Represents a Wiki.js page. - + This model contains all the data for a wiki page including content, metadata, and computed properties. """ - + id: int = Field(..., description="Unique page identifier") title: str = Field(..., description="Page title") path: str = Field(..., description="Page path/slug") content: Optional[str] = Field(None, description="Page content") - + # Optional fields that may be present description: Optional[str] = Field(None, description="Page description") is_published: bool = Field(True, description="Whether page is published") is_private: bool = Field(False, description="Whether page is private") - + # Metadata tags: List[str] = Field(default_factory=list, description="Page tags") locale: str = Field("en", description="Page locale") - + # Author information author_id: Optional[int] = Field(None, alias="authorId") author_name: Optional[str] = Field(None, alias="authorName") author_email: Optional[str] = Field(None, alias="authorEmail") - - # Editor information + + # Editor information editor: Optional[str] = Field(None, description="Editor used") - + @validator("path") def validate_path(cls, v): """Validate page path format.""" if not v: raise ValueError("Path cannot be empty") - + # Remove leading/trailing slashes and normalize v = v.strip("/") - + # Check for valid characters (letters, numbers, hyphens, underscores, slashes) if not re.match(r"^[a-zA-Z0-9\-_/]+$", v): raise ValueError("Path contains invalid characters") - + return v - + @validator("title") def validate_title(cls, v): """Validate page title.""" if not v or not v.strip(): raise ValueError("Title cannot be empty") - + # Limit title length if len(v) > 255: raise ValueError("Title cannot exceed 255 characters") - + return v.strip() - + @property def word_count(self) -> int: """Calculate word count from content.""" if not self.content: return 0 - + # Simple word count - split on whitespace words = self.content.split() return len(words) - + @property def reading_time(self) -> int: """Estimate reading time in minutes (assuming 200 words per minute).""" return max(1, self.word_count // 200) - + @property def url_path(self) -> str: """Get the full URL path for this page.""" return f"/{self.path}" - + def extract_headings(self) -> List[str]: """Extract markdown headings from content. - + Returns: List of heading text (without # markers) """ if not self.content: return [] - + headings = [] for line in self.content.split("\n"): line = line.strip() @@ -102,15 +101,15 @@ class Page(TimestampedModel): heading = re.sub(r"^#+\s*", "", line).strip() if heading: headings.append(heading) - + return headings - + def has_tag(self, tag: str) -> bool: """Check if page has a specific tag. - + Args: tag: Tag to check for - + Returns: True if page has the tag """ @@ -119,70 +118,70 @@ class Page(TimestampedModel): class PageCreate(BaseModel): """Data model for creating a new page.""" - + title: str = Field(..., description="Page title") path: str = Field(..., description="Page path/slug") content: str = Field(..., description="Page content") - + # Optional fields description: Optional[str] = Field(None, description="Page description") is_published: bool = Field(True, description="Whether to publish immediately") is_private: bool = Field(False, description="Whether page should be private") - + tags: List[str] = Field(default_factory=list, description="Page tags") locale: str = Field("en", description="Page locale") editor: str = Field("markdown", description="Editor to use") - + @validator("path") def validate_path(cls, v): """Validate page path format.""" if not v: raise ValueError("Path cannot be empty") - + # Remove leading/trailing slashes and normalize v = v.strip("/") - + # Check for valid characters if not re.match(r"^[a-zA-Z0-9\-_/]+$", v): raise ValueError("Path contains invalid characters") - + return v - + @validator("title") def validate_title(cls, v): """Validate page title.""" if not v or not v.strip(): raise ValueError("Title cannot be empty") - + if len(v) > 255: raise ValueError("Title cannot exceed 255 characters") - + return v.strip() class PageUpdate(BaseModel): """Data model for updating an existing page.""" - + # All fields optional for partial updates title: Optional[str] = Field(None, description="Page title") content: Optional[str] = Field(None, description="Page content") description: Optional[str] = Field(None, description="Page description") - + is_published: Optional[bool] = Field(None, description="Publication status") is_private: Optional[bool] = Field(None, description="Privacy status") - + tags: Optional[List[str]] = Field(None, description="Page tags") - + @validator("title") def validate_title(cls, v): """Validate page title if provided.""" if v is not None: if not v.strip(): raise ValueError("Title cannot be empty") - + if len(v) > 255: raise ValueError("Title cannot exceed 255 characters") - + return v.strip() - - return v \ No newline at end of file + + return v diff --git a/wikijs/utils/__init__.py b/wikijs/utils/__init__.py index bcc67bd..621665b 100644 --- a/wikijs/utils/__init__.py +++ b/wikijs/utils/__init__.py @@ -1,23 +1,23 @@ """Utility functions for wikijs-python-sdk.""" from .helpers import ( + build_api_url, + chunk_list, + extract_error_message, normalize_url, + parse_wiki_response, + safe_get, sanitize_path, validate_url, - build_api_url, - parse_wiki_response, - extract_error_message, - chunk_list, - safe_get, ) __all__ = [ "normalize_url", - "sanitize_path", + "sanitize_path", "validate_url", "build_api_url", "parse_wiki_response", "extract_error_message", "chunk_list", "safe_get", -] \ No newline at end of file +] diff --git a/wikijs/utils/helpers.py b/wikijs/utils/helpers.py index ed0a242..7dea768 100644 --- a/wikijs/utils/helpers.py +++ b/wikijs/utils/helpers.py @@ -1,7 +1,7 @@ """Helper utilities for wikijs-python-sdk.""" import re -from typing import Any, Dict, Optional +from typing import Any, Dict from urllib.parse import urljoin, urlparse from ..exceptions import APIError, ValidationError @@ -9,37 +9,37 @@ from ..exceptions import APIError, ValidationError def normalize_url(base_url: str) -> str: """Normalize a base URL for API usage. - + Args: base_url: Base URL to normalize - + Returns: Normalized URL without trailing slash - + Raises: ValidationError: If URL is invalid """ if not base_url: raise ValidationError("Base URL cannot be empty") - + # Add https:// if no scheme provided if not base_url.startswith(("http://", "https://")): base_url = f"https://{base_url}" - + # Validate URL format if not validate_url(base_url): raise ValidationError(f"Invalid URL format: {base_url}") - + # Remove trailing slash return base_url.rstrip("/") def validate_url(url: str) -> bool: """Validate URL format. - + Args: url: URL to validate - + Returns: True if URL is valid """ @@ -52,72 +52,72 @@ def validate_url(url: str) -> bool: def sanitize_path(path: str) -> str: """Sanitize a wiki page path. - + Args: path: Path to sanitize - + Returns: Sanitized path - + Raises: ValidationError: If path is invalid """ if not path: raise ValidationError("Path cannot be empty") - + # Remove leading/trailing slashes and whitespace path = path.strip().strip("/") - + # Replace spaces with hyphens path = re.sub(r"\s+", "-", path) - + # Remove invalid characters, keep only alphanumeric, hyphens, underscores, slashes path = re.sub(r"[^a-zA-Z0-9\-_/]", "", path) - + # Remove multiple consecutive hyphens or slashes path = re.sub(r"[-/]+", lambda m: m.group(0)[0], path) - + if not path: raise ValidationError("Path contains no valid characters") - + return path def build_api_url(base_url: str, endpoint: str) -> str: """Build full API URL from base URL and endpoint. - + Args: base_url: Base URL (already normalized) endpoint: API endpoint path - + Returns: Full API URL """ # Ensure endpoint starts with / if not endpoint.startswith("/"): endpoint = f"/{endpoint}" - + # Wiki.js API is typically at /graphql, but we'll use REST-style for now api_base = f"{base_url}/api" - + return urljoin(api_base, endpoint.lstrip("/")) def parse_wiki_response(response_data: Dict[str, Any]) -> Dict[str, Any]: """Parse Wiki.js API response data. - + Args: response_data: Raw response data from API - + Returns: Parsed response data - + Raises: APIError: If response indicates an error """ if not isinstance(response_data, dict): return response_data - + # Check for error indicators if "error" in response_data: error_info = response_data["error"] @@ -127,26 +127,30 @@ def parse_wiki_response(response_data: Dict[str, Any]) -> Dict[str, Any]: else: message = str(error_info) code = None - + raise APIError(f"API Error: {message}", details={"code": code}) - + # Handle GraphQL-style errors if "errors" in response_data: errors = response_data["errors"] if errors: first_error = errors[0] if isinstance(errors, list) else errors - message = first_error.get("message", "GraphQL error") if isinstance(first_error, dict) else str(first_error) + message = ( + first_error.get("message", "GraphQL error") + if isinstance(first_error, dict) + else str(first_error) + ) raise APIError(f"GraphQL Error: {message}", details={"errors": errors}) - + return response_data def extract_error_message(response: Any) -> str: """Extract error message from response. - - Args: + + Args: response: Response object or data - + Returns: Error message string """ @@ -160,50 +164,52 @@ def extract_error_message(response: Any) -> str: return str(data[field]) except Exception: pass - + if hasattr(response, "text"): - return response.text[:200] + "..." if len(response.text) > 200 else response.text - + return ( + response.text[:200] + "..." if len(response.text) > 200 else response.text + ) + return str(response) def chunk_list(items: list, chunk_size: int) -> list: """Split list into chunks of specified size. - + Args: items: List to chunk chunk_size: Size of each chunk - + Returns: List of chunks """ if chunk_size <= 0: raise ValueError("Chunk size must be positive") - - return [items[i:i + chunk_size] for i in range(0, len(items), chunk_size)] + + return [items[i : i + chunk_size] for i in range(0, len(items), chunk_size)] def safe_get(data: Dict[str, Any], key: str, default: Any = None) -> Any: """Safely get value from dictionary with dot notation support. - + Args: data: Dictionary to get value from key: Key (supports dot notation like "user.name") default: Default value if key not found - + Returns: Value or default """ if "." not in key: return data.get(key, default) - + keys = key.split(".") current = data - + for k in keys: if isinstance(current, dict) and k in current: current = current[k] else: return default - - return current \ No newline at end of file + + return current diff --git a/wikijs/version.py b/wikijs/version.py index 62e7e77..3234af5 100644 --- a/wikijs/version.py +++ b/wikijs/version.py @@ -4,4 +4,4 @@ __version__ = "0.1.0" __version_info__ = (0, 1, 0) # Version history -# 0.1.0 - MVP Release: Basic Wiki.js integration with Pages API \ No newline at end of file +# 0.1.0 - MVP Release: Basic Wiki.js integration with Pages API