knowledge/tests/service/test_kg_node_service.py

import pytest
from service.kg_node_service import KGNodeService
from model.kg_node import KGNode
from sqlalchemy.exc import IntegrityError

@pytest.fixture(scope="module")
def kg_node_service():
    from db.session import get_db
    return KGNodeService(next(get_db()))

@pytest.fixture
def test_node_data():
    return {
        "name": "测试节点",
        "category": "测试类别",
        "version": "1.0"
    }

class TestKGNodeServiceCRUD:
    def test_create_and_get_node(self, kg_node_service, test_node_data):
        created = kg_node_service.create_node(test_node_data)
        assert created.id is not None
        retrieved = kg_node_service.get_node(created.id)
        assert retrieved.name == test_node_data['name']

    def test_update_node(self, kg_node_service, test_node_data):
        node = kg_node_service.create_node(test_node_data)
        updated = kg_node_service.update_node(node.id, {"name": "更新后的节点"})
        assert updated.name == "更新后的节点"

    def test_delete_node(self, kg_node_service, test_node_data):
        node = kg_node_service.create_node(test_node_data)
        assert kg_node_service.delete_node(node.id) is None
        with pytest.raises(ValueError):
            kg_node_service.get_node(node.id)

    def test_duplicate_node(self, kg_node_service, test_node_data):
        kg_node_service.create_node(test_node_data)
        with pytest.raises(ValueError):
            kg_node_service.create_node(test_node_data)

class TestPaginatedSearch:
    def test_paginated_search(self, kg_node_service, test_node_data):
        results = kg_node_service.paginated_search({
            'keyword': '感染性',
            'pageNo': 1,
            'limit': 10
        })
        assert len(results['records']) > 0
        assert results['pagination']['pageNo'] == 1

class TestBatchProcess:
    def test_batch_process_er_nodes(self, kg_node_service, test_node_data):
        kg_node_service.batch_process_er_nodes()