tests/test_audit_service.py

"""
Comprehensive Tests for Audit Service

Tests cover:
1. Client event logging
2. Server event logging
3. Request metadata extraction
4. Async logging
5. Error handling
6. AuditLog model integration

Uses mocked database and request objects.
"""
import pytest
from datetime import datetime
from unittest.mock import MagicMock, AsyncMock, patch
from fastapi import Request


# ============================================================================
# 1. Client Event Logging Tests
# ============================================================================

class TestClientEventLogging:
    """Test client-side event logging."""
    
    @pytest.mark.asyncio
    async def test_log_client_event_success(self, db_session):
        """Log successful client event."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        # Create mock request
        mock_request = MagicMock()
        mock_request.client.host = "192.168.1.1"
        mock_request.headers.get.side_effect = lambda k, default=None: {
            "user-agent": "Mozilla/5.0",
            "referer": "https://example.com"
        }.get(k.lower(), default)
        
        # Log client event
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="page_view",
            status="success",
            client_user_id="temp_123",
            details={"page": "/home"},
            request=mock_request
        )
        
        # Verify log was created
        result = await db_session.execute(
            select(AuditLog).where(AuditLog.action == "page_view")
        )
        log = result.scalar_one_or_none()
        
        assert log is not None
        assert log.log_type == "client"
        assert log.action == "page_view"
        assert log.status == "success"
        assert log.client_user_id == "temp_123"
        assert log.ip_address == "192.168.1.1"
    
    @pytest.mark.asyncio
    async def test_log_client_event_with_user(self, db_session):
        """Log client event with authenticated user."""
        from services.audit_service import AuditService
        from core.models import User, AuditLog
        from sqlalchemy import select
        
        # Create user
        user = User(user_id="usr_audit", email="audit@example.com")
        db_session.add(user)
        await db_session.commit()
        
        mock_request = MagicMock()
        mock_request.client.host = "10.0.0.1"
        mock_request.headers.get.return_value = None
        
        # Log with user_id
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="login",
            status="success",
            user_id=user.id,
            client_user_id="temp_456",
            request=mock_request
        )
        
        result = await db_session.execute(
            select(AuditLog).where(AuditLog.user_id == user.id)
        )
        log = result.scalar_one_or_none()
        
        assert log.user_id == user.id
        assert log.client_user_id == "temp_456"


# ============================================================================
# 2. Server Event Logging Tests
# ============================================================================

class TestServerEventLogging:
    """Test server-side event logging."""
    
    @pytest.mark.asyncio
    async def test_log_server_event(self, db_session):
        """Log server event."""
        from services.audit_service import AuditService
        from core.models import User, AuditLog
        from sqlalchemy import select
        
        user = User(user_id="usr_server", email="server@example.com")
        db_session.add(user)
        await db_session.commit()
        
        await AuditService.log_event(
            db=db_session,
            log_type="server",
            action="credit_deduction",
            status="success",
            user_id=user.id,
            details={"amount": 10, "reason": "video_generation"}
        )
        
        result = await db_session.execute(
            select(AuditLog).where(AuditLog.action == "credit_deduction")
        )
        log = result.scalar_one_or_none()
        
        assert log.log_type == "server"
        assert log.details["amount"] == 10
    
    @pytest.mark.asyncio
    async def test_log_server_failure(self, db_session):
        """Log server failure event."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        await AuditService.log_event(
            db=db_session,
            log_type="server",
            action="job_processing",
            status="failure",
            error_message="API quota exceeded",
            details={"job_id": "job_123"}
        )
        
        result = await db_session.execute(
            select(AuditLog).where(AuditLog.status == "failure")
        )
        log = result.scalar_one_or_none()
        
        assert log.error_message == "API quota exceeded"
        assert log.status == "failure"


# ============================================================================
# 3. Request Metadata Extraction Tests
# ============================================================================

