cma-management/backend/tests/test_org.py

"""组织管理模块测试 — 组织节点CRUD + 树结构 + 切换状态"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from tests.conftest import create_test_user, get_token_for_user, auth_header
from app.models import OrgNode


class TestOrgNodes:
    """组织节点CRUD测试"""

    BASE = "/api/cma/org"

    def test_list_nodes_empty(self, client: TestClient, db: Session):
        """空列表"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.get(f"{self.BASE}/nodes", headers=auth_header(token))
        assert resp.status_code == 200
        # org_nodes may have seed data from init_db, just check response is valid
        assert "data" in resp.json()

    def test_tree_empty(self, client: TestClient, db: Session):
        """空树"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.get(f"{self.BASE}/tree", headers=auth_header(token))
        assert resp.status_code == 200
        assert "data" in resp.json()

    def test_create_root_node(self, client: TestClient, db: Session):
        """创建根节点（集团）"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.post(
            f"{self.BASE}/nodes",
            headers=auth_header(token),
            json={"name": "集团公司", "code": "GROUP", "level": 1, "sort_order": 1},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["name"] == "集团公司"
        assert data["code"] == "GROUP"
        assert data["level"] == 1

    def test_create_child_node(self, client: TestClient, db: Session):
        """创建子节点"""
        create_test_user(db)
        token = get_token_for_user(client)

        # 先创建父节点
        parent_resp = client.post(
            f"{self.BASE}/nodes",
            headers=auth_header(token),
            json={"name": "集团", "code": "G01", "level": 1},
        )
        parent_id = parent_resp.json()["id"]

        # 创建子节点
        resp = client.post(
            f"{self.BASE}/nodes",
            headers=auth_header(token),
            json={"parent_id": parent_id, "name": "华东事业部", "code": "BU_EAST", "level": 2},
        )
        assert resp.status_code == 200
        assert resp.json()["name"] == "华东事业部"
        assert resp.json()["level"] == 2

    def test_tree_structure(self, client: TestClient, db: Session):
        """返回正确的树结构"""
        create_test_user(db)
        token = get_token_for_user(client)

        # 创建层级数据
        r1 = client.post(
            f"{self.BASE}/nodes",
            headers=auth_header(token),
            json={"name": "集团", "code": "G01", "level": 1, "sort_order": 1},
        ).json()
        p_id = r1["id"]

        client.post(
            f"{self.BASE}/nodes",
            headers=auth_header(token),
            json={"parent_id": p_id, "name": "事业部A", "code": "BU_A", "level": 2},
        )
        client.post(
            f"{self.BASE}/nodes",
            headers=auth_header(token),
            json={"parent_id": p_id, "name": "事业部B", "code": "BU_B", "level": 2},
        )

        resp = client.get(f"{self.BASE}/tree", headers=auth_header(token)).json()
        tree = resp["data"]
        # 至少有一个根节点，且我们创建的根节点在列表中
        assert len(tree) >= 1
        created_root = [t for t in tree if t["label"] == "集团"]
        assert len(created_root) == 1
        assert len(created_root[0]["children"]) == 2

    def test_update_node(self, client: TestClient, db: Session):
        """修改节点"""
        create_test_user(db)
        token = get_token_for_user(client)

        node = OrgNode(name="旧名称", code="OLD", level=1, sort_order=1)
        db.add(node)
        db.commit()

        resp = client.put(
            f"{self.BASE}/nodes/{node.id}",
            headers=auth_header(token),
            json={"name": "新名称", "code": "NEW"},
        )
        assert resp.status_code == 200
        assert resp.json()["name"] == "新名称"
        assert resp.json()["code"] == "NEW"

    def test_delete_leaf_node(self, client: TestClient, db: Session):
        """删除叶子节点"""
        create_test_user(db)
        token = get_token_for_user(client)

        node = OrgNode(name="待删除节点", code="DEL", level=3)
        db.add(node)
        db.commit()

        resp = client.delete(f"{self.BASE}/nodes/{node.id}", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["message"] == "已删除"

    def test_delete_node_with_children_blocked(self, client: TestClient, db: Session):
        """删除有子节点的节点被阻止"""
        create_test_user(db)
        token = get_token_for_user(client)

        parent = OrgNode(name="父节点", code="PARENT", level=1)
        db.add(parent)
        db.flush()
        child = OrgNode(name="子节点", code="CHILD", level=2, parent_id=parent.id)
        db.add(child)
        db.commit()

        resp = client.delete(f"{self.BASE}/nodes/{parent.id}", headers=auth_header(token))
        assert resp.status_code == 400
        assert "子节点" in resp.json()["detail"]

    def test_toggle_node(self, client: TestClient, db: Session):
        """切换启用/禁用"""
        create_test_user(db)
        token = get_token_for_user(client)

        node = OrgNode(name="可切换节点", code="TOGGLE", level=2, enabled=1)
        db.add(node)
        db.commit()

        # 禁用它
        resp = client.put(f"{self.BASE}/nodes/{node.id}/toggle", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["enabled"] == 0

        # 再启用
        resp = client.put(f"{self.BASE}/nodes/{node.id}/toggle", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["enabled"] == 1

    def test_delete_not_found(self, client: TestClient, db: Session):
        """删除不存在的节点"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.delete(f"{self.BASE}/nodes/99999", headers=auth_header(token))
        assert resp.status_code == 404