From 2984f11181bf629e9b6647d438051284dcefd490 Mon Sep 17 00:00:00 2001
From: logan <admin@attacker.club>
Date: Wed, 3 Jun 2026 02:24:40 +0800
Subject: [PATCH] =?UTF-8?q?refactor:=20tests=E7=9B=AE=E5=BD=95=E7=B2=BE?=
 =?UTF-8?q?=E7=AE=80=E9=87=8D=E6=9E=84=20+=20=E8=84=8F=E6=95=B0=E6=8D=AE?=
 =?UTF-8?q?=E6=B8=85=E7=90=86=E8=84=9A=E6=9C=AC?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 删除16个旧Python测试文件(970行→400行)
- 重写conftest.py适配真实种子用户(admin/test123/test234/testtest/logan)
- 5个精简测试文件覆盖核心API:
  - test_auth.py: 登录/鉴权/种子用户验证 (6 tests)
  - test_instances.py: 实例列表/详情/连接测试 (6 tests)
  - test_tickets.py: 工单CRUD/只读SQL拦截/auditor权限 (7 tests)
  - test_workflow.py: E2E完整审批流/拒绝/撤销 (3 tests)
  - test_permissions.py: 角色权限/执行权限/workflow权限 (6 tests)
- 新增 cleanup_test_data.py 清理测试脏数据(支持--dry-run)
- 全量29测试3.27s通过
---
 tests/cleanup_test_data.py              |  87 +++++++++
 tests/conftest.py                       | 226 ++++++------------------
 tests/test_announcement.py              |  16 --
 tests/test_audit_rules.py               |  53 ------
 tests/test_auth.py                      |  32 ++--
 tests/test_instance_detail.py           |  42 -----
 tests/test_instance_fields.py           |  33 ----
 tests/test_instances.py                 | 111 +++---------
 tests/test_integration_api_crud.py      |  92 ----------
 tests/test_integration_audit_policy.py  |  20 ---
 tests/test_integration_config_notify.py |  35 ----
 tests/test_integration_query.py         |  23 ---
 tests/test_integration_ticket_flow.py   |  35 ----
 tests/test_ldap_config.py               |  32 ----
 tests/test_notification.py              |  38 ----
 tests/test_permissions.py               |  63 +++++++
 tests/test_redis.py                     |  38 ----
 tests/test_schema_drift.py              |  19 --
 tests/test_settings.py                  |  23 ---
 tests/test_tickets.py                   | 135 +++++++-------
 tests/test_users.py                     |  76 --------
 tests/test_workflow.py                  | 157 +++++++++-------
 22 files changed, 408 insertions(+), 978 deletions(-)
 create mode 100644 tests/cleanup_test_data.py
 delete mode 100644 tests/test_announcement.py
 delete mode 100644 tests/test_audit_rules.py
 delete mode 100644 tests/test_instance_detail.py
 delete mode 100644 tests/test_instance_fields.py
 delete mode 100644 tests/test_integration_api_crud.py
 delete mode 100644 tests/test_integration_audit_policy.py
 delete mode 100644 tests/test_integration_config_notify.py
 delete mode 100644 tests/test_integration_query.py
 delete mode 100644 tests/test_integration_ticket_flow.py
 delete mode 100644 tests/test_ldap_config.py
 delete mode 100644 tests/test_notification.py
 create mode 100644 tests/test_permissions.py
 delete mode 100644 tests/test_redis.py
 delete mode 100644 tests/test_schema_drift.py
 delete mode 100644 tests/test_settings.py
 delete mode 100644 tests/test_users.py

diff --git a/tests/cleanup_test_data.py b/tests/cleanup_test_data.py
new file mode 100644
index 0000000..87d9bd6
--- /dev/null
+++ b/tests/cleanup_test_data.py
@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+"""
+Cleanup test data from SQLite database.
+Removes tickets, approvals, and test-created tables, preserving seed data.
+
+Usage:
+    python tests/cleanup_test_data.py          # full cleanup
+    python tests/cleanup_test_data.py --dry-run # preview only
+"""
+import sqlite3
+import sys
+import os
+
+DB_PATH = os.path.join(os.path.dirname(__file__), '..', 'backend', 'sqlite3.db')
+
+# Tables in remote MySQL that were created by tests (DROP IF EXISTS)
+TEST_MYSQL_TABLES = [
+    "_test_tbl", "_e2e_test", "_reject_test", "_cancel_test", "_perm_test",
+    "orders", "_ghost_test_del", "flow_test", "ghost_test",
+]
+
+
+def cleanup(dry_run=False):
+    db = sqlite3.connect(DB_PATH)
+    cur = db.cursor()
+
+    actions = []
+
+    # 1. Count and delete ticket-related data
+    for table in ["ticket_sql_items", "ticket_approvals", "tickets"]:
+        cur.execute(f"SELECT COUNT(*) FROM {table}")
+        count = cur.fetchone()[0]
+        if count > 0:
+            actions.append(f"DELETE {count} rows FROM {table}")
+            if not dry_run:
+                cur.execute(f"DELETE FROM {table}")
+
+    # 2. Count query history
+    cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='query_history'")
+    if cur.fetchone():
+        cur.execute("SELECT COUNT(*) FROM query_history")
+        count = cur.fetchone()[0]
+        if count > 0:
+            actions.append(f"DELETE {count} rows FROM query_history")
+            if not dry_run:
+                cur.execute("DELETE FROM query_history")
+
+    # 3. Reset auto-increment counters
+    if not dry_run:
+        for table in ["ticket_sql_items", "ticket_approvals", "tickets"]:
+            cur.execute(f"DELETE FROM sqlite_sequence WHERE name='{table}'")
+
+    if not dry_run:
+        db.commit()
+    db.close()
+
+    return actions
+
+
+def main():
+    dry_run = "--dry-run" in sys.argv
+    if dry_run:
+        print("🔍 DRY RUN — preview only\n")
+
+    actions = cleanup(dry_run)
+
+    if not actions:
+        print("✅ 数据库干净，无需清理")
+        return
+
+    for a in actions:
+        prefix = "  WOULD " if dry_run else "  ✅ "
+        print(f"{prefix}{a}")
+
+    if dry_run:
+        print(f"\n共 {len(actions)} 项待清理 (使用不带 --dry-run 执行)")
+    else:
+        print(f"\n✅ 已清理 {len(actions)} 项")
+
+    # Print note about MySQL remote tables
+    print(f"\n⚠️  注意: 远程 MySQL 测试表需手动清理:")
+    for t in TEST_MYSQL_TABLES:
+        print(f"    DROP TABLE IF EXISTS opsxsql_test.{t};")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/conftest.py b/tests/conftest.py
index 5193fe8..1716a4f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,212 +1,92 @@
-"""Integration test conftest — connects to real Go backend at localhost:9000"""
-from typing import AsyncGenerator
+"""Integration test conftest — Go/Gin backend at localhost:9000"""
 import pytest
 import pytest_asyncio
 from httpx import AsyncClient
 
 BASE_URL = "http://localhost:9000"
 
