graph_rag_server/tests/unit/application/vector_search/test_dtos.py

"""
向量搜索 DTOs 单元测试

测试向量搜索应用层的数据传输对象（DTOs）。
验证 DTO 与领域实体之间的转换、序列化和反序列化功能。
"""

import pytest
from datetime import datetime

from src.application.vector_search.dtos import (
    DocumentDTO,
    SearchResultDTO,
    SearchResponseDTO
)
from src.domain.vector_search.entities import Document, SearchResult
from src.domain.vector_search.value_objects import Vector
from src.domain.shared.value_objects import EntityId, Timestamp


class TestDocumentDTO:
    """测试 DocumentDTO 类"""
    
    def test_from_entity_basic(self):
        """测试从领域实体创建 DTO（基本情况）"""
        # Arrange
        doc = Document(
            id=EntityId(value="doc_123"),
            content="Test content",
            embedding=None,
            metadata={"source": "test", "author": "John"},
            created_at=Timestamp.from_iso_string("2024-01-01T00:00:00"),
            updated_at=Timestamp.from_iso_string("2024-01-01T12:00:00")
        )
        
        # Act
        dto = DocumentDTO.from_entity(doc)
        
        # Assert
        assert dto.id == "doc_123"
        assert dto.content == "Test content"
        assert dto.metadata == {"source": "test", "author": "John"}
        assert dto.embedding is None
        assert dto.score is None
        assert dto.created_at == "2024-01-01T00:00:00"
        assert dto.updated_at == "2024-01-01T12:00:00"
    
    def test_from_entity_with_embedding(self):
        """测试从领域实体创建 DTO（包含嵌入向量）"""
        # Arrange
        vector = Vector(dimensions=[1.0, 2.0, 3.0])
        doc = Document(
            id=EntityId(value="doc_456"),
            content="Test content with embedding",
            embedding=vector,
            metadata={},
            created_at=Timestamp.now(),
            updated_at=Timestamp.now()
        )
        
        # Act
        dto = DocumentDTO.from_entity(doc, include_embedding=True)
        
        # Assert
        assert dto.embedding == [1.0, 2.0, 3.0]
    
    def test_from_entity_without_embedding(self):
        """测试从领域实体创建 DTO（不包含嵌入向量）"""
        # Arrange
        vector = Vector(dimensions=[1.0, 2.0, 3.0])
        doc = Document(
            id=EntityId(value="doc_789"),
            content="Test content",
            embedding=vector,
            metadata={},
            created_at=Timestamp.now(),
            updated_at=Timestamp.now()
        )
        
        # Act
        dto = DocumentDTO.from_entity(doc, include_embedding=False)
        
        # Assert
        assert dto.embedding is None
    
    def test_from_entity_with_score(self):
        """测试从领域实体创建 DTO（包含分数）"""
        # Arrange
        doc = Document(
            id=EntityId.generate(),
            content="Test content",
            embedding=None,
            metadata={},
            created_at=Timestamp.now(),
            updated_at=Timestamp.now()
        )
        
        # Act
        dto = DocumentDTO.from_entity(doc, score=0.95)
        
        # Assert
        assert dto.score == 0.95
    
    def test_to_entity_basic(self):
        """测试将 DTO 转换为领域实体（基本情况）"""
        # Arrange
        dto = DocumentDTO(
            id="doc_123",
            content="Test content",
            metadata={"source": "test"},
            embedding=None,
            score=None,
            created_at="2024-01-01T00:00:00",
            updated_at="2024-01-01T12:00:00"
        )
        
        # Act
        doc = dto.to_entity()
        
        # Assert
        assert str(doc.id) == "doc_123"
        assert doc.content == "Test content"
        assert doc.metadata == {"source": "test"}
        assert doc.embedding is None
        assert doc.created_at.to_iso_string() == "2024-01-01T00:00:00"
        assert doc.updated_at.to_iso_string() == "2024-01-01T12:00:00"
    
    def test_to_entity_with_embedding(self):
        """测试将 DTO 转换为领域实体（包含嵌入向量）"""
        # Arrange
        dto = DocumentDTO(
            id="doc_456",
            content="Test content",
            metadata={},
            embedding=[1.0, 2.0, 3.0],
            created_at="2024-01-01T00:00:00",
            updated_at="2024-01-01T00:00:00"
        )
        
        # Act
        doc = dto.to_entity()
        
        # Assert
        assert doc.embedding is not None
        assert doc.embedding.dimensions == [1.0, 2.0, 3.0]
    
    def test_to_dict_basic(self):
        """测试将 DTO 转换为字典（基本情况）"""
        # Arrange
        dto = DocumentDTO(
            id="doc_123",
            content="Test content",
            metadata={"source": "test"},
            created_at="2024-01-01T00:00:00",
            updated_at="2024-01-01T12:00:00"
        )
        
        # Act
        data = dto.to_dict()
        
        # Assert
        assert data["id"] == "doc_123"
        assert data["content"] == "Test content"
        assert data["metadata"] == {"source": "test"}
        assert "embedding" not in data
        assert "score" not in data
        assert data["created_at"] == "2024-01-01T00:00:00"
        assert data["updated_at"] == "2024-01-01T12:00:00"
    
    def test_to_dict_with_optional_fields(self):
        """测试将 DTO 转换为字典（包含可选字段）"""
        # Arrange
        dto = DocumentDTO(
            id="doc_456",
            content="Test content",
            metadata={},
            embedding=[1.0, 2.0, 3.0],
            score=0.95,
            created_at="2024-01-01T00:00:00",
            updated_at="2024-01-01T00:00:00"
        )
        
        # Act
        data = dto.to_dict()
        
        # Assert
        assert data["embedding"] == [1.0, 2.0, 3.0]
        assert data["score"] == 0.95
    
    def test_from_dict_basic(self):
        """测试从字典创建 DTO（基本情况）"""
        # Arrange
        data = {
            "id": "doc_123",
            "content": "Test content",
            "metadata": {"source": "test"},
            "created_at": "2024-01-01T00:00:00",
            "updated_at": "2024-01-01T12:00:00"
        }
        
        # Act
        dto = DocumentDTO.from_dict(data)
        
        # Assert
        assert dto.id == "doc_123"
        assert dto.content == "Test content"
        assert dto.metadata == {"source": "test"}
        assert dto.embedding is None
        assert dto.score is None
    
    def test_from_dict_with_optional_fields(self):
        """测试从字典创建 DTO（包含可选字段）"""
        # Arrange
        data = {
            "id": "doc_456",
            "content": "Test content",
            "metadata": {},
            "embedding": [1.0, 2.0, 3.0],
            "score": 0.95,
            "created_at": "2024-01-01T00:00:00",
            "updated_at": "2024-01-01T00:00:00"
        }
        
        # Act
        dto = DocumentDTO.from_dict(data)
        
        # Assert
        assert dto.embedding == [1.0, 2.0, 3.0]
        assert dto.score == 0.95
    
    def test_roundtrip_entity_to_dto_to_entity(self):
        """测试实体 -> DTO -> 实体的往返转换"""
        # Arrange
        original_doc = Document(
            id=EntityId(value="doc_roundtrip"),
            content="Roundtrip test content",
            embedding=Vector(dimensions=[1.0, 2.0, 3.0]),
            metadata={"test": "roundtrip"},
            created_at=Timestamp.from_iso_string("2024-01-01T00:00:00"),
            updated_at=Timestamp.from_iso_string("2024-01-01T12:00:00")
        )
        
        # Act
        dto = DocumentDTO.from_entity(original_doc, include_embedding=True)
        converted_doc = dto.to_entity()
        
        # Assert
        assert str(converted_doc.id) == str(original_doc.id)
        assert converted_doc.content == original_doc.content
        assert converted_doc.embedding.dimensions == original_doc.embedding.dimensions
        assert converted_doc.metadata == original_doc.metadata
        assert converted_doc.created_at.to_iso_string() == original_doc.created_at.to_iso_string()
        assert converted_doc.updated_at.to_iso_string() == original_doc.updated_at.to_iso_string()
    
    def test_roundtrip_dict_to_dto_to_dict(self):
        """测试字典 -> DTO -> 字典的往返转换"""
        # Arrange
        original_data = {
            "id": "doc_dict_roundtrip",
            "content": "Dict roundtrip test",
            "metadata": {"key": "value"},
            "embedding": [1.0, 2.0, 3.0],
            "score": 0.88,
            "created_at": "2024-01-01T00:00:00",
            "updated_at": "2024-01-01T12:00:00"
        }
        
        # Act
        dto = DocumentDTO.from_dict(original_data)
        converted_data = dto.to_dict()
        
        # Assert
        assert converted_data == original_data
    
    def test_metadata_is_copied(self):
        """测试元数据被复制而不是引用"""
        # Arrange
        original_metadata = {"key": "value"}
        doc = Document(
            id=EntityId.generate(),
            content="Test",
            embedding=None,
            metadata=original_metadata,
            created_at=Timestamp.now(),
            updated_at=Timestamp.now()
        )
        
        # Act
        dto = DocumentDTO.from_entity(doc)
        dto.metadata["new_key"] = "new_value"
        
        # Assert
        assert "new_key" not in original_metadata


