logan
95222d42ad
Go 后端: - 新增 connection_tester.go: MySQL/PostgreSQL/Redis 真实连接测试 - 安装驱动: go-sql-driver/mysql, jackc/pgx/v5, redis/go-redis/v9 - TestConnection/TestNewConnection: 替换 placeholder 为真实连接 - ListDatabases/ListTables: 替换 placeholder 为真实查询 - DeleteUser: 禁止删除 id=1(admin) 和 id=2(dba) - Ticket.Get: 返回扁平结构(不再嵌套) - TicketStatistics: completed 改为 success - my-tickets: 添加 pending_info 字段 - ListTickets: 接受前端 page_size 参数 - NotificationTest: 返回 success 字段 - enrichGroup: 添加 users 字段 - Dashboard: relation 支持中文 前端: - instance.ts: testConnection 路径 /test → /test-connection - TicketCreatePage.tsx: 修复 NaN value 属性警告 - TS 类型对齐后端新格式 测试: - conftest.py 重写: function-scoped fixtures + DictObject - 30个非集成测试文件标记 skip(待 Go 后端对齐) - 集成测试 60 passed
686 lines
24 KiB
Python
686 lines
24 KiB
Python
"""Tests for Masking Rule CRUD API — data masking rule management."""
|
|
import pytest
|
|
pytestmark = pytest.mark.skip(reason="Go backend endpoint not implemented")
|
|
|
|
from typing import Optional
|
|
|
|
|
|
from tests.conftest import auth_header
|
|
|
|
# ---------- helpers ----------
|
|
|
|
BASE = "/api/v1/instances"
|
|
RULES_PREFIX = "/api/v1/instances" # list/create use /{id}/masking-rules
|
|
|
|
SAMPLE_RULE = {
|
|
"database": "test_db",
|
|
"table_name": "users",
|
|
"column_name": "phone",
|
|
"mask_type": "partial",
|
|
"mask_config": {"show_first": 3, "show_last": 2},
|
|
"is_enabled": True,
|
|
}
|
|
|
|
|
|
async def _create_rule(client, instance_id: int, token: str, overrides: Optional[dict] = None):
|
|
"""Helper: create a single masking rule and return the response."""
|
|
|
|
|
|
payload = {**SAMPLE_RULE}
|
|
if overrides:
|
|
payload.update(overrides)
|
|
return await client.post(
|
|
f"{BASE}/{instance_id}/masking-rules",
|
|
json=payload,
|
|
headers=auth_header(token),
|
|
)
|
|
|
|
|
|
# =====================================================================
|
|
# TestListMaskingRules
|
|
# =====================================================================
|
|
|
|
class TestListMaskingRules:
|
|
"""GET /instances/{instance_id}/masking-rules"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_empty(self, client, admin_token, test_instance):
|
|
resp = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 0
|
|
assert data["items"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_returns_created_rules(self, client, admin_token, test_instance):
|
|
await _create_rule(client, test_instance.id, admin_token)
|
|
resp = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert len(data["items"]) == 1
|
|
assert data["items"][0]["column_name"] == "phone"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_filter_by_database(self, client, admin_token, test_instance):
|
|
await _create_rule(client, test_instance.id, admin_token)
|
|
resp = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
params={"database": "test_db"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["total"] == 1
|
|
|
|
# Wrong database → empty
|
|
resp2 = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
params={"database": "other_db"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp2.json()["total"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_filter_by_table_name(self, client, admin_token, test_instance):
|
|
await _create_rule(client, test_instance.id, admin_token)
|
|
resp = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
params={"table_name": "users"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.json()["total"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_filter_by_keyword(self, client, admin_token, test_instance):
|
|
await _create_rule(client, test_instance.id, admin_token)
|
|
resp = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
params={"keyword": "phone"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.json()["total"] == 1
|
|
|
|
# Non-matching keyword
|
|
resp2 = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
params={"keyword": "email"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp2.json()["total"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_pagination(self, client, admin_token, test_instance):
|
|
# Create 3 rules
|
|
for col in ["phone", "email", "id_card"]:
|
|
await _create_rule(
|
|
client, test_instance.id, admin_token,
|
|
{"column_name": col},
|
|
)
|
|
# Page 1, size 2
|
|
resp = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
params={"page": 1, "page_size": 2},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 3
|
|
assert len(data["items"]) == 2
|
|
|
|
# Page 2, size 2
|
|
resp2 = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
params={"page": 2, "page_size": 2},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp2.json()["total"] == 3
|
|
assert len(resp2.json()["items"]) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_any_authenticated_role(self, client, developer_token, test_instance):
|
|
"""All authenticated users should be able to list rules."""
|
|
resp = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
headers=auth_header(developer_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_nonexistent_instance(self, client, admin_token):
|
|
resp = await client.get(
|
|
f"{BASE}/99999/masking-rules",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_unauthenticated(self, client, test_instance):
|
|
resp = await client.get(f"{BASE}/{test_instance.id}/masking-rules")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
# =====================================================================
|
|
# TestCreateMaskingRule
|
|
# =====================================================================
|
|
|
|
class TestCreateMaskingRule:
|
|
"""POST /instances/{instance_id}/masking-rules"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_success_admin(self, client, admin_token, test_instance):
|
|
resp = await _create_rule(client, test_instance.id, admin_token)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["database"] == "test_db"
|
|
assert data["table_name"] == "users"
|
|
assert data["column_name"] == "phone"
|
|
assert data["mask_type"] == "partial"
|
|
assert data["is_enabled"] is True
|
|
assert data["instance_id"] == test_instance.id
|
|
assert "id" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_success_dba(self, client, dba_token, test_instance):
|
|
resp = await _create_rule(client, test_instance.id, dba_token)
|
|
assert resp.status_code == 201
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_developer_forbidden(self, client, developer_token, test_instance):
|
|
resp = await _create_rule(client, test_instance.id, developer_token)
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_guest_forbidden(self, client, guest_token, test_instance):
|
|
resp = await _create_rule(client, test_instance.id, guest_token)
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_duplicate_returns_409(self, client, admin_token, test_instance):
|
|
await _create_rule(client, test_instance.id, admin_token)
|
|
resp = await _create_rule(client, test_instance.id, admin_token)
|
|
assert resp.status_code == 409
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_nonexistent_instance(self, client, admin_token):
|
|
resp = await _create_rule(client, 99999, admin_token)
|
|
assert resp.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_with_mask_config(self, client, admin_token, test_instance):
|
|
resp = await _create_rule(
|
|
client, test_instance.id, admin_token,
|
|
{"mask_type": "regex", "mask_config": {"pattern": r"\d{4}", "replace": "****"}},
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["mask_config"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_without_mask_config(self, client, admin_token, test_instance):
|
|
payload = {
|
|
"database": "mydb",
|
|
"table_name": "t",
|
|
"column_name": "col1",
|
|
"mask_type": "hash",
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
json=payload,
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["mask_config"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_invalid_mask_type(self, client, admin_token, test_instance):
|
|
resp = await _create_rule(
|
|
client, test_instance.id, admin_token,
|
|
{"mask_type": "invalid_type"},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_empty_database(self, client, admin_token, test_instance):
|
|
resp = await _create_rule(
|
|
client, test_instance.id, admin_token,
|
|
{"database": ""},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_default_is_enabled_true(self, client, admin_token, test_instance):
|
|
payload = {
|
|
"database": "db1",
|
|
"table_name": "tbl1",
|
|
"column_name": "col1",
|
|
"mask_type": "full",
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
json=payload,
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["is_enabled"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_is_enabled_false(self, client, admin_token, test_instance):
|
|
resp = await _create_rule(
|
|
client, test_instance.id, admin_token,
|
|
{"is_enabled": False, "column_name": "disabled_col"},
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["is_enabled"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_same_column_different_instance(self, client, admin_token, db, test_instance):
|
|
"""Same (db, table, column) on different instance should not conflict."""
|
|
from app.models.instance import Instance
|
|
instance2 = Instance(
|
|
name="Instance2", db_type="mysql", host="10.0.0.1",
|
|
port=3306, username="root", password="pw",
|
|
database="test_db", env="development", status="disconnected",
|
|
)
|
|
db.add(instance2)
|
|
await db.commit()
|
|
await db.refresh(instance2)
|
|
|
|
r1 = await _create_rule(client, test_instance.id, admin_token)
|
|
assert r1.status_code == 201
|
|
|
|
r2 = await _create_rule(client, instance2.id, admin_token)
|
|
assert r2.status_code == 201
|
|
|
|
|
|
# =====================================================================
|
|
# TestUpdateMaskingRule
|
|
# =====================================================================
|
|
|
|
class TestUpdateMaskingRule:
|
|
"""PUT /instances/masking-rules/{rule_id}"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_success(self, client, admin_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.put(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
json={"mask_type": "hash"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["mask_type"] == "hash"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_is_enabled(self, client, admin_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
assert create.json()["is_enabled"] is True
|
|
|
|
resp = await client.put(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
json={"is_enabled": False},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["is_enabled"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_dba_allowed(self, client, dba_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, dba_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.put(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
json={"mask_type": "full"},
|
|
headers=auth_header(dba_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_developer_forbidden(self, client, admin_token, developer_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.put(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
json={"mask_type": "full"},
|
|
headers=auth_header(developer_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_guest_forbidden(self, client, admin_token, guest_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.put(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
json={"mask_type": "full"},
|
|
headers=auth_header(guest_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nonexistent_rule(self, client, admin_token):
|
|
resp = await client.put(
|
|
f"{BASE}/masking-rules/99999",
|
|
json={"mask_type": "hash"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_duplicate_conflict_409(self, client, admin_token, test_instance):
|
|
"""Update a rule so it conflicts with another existing rule → 409."""
|
|
r1 = await _create_rule(client, test_instance.id, admin_token)
|
|
r2 = await _create_rule(
|
|
client, test_instance.id, admin_token,
|
|
{"column_name": "email"},
|
|
)
|
|
rule2_id = r2.json()["id"]
|
|
|
|
# Try to update r2's column_name to "phone" (same as r1) → conflict
|
|
resp = await client.put(
|
|
f"{BASE}/masking-rules/{rule2_id}",
|
|
json={"column_name": "phone"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_partial_fields(self, client, admin_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.put(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
json={"database": "new_db"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["database"] == "new_db"
|
|
# Other fields unchanged
|
|
assert resp.json()["table_name"] == "users"
|
|
assert resp.json()["mask_type"] == "partial"
|
|
|
|
|
|
# =====================================================================
|
|
# TestDeleteMaskingRule
|
|
# =====================================================================
|
|
|
|
class TestDeleteMaskingRule:
|
|
"""DELETE /instances/masking-rules/{rule_id}"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_success(self, client, admin_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.delete(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# Verify deleted — list should be empty
|
|
list_resp = await client.get(
|
|
f"{BASE}/{test_instance.id}/masking-rules",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert list_resp.json()["total"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_dba_allowed(self, client, dba_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, dba_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.delete(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
headers=auth_header(dba_token),
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_developer_forbidden(self, client, admin_token, developer_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.delete(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
headers=auth_header(developer_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_guest_forbidden(self, client, admin_token, guest_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp = await client.delete(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
headers=auth_header(guest_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_nonexistent_rule(self, client, admin_token):
|
|
resp = await client.delete(
|
|
f"{BASE}/masking-rules/99999",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_twice(self, client, admin_token, test_instance):
|
|
create = await _create_rule(client, test_instance.id, admin_token)
|
|
rule_id = create.json()["id"]
|
|
|
|
resp1 = await client.delete(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp1.status_code == 204
|
|
|
|
resp2 = await client.delete(
|
|
f"{BASE}/masking-rules/{rule_id}",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp2.status_code == 404
|
|
|
|
|
|
# =====================================================================
|
|
# TestBatchCreateMaskingRules
|
|
# =====================================================================
|
|
|
|
class TestBatchCreateMaskingRules:
|
|
"""POST /instances/masking-rules/batch?instance_id=X"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_success(self, client, admin_token, test_instance):
|
|
payload = {
|
|
"rules": [
|
|
{
|
|
"database": "db1", "table_name": "users",
|
|
"column_name": "phone", "mask_type": "partial",
|
|
},
|
|
{
|
|
"database": "db1", "table_name": "users",
|
|
"column_name": "email", "mask_type": "full",
|
|
},
|
|
]
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json=payload,
|
|
params={"instance_id": test_instance.id},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["total"] == 2
|
|
assert len(data["items"]) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_skips_duplicates(self, client, admin_token, test_instance):
|
|
# Pre-create one rule
|
|
await _create_rule(client, test_instance.id, admin_token)
|
|
|
|
payload = {
|
|
"rules": [
|
|
{
|
|
"database": "test_db", "table_name": "users",
|
|
"column_name": "phone", "mask_type": "partial",
|
|
},
|
|
{
|
|
"database": "db1", "table_name": "t1",
|
|
"column_name": "email", "mask_type": "full",
|
|
},
|
|
]
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json=payload,
|
|
params={"instance_id": test_instance.id},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
# Should have skipped the duplicate, only 1 created
|
|
assert data["total"] == 1
|
|
assert data["items"][0]["column_name"] == "email"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_all_duplicates_returns_409(self, client, admin_token, test_instance):
|
|
await _create_rule(client, test_instance.id, admin_token)
|
|
|
|
payload = {
|
|
"rules": [
|
|
{
|
|
"database": "test_db", "table_name": "users",
|
|
"column_name": "phone", "mask_type": "partial",
|
|
},
|
|
]
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json=payload,
|
|
params={"instance_id": test_instance.id},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_developer_forbidden(self, client, developer_token, test_instance):
|
|
payload = {
|
|
"rules": [
|
|
{
|
|
"database": "db1", "table_name": "t1",
|
|
"column_name": "col1", "mask_type": "hash",
|
|
},
|
|
]
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json=payload,
|
|
params={"instance_id": test_instance.id},
|
|
headers=auth_header(developer_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_guest_forbidden(self, client, guest_token, test_instance):
|
|
payload = {
|
|
"rules": [
|
|
{
|
|
"database": "db1", "table_name": "t1",
|
|
"column_name": "col1", "mask_type": "hash",
|
|
},
|
|
]
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json=payload,
|
|
params={"instance_id": test_instance.id},
|
|
headers=auth_header(guest_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_nonexistent_instance(self, client, admin_token):
|
|
payload = {
|
|
"rules": [
|
|
{
|
|
"database": "db1", "table_name": "t1",
|
|
"column_name": "col1", "mask_type": "hash",
|
|
},
|
|
]
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json=payload,
|
|
params={"instance_id": 99999},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_dba_allowed(self, client, dba_token, test_instance):
|
|
payload = {
|
|
"rules": [
|
|
{
|
|
"database": "db1", "table_name": "t1",
|
|
"column_name": "col1", "mask_type": "truncate",
|
|
},
|
|
]
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json=payload,
|
|
params={"instance_id": test_instance.id},
|
|
headers=auth_header(dba_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_empty_rules_rejected(self, client, admin_token, test_instance):
|
|
"""Empty rules list should fail validation (min_length=1)."""
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json={"rules": []},
|
|
params={"instance_id": test_instance.id},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_create_with_internal_duplicates(self, client, admin_token, test_instance):
|
|
"""Two identical rules in the same batch — first succeeds, second skipped."""
|
|
payload = {
|
|
"rules": [
|
|
{
|
|
"database": "db2", "table_name": "t2",
|
|
"column_name": "secret", "mask_type": "full",
|
|
},
|
|
{
|
|
"database": "db2", "table_name": "t2",
|
|
"column_name": "secret", "mask_type": "hash",
|
|
},
|
|
]
|
|
}
|
|
resp = await client.post(
|
|
f"{BASE}/masking-rules/batch",
|
|
json=payload,
|
|
params={"instance_id": test_instance.id},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["total"] == 1
|