knowledge/tests/service/test_trunks_service.py

import pytest
from service.trunks_service import TrunksService
from model.trunks_model import Trunks
from sqlalchemy.exc import IntegrityError

@pytest.fixture(scope="module")
def trunks_service():
    return TrunksService()

@pytest.fixture
def test_trunk_data():
    return {
        "content": """本切片内容来自:《急诊与灾难医学（第4版）》 ，第二章急性发热 ，第一节概述
第二章      急性发热
发热（fever）是机体在内、外致热原作用下，或由于各种病因导致体温调节中枢功能障碍，而出现 的以体温升高超出正常范围为主要表现的临床症状。通常体表温度≥37.3℃可诊断为发热。热程在 2 周以内者为急性发热。
第一节  | 概   述
急性发热可分为感染性发热和非感染性发热。感染性发热较为多见，常见的病原体包括细菌、病 毒、衣原体、支原体、立克次体、螺旋体、真菌、原虫、蠕虫等，涉及的部位可由浅表组织到深部组织。非 感染性发热的病因包括结缔组织病、超敏反应性疾病、过敏性疾病、肿瘤性疾病、内分泌和代谢性疾 病、中枢神经系统疾病、散热障碍、创伤、烧伤、手术以及其他不明原因。
急性发热起病急骤，常有受凉、劳累或进不洁饮食史。发热及其伴随症状和体征多种多样，详细 询问病史，连续观察热程、热型，仔细查体，完善实验室检查，是发热诊断和鉴别诊断的重要依据。而 体温的高低并不是判断疾病危重程度的唯一依据。
发热是一个病因较为复杂的临床症状，而不是一种疾病，是机体对致病因素的一种全身性的代偿 反应。发热的治疗包括正确使用物理降温和解热药物，合理应用抗生素以及糖皮质激素。""",
        "file_path": "test_path.pdf"
    }

class TestTrunksServiceCRUD:
    def test_create_and_get_trunk(self, trunks_service, test_trunk_data):
        # 测试创建和查询
        created = trunks_service.create_trunk(test_trunk_data)
        assert created.id is not None

    def test_update_trunk(self, trunks_service, test_trunk_data):
        trunk = trunks_service.create_trunk(test_trunk_data)
        updated = trunks_service.update_trunk(trunk.id, {"content": "更新内容"})
        assert updated.content == "更新内容"

    def test_delete_trunk(self, trunks_service, test_trunk_data):
        trunk = trunks_service.create_trunk(test_trunk_data)
        assert trunks_service.delete_trunk(trunk.id)
        assert trunks_service.get_trunk_by_id(trunk.id) is None

class TestSearchOperations:
    def test_vector_search(self, trunks_service, test_trunk_data):
        trunks_service.create_trunk(test_trunk_data)
        results = trunks_service.search_by_vector("各种致病因素作用引起的有效循环血容量急剧减少")
        print("搜索结果:", results)
        assert len(results) > 0

    # def test_fulltext_search(self, trunks_service, test_trunk_data):
    #     trunks_service.create_trunk(test_trunk_data)
    #     results = trunks_service.fulltext_search("测试")
    #     assert len(results) > 0

class TestExceptionCases:
    def test_duplicate_id(self, trunks_service, test_trunk_data):
        with pytest.raises(IntegrityError):
            trunk1 = trunks_service.create_trunk(test_trunk_data)
            test_trunk_data["id"] = trunk1.id
            trunks_service.create_trunk(test_trunk_data)

    def test_invalid_vector_dimension(self, trunks_service, test_trunk_data):
        with pytest.raises(ValueError):
            invalid_data = test_trunk_data.copy()
            invalid_data["embedding"] = [0.1]*100
            trunks_service.create_trunk(invalid_data)

@pytest.fixture
def trunk_factory():
    class TrunkFactory:
        @staticmethod
        def create(**overrides):
            defaults = {
                "content": "工厂内容",
                "file_path": "factory_path.pdf"
            }
            return {**defaults, **overrides}
    return TrunkFactory()


class TestBatchCreateFromDirectory:
    def test_batch_create_from_directory(self, trunks_service, test_data_dir):
        # 使用现有目录路径
        base_path = Path(r'E:\project\vscode\files')
        
        # 遍历目录并创建trunk
        created_ids = []
        for txt_path in base_path.glob('**/*_split_*.txt'):
            relative_path = txt_path.relative_to(base_path.parent.parent)
            with open(txt_path, 'r', encoding='utf-8') as f:
                trunk_data = {
                    "content": f.read(),
                    "file_path": str(relative_path).replace('\\', '/')
                }
                trunk = trunks_service.create_trunk(trunk_data)
                created_ids.append(trunk.id)

        # 验证数据库记录
        for trunk_id in created_ids:
            db_trunk = trunks_service.get_trunk_by_id(trunk_id)
            assert db_trunk is not None
            assert ".txt" in db_trunk.file_path
            assert "_split_" in db_trunk.file_path
            assert len(db_trunk.content) > 0