class TestSearchResultDTO:
    """测试 SearchResultDTO 类"""
    
    def test_from_entity_basic(self):
        """测试从领域实体创建搜索结果 DTO"""
        # Arrange
        doc = Document(
            id=EntityId(value="doc_123"),
            content="Test content",
            embedding=None,
            metadata={"source": "test"},
            created_at=Timestamp.now(),
            updated_at=Timestamp.now()
        )
        search_result = SearchResult(
            document=doc,
            score=0.95,
            rank=0
        )
        
        # Act
        dto = SearchResultDTO.from_entity(search_result)
        
        # Assert
        assert dto.score == 0.95
        assert dto.rank == 0
        assert dto.document.id == "doc_123"
        assert dto.document.content == "Test content"
        assert dto.document.score == 0.95  # 分数应该传递到文档 DTO
    
    def test_from_entity_with_embedding(self):
        """测试从领域实体创建搜索结果 DTO（包含嵌入向量）"""
        # Arrange
        vector = Vector(dimensions=[1.0, 2.0, 3.0])
        doc = Document(
            id=EntityId.generate(),
            content="Test content",
            embedding=vector,
            metadata={},
            created_at=Timestamp.now(),
            updated_at=Timestamp.now()
        )
        search_result = SearchResult(document=doc, score=0.88, rank=1)
        
        # Act
        dto = SearchResultDTO.from_entity(search_result, include_embedding=True)
        
        # Assert
        assert dto.document.embedding == [1.0, 2.0, 3.0]
    
    def test_to_dict(self):
        """测试将搜索结果 DTO 转换为字典"""
        # Arrange
        doc_dto = DocumentDTO(
            id="doc_123",
            content="Test content",
            metadata={"source": "test"},
            created_at="2024-01-01T00:00:00",
            updated_at="2024-01-01T00:00:00"
        )
        result_dto = SearchResultDTO(
            document=doc_dto,
            score=0.95,
            rank=0
        )
        
        # Act
        data = result_dto.to_dict()
        
        # Assert
        assert data["score"] == 0.95
        assert data["rank"] == 0
        assert data["document"]["id"] == "doc_123"
        assert data["document"]["content"] == "Test content"
    
    def test_from_dict(self):
        """测试从字典创建搜索结果 DTO"""
        # Arrange
        data = {
            "document": {
                "id": "doc_123",
                "content": "Test content",
                "metadata": {"source": "test"},
                "created_at": "2024-01-01T00:00:00",
                "updated_at": "2024-01-01T00:00:00"
            },
            "score": 0.95,
            "rank": 0
        }
        
        # Act
        dto = SearchResultDTO.from_dict(data)
        
        # Assert
        assert dto.score == 0.95
        assert dto.rank == 0
        assert dto.document.id == "doc_123"
    
    def test_roundtrip_dict_to_dto_to_dict(self):
        """测试字典 -> DTO -> 字典的往返转换"""
        # Arrange
        original_data = {
            "document": {
                "id": "doc_roundtrip",
                "content": "Roundtrip test",
                "metadata": {},
                "created_at": "2024-01-01T00:00:00",
                "updated_at": "2024-01-01T00:00:00"
            },
            "score": 0.88,
            "rank": 2
        }
        
        # Act
        dto = SearchResultDTO.from_dict(original_data)
        converted_data = dto.to_dict()
        
        # Assert
        assert converted_data == original_data


