Simplenote MCP Server

"""Tests for security alerting system."""

from datetime import datetime, timedelta
from unittest.mock import AsyncMock, Mock, patch

import pytest

from simplenote_mcp.server.alerting import (
    AlertSeverity,
    AlertType,
    SecurityAlert,
    SecurityAlerter,
    alert_authentication_failure,
    alert_dangerous_input,
    alert_rate_limit_violation,
    alert_suspicious_pattern,
    get_alerter,
)
from simplenote_mcp.server.errors import SecurityError


class TestSecurityAlert:
    """Test SecurityAlert class."""

    def test_alert_creation(self):
        """Test creating a security alert."""
        alert = SecurityAlert(
            alert_type=AlertType.DANGEROUS_INPUT,
            severity=AlertSeverity.HIGH,
            message="Test alert message",
            context={"key": "value"},
            user_id="test_user",
            client_info={"ip": "127.0.0.1"},
        )

        assert alert.alert_type == AlertType.DANGEROUS_INPUT
        assert alert.severity == AlertSeverity.HIGH
        assert alert.message == "Test alert message"
        assert alert.context == {"key": "value"}
        assert alert.user_id == "test_user"
        assert alert.client_info == {"ip": "127.0.0.1"}
        assert isinstance(alert.timestamp, datetime)
        assert alert.alert_id.startswith("dangerous_input_")

    def test_alert_to_dict(self):
        """Test converting alert to dictionary."""
        alert = SecurityAlert(
            alert_type=AlertType.RATE_LIMIT_VIOLATION,
            severity=AlertSeverity.MEDIUM,
            message="Rate limit exceeded",
            context={"requests": 100},
        )

        alert_dict = alert.to_dict()

        assert alert_dict["alert_type"] == "rate_limit_violation"
        assert alert_dict["severity"] == "MEDIUM"
        assert alert_dict["message"] == "Rate limit exceeded"
        assert alert_dict["context"] == {"requests": 100}
        assert "timestamp" in alert_dict
        assert "alert_id" in alert_dict

    def test_alert_string_representation(self):
        """Test string representation of alert."""
        alert = SecurityAlert(
            alert_type=AlertType.AUTHENTICATION_FAILURE,
            severity=AlertSeverity.LOW,
            message="Login failed",
            context={},
        )

        alert_str = str(alert)
        assert alert_str == "[LOW] authentication_failure: Login failed"


