diff --git a/pyproject.toml b/pyproject.toml index fc6690b..52bc987 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -127,7 +127,7 @@ addopts = [ "--cov-report=term-missing", "--cov-report=html", "--cov-report=xml", - "--cov-fail-under=82", + "--cov-fail-under=84", ] markers = [ "unit: Unit tests", diff --git a/tests/endpoints/test_groups_extended.py b/tests/endpoints/test_groups_extended.py new file mode 100644 index 0000000..1fc7697 --- /dev/null +++ b/tests/endpoints/test_groups_extended.py @@ -0,0 +1,277 @@ +"""Extended tests for Groups endpoint to improve coverage.""" + +from unittest.mock import Mock +import pytest + +from wikijs.endpoints import GroupsEndpoint +from wikijs.exceptions import APIError, ValidationError +from wikijs.models import GroupCreate, GroupUpdate + + +class TestGroupsEndpointExtended: + """Extended test suite for GroupsEndpoint.""" + + @pytest.fixture + def client(self): + """Create mock client.""" + mock_client = Mock() + mock_client.base_url = "https://wiki.example.com" + mock_client._request = Mock() + return mock_client + + @pytest.fixture + def endpoint(self, client): + """Create GroupsEndpoint instance.""" + return GroupsEndpoint(client) + + def test_list_groups_api_error(self, endpoint): + """Test list groups with API error.""" + mock_response = {"errors": [{"message": "API error"}]} + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="API error"): + endpoint.list() + + def test_get_group_not_found(self, endpoint): + """Test get group when not found.""" + mock_response = { + "data": { + "groups": { + "single": None + } + } + } + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="not found"): + endpoint.get(999) + + def test_get_group_api_error(self, endpoint): + """Test get group with API error.""" + mock_response = {"errors": [{"message": "Access denied"}]} + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Access denied"): + endpoint.get(1) + + def test_create_group_with_dict(self, endpoint): + """Test creating group with dictionary.""" + group_dict = {"name": "Editors", "permissions": ["read:pages"]} + + mock_response = { + "data": { + "groups": { + "create": { + "responseResult": {"succeeded": True}, + "group": { + "id": 2, + "name": "Editors", + "isSystem": False, + "redirectOnLogin": "/", + "permissions": ["read:pages"], + "pageRules": [], + "createdAt": "2024-01-01T00:00:00Z", + "updatedAt": "2024-01-01T00:00:00Z", + }, + } + } + } + } + endpoint._post = Mock(return_value=mock_response) + + group = endpoint.create(group_dict) + assert group.name == "Editors" + + def test_create_group_validation_error(self, endpoint): + """Test create group with validation error.""" + with pytest.raises(ValidationError): + endpoint.create("invalid_type") + + def test_create_group_api_error(self, endpoint): + """Test create group with API error.""" + group_data = GroupCreate(name="Editors") + mock_response = {"errors": [{"message": "Group exists"}]} + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Group exists"): + endpoint.create(group_data) + + def test_create_group_failed_result(self, endpoint): + """Test create group with failed result.""" + group_data = GroupCreate(name="Editors") + mock_response = { + "data": { + "groups": { + "create": { + "responseResult": {"succeeded": False, "message": "Creation failed"} + } + } + } + } + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Creation failed"): + endpoint.create(group_data) + + def test_update_group_validation_error_invalid_id(self, endpoint): + """Test update group with invalid ID.""" + update_data = GroupUpdate(name="Updated") + + with pytest.raises(ValidationError): + endpoint.update(0, update_data) + + def test_update_group_with_dict(self, endpoint): + """Test updating group with dictionary.""" + update_dict = {"name": "Updated Name"} + + mock_response = { + "data": { + "groups": { + "update": { + "responseResult": {"succeeded": True}, + "group": { + "id": 1, + "name": "Updated Name", + "isSystem": False, + "redirectOnLogin": "/", + "permissions": [], + "pageRules": [], + "createdAt": "2024-01-01T00:00:00Z", + "updatedAt": "2024-01-02T00:00:00Z", + }, + } + } + } + } + endpoint._post = Mock(return_value=mock_response) + + group = endpoint.update(1, update_dict) + assert group.name == "Updated Name" + + def test_update_group_validation_error_invalid_data(self, endpoint): + """Test update group with invalid data.""" + with pytest.raises(ValidationError): + endpoint.update(1, "invalid_type") + + def test_update_group_api_error(self, endpoint): + """Test update group with API error.""" + update_data = GroupUpdate(name="Updated") + mock_response = {"errors": [{"message": "Update failed"}]} + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Update failed"): + endpoint.update(1, update_data) + + def test_update_group_failed_result(self, endpoint): + """Test update group with failed result.""" + update_data = GroupUpdate(name="Updated") + mock_response = { + "data": { + "groups": { + "update": { + "responseResult": {"succeeded": False, "message": "Update denied"} + } + } + } + } + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Update denied"): + endpoint.update(1, update_data) + + def test_delete_group_validation_error(self, endpoint): + """Test delete group with invalid ID.""" + with pytest.raises(ValidationError): + endpoint.delete(-1) + + def test_delete_group_api_error(self, endpoint): + """Test delete group with API error.""" + mock_response = {"errors": [{"message": "Delete failed"}]} + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Delete failed"): + endpoint.delete(1) + + def test_delete_group_failed_result(self, endpoint): + """Test delete group with failed result.""" + mock_response = { + "data": { + "groups": { + "delete": { + "responseResult": {"succeeded": False, "message": "Cannot delete system group"} + } + } + } + } + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Cannot delete system group"): + endpoint.delete(1) + + def test_assign_user_validation_error_invalid_group(self, endpoint): + """Test assign user with invalid group ID.""" + with pytest.raises(ValidationError): + endpoint.assign_user(0, 1) + + def test_assign_user_validation_error_invalid_user(self, endpoint): + """Test assign user with invalid user ID.""" + with pytest.raises(ValidationError): + endpoint.assign_user(1, 0) + + def test_assign_user_api_error(self, endpoint): + """Test assign user with API error.""" + mock_response = {"errors": [{"message": "Assignment failed"}]} + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Assignment failed"): + endpoint.assign_user(1, 5) + + def test_assign_user_failed_result(self, endpoint): + """Test assign user with failed result.""" + mock_response = { + "data": { + "groups": { + "assignUser": { + "responseResult": {"succeeded": False, "message": "User already in group"} + } + } + } + } + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="User already in group"): + endpoint.assign_user(1, 5) + + def test_unassign_user_validation_error_invalid_group(self, endpoint): + """Test unassign user with invalid group ID.""" + with pytest.raises(ValidationError): + endpoint.unassign_user(-1, 1) + + def test_unassign_user_validation_error_invalid_user(self, endpoint): + """Test unassign user with invalid user ID.""" + with pytest.raises(ValidationError): + endpoint.unassign_user(1, -1) + + def test_unassign_user_api_error(self, endpoint): + """Test unassign user with API error.""" + mock_response = {"errors": [{"message": "Unassignment failed"}]} + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="Unassignment failed"): + endpoint.unassign_user(1, 5) + + def test_unassign_user_failed_result(self, endpoint): + """Test unassign user with failed result.""" + mock_response = { + "data": { + "groups": { + "unassignUser": { + "responseResult": {"succeeded": False, "message": "User not in group"} + } + } + } + } + endpoint._post = Mock(return_value=mock_response) + + with pytest.raises(APIError, match="User not in group"): + endpoint.unassign_user(1, 5) diff --git a/tests/test_coverage_boost.py b/tests/test_coverage_boost.py new file mode 100644 index 0000000..beeba1b --- /dev/null +++ b/tests/test_coverage_boost.py @@ -0,0 +1,204 @@ +"""Targeted tests to reach 85% coverage.""" + +import pytest +import time +from unittest.mock import Mock, patch +from wikijs.cache.memory import MemoryCache +from wikijs.ratelimit import RateLimiter +from wikijs.metrics import MetricsCollector +from wikijs.client import WikiJSClient +from wikijs.exceptions import AuthenticationError, APIError + + +class TestCacheEdgeCases: + """Test cache edge cases to cover missing lines.""" + + def test_invalidate_resource_with_malformed_keys(self): + """Test invalidate_resource with malformed cache keys (covers line 132).""" + cache = MemoryCache(ttl=300) + + # Add some normal keys + cache._cache["page:123"] = ({"data": "test1"}, time.time() + 300) + cache._cache["page:456"] = ({"data": "test2"}, time.time() + 300) + + # Add a malformed key without colon separator (covers line 132) + cache._cache["malformedkey"] = ({"data": "test3"}, time.time() + 300) + + # Invalidate page type - should skip malformed key + cache.invalidate_resource("page") + + # Normal keys should be gone + assert "page:123" not in cache._cache + assert "page:456" not in cache._cache + + # Malformed key should still exist (was skipped by continue on line 132) + assert "malformedkey" in cache._cache + + +class TestMetricsEdgeCases: + """Test metrics edge cases.""" + + def test_record_request_with_server_error(self): + """Test recording request with 5xx status code (covers line 64).""" + collector = MetricsCollector() + + # Record a server error (5xx) + collector.record_request( + endpoint="/test", + method="GET", + status_code=500, + duration_ms=150.0, + error="Internal Server Error" + ) + + # Check counters + assert collector._counters["total_requests"] == 1 + assert collector._counters["total_errors"] == 1 + assert collector._counters["total_server_errors"] == 1 # Line 64 + + def test_percentile_with_empty_data(self): + """Test _percentile with empty data (covers line 135).""" + collector = MetricsCollector() + + # Get percentile with no recorded requests + result = collector._percentile([], 95) + + # Should return 0.0 for empty data (line 135) + assert result == 0.0 + + +class TestClientErrorHandling: + """Test client error handling paths.""" + + @patch('wikijs.client.requests.Session') + def test_connection_unexpected_format(self, mock_session_class): + """Test test_connection with unexpected response format (covers line 287).""" + mock_session = Mock() + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"unexpected": "format"} + mock_session.request.return_value = mock_response + mock_session_class.return_value = mock_session + + client = WikiJSClient("https://wiki.example.com", auth="test-key") + + with pytest.raises(APIError, match="Unexpected response format"): + client.test_connection() + + @patch('wikijs.client.requests.Session') + def test_connection_reraises_auth_error(self, mock_session_class): + """Test test_connection re-raises AuthenticationError (covers line 295).""" + mock_session = Mock() + + def raise_auth_error(*args, **kwargs): + raise AuthenticationError("Auth failed") + + mock_session.request.side_effect = raise_auth_error + mock_session_class.return_value = mock_session + + client = WikiJSClient("https://wiki.example.com", auth="test-key") + + with pytest.raises(AuthenticationError, match="Auth failed"): + client.test_connection() + + @patch('wikijs.client.requests.Session') + def test_connection_reraises_api_error(self, mock_session_class): + """Test test_connection re-raises APIError (covers line 307).""" + mock_session = Mock() + + def raise_api_error(*args, **kwargs): + raise APIError("API failed") + + mock_session.request.side_effect = raise_api_error + mock_session_class.return_value = mock_session + + client = WikiJSClient("https://wiki.example.com", auth="test-key") + + with pytest.raises(APIError, match="API failed"): + client.test_connection() + + +class TestLoggingEdgeCases: + """Test logging edge cases.""" + + def test_setup_logging_with_json_format(self): + """Test setup_logging with JSON format.""" + from wikijs.logging import setup_logging + + logger = setup_logging(level="DEBUG", format_type="json") + assert logger.level == 10 # DEBUG = 10 + + def test_json_formatter_with_exception_info(self): + """Test JSON formatter with exception info (covers line 33).""" + import logging + import sys + from wikijs.logging import JSONFormatter + + formatter = JSONFormatter() + + # Create a log record with actual exception info + try: + 1 / 0 + except ZeroDivisionError: + exc_info = sys.exc_info() + + record = logging.LogRecord( + name="test", + level=logging.ERROR, + pathname="test.py", + lineno=1, + msg="Error occurred", + args=(), + exc_info=exc_info + ) + + result = formatter.format(record) + assert "exception" in result + assert "ZeroDivisionError" in result + + def test_json_formatter_with_extra_fields(self): + """Test JSON formatter with extra fields (covers line 37).""" + import logging + from wikijs.logging import JSONFormatter + + formatter = JSONFormatter() + + record = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="test.py", + lineno=1, + msg="Test message", + args=(), + exc_info=None + ) + + # Add extra attribute + record.extra = {"user_id": 123, "request_id": "abc"} + + result = formatter.format(record) + assert "user_id" in result + assert "123" in result + + def test_setup_logging_with_file_output(self): + """Test setup_logging with file output (covers line 65).""" + import tempfile + import os + from wikijs.logging import setup_logging + + # Create temporary file + with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.log') as f: + log_file = f.name + + try: + logger = setup_logging(level="INFO", format_type="text", output_file=log_file) + logger.info("Test message") + + # Verify file was created and has content + assert os.path.exists(log_file) + with open(log_file, 'r') as f: + content = f.read() + assert "Test message" in content + finally: + if os.path.exists(log_file): + os.unlink(log_file)