""" 向量搜索 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