class TestSecurityAlerter:
    """Test SecurityAlerter class."""

    def setup_method(self):
        """Set up test environment."""
        self.test_config = {
            "failed_auth_threshold": 3,
            "rate_limit_threshold": 2,
            "suspicious_pattern_threshold": 2,
            "time_window_minutes": 5,
            "enable_file_alerts": False,  # Disable file alerts for tests
            "enable_email_alerts": False,
            "enable_webhook_alerts": False,
        }
        self.alerter = SecurityAlerter(self.test_config)

    @pytest.mark.asyncio
    async def test_create_alert(self):
        """Test creating an alert."""
        alert = await self.alerter.create_alert(
            AlertType.DANGEROUS_INPUT,
            AlertSeverity.HIGH,
            "Test dangerous input",
            {"pattern": "SELECT * FROM"},
            user_id="test_user",
        )

        assert alert.alert_type == AlertType.DANGEROUS_INPUT
        assert alert.severity == AlertSeverity.HIGH
        assert alert.message == "Test dangerous input"
        assert alert.user_id == "test_user"

        # Check it was added to history
        assert len(self.alerter.alert_history) == 1
        assert self.alerter.alert_history[0] == alert

    @pytest.mark.asyncio
    async def test_authentication_failure_pattern(self):
        """Test authentication failure pattern detection."""
        user_id = "test_user"

        # Create multiple authentication failures
        for i in range(self.test_config["failed_auth_threshold"]):
            await self.alerter.create_alert(
                AlertType.AUTHENTICATION_FAILURE,
                AlertSeverity.MEDIUM,
                f"Auth failure {i}",
                {"attempt": i},
                user_id=user_id,
            )

        # Should have created additional alert for repeated failures
        # Initial alerts + 1 escalation alert
        expected_alerts = self.test_config["failed_auth_threshold"] + 1
        assert len(self.alerter.alert_history) == expected_alerts

        # Check the escalation alert
        escalation_alert = self.alerter.alert_history[-1]
        assert escalation_alert.alert_type == AlertType.REPEATED_FAILURES
        assert escalation_alert.severity == AlertSeverity.HIGH

    @pytest.mark.asyncio
    async def test_rate_limit_violation_pattern(self):
        """Test rate limit violation pattern detection."""
        user_id = "test_user"

        # Create multiple rate limit violations
        for i in range(self.test_config["rate_limit_threshold"]):
            await self.alerter.create_alert(
                AlertType.RATE_LIMIT_VIOLATION,
                AlertSeverity.MEDIUM,
                f"Rate limit violation {i}",
                {"requests": 100 + i},
                user_id=user_id,
            )

        # Should have created additional alert for anomalous behavior
        expected_alerts = self.test_config["rate_limit_threshold"] + 1
        assert len(self.alerter.alert_history) == expected_alerts

        # Check the escalation alert
        escalation_alert = self.alerter.alert_history[-1]
        assert escalation_alert.alert_type == AlertType.ANOMALOUS_BEHAVIOR
        assert escalation_alert.severity == AlertSeverity.HIGH

    @pytest.mark.asyncio
    async def test_suspicious_input_pattern(self):
        """Test suspicious input pattern detection."""
        user_id = "test_user"

        # Create multiple dangerous input alerts
        for i in range(self.test_config["suspicious_pattern_threshold"]):
            await self.alerter.create_alert(
                AlertType.DANGEROUS_INPUT,
                AlertSeverity.HIGH,
                f"Dangerous input {i}",
                {"pattern": f"pattern_{i}"},
                user_id=user_id,
            )

        # Should have created additional alert for security threshold exceeded
        expected_alerts = self.test_config["suspicious_pattern_threshold"] + 1
        assert len(self.alerter.alert_history) == expected_alerts

        # Check the escalation alert
        escalation_alert = self.alerter.alert_history[-1]
        assert escalation_alert.alert_type == AlertType.SECURITY_THRESHOLD_EXCEEDED
        assert escalation_alert.severity == AlertSeverity.CRITICAL

    def test_get_recent_alerts(self):
        """Test getting recent alerts."""
        # Add some test alerts manually
        old_alert = SecurityAlert(
            AlertType.DANGEROUS_INPUT,
            AlertSeverity.LOW,
            "Old alert",
            {},
        )
        old_alert.timestamp = datetime.utcnow() - timedelta(hours=2)

        recent_alert = SecurityAlert(
            AlertType.RATE_LIMIT_VIOLATION,
            AlertSeverity.HIGH,
            "Recent alert",
            {},
        )

        self.alerter.alert_history.extend([old_alert, recent_alert])

        # Get recent alerts (last 60 minutes)
        recent_alerts = self.alerter.get_recent_alerts(minutes=60)
        assert len(recent_alerts) == 1
        assert recent_alerts[0] == recent_alert

        # Get alerts with severity filter
        high_alerts = self.alerter.get_recent_alerts(
            minutes=180, severity=AlertSeverity.HIGH
        )
        assert len(high_alerts) == 1
        assert high_alerts[0] == recent_alert

        # Get alerts with type filter
        dangerous_alerts = self.alerter.get_recent_alerts(
            minutes=180, alert_type=AlertType.DANGEROUS_INPUT
        )
        assert len(dangerous_alerts) == 1
        assert dangerous_alerts[0] == old_alert

    def test_get_alert_summary(self):
        """Test getting alert summary."""
        # Add test alerts
        alerts = [
            SecurityAlert(
                AlertType.DANGEROUS_INPUT, AlertSeverity.HIGH, "Alert 1", {}, "user1"
            ),
            SecurityAlert(
                AlertType.RATE_LIMIT_VIOLATION,
                AlertSeverity.MEDIUM,
                "Alert 2",
                {},
                "user2",
            ),
            SecurityAlert(
                AlertType.DANGEROUS_INPUT, AlertSeverity.HIGH, "Alert 3", {}, "user1"
            ),
        ]

        self.alerter.alert_history.extend(alerts)

        summary = self.alerter.get_alert_summary(minutes=60)

        assert summary["total_alerts"] == 3
        assert summary["by_severity"]["HIGH"] == 2
        assert summary["by_severity"]["MEDIUM"] == 1
        assert summary["by_type"]["dangerous_input"] == 2
        assert summary["by_type"]["rate_limit_violation"] == 1
        assert len(summary["affected_users"]) == 2
        assert "user1" in summary["affected_users"]
        assert "user2" in summary["affected_users"]
        assert summary["latest_alert"] is not None

    def test_cleanup_old_data(self):
        """Test cleaning up old tracking data."""
        # Add some test data
        self.alerter.failure_counts["user1"] = 5
        self.alerter.failure_counts["user2"] = 3

        # Add many entries to trigger cleanup
        for i in range(1001):
            self.alerter.failure_counts[f"user_{i}"] = 1

        self.alerter.cleanup_old_data(days=7)

        # Should have cleared the data
        assert len(self.alerter.failure_counts) == 0


