logan
d5d19d6be4
- 项目名称 Web-Lite Pro → SQL-Platform 全局替换(13处) - Instance 模型新增: access_mode, workflow, excluded_databases, mask_fields - 创建数据源表单: 分区卡片(数据源信息/连接模式/流程/高级配置) - 后端 JSON 字段序列化/反序列化 - 新增 5 个 Instance 字段测试 (133 total)
125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
"""测试 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": "pro-cex-rds-只读库",
|
|
"db_type": "mysql",
|
|
"host": "pro-rds-ro.51aipay.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
|