from datetime import datetime, timezone from typing import Tuple import pytest import aiops.entity.audit as auditent import aiops.models.audit as auditmod from aiops.service.audit.repo.convertor import E2M, M2E @pytest.fixture def instances() -> Tuple[auditmod.AuditLog, auditent.AuditLog]: """Provides a sample model and entity for testing.""" now = datetime.now(timezone.utc) details_data = {"user_agent": "test-client", "ip_address": "127.0.0.1"} model_instance = auditmod.AuditLog( id=1, created_at=now, type="chatbot_session", object_id="session-12345", operation="start", operator="user-test", operate_at=now, details=details_data, ) entity_instance = auditent.AuditLog( id=1, created_at=now, type=auditent.AuditLogType.CHATBOT_SESSION, object_id="session-12345", operation="start", operator="user-test", operate_at=now, details=details_data, ) return model_instance, entity_instance class TestAuditLogConvertor: def test_m2e_audit_log_conversion( self, instances: Tuple[auditmod.AuditLog, auditent.AuditLog], ) -> None: """ Tests the conversion from a SQLAlchemy Model to a Pydantic Entity. """ model, expected_entity = instances result_entity = M2E.audit_log(model) assert isinstance(result_entity, auditent.AuditLog) assert result_entity.id == expected_entity.id assert result_entity.created_at == expected_entity.created_at assert result_entity.type == expected_entity.type assert result_entity.object_id == expected_entity.object_id assert result_entity.operator == expected_entity.operator assert result_entity.operate_at == expected_entity.operate_at assert result_entity.details == expected_entity.details def test_m2e_audit_log_with_none_details(self) -> None: """ Tests the M2E conversion specifically when the 'details' field is None. """ model_instance = auditmod.AuditLog( id=2, created_at=datetime.now(timezone.utc), type="chatbot_session", operation="start", object_id="session-abcde", operator="system", operate_at=datetime.now(timezone.utc), details=None, ) result_entity = M2E.audit_log(model_instance) assert isinstance(result_entity, auditent.AuditLog) assert result_entity.id == 2 assert result_entity.details is None def test_e2m_audit_log_conversion( self, instances: Tuple[auditmod.AuditLog, auditent.AuditLog], ) -> None: """ Tests the conversion from a Pydantic Entity to a SQLAlchemy Model. """ _, entity = instances result_model = E2M.audit_log(entity) assert isinstance(result_model, auditmod.AuditLog) assert result_model.id == entity.id assert ( result_model.type == entity.type.value ) # 检查 Enum 是否转换为 string value assert result_model.object_id == entity.object_id assert result_model.operator == entity.operator assert result_model.operate_at == entity.operate_at assert result_model.details == entity.details def test_e2m_audit_log_with_none_details(self) -> None: """ Tests the E2M conversion specifically when the 'details' field is None. """ entity_instance = auditent.AuditLog( id=3, created_at=datetime.now(timezone.utc), type=auditent.AuditLogType.CHATBOT_SESSION, object_id="session-fghij", operation="start", operator="admin", operate_at=datetime.now(timezone.utc), details=None, ) result_model = E2M.audit_log(entity_instance) assert isinstance(result_model, auditmod.AuditLog) assert result_model.id == 3 assert result_model.details is None def test_e2m_new_entity_without_id(self) -> None: """ Tests the special logic where a new entity (id=0) does not set the model's id. """ new_entity = auditent.AuditLog( id=0, created_at=datetime.now(timezone.utc), type=auditent.AuditLogType.CHATBOT_SESSION, object_id="session-new", operation="start", operator="new-user", operate_at=datetime.now(timezone.utc), details={"status": "new"}, ) result_model = E2M.audit_log(new_entity) assert not result_model.id == 0