class TestAlertingFunctions:
    """Test alerting convenience functions."""

    @pytest.mark.asyncio
    async def test_alert_authentication_failure(self):
        """Test authentication failure alert function."""
        with patch("simplenote_mcp.server.alerting.get_alerter") as mock_get_alerter:
            mock_alerter = Mock()
            mock_alerter.create_alert = AsyncMock()
            mock_get_alerter.return_value = mock_alerter

            await alert_authentication_failure(
                "test_user", "Invalid password", {"ip": "127.0.0.1"}
            )

            mock_alerter.create_alert.assert_called_once()
            call_args = mock_alerter.create_alert.call_args
            assert call_args[0][0] == AlertType.AUTHENTICATION_FAILURE
            assert call_args[0][1] == AlertSeverity.MEDIUM
            assert "test_user" in call_args[0][2]
            assert call_args[1]["user_id"] == "test_user"

    @pytest.mark.asyncio
    async def test_alert_rate_limit_violation(self):
        """Test rate limit violation alert function."""
        with patch("simplenote_mcp.server.alerting.get_alerter") as mock_get_alerter:
            mock_alerter = Mock()
            mock_alerter.create_alert = AsyncMock()
            mock_get_alerter.return_value = mock_alerter

            await alert_rate_limit_violation("test_user", 150, 100, {"source": "api"})

            mock_alerter.create_alert.assert_called_once()
            call_args = mock_alerter.create_alert.call_args
            assert call_args[0][0] == AlertType.RATE_LIMIT_VIOLATION
            assert call_args[0][1] == AlertSeverity.MEDIUM
            assert "150/100" in call_args[0][2]

    @pytest.mark.asyncio
    async def test_alert_dangerous_input(self):
        """Test dangerous input alert function."""
        with patch("simplenote_mcp.server.alerting.get_alerter") as mock_get_alerter:
            mock_alerter = Mock()
            mock_alerter.create_alert = AsyncMock()
            mock_get_alerter.return_value = mock_alerter

            await alert_dangerous_input(
                "test_user", "note_content", "SELECT * FROM users", {"client": "web"}
            )

            mock_alerter.create_alert.assert_called_once()
            call_args = mock_alerter.create_alert.call_args
            assert call_args[0][0] == AlertType.DANGEROUS_INPUT
            assert call_args[0][1] == AlertSeverity.HIGH
            assert "note_content" in call_args[0][2]

    @pytest.mark.asyncio
    async def test_alert_suspicious_pattern(self):
        """Test suspicious pattern alert function."""
        with patch("simplenote_mcp.server.alerting.get_alerter") as mock_get_alerter:
            mock_alerter = Mock()
            mock_alerter.create_alert = AsyncMock()
            mock_get_alerter.return_value = mock_alerter

            await alert_suspicious_pattern(
                "test_user",
                "Rapid sequential requests",
                {"request_count": 200, "time_window": "1 minute"},
                {"user_agent": "bot"},
            )

            mock_alerter.create_alert.assert_called_once()
            call_args = mock_alerter.create_alert.call_args
            assert call_args[0][0] == AlertType.SUSPICIOUS_PATTERN
            assert call_args[0][1] == AlertSeverity.MEDIUM
            assert "Rapid sequential requests" in call_args[0][2]


class TestGlobalAlerter:
    """Test global alerter functionality."""

    def test_get_global_alerter(self):
        """Test getting global alerter instance."""
        alerter1 = get_alerter()
        alerter2 = get_alerter()

        # Should return the same instance
        assert alerter1 is alerter2
        assert isinstance(alerter1, SecurityAlerter)

    def test_global_alerter_functionality(self):
        """Test global alerter works correctly."""
        alerter = get_alerter()

        # Should have default configuration
        assert alerter.thresholds["failed_auth_threshold"] == 5
        assert alerter.thresholds["rate_limit_threshold"] == 3
        assert alerter.enable_file_alerts is True


class TestAlertIntegration:
    """Test integration between alerting and other security components."""

    @pytest.mark.asyncio
    async def test_security_event_triggers_alert(self):
        """Test that security events trigger alerts correctly."""
        from simplenote_mcp.server.security import security_validator

        # Test dangerous content detection triggers alert
        with pytest.raises(SecurityError):  # SecurityError expected
            security_validator.validate_note_content(
                "SELECT * FROM users WHERE 1=1", "test content"
            )

        # Check that alert was triggered (this tests the integration)
        # In a real test, we would mock the alerting system

    def test_rate_limit_triggers_alert(self):
        """Test that rate limit violations trigger alerts."""
        # Trigger multiple rate limit violations
        # This would test the integration with alerting
        # In a real test, we would mock the alerting system