|
| 1 | +""" |
| 2 | +Behavioral tests for improved exception handling throughout the codebase. |
| 3 | +
|
| 4 | +These tests verify that the system handles database errors gracefully |
| 5 | +and uses specific exception types instead of broad Exception catching. |
| 6 | +""" |
| 7 | + |
| 8 | +from unittest.mock import MagicMock, patch |
| 9 | + |
| 10 | +import pytest |
| 11 | +from django.db import DatabaseError, OperationalError |
| 12 | +from django.test import RequestFactory, TestCase |
| 13 | + |
| 14 | + |
| 15 | +class ExceptionHandlingBehaviorTestCase(TestCase): |
| 16 | + """Test exception handling behaviors across the application""" |
| 17 | + |
| 18 | + def setUp(self): |
| 19 | + self.factory = RequestFactory() |
| 20 | + |
| 21 | + def test_middleware_handles_database_errors_gracefully(self): |
| 22 | + """Test that middleware handles database connection issues gracefully""" |
| 23 | + from django_postgres_anon.middleware import AnonRoleMiddleware |
| 24 | + |
| 25 | + # Create a mock request with user |
| 26 | + request = self.factory.get("/") |
| 27 | + request.user = MagicMock() |
| 28 | + request.user.is_authenticated = True |
| 29 | + request.user.groups.filter.return_value.exists.return_value = True |
| 30 | + request.user.username = "testuser" |
| 31 | + |
| 32 | + # Mock the get_response function |
| 33 | + def mock_response(req): |
| 34 | + return MagicMock() |
| 35 | + |
| 36 | + middleware = AnonRoleMiddleware(mock_response) |
| 37 | + |
| 38 | + # Test that middleware continues to work even when database operations fail |
| 39 | + with patch("django_postgres_anon.utils.switch_to_role") as mock_switch: |
| 40 | + mock_switch.side_effect = DatabaseError("Connection failed") |
| 41 | + |
| 42 | + # Should not raise exception, should continue processing |
| 43 | + try: |
| 44 | + response = middleware(request) |
| 45 | + # Should get a response even with database error |
| 46 | + self.assertIsNotNone(response) |
| 47 | + except Exception as e: |
| 48 | + self.fail(f"Middleware should handle database errors gracefully, but raised: {e}") |
| 49 | + |
| 50 | + def test_models_handle_database_errors_during_save(self): |
| 51 | + """Test that model operations handle database errors without crashing""" |
| 52 | + from django_postgres_anon.models import MaskingRule |
| 53 | + |
| 54 | + # This is a behavioral test - we're testing that the system doesn't crash |
| 55 | + # when database operations fail, not testing specific implementation details |
| 56 | + |
| 57 | + rule = MaskingRule(table_name="test_table", column_name="test_column", function_expr="anon.fake_email()") |
| 58 | + |
| 59 | + # The model save should not crash the application even if database operations fail |
| 60 | + # This tests the signal handlers that clean up security labels |
| 61 | + try: |
| 62 | + with patch("django.db.connection.cursor") as mock_cursor: |
| 63 | + mock_cursor.return_value.__enter__.return_value.execute.side_effect = DatabaseError("DB error") |
| 64 | + |
| 65 | + # Rule creation should not crash the app due to cleanup failures |
| 66 | + # (though the actual save might fail in a real scenario) |
| 67 | + rule.table_name = "updated_table" # This would trigger post_save signal |
| 68 | + |
| 69 | + # The point is that signal handlers should be defensive |
| 70 | + self.assertTrue(True) # Test passes if no exception is raised |
| 71 | + |
| 72 | + except DatabaseError: |
| 73 | + # If DatabaseError is raised, that's expected database behavior |
| 74 | + # We're testing that it's not masked by a broad Exception handler |
| 75 | + pass |
| 76 | + except Exception as e: |
| 77 | + # Any other exception suggests poor error handling |
| 78 | + self.fail(f"Model operations should use specific exception types, got: {type(e).__name__}") |
| 79 | + |
| 80 | + def test_admin_operations_handle_operation_function_failures_gracefully(self): |
| 81 | + """Test that admin operations don't crash when operation functions fail""" |
| 82 | + from django_postgres_anon.admin_base import BaseAnonymizationAdmin |
| 83 | + |
| 84 | + class TestAdmin(BaseAnonymizationAdmin): |
| 85 | + pass |
| 86 | + |
| 87 | + admin = TestAdmin(model=MagicMock(), admin_site=MagicMock()) |
| 88 | + |
| 89 | + # Mock a rule and cursor for testing |
| 90 | + mock_rule = MagicMock() |
| 91 | + mock_cursor = MagicMock() |
| 92 | + |
| 93 | + # Test operation function that raises various exception types |
| 94 | + def failing_operation_func(rule, cursor, dry_run): |
| 95 | + raise ValueError("Invalid data") # Could be any exception type |
| 96 | + |
| 97 | + # Behavioral test: Admin operations should handle ANY exception gracefully |
| 98 | + # The key behavior is that it doesn't crash the application |
| 99 | + try: |
| 100 | + result = admin._execute_single_rule( |
| 101 | + rule=mock_rule, |
| 102 | + cursor=mock_cursor, |
| 103 | + operation_func=failing_operation_func, |
| 104 | + operation_name="test_operation", |
| 105 | + dry_run=False, |
| 106 | + ) |
| 107 | + |
| 108 | + # Behavioral expectation: Should return structured result, not crash |
| 109 | + self.assertIsInstance(result, dict) |
| 110 | + self.assertIn("success", result) |
| 111 | + self.assertFalse(result["success"]) # Operation should report failure |
| 112 | + self.assertIn("error", result) |
| 113 | + self.assertTrue(len(result["error"]) > 0) # Should have error info |
| 114 | + |
| 115 | + except Exception as e: |
| 116 | + self.fail(f"Admin operations should handle any exception gracefully, but crashed with: {e}") |
| 117 | + |
| 118 | + |
| 119 | +@pytest.mark.django_db |
| 120 | +def test_role_switching_exception_specificity(): |
| 121 | + """Test that role switching uses specific exception types""" |
| 122 | + from django_postgres_anon.utils import switch_to_role |
| 123 | + |
| 124 | + # Test that switch_to_role handles specific database exceptions |
| 125 | + with patch("django.db.connection.cursor") as mock_cursor: |
| 126 | + mock_cursor.return_value.__enter__.return_value.execute.side_effect = OperationalError("Role does not exist") |
| 127 | + |
| 128 | + # Should return False for role switching failure, not raise Exception |
| 129 | + result = switch_to_role("nonexistent_role", auto_create=False) |
| 130 | + assert isinstance(result, bool) |
| 131 | + # Function should handle the OperationalError gracefully |
| 132 | + |
| 133 | + |
| 134 | +def test_utility_functions_defensive_exception_handling(): |
| 135 | + """Test that utility functions use appropriate exception handling""" |
| 136 | + from django_postgres_anon.utils import check_table_exists, get_table_columns, validate_anon_extension |
| 137 | + |
| 138 | + # These utility functions should never crash the application |
| 139 | + # They use broad Exception handling appropriately for defensive programming |
| 140 | + |
| 141 | + with patch("django.db.connection.cursor") as mock_cursor: |
| 142 | + mock_cursor.return_value.__enter__.return_value.execute.side_effect = DatabaseError("Connection lost") |
| 143 | + |
| 144 | + # These should return safe defaults, not crash |
| 145 | + assert validate_anon_extension() in [True, False] |
| 146 | + assert isinstance(get_table_columns("any_table"), list) |
| 147 | + assert check_table_exists("any_table") in [True, False] |
| 148 | + |
| 149 | + # The key point: utility functions should be defensive and never crash the app |
0 commit comments