-
-class DictObject(dict):
-    """Dict that supports dot-notation access: obj.key == obj['key']."""
-    def __getattr__(self, key):
-        try:
-            return self[key]
-        except KeyError:
-            raise AttributeError(f"'{type(self).__name__}' has no attribute '{key}'")
-    def __setattr__(self, key, value):
-        self[key] = value
-
-# Token cache — avoid re-login for every test
-_token_cache: dict[str, str] = {}
+# ═══════════════════════════════════════════════
+# Real test users (seeded by main.go seed_data)
+# ═══════════════════════════════════════════════
+USERS = {
+    "admin":     {"password": "admin123", "role": "admin"},
+    "test123":   {"password": "test123",  "role": "developer"},
+    "test234":   {"password": "test234",  "role": "developer"},
+    "testtest":  {"password": "testtest", "role": "developer"},
+    "logan":     {"password": "zxc987",   "role": "dba"},
+}
 
 
-async def _get_token(client: AsyncClient, username: str, password: str = "admin123") -> str:
+def auth_header(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _login(client: AsyncClient, username: str, password: str) -> str:
     resp = await client.post("/api/v1/auth/login", json={
-        "username": username,
-        "password": password,
+        "username": username, "password": password,
     })
-    if resp.status_code == 200:
-        return resp.json()["access_token"]
-    return ""
+    assert resp.status_code == 200, f"Login failed for {username}: {resp.status_code} {resp.text}"
+    return resp.json()["access_token"]
 
 
-async def _ensure_admin_exists(client: AsyncClient):
-    """Ensure admin and dba users exist — recreate if deleted by tests."""
-    # Try login first
-    admin_token = await _get_token(client, "admin")
-    if admin_token:
-        return admin_token
-    
-    # Admin doesn't exist — use dev1 to create it
-    dev_token = await _get_token(client, "dev1")
-    if not dev_token:
-        return ""
-    
-    # Create admin user via API
-    resp = await client.post("/api/v1/users", json={
-        "username": "admin",
-        "display_name": "超级管理员",
-        "role": "admin",
-        "password": "admin123",
-    }, headers={"Authorization": f"Bearer {dev_token}"})
-    
-    if resp.status_code in (200, 201):
-        pass  # created
-    elif resp.status_code == 400:
-        # Username exists but password wrong — can't fix via API
-        pass
-    
-    # Try login again
-    return await _get_token(client, "admin")
-
+# ═══════════════════════════════════════════════
+# Fixtures
+# ═══════════════════════════════════════════════
 
 @pytest_asyncio.fixture
-async def client() -> AsyncGenerator[AsyncClient, None]:
-    """Function-scoped client — fresh for each test, no event loop issues."""
+async def client():
     async with AsyncClient(base_url=BASE_URL, timeout=30.0) as c:
         yield c
 
 
 @pytest_asyncio.fixture
-async def admin_token(client: AsyncClient) -> str:
-    token = await _get_token(client, "admin")
-    assert token, "Admin login failed"
-    return token
+async def admin_token(client):
+    return await _login(client, "admin", "admin123")
 
 
 @pytest_asyncio.fixture
-async def dba_token(client: AsyncClient) -> str:
-    token = await _get_token(client, "dba")
-    if token:
-        return token
-    # Fallback
-    return await _get_token(client, "admin")
+async def dba_token(client):
+    return await _login(client, "logan", "zxc987")
 
 
 @pytest_asyncio.fixture
-async def developer_token(client: AsyncClient) -> str:
-    token = await _get_token(client, "dev1")
-    assert token, "Developer login failed"
-    return token
+async def developer_token(client):
+    return await _login(client, "testtest", "testtest")
 
 
 @pytest_asyncio.fixture
-async def guest_token(client: AsyncClient) -> str:
-    token = await _get_token(client, "auditor")
-    if token:
-        return token
-    pytest.skip("guest/auditor user not available")
-    return ""
-
-
-# ══════════════════════════════════════════════════════════════
-# Helper fixtures for non-integration test files
-# ══════════════════════════════════════════════════════════════
-
-@pytest_asyncio.fixture
-async def admin_user(client: AsyncClient, admin_token: str) -> dict:
-    resp = await client.get("/api/v1/auth/me", headers=auth_header(admin_token))
-    if resp.status_code == 200:
-        return resp.json()
-    return {"id": 1, "username": "admin", "role": "admin"}
+async def approver1_token(client):
+    """test123 — 初审人"""
+    return await _login(client, "test123", "test123")
 
 
 @pytest_asyncio.fixture
-async def disabled_user(client: AsyncClient, admin_token: str) -> dict:
-    return DictObject({"id": 999, "username": "disabled_test_user", "role": "developer", "is_active": False})
+async def approver2_token(client):
+    """test234 — 复审人"""
+    return await _login(client, "test234", "test234")
 
 
 @pytest_asyncio.fixture
-async def test_mysql_instance(client: AsyncClient, admin_token: str) -> dict:
+async def mysql_instance(client, admin_token):
+    """获取第一个 MySQL 实例"""
     resp = await client.get("/api/v1/instances", headers=auth_header(admin_token))
-    if resp.status_code == 200:
-        data = resp.json()
-        items = data.get("items", data) if isinstance(data, dict) else data
-        for inst in items:
-            if inst.get("db_type") == "mysql":
-                return inst
-    return DictObject({"id": 3, "name": "测试MySQL", "db_type": "mysql", "host": "127.0.0.1", "port": 3306})
+    assert resp.status_code == 200
+    data = resp.json()
+    items = data.get("items", data) if isinstance(data, dict) else data
+    for inst in items:
+        if inst.get("db_type") == "mysql":
+            return inst
+    pytest.skip("No MySQL instance found")
 
 
 @pytest_asyncio.fixture
-async def test_redis_instance(client: AsyncClient, admin_token: str) -> dict:
+async def redis_instance(client, admin_token):
+    """获取第一个 Redis 实例"""
     resp = await client.get("/api/v1/instances", headers=auth_header(admin_token))
-    if resp.status_code == 200:
-        data = resp.json()
-        items = data.get("items", data) if isinstance(data, dict) else data
-        for inst in items:
-            if inst.get("db_type") == "redis":
-                return inst
-    return DictObject({"id": 2, "name": "测试Redis", "db_type": "redis", "host": "127.0.0.1", "port": 6379})
-
-
-@pytest_asyncio.fixture
-async def test_ticket(client: AsyncClient, admin_token: str, test_mysql_instance: dict) -> dict:
-    resp = await client.post("/api/v1/tickets", json={
-        "title": "conftest测试工单", "db_type": "mysql",
-        "instance_id": test_mysql_instance["id"],
-        "target_database": "test",
-        "sql_content": "SELECT 1", "reason": "conftest测试",
-    }, headers=auth_header(admin_token))
-    if resp.status_code in (200, 201):
-        return resp.json()
-    return DictObject({"id": 1, "title": "conftest测试工单", "status": "pending"})
-
-
-
-@pytest_asyncio.fixture
-async def test_instance(client: AsyncClient, admin_token: str) -> dict:
-    """Get first MySQL test instance."""
-    resp = await client.get("/api/v1/instances", headers=auth_header(admin_token))
-    if resp.status_code == 200:
-        data = resp.json()
-        items = data.get("items", data) if isinstance(data, dict) else data
-        for inst in items:
-            if inst.get("db_type") == "mysql":
-                return DictObject(inst)
-    return DictObject({"id": 3, "name": "测试MySQL", "db_type": "mysql", "host": "127.0.0.1", "port": 3306})
-
-
-@pytest_asyncio.fixture
-async def db(client: AsyncClient, admin_token: str) -> dict:
-    """Database fixture — get first MySQL instance."""
-    resp = await client.get("/api/v1/instances", headers=auth_header(admin_token))
-    if resp.status_code == 200:
-        data = resp.json()
-        items = data.get("items", data) if isinstance(data, dict) else data
-        for inst in items:
-            if inst.get("db_type") == "mysql":
-                return DictObject(inst)
-    return DictObject({"id": 3, "name": "测试MySQL", "db_type": "mysql", "host": "127.0.0.1", "port": 3306})
-
-
-@pytest_asyncio.fixture
-async def dba_user(client: AsyncClient, dba_token: str) -> dict:
-    """Get DBA user info."""
-    resp = await client.get("/api/v1/auth/me", headers=auth_header(dba_token))
-    if resp.status_code == 200:
-        return resp.json()
-    return DictObject({"id": 2, "username": "dba", "role": "dba"})
-
-
-@pytest_asyncio.fixture
-async def developer_user(client: AsyncClient, developer_token: str) -> dict:
-    """Get developer user info."""
-    resp = await client.get("/api/v1/auth/me", headers=auth_header(developer_token))
-    if resp.status_code == 200:
-        return resp.json()
-    return DictObject({"id": 3, "username": "dev1", "role": "developer"})
-
-
-@pytest_asyncio.fixture
-async def guest_user(client: AsyncClient, guest_token: str) -> dict:
-    """Get guest/auditor user info."""
-    resp = await client.get("/api/v1/auth/me", headers=auth_header(guest_token))
-    if resp.status_code == 200:
-        return resp.json()
-    return DictObject({"id": 5, "username": "auditor", "role": "auditor"})
-
-
-def auth_header(token: str) -> dict:
-    return {"Authorization": f"Bearer {token}"}
+    assert resp.status_code == 200
+    data = resp.json()
+    items = data.get("items", data) if isinstance(data, dict) else data
+    for inst in items:
+        if inst.get("db_type") == "redis":
+            return inst
+    pytest.skip("No Redis instance found")
diff --git a/tests/test_announcement.py b/tests/test_announcement.py
deleted file mode 100644
index ff0e7fa..0000000
--- a/tests/test_announcement.py
+++ /dev/null
@@ -1,16 +0,0 @@
-"""Announcement list tests — Go/Gin backend (read-only endpoint)."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestAnnouncements:
-    async def test_list_announcements(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/announcements", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        assert isinstance(data, list)
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/announcements")
-        assert resp.status_code == 401
diff --git a/tests/test_audit_rules.py b/tests/test_audit_rules.py
deleted file mode 100644
index 75962aa..0000000
--- a/tests/test_audit_rules.py
+++ /dev/null
@@ -1,53 +0,0 @@
-"""Audit rules read-only tests — Go/Gin backend.
-Go backend only has GET /audit-rules (list), no CRUD."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestListAuditRules:
-    async def test_list_rules(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/audit-rules", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        assert isinstance(data, list)
-        assert len(data) > 0
-
-    async def test_list_rules_has_required_fields(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/audit-rules", headers=auth_header(admin_token))
-        data = resp.json()
-        if data:
-            rule = data[0]
-            assert "id" in rule
-            assert "name" in rule
-            assert "rule_type" in rule
-            assert "severity" in rule
-            assert "db_type" in rule
-            assert "is_enabled" in rule
-
-    async def test_filter_db_type(self, client: AsyncClient, admin_token: str):
-        """Go backend does not support server-side filter, verify data has db_type field."""
-        resp = await client.get("/api/v1/audit-rules", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        mysql_rules = [r for r in data if r["db_type"] == "mysql"]
-        assert len(mysql_rules) > 0, "Should have at least one mysql rule"
-
-    async def test_filter_rule_type(self, client: AsyncClient, admin_token: str):
-        """Go backend does not support server-side filter, verify rule_type field exists."""
-        resp = await client.get("/api/v1/audit-rules", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        ddl_rules = [r for r in data if r["rule_type"] == "DDL"]
-        assert len(ddl_rules) > 0, "Should have at least one DDL rule"
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/audit-rules")
-        assert resp.status_code == 401
-
-    async def test_all_users_can_list(self, client: AsyncClient, admin_token: str, developer_token: str):
-        """audit-rules is under auth group (no role restriction), all authenticated users can access."""
-        resp = await client.get("/api/v1/audit-rules", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        resp2 = await client.get("/api/v1/audit-rules", headers=auth_header(developer_token))
-        assert resp2.status_code == 200
diff --git a/tests/test_auth.py b/tests/test_auth.py
index c7f583f..574f22f 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -1,39 +1,51 @@
-"""Auth endpoint tests — Go/Gin backend."""
+"""Auth API tests"""
 import pytest
 from httpx import AsyncClient
-from tests.conftest import auth_header
+def auth_header(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
 
 
 class TestLogin:
     async def test_admin_login(self, client: AsyncClient):
         resp = await client.post("/api/v1/auth/login", json={
-            "username": "admin", "password": "admin123"
+            "username": "admin", "password": "admin123",
         })
         assert resp.status_code == 200
-        data = resp.json()
-        assert "access_token" in data
+        assert "access_token" in resp.json()
+
+    async def test_all_seed_users_login(self, client: AsyncClient):
+        """验证所有种子用户都能登录"""
+        users = {
+            "admin": "admin123", "test123": "test123",
+            "test234": "test234", "testtest": "testtest", "logan": "zxc987",
+        }
+        for username, password in users.items():
+            resp = await client.post("/api/v1/auth/login", json={
+                "username": username, "password": password,
+            })
+            assert resp.status_code == 200, f"{username} login failed"
+            assert "access_token" in resp.json()
 
     async def test_wrong_password(self, client: AsyncClient):
         resp = await client.post("/api/v1/auth/login", json={
-            "username": "admin", "password": "wrong"
+            "username": "admin", "password": "wrong",
         })
         assert resp.status_code == 401
 
     async def test_nonexistent_user(self, client: AsyncClient):
         resp = await client.post("/api/v1/auth/login", json={
-            "username": "nobody", "password": "test"
+            "username": "nobody", "password": "xxx",
         })
         assert resp.status_code == 401
 
 
 class TestGetMe:
-    async def test_get_me(self, client: AsyncClient, admin_token: str):
+    async def test_get_me(self, client: AsyncClient, admin_token):
         resp = await client.get("/api/v1/auth/me", headers=auth_header(admin_token))
         assert resp.status_code == 200
         data = resp.json()
         assert data["username"] == "admin"
-        assert "role" in data
-        assert "id" in data
+        assert data["role"] == "admin"
 
     async def test_no_token(self, client: AsyncClient):
         resp = await client.get("/api/v1/auth/me")
diff --git a/tests/test_instance_detail.py b/tests/test_instance_detail.py
deleted file mode 100644
index cd74be2..0000000
--- a/tests/test_instance_detail.py
+++ /dev/null
@@ -1,42 +0,0 @@
-"""Instance detail endpoint tests — Go/Gin backend."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestInstanceDetail:
-    async def test_detail_databases(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        inst = test_mysql_instance
-        resp = await client.get(
-            f"/api/v1/instances/{inst['id']}/detail/databases",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code in (200, 500)  # 500 if MySQL unreachable
-
-    async def test_detail_processlist(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        inst = test_mysql_instance
-        resp = await client.get(
-            f"/api/v1/instances/{inst['id']}/detail/processlist",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code in (200, 500)  # 500 if MySQL unreachable
-
-    async def test_detail_status(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        inst = test_mysql_instance
-        resp = await client.get(
-            f"/api/v1/instances/{inst['id']}/detail/status",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code in (200, 500)  # 500 if MySQL unreachable
-
-    async def test_detail_users(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        inst = test_mysql_instance
-        resp = await client.get(
-            f"/api/v1/instances/{inst['id']}/detail/users",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/instances/4/detail/databases")
-        assert resp.status_code == 401
diff --git a/tests/test_instance_fields.py b/tests/test_instance_fields.py
deleted file mode 100644
index 961e079..0000000
--- a/tests/test_instance_fields.py
+++ /dev/null
@@ -1,33 +0,0 @@
-"""Instance fields (database/table listing) tests — Go/Gin backend."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestInstanceFields:
-    async def test_list_databases(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        resp = await client.get(
-            f"/api/v1/instances/{test_mysql_instance['id']}/databases",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-
-    async def test_list_tables(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        # First get a database name
-        db_resp = await client.get(
-            f"/api/v1/instances/{test_mysql_instance['id']}/databases",
-            headers=auth_header(admin_token),
-        )
-        dbs = db_resp.json()
-        if not dbs:
-            pytest.skip("No databases available")
-        db_name = dbs[0] if isinstance(dbs[0], str) else dbs[0].get("name", "test")
-        resp = await client.get(
-            f"/api/v1/instances/{test_mysql_instance['id']}/databases/{db_name}/tables",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/instances/4/databases")
-        assert resp.status_code == 401
diff --git a/tests/test_instances.py b/tests/test_instances.py
index 5c15248..e728b8c 100644
--- a/tests/test_instances.py
+++ b/tests/test_instances.py
@@ -1,107 +1,52 @@
-"""Instance CRUD + listing tests — Go/Gin backend."""
+"""Instance API tests"""
 import pytest
 from httpx import AsyncClient
-from tests.conftest import auth_header
+def auth_header(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
 
 
 class TestListInstances:
-    async def test_list_instances(self, client: AsyncClient, admin_token: str):
+    async def test_admin_sees_all(self, client: AsyncClient, admin_token):
         resp = await client.get("/api/v1/instances", headers=auth_header(admin_token))
         assert resp.status_code == 200
         data = resp.json()
-        assert isinstance(data, list)
+        items = data.get("items", data) if isinstance(data, dict) else data
+        assert len(items) >= 3  # MySQL + 2 Redis
 
-    async def test_list_filter_db_type(self, client: AsyncClient, admin_token: str):
-        """Go backend does not support ?db_type= filter, filter client-side."""
-        resp = await client.get("/api/v1/instances", headers=auth_header(admin_token))
+    async def test_developer_sees_group_instances(self, client: AsyncClient, developer_token):
+        """testtest 在 test 权限组，应看到绑定的实例"""
+        resp = await client.get("/api/v1/instances", headers=auth_header(developer_token))
         assert resp.status_code == 200
         data = resp.json()
-        mysql_instances = [i for i in data if i["db_type"] == "mysql"]
-        if not mysql_instances:
-            pytest.skip("No mysql instances")
+        items = data.get("items", data) if isinstance(data, dict) else data
+        assert len(items) >= 1
 
-    async def test_unauthenticated(self, client: AsyncClient):
+    async def test_no_auth(self, client: AsyncClient):
         resp = await client.get("/api/v1/instances")
         assert resp.status_code == 401
 
 
 class TestGetInstance:
-    async def test_get_existing(self, client: AsyncClient, admin_token: str):
-        # List first to get a valid id
-        lst = await client.get("/api/v1/instances", headers=auth_header(admin_token))
-        instances = lst.json()
-        if not instances:
-            pytest.skip("No instances in database")
-        inst_id = instances[0]["id"]
-        resp = await client.get(f"/api/v1/instances/{inst_id}", headers=auth_header(admin_token))
+    async def test_get_mysql_instance(self, client: AsyncClient, admin_token, mysql_instance):
+        resp = await client.get(f"/api/v1/instances/{mysql_instance['id']}", headers=auth_header(admin_token))
         assert resp.status_code == 200
-        assert resp.json()["id"] == inst_id
-
-    async def test_get_not_found(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/instances/99999", headers=auth_header(admin_token))
-        assert resp.status_code in (404, 400)
+        assert resp.json()["db_type"] == "mysql"
 
 
-class TestCreateInstance:
-    async def test_create_instance(self, client: AsyncClient, admin_token: str):
-        import uuid
-        resp = await client.post("/api/v1/instances", headers=auth_header(admin_token), json={
-            "name": f"test-{uuid.uuid4().hex[:6]}",
-            "db_type": "mysql",
-            "host": "10.0.0.1",
-            "port": 3306,
-            "username": "root",
-        })
-        assert resp.status_code == 201
+class TestTestConnection:
+    async def test_mysql_connection(self, client: AsyncClient, admin_token, mysql_instance):
+        resp = await client.post(
+            f"/api/v1/instances/{mysql_instance['id']}/test-connection",
+            headers=auth_header(admin_token), json={},
+        )
+        assert resp.status_code == 200
         data = resp.json()
-        assert "id" in data
+        assert data.get("success") is True
 
-    async def test_developer_cannot_create(self, client: AsyncClient, developer_token: str):
-        resp = await client.post("/api/v1/instances", headers=auth_header(developer_token), json={
-            "name": "nope", "db_type": "mysql", "host": "10.0.0.1",
-            "port": 3306, "username": "root",
-        })
-        assert resp.status_code == 403
-
-
-class TestUpdateInstance:
-    async def test_update_instance(self, client: AsyncClient, admin_token: str):
-        lst = await client.get("/api/v1/instances", headers=auth_header(admin_token))
-        instances = lst.json()
-        if not instances:
-            pytest.skip("No instances")
-        inst_id = instances[0]["id"]
-        resp = await client.put(
-            f"/api/v1/instances/{inst_id}",
-            headers=auth_header(admin_token),
-            json={"name": "updated-test"},
+    async def test_redis_connection(self, client: AsyncClient, admin_token, redis_instance):
+        resp = await client.post(
+            f"/api/v1/instances/{redis_instance['id']}/test-connection",
+            headers=auth_header(admin_token), json={},
         )
         assert resp.status_code == 200
-
-    async def test_update_not_found(self, client: AsyncClient, admin_token: str):
-        resp = await client.put(
-            "/api/v1/instances/99999",
-            headers=auth_header(admin_token),
-            json={"name": "x"},
-        )
-        assert resp.status_code in (404, 400)
-
-
-class TestDeleteInstance:
-    async def test_delete_not_found(self, client: AsyncClient, admin_token: str):
-        resp = await client.delete("/api/v1/instances/99999", headers=auth_header(admin_token))
-        assert resp.status_code in (200, 404)
-
-
-class TestInstanceConnections:
-    async def test_list_databases(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        inst = test_mysql_instance
-        resp = await client.get(
-            f"/api/v1/instances/{inst['id']}/databases",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-
-    async def test_accessible_instances(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/instances/accessible", headers=auth_header(admin_token))
-        assert resp.status_code == 200
+        assert resp.json().get("success") is True
diff --git a/tests/test_integration_api_crud.py b/tests/test_integration_api_crud.py
deleted file mode 100644
index a6eb784..0000000
--- a/tests/test_integration_api_crud.py
+++ /dev/null
@@ -1,92 +0,0 @@
-"""Integration: API CRUD smoke test — Go/Gin backend.
-Validates core API flow: auth → list → create → read → update → delete."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestInstanceFlow:
-    """Full instance lifecycle test."""
-
-    async def test_instance_crud_flow(self, client: AsyncClient, admin_token: str):
-        import uuid
-        suffix = uuid.uuid4().hex[:6]
-
-        # Create
-        resp = await client.post("/api/v1/instances", headers=auth_header(admin_token), json={
-            "name": f"crud-test-{suffix}",
-            "db_type": "mysql",
-            "host": "10.0.0.99",
-            "port": 3306,
-            "username": "root",
-        })
-        assert resp.status_code == 201
-        inst = resp.json()
-        inst_id = inst["id"]
-
-        # Read
-        resp = await client.get(f"/api/v1/instances/{inst_id}", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        assert resp.json()["name"] == f"crud-test-{suffix}"
-
-        # Update
-        resp = await client.put(
-            f"/api/v1/instances/{inst_id}",
-            headers=auth_header(admin_token),
-            json={"name": f"updated-{suffix}"},
-        )
-        assert resp.status_code == 200
-
-        # Delete
-        resp = await client.delete(f"/api/v1/instances/{inst_id}", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-
-        # Verify deleted
-        resp = await client.get(f"/api/v1/instances/{inst_id}", headers=auth_header(admin_token))
-        assert resp.status_code in (404, 400)
-
-
-class TestWorkflowFlow:
-    """Full workflow lifecycle test."""
-
-    async def test_workflow_crud_flow(self, client: AsyncClient, admin_token: str):
-        import uuid
-        suffix = uuid.uuid4().hex[:6]
-
-        # Create
-        resp = await client.post("/api/v1/workflows", headers=auth_header(admin_token), json={
-            "name": f"crud-wf-{suffix}",
-            "description": "CRUD test",
-            "is_active": True,
-        })
-        assert resp.status_code == 201
-        wf_id = resp.json()["id"]
-
-        # Read
-        resp = await client.get(f"/api/v1/workflows/{wf_id}", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        assert resp.json()["name"] == f"crud-wf-{suffix}"
-
-        # Update
-        resp = await client.put(
-            f"/api/v1/workflows/{wf_id}",
-            headers=auth_header(admin_token),
-            json={"name": f"updated-wf-{suffix}", "description": "Updated"},
-        )
-        assert resp.status_code == 200
-
-        # Delete
-        resp = await client.delete(f"/api/v1/workflows/{wf_id}", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-
-
-class TestDashboardEndpoints:
-    """Dashboard endpoint smoke tests."""
-
-    async def test_dashboard_stats(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/dashboard/stats", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-
-    async def test_my_tickets(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/dashboard/my-tickets", headers=auth_header(admin_token))
-        assert resp.status_code == 200
diff --git a/tests/test_integration_audit_policy.py b/tests/test_integration_audit_policy.py
deleted file mode 100644
index 5258511..0000000
--- a/tests/test_integration_audit_policy.py
+++ /dev/null
@@ -1,20 +0,0 @@
-"""Integration: audit policy — Go/Gin backend.
-Tests SQL audit via ticket audit endpoint."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestAuditPolicy:
-    async def test_audit_sql(self, client: AsyncClient, admin_token: str):
-        resp = await client.post("/api/v1/tickets/audit", headers=auth_header(admin_token), json={
-            "sql_content": "SELECT * FROM users",
-            "db_type": "mysql",
-        })
-        assert resp.status_code == 200
-
-    async def test_audit_sql_no_auth(self, client: AsyncClient):
-        resp = await client.post("/api/v1/tickets/audit", json={
-            "sql": "SELECT 1", "db_type": "mysql",
-        })
-        assert resp.status_code == 401
diff --git a/tests/test_integration_config_notify.py b/tests/test_integration_config_notify.py
deleted file mode 100644
index d6f0d3c..0000000
--- a/tests/test_integration_config_notify.py
+++ /dev/null
@@ -1,35 +0,0 @@
-"""Integration: notification config and test — Go/Gin backend."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestNotificationConfig:
-    async def test_full_notification_flow(self, client: AsyncClient, admin_token: str):
-        import uuid
-        suffix = uuid.uuid4().hex[:6]
-
-        # Create
-        resp = await client.post("/api/v1/notifications", headers=auth_header(admin_token), json={
-            "name": f"test-ch-{suffix}",
-            "channel_type": "webhook_feishu",
-            "config": {"webhook_url": "https://example.com/test"},
-            "is_enabled": False,
-        })
-        assert resp.status_code == 200
-        ch_id = resp.json()["id"]
-
-        # Update
-        resp = await client.put(
-            f"/api/v1/notifications/{ch_id}",
-            headers=auth_header(admin_token),
-            json={"name": f"updated-ch-{suffix}"},
-        )
-        assert resp.status_code == 200
-
-        # Delete
-        resp = await client.delete(
-            f"/api/v1/notifications/{ch_id}",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
diff --git a/tests/test_integration_query.py b/tests/test_integration_query.py
deleted file mode 100644
index f2c5a30..0000000
--- a/tests/test_integration_query.py
+++ /dev/null
@@ -1,23 +0,0 @@
-"""Integration: query execution test — Go/Gin backend."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestQueryExecute:
-    async def test_execute_select(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        resp = await client.post("/api/v1/query/execute", headers=auth_header(admin_token), json={
-            "instance_id": test_mysql_instance["id"],
-            "database": "test",
-            "sql": "SELECT 1 AS val",
-        })
-        assert resp.status_code == 200
-        data = resp.json()
-        # Should have columns and rows
-        assert "columns" in data or "rows" in data or "data" in data
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.post("/api/v1/query/execute", json={
-            "instance_id": 1, "database": "test", "sql": "SELECT 1",
-        })
-        assert resp.status_code == 401
diff --git a/tests/test_integration_ticket_flow.py b/tests/test_integration_ticket_flow.py
deleted file mode 100644
index 3cc4cf3..0000000
--- a/tests/test_integration_ticket_flow.py
+++ /dev/null
@@ -1,35 +0,0 @@
-"""Integration: ticket flow test — Go/Gin backend."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestTicketFlow:
-    async def test_create_and_get_ticket(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        import uuid
-        suffix = uuid.uuid4().hex[:6]
-        # Create
-        resp = await client.post("/api/v1/tickets", headers=auth_header(admin_token), json={
-            "title": f"flow-test-{suffix}",
-            "db_type": "mysql",
-            "instance_id": test_mysql_instance["id"],
-            "target_database": "test",
-            "sql_content": "SELECT 1",
-            "reason": "Integration test",
-        })
-        assert resp.status_code == 201
-        ticket_id = resp.json()["id"]
-
-        # Get
-        resp = await client.get(f"/api/v1/tickets/{ticket_id}", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        assert resp.json()["id"] == ticket_id
-
-    async def test_ticket_timeline(self, client: AsyncClient, admin_token: str):
-        lst = await client.get("/api/v1/tickets", headers=auth_header(admin_token))
-        items = lst.json().get("items", [])
-        if not items:
-            pytest.skip("No tickets")
-        tid = items[0]["id"]
-        resp = await client.get(f"/api/v1/tickets/{tid}/timeline", headers=auth_header(admin_token))
-        assert resp.status_code == 200
diff --git a/tests/test_ldap_config.py b/tests/test_ldap_config.py
deleted file mode 100644
index 0caf5cc..0000000
--- a/tests/test_ldap_config.py
+++ /dev/null
@@ -1,32 +0,0 @@
-"""LDAP config tests — Go/Gin backend (admin only)."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestLDAPConfig:
-    async def test_get_config(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/ldap-config", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        assert "ldap_host" in data
-        assert "ldap_port" in data
-
-    async def test_update_config(self, client: AsyncClient, admin_token: str):
-        resp = await client.put("/api/v1/ldap-config", headers=auth_header(admin_token), json={
-            "ldap_host": "8.217.205.92",
-            "ldap_port": 389,
-            "admin_dn": "cn=admin,dc=opsx,dc=vip",
-            "admin_password": "admin123",
-            "base_dn": "dc=opsx,dc=vip",
-            "use_ldaps": False,
-        })
-        assert resp.status_code == 200
-
-    async def test_developer_forbidden(self, client: AsyncClient, developer_token: str):
-        resp = await client.get("/api/v1/ldap-config", headers=auth_header(developer_token))
-        assert resp.status_code == 403
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/ldap-config")
-        assert resp.status_code == 401
diff --git a/tests/test_notification.py b/tests/test_notification.py
deleted file mode 100644
index 0e4d5be..0000000
--- a/tests/test_notification.py
+++ /dev/null
@@ -1,38 +0,0 @@
-"""Notification channel tests — Go/Gin backend."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestListNotifications:
-    async def test_list(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/notifications", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        assert isinstance(data, list)
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/notifications")
-        assert resp.status_code == 401
-
-
-class TestCreateNotification:
-    async def test_create_feishu(self, client: AsyncClient, admin_token: str):
-        import uuid
-        resp = await client.post("/api/v1/notifications", headers=auth_header(admin_token), json={
-            "name": f"test-feishu-{uuid.uuid4().hex[:6]}",
-            "channel_type": "webhook_feishu",
-            "config": {"webhook_url": "https://example.com/feishu"},
-            "is_enabled": False,
-        })
-        assert resp.status_code == 200
-
-    async def test_create_dingtalk(self, client: AsyncClient, admin_token: str):
-        import uuid
-        resp = await client.post("/api/v1/notifications", headers=auth_header(admin_token), json={
-            "name": f"test-ding-{uuid.uuid4().hex[:6]}",
-            "channel_type": "dingtalk",
-            "config": {"webhook_url": "https://example.com/dingtalk"},
-            "is_enabled": False,
-        })
-        assert resp.status_code == 200
diff --git a/tests/test_permissions.py b/tests/test_permissions.py
new file mode 100644
index 0000000..662653a
--- /dev/null
+++ b/tests/test_permissions.py
@@ -0,0 +1,63 @@
+"""Permission tests — verify role-based access control"""
+import pytest
+from httpx import AsyncClient
+def auth_header(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
+
+
+class TestTicketPermissions:
+    """验证各角色对工单操作的权限"""
+
+    async def test_developer_can_see_approver_tickets(self, client: AsyncClient,
+                                                       developer_token, approver1_token):
+        """test123(developer) 作为审批人能看到待审工单"""
+        # test123 查看工单列表
+        resp = await client.get("/api/v1/tickets", headers=auth_header(approver1_token))
+        assert resp.status_code == 200
+        data = resp.json()
+        # 应能看到工单（自己的 + 作为审批人的）
+        assert "items" in data
+
+    async def test_only_admin_dba_can_execute(self, client: AsyncClient,
+                                               developer_token, admin_token, mysql_instance):
+        """developer 不能执行工单"""
+        # 创建工单
+        resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
+            "title": "权限测试-执行",
+            "db_type": "mysql",
+            "instance_id": mysql_instance["id"],
+            "target_database": "opsxsql_test",
+            "sql_content": "CREATE TABLE IF NOT EXISTS _perm_test (id INT)",
+            "workflow_id": 4,
+        })
+        assert resp.status_code in (200, 201)
+        tid = resp.json()["id"]
+
+        # developer 尝试执行 → 应被拒绝
+        resp = await client.post(f"/api/v1/tickets/{tid}/execute", headers=auth_header(developer_token), json={})
+        assert resp.status_code in (403, 400)
+
+
+class TestWorkflowPermissions:
+    """审批流程管理权限"""
+
+    async def test_admin_can_list_workflows(self, client: AsyncClient, admin_token):
+        resp = await client.get("/api/v1/workflows", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+
+    async def test_developer_can_list_workflows(self, client: AsyncClient, developer_token):
+        """developer 应该能看到 workflow 列表（前端提交时需要选 workflow）"""
+        resp = await client.get("/api/v1/workflows", headers=auth_header(developer_token))
+        assert resp.status_code == 200
+
+
+class TestDashboardPermissions:
+    """Dashboard 统计数据权限"""
+
+    async def test_admin_statistics(self, client: AsyncClient, admin_token):
+        resp = await client.get("/api/v1/tickets/statistics", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+
+    async def test_developer_statistics(self, client: AsyncClient, developer_token):
+        resp = await client.get("/api/v1/tickets/statistics", headers=auth_header(developer_token))
+        assert resp.status_code == 200
diff --git a/tests/test_redis.py b/tests/test_redis.py
deleted file mode 100644
index d316c0e..0000000
--- a/tests/test_redis.py
+++ /dev/null
@@ -1,38 +0,0 @@
-"""Redis instance endpoint tests — Go/Gin backend."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestRedisInfo:
-    async def test_redis_info(self, client: AsyncClient, admin_token: str, test_redis_instance):
-        resp = await client.get(
-            f"/api/v1/redis/{test_redis_instance['id']}/info",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-
-    async def test_redis_dbsize(self, client: AsyncClient, admin_token: str, test_redis_instance):
-        resp = await client.get(
-            f"/api/v1/redis/{test_redis_instance['id']}/dbsize",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-
-    async def test_redis_slowlog(self, client: AsyncClient, admin_token: str, test_redis_instance):
-        resp = await client.get(
-            f"/api/v1/redis/{test_redis_instance['id']}/slowlog",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-
-    async def test_redis_clients(self, client: AsyncClient, admin_token: str, test_redis_instance):
-        resp = await client.get(
-            f"/api/v1/redis/{test_redis_instance['id']}/clients",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/redis/2/info")
-        assert resp.status_code == 401
diff --git a/tests/test_schema_drift.py b/tests/test_schema_drift.py
deleted file mode 100644
index 6084931..0000000
--- a/tests/test_schema_drift.py
+++ /dev/null
@@ -1,19 +0,0 @@
-"""Schema drift tests — Go backend does not have schema-drift endpoints yet."""
-import pytest
-
-pytestmark = pytest.mark.skip(reason="Schema drift endpoints not implemented in Go backend")
-
-
-class TestSchemaDriftCapture:
-    async def test_capture_success(self, client):
-        pass
-
-
-class TestSchemaDriftCheck:
-    async def test_check_success(self, client):
-        pass
-
-
-class TestSchemaDriftHistory:
-    async def test_history(self, client):
-        pass
diff --git a/tests/test_settings.py b/tests/test_settings.py
deleted file mode 100644
index 105b8d3..0000000
--- a/tests/test_settings.py
+++ /dev/null
@@ -1,23 +0,0 @@
-"""Settings endpoint tests — Go/Gin backend."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestListSettings:
-    async def test_list_settings(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/settings", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        assert isinstance(data, dict)
-        # Should have grouped settings
-        assert len(data) > 0
-
-
-class TestUpdateSettings:
-    async def test_update_setting(self, client: AsyncClient, admin_token: str):
-        # Go backend expects array of {key, value} objects
-        resp = await client.put("/api/v1/settings", headers=auth_header(admin_token), json=[
-            {"key": "data_masking_enabled", "value": "true"},
-        ])
-        assert resp.status_code == 200
diff --git a/tests/test_tickets.py b/tests/test_tickets.py
index c95a491..4ba4341 100644
--- a/tests/test_tickets.py
+++ b/tests/test_tickets.py
@@ -1,92 +1,83 @@
-"""Ticket CRUD + workflow tests — Go/Gin backend."""
+"""Ticket API tests — CRUD + status filter"""
 import pytest
 from httpx import AsyncClient
-from tests.conftest import auth_header
+def auth_header(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
 
 
 class TestListTickets:
-    async def test_list_tickets(self, client: AsyncClient, admin_token: str):
+    async def test_admin_sees_all(self, client: AsyncClient, admin_token):
         resp = await client.get("/api/v1/tickets", headers=auth_header(admin_token))
         assert resp.status_code == 200
         data = resp.json()
         assert "items" in data
-        assert isinstance(data["items"], list)
-
-    async def test_list_with_pagination(self, client: AsyncClient, admin_token: str):
-        resp = await client.get(
-            "/api/v1/tickets?page=1&page_size=5",
-            headers=auth_header(admin_token),
-        )
-        assert resp.status_code == 200
-        data = resp.json()
-        assert "items" in data
         assert "total" in data
 
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/tickets")
-        assert resp.status_code == 401
+    async def test_developer_sees_own_and_approver(self, client: AsyncClient, developer_token):
+        """developer 能看到自己提交的 + 自己作为审批人的工单"""
+        resp = await client.get("/api/v1/tickets", headers=auth_header(developer_token))
+        assert resp.status_code == 200
+
+    async def test_filter_by_status(self, client: AsyncClient, admin_token):
+        resp = await client.get("/api/v1/tickets?status=pending", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+        items = resp.json().get("items", [])
+        for t in items:
+            assert t["status"] == "pending"
+
+    async def test_pagination(self, client: AsyncClient, admin_token):
+        resp = await client.get("/api/v1/tickets?page=1&page_size=5", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["page"] == 1
+        assert data["page_size"] == 5
 
 
 class TestCreateTicket:
-    async def test_create_ticket(self, client: AsyncClient, admin_token: str, test_mysql_instance):
-        resp = await client.post("/api/v1/tickets", headers=auth_header(admin_token), json={
-            "title": "Test ticket",
+    async def test_create_mysql_ticket(self, client: AsyncClient, developer_token, mysql_instance):
+        resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
+            "title": "测试工单-CREATE TABLE",
             "db_type": "mysql",
-            "instance_id": test_mysql_instance["id"],
-            "target_database": "test",
-            "sql_content": "SELECT 1",
-            "reason": "Test",
+            "instance_id": mysql_instance["id"],
+            "target_database": "opsxsql_test",
+            "sql_content": "CREATE TABLE IF NOT EXISTS _test_tbl (id INT)",
+            "workflow_id": 4,
         })
-        assert resp.status_code == 201
+        assert resp.status_code in (200, 201)
         data = resp.json()
-        assert "id" in data
+        assert data["status"] == "pending"
+        assert "ticket_no" in data
+        return data
+
+    async def test_reject_readonly_sql(self, client: AsyncClient, developer_token, mysql_instance):
+        """SELECT/SHOW 等只读 SQL 应被拦截"""
+        resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
+            "title": "只读SQL",
+            "db_type": "mysql",
+            "instance_id": mysql_instance["id"],
+            "target_database": "opsxsql_test",
+            "sql_content": "SELECT * FROM orders",
+            "workflow_id": 4,
+        })
+        assert resp.status_code == 400
+
+    async def test_auditor_cannot_create(self, client: AsyncClient):
+        """auditor 角色不能提交工单"""
+        
+        resp = await client.post("/api/v1/auth/login", json={
+            "username": "wangwu", "password": "admin123",
+        })
+        if resp.status_code != 200:
+            pytest.skip("auditor user not available")
+        token = resp.json()["access_token"]
+        resp2 = await client.post("/api/v1/tickets", headers=auth_header(token), json={
+            "title": "auditor提交",
+            "db_type": "mysql", "instance_id": 5,
+            "target_database": "opsxsql_test",
+            "sql_content": "CREATE TABLE _t(x INT)",
+        })
+        assert resp2.status_code in (403, 401)
 
     async def test_no_auth(self, client: AsyncClient):
-        resp = await client.post("/api/v1/tickets", json={
-            "title": "No auth", "db_type": "mysql",
-            "instance_id": 1, "target_database": "test",
-            "sql_content": "SELECT 1", "reason": "test",
-        })
+        resp = await client.post("/api/v1/tickets", json={"title": "x"})
         assert resp.status_code == 401
-
-
-class TestGetTicket:
-    async def test_get_ticket(self, client: AsyncClient, admin_token: str):
-        # List first
-        lst = await client.get("/api/v1/tickets", headers=auth_header(admin_token))
-        items = lst.json().get("items", [])
-        if not items:
-            pytest.skip("No tickets in database")
-        tid = items[0]["id"]
-        resp = await client.get(f"/api/v1/tickets/{tid}", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        assert resp.json()["id"] == tid
-
-
-class TestReviewTicket:
-    async def test_approve_ticket(self, client: AsyncClient, admin_token: str):
-        # Find a pending ticket
-        lst = await client.get(
-            "/api/v1/tickets?status=pending&page=1&page_size=5",
-            headers=auth_header(admin_token),
-        )
-        items = lst.json().get("items", [])
-        pending = [t for t in items if t.get("status") == "pending"]
-        if not pending:
-            pytest.skip("No pending tickets to approve")
-        tid = pending[0]["id"]
-        resp = await client.post(
-            f"/api/v1/tickets/{tid}/approve",
-            headers=auth_header(admin_token),
-            json={"action": "approve", "comment": "Test approve"},
-        )
-        assert resp.status_code in (200, 400)  # 400 if already approved
-
-    async def test_developer_cannot_approve(self, client: AsyncClient, developer_token: str):
-        resp = await client.post(
-            "/api/v1/tickets/1/approve",
-            headers=auth_header(developer_token),
-            json={"action": "approve"},
-        )
-        # May be 403 or 400 depending on ticket state
-        assert resp.status_code in (200, 400, 403, 404)
diff --git a/tests/test_users.py b/tests/test_users.py
deleted file mode 100644
index ba95fa0..0000000
--- a/tests/test_users.py
+++ /dev/null
@@ -1,76 +0,0 @@
-"""User CRUD tests — Go/Gin backend (admin only)."""
-import pytest
-from httpx import AsyncClient
-from tests.conftest import auth_header
-
-
-class TestListUsers:
-    async def test_admin_list_users(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/users", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        assert isinstance(data, list)
-        assert len(data) > 0
-        assert "id" in data[0]
-        assert "username" in data[0]
-        assert "role" in data[0]
-
-    async def test_developer_forbidden(self, client: AsyncClient, developer_token: str):
-        resp = await client.get("/api/v1/users", headers=auth_header(developer_token))
-        assert resp.status_code == 403
-
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/users")
-        assert resp.status_code == 401
-
-
-class TestCreateUser:
-    async def test_create_user_success(self, client: AsyncClient, admin_token: str):
-        import uuid
-        uname = f"test_{uuid.uuid4().hex[:8]}"
-        resp = await client.post("/api/v1/users", headers=auth_header(admin_token), json={
-            "username": uname,
-            "password": "test123456",
-            "role": "developer",
-            "display_name": "Test User",
-        })
-        assert resp.status_code == 201
-        data = resp.json()
-        assert data["username"] == uname
-        assert data["role"] == "developer"
-
-    async def test_create_duplicate_username(self, client: AsyncClient, admin_token: str):
-        resp = await client.post("/api/v1/users", headers=auth_header(admin_token), json={
-            "username": "admin",
-            "password": "test123456",
-            "role": "developer",
-            "display_name": "Duplicate",
-        })
-        # Should fail with 400 or 409
-        assert resp.status_code in (400, 409)
-
-    async def test_developer_cannot_create(self, client: AsyncClient, developer_token: str):
-        resp = await client.post("/api/v1/users", headers=auth_header(developer_token), json={
-            "username": "nope", "password": "test123456", "role": "developer",
-        })
-        assert resp.status_code == 403
-
-
-class TestUpdateUser:
-    async def test_update_display_name(self, client: AsyncClient, admin_token: str):
-        # Get admin user id
-        me = await client.get("/api/v1/auth/me", headers=auth_header(admin_token))
-        uid = me.json()["id"]
-        resp = await client.put(
-            f"/api/v1/users/{uid}",
-            headers=auth_header(admin_token),
-            json={"display_name": "Updated Name"},
-        )
-        assert resp.status_code == 200
-
-
-class TestDeleteUser:
-    async def test_delete_nonexistent(self, client: AsyncClient, admin_token: str):
-        resp = await client.delete("/api/v1/users/99999", headers=auth_header(admin_token))
-        # May return 404 or 200 with message
-        assert resp.status_code in (200, 404)
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index e2cf8d4..ad72f45 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -1,78 +1,105 @@
-"""Workflow CRUD tests — Go/Gin backend."""
+"""Full E2E ticket workflow test — submit→approve→approve→execute"""
 import pytest
 from httpx import AsyncClient
-from tests.conftest import auth_header
+def auth_header(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
 
 
-class TestListWorkflows:
-    async def test_list_workflows(self, client: AsyncClient, admin_token: str):
-        resp = await client.get("/api/v1/workflows", headers=auth_header(admin_token))
-        assert resp.status_code == 200
-        data = resp.json()
-        assert isinstance(data, list)
+class TestE2EWorkflow:
+    """完整审批流程: testtest提交 → test123初审 → test234复审 → admin执行"""
 
-    async def test_no_auth(self, client: AsyncClient):
-        resp = await client.get("/api/v1/workflows")
-        assert resp.status_code == 401
-
-
-class TestCreateWorkflow:
-    async def test_create_success(self, client: AsyncClient, admin_token: str):
-        import uuid
-        resp = await client.post("/api/v1/workflows", headers=auth_header(admin_token), json={
-            "name": f"wf-test-{uuid.uuid4().hex[:6]}",
-            "description": "Test workflow",
-            "is_active": True,
+    async def test_full_approval_flow(self, client: AsyncClient,
+                                       developer_token, approver1_token,
+                                       approver2_token, admin_token,
+                                       mysql_instance):
+        # Step 1: developer 提交工单
+        resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
+            "title": "E2E测试-完整审批流",
+            "db_type": "mysql",
+            "instance_id": mysql_instance["id"],
+            "target_database": "opsxsql_test",
+            "sql_content": "CREATE TABLE IF NOT EXISTS _e2e_test (id INT PRIMARY KEY, val VARCHAR(50))",
+            "workflow_id": 4,
         })
-        assert resp.status_code == 201
-        data = resp.json()
-        assert "id" in data
-        assert "message" in data
+        assert resp.status_code in (200, 201)
+        ticket = resp.json()
+        tid = ticket["id"]
+        assert ticket["status"] == "pending"
 
-    async def test_create_with_category(self, client: AsyncClient, admin_token: str):
-        import uuid
-        resp = await client.post("/api/v1/workflows", headers=auth_header(admin_token), json={
-            "name": f"wf-cat-{uuid.uuid4().hex[:6]}",
-            "description": "With category",
-            "category": "ticket",
-            "is_active": True,
+        # Step 1.5: 验证审批记录已生成
+        resp = await client.get(f"/api/v1/tickets/{tid}/approvals", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+        approvals = resp.json()
+        assert len(approvals) >= 2  # 至少有初审+复审+执行
+
+        # Step 2: test123 初审通过
+        resp = await client.post(f"/api/v1/tickets/{tid}/review", headers=auth_header(approver1_token), json={
+            "action": "approve", "comment": "初审通过",
         })
-        assert resp.status_code == 201
-
-
-class TestGetWorkflow:
-    async def test_get_existing(self, client: AsyncClient, admin_token: str):
-        # List first
-        lst = await client.get("/api/v1/workflows", headers=auth_header(admin_token))
-        wfs = lst.json()
-        if not wfs:
-            pytest.skip("No workflows")
-        wf_id = wfs[0]["id"]
-        resp = await client.get(f"/api/v1/workflows/{wf_id}", headers=auth_header(admin_token))
         assert resp.status_code == 200
-        data = resp.json()
-        assert data["id"] == wf_id
-        assert "name" in data
-        assert "steps" in data
-        assert "is_active" in data
+        assert resp.json()["ticket"]["status"] == "reviewing"
 
-
-class TestUpdateWorkflow:
-    async def test_update_name(self, client: AsyncClient, admin_token: str):
-        lst = await client.get("/api/v1/workflows", headers=auth_header(admin_token))
-        wfs = lst.json()
-        if not wfs:
-            pytest.skip("No workflows")
-        wf_id = wfs[0]["id"]
-        resp = await client.put(
-            f"/api/v1/workflows/{wf_id}",
-            headers=auth_header(admin_token),
-            json={"name": "updated-name", "description": "updated desc"},
-        )
+        # Step 3: test234 复审通过
+        resp = await client.post(f"/api/v1/tickets/{tid}/review", headers=auth_header(approver2_token), json={
+            "action": "approve", "comment": "复审通过",
+        })
         assert resp.status_code == 200
+        assert resp.json()["ticket"]["status"] == "approved"
 
+        # Step 4: admin 执行
+        resp = await client.post(f"/api/v1/tickets/{tid}/execute", headers=auth_header(admin_token), json={})
+        assert resp.status_code in (200, 201)
+        ticket = resp.json()["ticket"]
+        assert ticket["status"] == "success"
 
-class TestDeleteWorkflow:
-    async def test_delete_not_found(self, client: AsyncClient, admin_token: str):
-        resp = await client.delete("/api/v1/workflows/99999", headers=auth_header(admin_token))
-        assert resp.status_code in (200, 404)
+        # Step 5: 验证执行结果是结构化 JSON
+        import json
+        exec_result = json.loads(ticket["execute_result"])
+        assert exec_result["status"] == "success"
+        assert "items" in exec_result
+        assert len(exec_result["items"]) >= 1
+        item = exec_result["items"][0]
+        assert item["status"] == "success"
+        assert "duration_ms" in item
+
+    async def test_reject_flow(self, client: AsyncClient,
+                                developer_token, approver1_token, mysql_instance):
+        """审批拒绝流程"""
+        resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
+            "title": "E2E测试-拒绝",
+            "db_type": "mysql",
+            "instance_id": mysql_instance["id"],
+            "target_database": "opsxsql_test",
+            "sql_content": "CREATE TABLE IF NOT EXISTS _reject_test (id INT)",
+            "workflow_id": 4,
+        })
+        assert resp.status_code in (200, 201)
+        tid = resp.json()["id"]
+
+        # 初审拒绝
+        resp = await client.post(f"/api/v1/tickets/{tid}/review", headers=auth_header(approver1_token), json={
+            "action": "reject", "comment": "不通过",
+        })
+        assert resp.status_code == 200
+        assert resp.json()["ticket"]["status"] == "rejected"
+
+    async def test_cancel_flow(self, client: AsyncClient,
+                                developer_token, mysql_instance):
+        """提交人撤销工单"""
+        resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
+            "title": "E2E测试-撤销",
+            "db_type": "mysql",
+            "instance_id": mysql_instance["id"],
+            "target_database": "opsxsql_test",
+            "sql_content": "CREATE TABLE IF NOT EXISTS _cancel_test (id INT)",
+            "workflow_id": 4,
+        })
+        assert resp.status_code in (200, 201)
+        tid = resp.json()["id"]
+
+        # 提交人撤销
+        resp = await client.post(f"/api/v1/tickets/{tid}/cancel", headers=auth_header(developer_token), json={
+            "reason": "不想执行了",
+        })
+        assert resp.status_code == 200
+        assert resp.json()["ticket"]["status"] == "cancelled"