class TestRequestMetadata:
    """Test extraction of request metadata."""
    
    @pytest.mark.asyncio
    async def test_extract_ip_address(self, db_session):
        """Extract IP address from request."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        mock_request = MagicMock()
        mock_request.client.host = "203.0.113.42"
        mock_request.headers.get.return_value = None
        
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="api_call",
            status="success",
            request=mock_request
        )
        
        result = await db_session.execute(select(AuditLog).where(AuditLog.action == "api_call"))
        log = result.scalar_one_or_none()
        
        assert log.ip_address == "203.0.113.42"
    
    @pytest.mark.asyncio
    async def test_extract_user_agent(self, db_session):
        """Extract user agent from request."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        mock_request = MagicMock()
        mock_request.client.host = "192.168.1.1"
        mock_request.headers.get.side_effect = lambda k, default=None: {
            "user-agent": "MyApp/1.0 (iOS)"
        }.get(k.lower(), default)
        
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="mobile_request",
            status="success",
            request=mock_request
        )
        
        result = await db_session.execute(select(AuditLog).where(AuditLog.action == "mobile_request"))
        log = result.scalar_one_or_none()
        
        assert "MyApp" in log.user_agent
    
    @pytest.mark.asyncio
    async def test_extract_referer(self, db_session):
        """Extract referer from request."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        mock_request = MagicMock()
        mock_request.client.host = "192.168.1.1"
        mock_request.headers.get.side_effect = lambda k, default=None: {
            "referer": "https://example.com/previous-page"
        }.get(k.lower(), default)
        
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="navigation",
            status="success",
            request=mock_request
        )
        
        result = await db_session.execute(select(AuditLog).where(AuditLog.action == "navigation"))
        log = result.scalar_one_or_none()
        
        assert "example.com" in log.refer_url


# ============================================================================
# 4. Error Handling Tests
# ============================================================================

class TestAuditErrorHandling:
    """Test error handling in audit service."""
    
    @pytest.mark.asyncio
    async def test_log_without_request(self, db_session):
        """Can log events without request object."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        # No request provided
        await AuditService.log_event(
            db=db_session,
            log_type="server",
            action="background_task",
            status="success"
        )
        
        result = await db_session.execute(select(AuditLog).where(AuditLog.action == "background_task"))
        log = result.scalar_one_or_none()
        
        assert log is not None
        assert log.ip_address is None  # No request means no IP
    
    @pytest.mark.asyncio
    async def test_log_with_missing_request_client(self, db_session):
        """Handle request without client attribute."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        mock_request = MagicMock()
        mock_request.client = None  # No client
        mock_request.headers.get.return_value = None
        
        # Should not crash
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="edge_case",
            status="success",
            request=mock_request
        )
        
        result = await db_session.execute(select(AuditLog).where(AuditLog.action == "edge_case"))
        log = result.scalar_one_or_none()
        
        assert log is not None


# ============================================================================
# 5. Details and Extra Data Tests
# ============================================================================

class TestAuditDetails:
    """Test storing structured details in audit logs."""
    
    @pytest.mark.asyncio
    async def test_store_complex_details(self, db_session):
        """Store complex JSON details."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        complex_details = {
            "user_action": "purchase",
            "items": ["credits_100", "credits_500"],
            "total_amount": 14900,
            "metadata": {
                "source": "web",
                "campaign": "summer_sale"
            }
        }
        
        await AuditService.log_event(
            db=db_session,
            log_type="server",
            action="purchase_attempt",
            status="success",
            details=complex_details
        )
        
        result = await db_session.execute(select(AuditLog).where(AuditLog.action == "purchase_attempt"))
        log = result.scalar_one_or_none()
        
        assert log.details["total_amount"] == 14900
        assert len(log.details["items"]) == 2
        assert log.details["metadata"]["campaign"] == "summer_sale"


# ============================================================================
# 6. Audit Query Tests
# ============================================================================

class TestAuditQueries:
    """Test querying audit logs."""
    
    @pytest.mark.asyncio
    async def test_query_by_user(self, db_session):
        """Query audit logs by user."""
        from services.audit_service import AuditService
        from core.models import User, AuditLog
        from sqlalchemy import select
        
        user = User(user_id="usr_query", email="query@example.com")
        db_session.add(user)
        await db_session.commit()
        
        # Create multiple logs for user
        for i in range(3):
            await AuditService.log_event(
                db=db_session,
                log_type="server",
                action=f"action_{i}",
                status="success",
                user_id=user.id
            )
        
        # Query user's logs
        result = await db_session.execute(
            select(AuditLog).where(AuditLog.user_id == user.id)
        )
        logs = result.scalars().all()
        
        assert len(logs) == 3
    
    @pytest.mark.asyncio
    async def test_query_by_action_type(self, db_session):
        """Query logs by action type."""
        from services.audit_service import AuditService
        from core.models import AuditLog
        from sqlalchemy import select
        
        # Create different action types
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="login",
            status="success"
        )
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="login",
            status="failure"
        )
        await AuditService.log_event(
            db=db_session,
            log_type="client",
            action="logout",
            status="success"
        )
        
        # Query only login actions
        result = await db_session.execute(
            select(AuditLog).where(AuditLog.action == "login")
        )
        logs = result.scalars().all()
        
        assert len(logs) == 2


if __name__ == "__main__":
    pytest.main([__file__, "-v"])