opsxsql/backend/tests/test_instance_fields.py

"""测试 Instance 新增字段：access_mode, workflow, excluded_databases, mask_fields"""
import pytest
from tests.conftest import auth_header


@pytest.mark.asyncio
async def test_create_instance_with_full_fields(client, admin_token):
    """创建数据源：完整字段"""
    res = await client.post("/api/v1/instances", json={
        "name": "prod-mysql-只读库",
        "db_type": "mysql",
        "host": "mysql.example.com",
        "port": 3306,
        "username": "admin",
        "password": "secret123",
        "env": "production",
        "access_mode": "read",
        "workflow": [
            {"stage": "提交阶段", "role": "提交人", "assignee": ""},
            {"stage": "开发负责人", "role": "开发负责人", "assignee": "lairdlandonloose"},
            {"stage": "执行人", "role": "执行人", "assignee": "zachlogan"},
        ],
        "excluded_databases": ["sys", "information_schema", "mysql"],
        "mask_fields": [
            {"table": "users", "field": "phone", "rule": "mask"},
            {"table": "users", "field": "id_card", "rule": "hide"},
        ],
    }, headers=auth_header(admin_token))
    assert res.status_code == 201
    data = res.json()
    assert data["access_mode"] == "read"
    assert data["workflow"] is not None
    assert len(data["workflow"]) == 3
    assert data["workflow"][1]["assignee"] == "lairdlandonloose"
    assert data["excluded_databases"] == ["sys", "information_schema", "mysql"]
    assert len(data["mask_fields"]) == 2
    assert data["mask_fields"][0]["table"] == "users"


@pytest.mark.asyncio
async def test_create_instance_default_access_mode(client, admin_token):
    """默认连接模式为 readwrite"""
    res = await client.post("/api/v1/instances", json={
        "name": "test-default",
        "db_type": "mysql",
        "host": "127.0.0.1",
        "port": 3306,
        "username": "root",
        "password": "test123",
    }, headers=auth_header(admin_token))
    assert res.status_code == 201
    assert res.json()["access_mode"] == "readwrite"


@pytest.mark.asyncio
async def test_list_instance_includes_new_fields(client, admin_token):
    """列表接口返回新字段"""
    # 创建
    await client.post("/api/v1/instances", json={
        "name": "list-test",
        "db_type": "postgres",
        "host": "10.0.0.1",
        "port": 5432,
        "username": "postgres",
        "password": "pwd",
        "access_mode": "write",
        "workflow": [{"stage": "审批", "role": "DBA", "assignee": "alice"}],
    }, headers=auth_header(admin_token))

    res = await client.get("/api/v1/instances", headers=auth_header(admin_token))
    assert res.status_code == 200
    items = res.json()
    found = [i for i in items if i["name"] == "list-test"]
    assert len(found) == 1
    assert found[0]["access_mode"] == "write"
    assert found[0]["workflow"] is not None
    assert found[0]["workflow"][0]["assignee"] == "alice"


@pytest.mark.asyncio
async def test_update_instance_workflow(client, admin_token):
    """更新流程配置"""
    create = await client.post("/api/v1/instances", json={
        "name": "update-wf",
        "db_type": "mysql",
        "host": "127.0.0.1",
        "port": 3306,
        "username": "root",
        "password": "test",
    }, headers=auth_header(admin_token))
    iid = create.json()["id"]

    res = await client.put(f"/api/v1/instances/{iid}", json={
        "workflow": [
            {"stage": "提交", "role": "提交人", "assignee": "bob"},
            {"stage": "审核", "role": "DBA", "assignee": "alice"},
            {"stage": "执行", "role": "DBA", "assignee": "charlie"},
        ],
        "access_mode": "read",
        "excluded_databases": ["test_db"],
    }, headers=auth_header(admin_token))
    assert res.status_code == 200
    data = res.json()
    assert len(data["workflow"]) == 3
    assert data["access_mode"] == "read"
    assert data["excluded_databases"] == ["test_db"]


@pytest.mark.asyncio
async def test_instance_null_json_fields(client, admin_token):
    """不传新字段时，返回 null"""
    res = await client.post("/api/v1/instances", json={
        "name": "minimal",
        "db_type": "mysql",
        "host": "127.0.0.1",
        "port": 3306,
        "username": "root",
        "password": "test",
    }, headers=auth_header(admin_token))
    assert res.status_code == 201
    data = res.json()
    assert data["workflow"] is None
    assert data["excluded_databases"] is None
    assert data["mask_fields"] is None