feat: 审核规则模版+设置页面整合+去掉角色切换demo
后端:
- AuditRule 新增 value 字段, rule_type 扩展 DDL/DML
- MySQL/PG/Redis 预设规则种子接口 POST /audit-rules/seed/{db_type}
- 公告管理 CRUD (announcements)
- 消息推送 CRUD+测试 (notifications: 钉钉/飞书/邮件)
- LDAP 配置 CRUD+测试连接 (ldap-config)
- 164 tests passing (+31 新增)
前端:
- AuditRulesPage 重写: DB类型Tab+DDL/DML分组+参数编辑
- SettingsPage: 系统设置主页(审核规则/公告/推送/LDAP)
- NotificationPage: 钉钉/飞书/SMTP配置
- LdapConfigPage: LDAP设置+测试连接
- AnnouncementPage: 公告Markdown编辑
- 去掉左下角角色切换demo
- 侧边栏合并为: 系统设置(含审核规则+公告+推送+LDAP)
This commit is contained in:
logan committed
1 parent
d5d19d6be4
commit
bab55d3927
27 files changed
+2881
-591
No files matched your search
@@ -12,6 +12,9 @@ from app.models.instance import Instance
|
||||
from app.models.ticket import Ticket
|
||||
from app.models.audit_log import AuditLog
|
||||
from app.models.audit_rule import AuditRule
|
||||
from app.models.announcement import Announcement
|
||||
from app.models.notification import NotificationChannel
|
||||
from app.models.ldap_config import LdapConfig
|
||||
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite://"
|
||||
|
||||
|
||||
@@ -0,0 +1,240 @@
|
||||
"""Tests for Audit Rule seed endpoint and value field (Phase 4.2+)."""
|
||||
import pytest
|
||||
from tests.conftest import auth_header
|
||||
|
||||
|
||||
class TestSeedAuditRules:
|
||||
@pytest.mark.asyncio
|
||||
async def test_seed_mysql_rules(self, client, admin_token):
|
||||
"""MySQL 种子规则批量创建"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules/seed/mysql",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) == 15 # MySQL 有 15 条预设规则
|
||||
names = {r["name"] for r in data}
|
||||
assert "DDLMultiToCommit" in names
|
||||
assert "DDLMaxKeyParts" in names
|
||||
assert "DMLMaxInsertRows" in names
|
||||
assert "DMLWhere" in names
|
||||
# 验证 value 字段
|
||||
key_parts = next(r for r in data if r["name"] == "DDLMaxKeyParts")
|
||||
assert key_parts["value"] == "8"
|
||||
max_key = next(r for r in data if r["name"] == "DDLMaxKey")
|
||||
assert max_key["value"] == "12"
|
||||
insert_rows = next(r for r in data if r["name"] == "DMLMaxInsertRows")
|
||||
assert insert_rows["value"] == "500"
|
||||
table_name = next(r for r in data if r["name"] == "MaxTableNameLen")
|
||||
assert table_name["value"] == "45"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_seed_postgres_rules(self, client, admin_token):
|
||||
"""PostgreSQL 种子规则批量创建"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules/seed/postgres",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) == 11
|
||||
names = {r["name"] for r in data}
|
||||
assert "PGDDLMultiToCommit" in names
|
||||
assert "PGDDLMaxKeyParts" in names
|
||||
assert "PGMaxTableNameLen" in names
|
||||
# 验证 value 字段
|
||||
key_parts = next(r for r in data if r["name"] == "PGDDLMaxKeyParts")
|
||||
assert key_parts["value"] == "8"
|
||||
table_len = next(r for r in data if r["name"] == "PGMaxTableNameLen")
|
||||
assert table_len["value"] == "63"
|
||||
insert_rows = next(r for r in data if r["name"] == "PGDMLMaxInsertRows")
|
||||
assert insert_rows["value"] == "500"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_seed_redis_rules(self, client, admin_token):
|
||||
"""Redis 种子规则批量创建"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules/seed/redis",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) == 4
|
||||
names = {r["name"] for r in data}
|
||||
assert "RedisAllowFlushDB" in names
|
||||
assert "RedisMaxKeyLen" in names
|
||||
assert "RedisRequireTTL" in names
|
||||
assert "RedisDefaultTTL" in names
|
||||
# 验证 value 字段
|
||||
key_len = next(r for r in data if r["name"] == "RedisMaxKeyLen")
|
||||
assert key_len["value"] == "256"
|
||||
ttl = next(r for r in data if r["name"] == "RedisDefaultTTL")
|
||||
assert ttl["value"] == "3600"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_seed_unsupported_db_type(self, client, admin_token):
|
||||
"""不支持的数据库类型应返回 400"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules/seed/oracle",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "不支持的数据库类型" in resp.json()["detail"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_seed_idempotent(self, client, admin_token):
|
||||
"""重复调用 seed 应幂等，不创建重复规则"""
|
||||
resp1 = await client.post(
|
||||
"/api/v1/audit-rules/seed/mysql",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp1.status_code == 200
|
||||
count1 = len(resp1.json())
|
||||
|
||||
resp2 = await client.post(
|
||||
"/api/v1/audit-rules/seed/mysql",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp2.status_code == 200
|
||||
count2 = len(resp2.json())
|
||||
assert count2 == 0 # 第二次调用不再创建
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_seed_forbidden_for_dba(self, client, dba_token):
|
||||
"""DBA 无权调用 seed 接口"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules/seed/mysql",
|
||||
headers=auth_header(dba_token),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_seed_forbidden_for_developer(self, client, developer_token):
|
||||
"""Developer 无权调用 seed 接口"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules/seed/mysql",
|
||||
headers=auth_header(developer_token),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
class TestValueFieldAndDDLType:
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_rule_with_value(self, client, admin_token):
|
||||
"""创建带 value 字段的规则"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules",
|
||||
json={
|
||||
"name": "test_with_value",
|
||||
"description": "测试value字段",
|
||||
"rule_type": "DDL",
|
||||
"severity": "warning",
|
||||
"db_type": "mysql",
|
||||
"value": "100",
|
||||
"message": "测试规则带value",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["value"] == "100"
|
||||
assert data["rule_type"] == "DDL"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_rule_with_dml_type(self, client, admin_token):
|
||||
"""创建 DML 类型规则"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules",
|
||||
json={
|
||||
"name": "test_dml_rule",
|
||||
"rule_type": "DML",
|
||||
"severity": "error",
|
||||
"db_type": "mysql",
|
||||
"message": "DML测试规则",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["rule_type"] == "DML"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_rule_null_value(self, client, admin_token):
|
||||
"""value 可以为 null"""
|
||||
resp = await client.post(
|
||||
"/api/v1/audit-rules",
|
||||
json={
|
||||
"name": "test_null_value",
|
||||
"rule_type": "DDL",
|
||||
"severity": "info",
|
||||
"message": "null value test",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["value"] is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_rule_value(self, client, admin_token):
|
||||
"""更新 value 字段"""
|
||||
create = await client.post(
|
||||
"/api/v1/audit-rules",
|
||||
json={
|
||||
"name": "update_value_test",
|
||||
"rule_type": "DDL",
|
||||
"severity": "warning",
|
||||
"value": "50",
|
||||
"message": "test",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
rule_id = create.json()["id"]
|
||||
|
||||
resp = await client.put(
|
||||
f"/api/v1/audit-rules/{rule_id}",
|
||||
json={"value": "200"},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["value"] == "200"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_rules_with_ddl_filter(self, client, admin_token):
|
||||
"""按 DDL 类型筛选规则"""
|
||||
# 先 seed MySQL
|
||||
await client.post(
|
||||
"/api/v1/audit-rules/seed/mysql",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
resp = await client.get(
|
||||
"/api/v1/audit-rules?rule_type=DDL",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) > 0
|
||||
assert all(r["rule_type"] == "DDL" for r in data)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_response_includes_value_field(self, client, admin_token):
|
||||
"""响应中包含 value 字段"""
|
||||
create = await client.post(
|
||||
"/api/v1/audit-rules",
|
||||
json={
|
||||
"name": "response_value_test",
|
||||
"rule_type": "DDL",
|
||||
"severity": "warning",
|
||||
"value": "42",
|
||||
"message": "test",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
rule_id = create.json()["id"]
|
||||
|
||||
resp = await client.get(
|
||||
f"/api/v1/audit-rules/{rule_id}",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert "value" in resp.json()
|
||||
assert resp.json()["value"] == "42"
|
||||
@@ -0,0 +1,379 @@
|
||||
"""Tests for Settings modules: Announcement, NotificationChannel, LDAP Config."""
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from tests.conftest import auth_header
|
||||
|
||||
|
||||
# ─── Announcement Tests ────────────────────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_announcement(client: AsyncClient, admin_token: str):
|
||||
resp = await client.post(
|
||||
"/api/v1/announcements",
|
||||
json={"title": "系统公告", "content": "# Hello\n这是公告内容"},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["title"] == "系统公告"
|
||||
assert data["content"] == "# Hello\n这是公告内容"
|
||||
assert data["is_published"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_announcements(client: AsyncClient, admin_token: str, developer_token: str):
|
||||
# Create two announcements
|
||||
for i in range(2):
|
||||
await client.post(
|
||||
"/api/v1/announcements",
|
||||
json={"title": f"公告{i}", "content": f"内容{i}"},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
|
||||
# Admin can list
|
||||
resp = await client.get("/api/v1/announcements", headers=auth_header(admin_token))
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()) == 2
|
||||
|
||||
# Developer can also list
|
||||
resp = await client.get("/api/v1/announcements", headers=auth_header(developer_token))
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_announcement(client: AsyncClient, admin_token: str):
|
||||
create = await client.post(
|
||||
"/api/v1/announcements",
|
||||
json={"title": "原始标题", "content": "原始内容"},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
aid = create.json()["id"]
|
||||
|
||||
resp = await client.put(
|
||||
f"/api/v1/announcements/{aid}",
|
||||
json={"title": "更新标题", "is_published": True},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["title"] == "更新标题"
|
||||
assert data["is_published"] is True
|
||||
assert data["published_at"] is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_announcement(client: AsyncClient, admin_token: str):
|
||||
create = await client.post(
|
||||
"/api/v1/announcements",
|
||||
json={"title": "待删除", "content": "内容"},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
aid = create.json()["id"]
|
||||
|
||||
resp = await client.delete(
|
||||
f"/api/v1/announcements/{aid}",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
# Verify deleted
|
||||
resp = await client.get("/api/v1/announcements", headers=auth_header(admin_token))
|
||||
assert all(a["id"] != aid for a in resp.json())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_announcement_non_admin_forbidden(client: AsyncClient, developer_token: str):
|
||||
# Create forbidden
|
||||
resp = await client.post(
|
||||
"/api/v1/announcements",
|
||||
json={"title": "test", "content": "test"},
|
||||
headers=auth_header(developer_token),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
# Update forbidden
|
||||
resp = await client.put(
|
||||
"/api/v1/announcements/1",
|
||||
json={"title": "test"},
|
||||
headers=auth_header(developer_token),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
# Delete forbidden
|
||||
resp = await client.delete(
|
||||
"/api/v1/announcements/1",
|
||||
headers=auth_header(developer_token),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ─── NotificationChannel Tests ─────────────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_notification_channel(client: AsyncClient, admin_token: str):
|
||||
resp = await client.post(
|
||||
"/api/v1/notifications",
|
||||
json={
|
||||
"name": "钉钉告警",
|
||||
"channel_type": "dingtalk",
|
||||
"config": {"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=xxx", "secret": "mysecret"},
|
||||
"is_enabled": True,
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "钉钉告警"
|
||||
assert data["channel_type"] == "dingtalk"
|
||||
# Password/secret should be masked
|
||||
assert data["config"]["secret"] == "******"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_notification_channels(client: AsyncClient, admin_token: str):
|
||||
resp = await client.get(
|
||||
"/api/v1/notifications",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert isinstance(resp.json(), list)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_notification_channel(client: AsyncClient, admin_token: str):
|
||||
create = await client.post(
|
||||
"/api/v1/notifications",
|
||||
json={
|
||||
"name": "飞书通知",
|
||||
"channel_type": "webhook_feishu",
|
||||
"config": {"webhook_url": "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"},
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
cid = create.json()["id"]
|
||||
|
||||
resp = await client.put(
|
||||
f"/api/v1/notifications/{cid}",
|
||||
json={"name": "飞书通知-更新", "is_enabled": False},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "飞书通知-更新"
|
||||
assert resp.json()["is_enabled"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_notification_channel(client: AsyncClient, admin_token: str):
|
||||
create = await client.post(
|
||||
"/api/v1/notifications",
|
||||
json={
|
||||
"name": "待删除",
|
||||
"channel_type": "email",
|
||||
"config": {
|
||||
"smtp_host": "smtp.example.com",
|
||||
"smtp_port": 465,
|
||||
"smtp_user": "user@example.com",
|
||||
"smtp_password": "pass123",
|
||||
"from_addr": "user@example.com",
|
||||
"use_tls": True,
|
||||
},
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
cid = create.json()["id"]
|
||||
|
||||
resp = await client.delete(
|
||||
f"/api/v1/notifications/{cid}",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_notification_test_endpoint(client: AsyncClient, admin_token: str):
|
||||
create = await client.post(
|
||||
"/api/v1/notifications",
|
||||
json={
|
||||
"name": "钉钉测试",
|
||||
"channel_type": "dingtalk",
|
||||
"config": {"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=test"},
|
||||
"is_enabled": True,
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
cid = create.json()["id"]
|
||||
|
||||
resp = await client.post(
|
||||
f"/api/v1/notifications/{cid}/test",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["success"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_notification_non_admin_forbidden(client: AsyncClient, developer_token: str):
|
||||
resp = await client.get(
|
||||
"/api/v1/notifications",
|
||||
headers=auth_header(developer_token),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_email_config_mask_password(client: AsyncClient, admin_token: str):
|
||||
create = await client.post(
|
||||
"/api/v1/notifications",
|
||||
json={
|
||||
"name": "邮件通道",
|
||||
"channel_type": "email",
|
||||
"config": {
|
||||
"smtp_host": "smtp.example.com",
|
||||
"smtp_port": 587,
|
||||
"smtp_user": "admin@example.com",
|
||||
"smtp_password": "super_secret_pass",
|
||||
"from_addr": "admin@example.com",
|
||||
"use_tls": True,
|
||||
},
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
data = create.json()
|
||||
assert data["config"]["smtp_password"] == "******"
|
||||
|
||||
|
||||
# ─── LDAP Config Tests ─────────────────────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_ldap_config(client: AsyncClient, admin_token: str):
|
||||
resp = await client.put(
|
||||
"/api/v1/ldap-config",
|
||||
json={
|
||||
"ldap_host": "ldap.example.com",
|
||||
"ldap_port": 389,
|
||||
"use_ldaps": False,
|
||||
"admin_dn": "cn=admin,dc=example,dc=com",
|
||||
"admin_password": "admin_password_123",
|
||||
"search_filter": "(uid=%s)",
|
||||
"base_dn": "ou=users,dc=example,dc=com",
|
||||
"user_mapping": {"username": "uid", "display_name": "cn", "email": "mail"},
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["ldap_host"] == "ldap.example.com"
|
||||
assert data["ldap_port"] == 389
|
||||
# Password should be masked
|
||||
assert data["admin_password"] == "******"
|
||||
assert data["user_mapping"]["username"] == "uid"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_ldap_config(client: AsyncClient, admin_token: str):
|
||||
# First create
|
||||
await client.put(
|
||||
"/api/v1/ldap-config",
|
||||
json={
|
||||
"ldap_host": "ldap.test.com",
|
||||
"ldap_port": 636,
|
||||
"use_ldaps": True,
|
||||
"admin_dn": "cn=admin,dc=test,dc=com",
|
||||
"admin_password": "testpass",
|
||||
"base_dn": "dc=test,dc=com",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
|
||||
resp = await client.get(
|
||||
"/api/v1/ldap-config",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["ldap_host"] == "ldap.test.com"
|
||||
assert data["use_ldaps"] is True
|
||||
assert data["admin_password"] == "******"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_ldap_config(client: AsyncClient, admin_token: str):
|
||||
# Create
|
||||
await client.put(
|
||||
"/api/v1/ldap-config",
|
||||
json={
|
||||
"ldap_host": "ldap.old.com",
|
||||
"ldap_port": 389,
|
||||
"admin_dn": "cn=admin,dc=old,dc=com",
|
||||
"admin_password": "oldpass",
|
||||
"base_dn": "dc=old,dc=com",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
|
||||
# Update
|
||||
resp = await client.put(
|
||||
"/api/v1/ldap-config",
|
||||
json={
|
||||
"ldap_host": "ldap.new.com",
|
||||
"ldap_port": 636,
|
||||
"use_ldaps": True,
|
||||
"admin_dn": "cn=admin,dc=new,dc=com",
|
||||
"admin_password": "newpass",
|
||||
"base_dn": "dc=new,dc=com",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["ldap_host"] == "ldap.new.com"
|
||||
assert resp.json()["use_ldaps"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ldap_config_not_found(client: AsyncClient, admin_token: str):
|
||||
resp = await client.get(
|
||||
"/api/v1/ldap-config",
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ldap_test_connection(client: AsyncClient, admin_token: str):
|
||||
# Create config first
|
||||
await client.put(
|
||||
"/api/v1/ldap-config",
|
||||
json={
|
||||
"ldap_host": "ldap.example.com",
|
||||
"ldap_port": 389,
|
||||
"admin_dn": "cn=admin,dc=example,dc=com",
|
||||
"admin_password": "admin123",
|
||||
"base_dn": "dc=example,dc=com",
|
||||
},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/ldap-config/test",
|
||||
json={"test_user": "testuser", "test_password": "testpass"},
|
||||
headers=auth_header(admin_token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
# Mock mode should return success
|
||||
assert data["success"] is True
|
||||
assert "Mock" in data["message"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ldap_non_admin_forbidden(client: AsyncClient, developer_token: str):
|
||||
resp = await client.get(
|
||||
"/api/v1/ldap-config",
|
||||
headers=auth_header(developer_token),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
Reference in new issue
Block a user