class TestSearchResponseDTO:
    """测试 SearchResponseDTO 类"""
    
    def test_to_dict_basic(self):
        """测试将搜索响应 DTO 转换为字典（基本情况）"""
        # Arrange
        doc_dto = DocumentDTO(
            id="doc_123",
            content="Test content",
            metadata={},
            created_at="2024-01-01T00:00:00",
            updated_at="2024-01-01T00:00:00"
        )
        result_dto = SearchResultDTO(
            document=doc_dto,
            score=0.95,
            rank=0
        )
        response = SearchResponseDTO(
            results=[result_dto],
            total=1,
            query_text="test query"
        )
        
        # Act
        data = response.to_dict()
        
        # Assert
        assert data["total"] == 1
        assert data["query_text"] == "test query"
        assert len(data["results"]) == 1
        assert data["results"][0]["score"] == 0.95
        assert "took_ms" not in data
    
    def test_to_dict_with_timing(self):
        """测试将搜索响应 DTO 转换为字典（包含耗时）"""
        # Arrange
        response = SearchResponseDTO(
            results=[],
            total=0,
            query_text="test query",
            took_ms=150
        )
        
        # Act
        data = response.to_dict()
        
        # Assert
        assert data["took_ms"] == 150
    
    def test_from_dict_basic(self):
        """测试从字典创建搜索响应 DTO"""
        # Arrange
        data = {
            "results": [
                {
                    "document": {
                        "id": "doc_123",
                        "content": "Test",
                        "metadata": {},
                        "created_at": "2024-01-01T00:00:00",
                        "updated_at": "2024-01-01T00:00:00"
                    },
                    "score": 0.95,
                    "rank": 0
                }
            ],
            "total": 1,
            "query_text": "test query"
        }
        
        # Act
        dto = SearchResponseDTO.from_dict(data)
        
        # Assert
        assert dto.total == 1
        assert dto.query_text == "test query"
        assert len(dto.results) == 1
        assert dto.results[0].score == 0.95
        assert dto.took_ms is None
    
    def test_from_dict_with_timing(self):
        """测试从字典创建搜索响应 DTO（包含耗时）"""
        # Arrange
        data = {
            "results": [],
            "total": 0,
            "query_text": "test query",
            "took_ms": 150
        }
        
        # Act
        dto = SearchResponseDTO.from_dict(data)
        
        # Assert
        assert dto.took_ms == 150
    
    def test_roundtrip_dict_to_dto_to_dict(self):
        """测试字典 -> DTO -> 字典的往返转换"""
        # Arrange
        original_data = {
            "results": [
                {
                    "document": {
                        "id": "doc_1",
                        "content": "Content 1",
                        "metadata": {},
                        "created_at": "2024-01-01T00:00:00",
                        "updated_at": "2024-01-01T00:00:00"
                    },
                    "score": 0.95,
                    "rank": 0
                },
                {
                    "document": {
                        "id": "doc_2",
                        "content": "Content 2",
                        "metadata": {},
                        "created_at": "2024-01-01T00:00:00",
                        "updated_at": "2024-01-01T00:00:00"
                    },
                    "score": 0.88,
                    "rank": 1
                }
            ],
            "total": 2,
            "query_text": "test query",
            "took_ms": 200
        }
        
        # Act
        dto = SearchResponseDTO.from_dict(original_data)
        converted_data = dto.to_dict()
        
        # Assert
        assert converted_data == original_data
    
    def test_empty_results(self):
        """测试空结果列表"""
        # Arrange
        response = SearchResponseDTO(
            results=[],
            total=0,
            query_text="no results query"
        )
        
        # Act
        data = response.to_dict()
        
        # Assert
        assert data["results"] == []
        assert data["total"] == 0