import importlib import os import re import sys import types import uuid import pytest from fastapi import HTTPException from fastapi.testclient import TestClient from sqlalchemy import delete, select, text, update from sqlalchemy.orm import Session os.environ["JWT_SECRET"] = "test-secret-must-be-32-characters!" os.environ["API_PAYLOAD_ENCRYPT"] = "false" PASSWORD_RULE = "密码至少 8 位，且同时包含字母和数字" LOGIN_FAILURE = "用户名或密码错误" KEEP_SELF = "不能停用或注销当前登录用户" KEEP_FULL_ADMIN = "至少保留一名拥有全部系统菜单的正常用户" def _install_stub(module_name: str, **attrs: object) -> None: try: module = importlib.import_module(module_name) except Exception: sys.modules.pop(module_name, None) module = types.ModuleType(module_name) sys.modules[module_name] = module for key, value in attrs.items(): setattr(module, key, value) return for key, value in attrs.items(): if not hasattr(module, key): setattr(module, key, value) def _encrypt_text(value: str) -> str: return f"enc:{value}" def _consume_captcha(captcha_id: str, code: str) -> bool: return False def _is_currently_locked(user) -> bool: return False def _record_failure(db, user) -> None: user.failed_login_count = int(user.failed_login_count or 0) + 1 def _clear_failures(db, user) -> None: user.failed_login_count = 0 user.locked_until = None _install_stub("app.core.field_crypto", encrypt_text=_encrypt_text) _install_stub("app.services.captcha", consume_captcha=_consume_captcha) _install_stub( "app.services.login_guard", is_currently_locked=_is_currently_locked, record_failure=_record_failure, clear_failures=_clear_failures, ) from app.api.admin.rbac import public_router, router from app.commands.create_superuser import seed_superuser from app.core.config import get_settings from app.core.masking import mask_mobile from app.core.security import create_access_token from app.db import Base, get_db, get_engine from app.main import create_app from app.models.rbac import AdminResource, AdminRole, AdminRoleResource, AdminUser, AdminUserRole, utcnow from app.services.rbac import count_full_admins, preserve_full_admin get_settings.cache_clear() def _safe_error(exc: Exception) -> str: raw = str(exc).splitlines()[0] if str(exc) else "" raw = re.sub(r"://\S+", "://***", raw) raw = re.sub(r"password=\S+", "password=***", raw, flags=re.I) return f"{exc.__class__.__name__}: {raw[:180]}" def _database_url() -> str: get_settings.cache_clear() return get_settings().database_url.strip() def _suffix(prefix: str) -> str: return f"{prefix}_{uuid.uuid4().hex[:10]}" def _staff_headers(subject: str = "1") -> dict[str, str]: token = create_access_token(subject, "staff") return {"Authorization": f"Bearer {token}"} def _mount(app) -> None: paths = {getattr(route, "path", "") for route in app.routes} if "/api/admin/users" not in paths: app.include_router(public_router, prefix="/api/admin") app.include_router(router, prefix="/api/admin") def _committed(username: str) -> bool: other = Session(bind=get_engine()) try: found = other.scalar(select(AdminUser.id).where(AdminUser.username == username)) return found is not None finally: other.close() def _delete_committed_user(username: str) -> None: other = Session(bind=get_engine()) try: row = other.scalar(select(AdminUser).where(AdminUser.username == username)) if row is None: return other.execute(delete(AdminUserRole).where(AdminUserRole.user_id == row.id)) other.delete(row) other.commit() finally: other.close() def _assert_isolated(connection, username: str) -> None: probe = Session( bind=connection, autoflush=False, expire_on_commit=False, join_transaction_mode="create_savepoint", ) now = utcnow() probe.add( AdminUser( username=username, password_hash="probe", display_name="probe", mobile_mask="", status="active", failed_login_count=0, created_at=now, updated_at=now, ) ) probe.commit() probe.close() if _committed(username): _delete_committed_user(username) pytest.fail("测试事务没有隔离，已停止，避免改动已有数据") @pytest.fixture def client(): if not _database_url(): pytest.skip("DATABASE_URL 未配置") os.environ["API_PAYLOAD_ENCRYPT"] = "false" get_settings.cache_clear() try: engine = get_engine() with engine.connect() as probe: probe.execute(text("SELECT 1")) Base.metadata.create_all(bind=engine) except Exception as exc: pytest.skip(f"数据库不可用：{_safe_error(exc)}") connection = engine.connect() transaction = connection.begin() probe_name = _suffix("probe") try: _assert_isolated(connection, probe_name) app = create_app() _mount(app) app.state.connection = connection def override_db(): db = Session( bind=connection, autoflush=False, expire_on_commit=False, join_transaction_mode="create_savepoint", ) try: yield db finally: db.close() app.dependency_overrides[get_db] = override_db with TestClient(app) as test_client: yield test_client finally: try: transaction.rollback() except Exception: pass connection.close() def _walk(nodes: list[dict]): for node in nodes: yield node yield from _walk(node.get("children") or []) def _no_password(payload) -> None: if isinstance(payload, dict): for key, value in payload.items(): assert "password" not in key.lower() assert key not in {"mobileCipher", "mobile_cipher", "password_hash"} _no_password(value) elif isinstance(payload, list): for item in payload: _no_password(item) def test_users_requires_staff_token(client: TestClient): missing = client.get("/api/admin/users") assert missing.status_code == 401 member = create_access_token("member-1", "member") rejected = client.get("/api/admin/users", headers={"Authorization": f"Bearer {member}"}) assert rejected.status_code == 403 def test_weak_password_rejected(client: TestClient): headers = _staff_headers() username = _suffix("weak") response = client.post( "/api/admin/users", headers=headers, json={ "username": username, "password": "abcdefgh", "displayName": "弱密码", "roleIds": [], }, ) assert response.status_code == 400 assert response.json()["detail"] == PASSWORD_RULE digits = client.post( "/api/admin/users", headers=headers, json={ "username": username, "password": "12345678", "displayName": "弱密码", "roleIds": [], }, ) assert digits.status_code == 400 assert digits.json()["detail"] == PASSWORD_RULE def test_cannot_cancel_self(client: TestClient): headers = _staff_headers("0") username = _suffix("self") created = client.post( "/api/admin/users", headers=headers, json={ "username": username, "password": "Abcdef12", "displayName": "本人", "roleIds": [], }, ) assert created.status_code == 200 user_id = created.json()["id"] own = _staff_headers(str(user_id)) cancelled = client.put( f"/api/admin/users/{user_id}", headers=own, json={"displayName": "本人", "status": "cancelled", "roleIds": []}, ) assert cancelled.status_code == 400 assert cancelled.json()["detail"] == KEEP_SELF listed = client.get("/api/admin/users", headers=headers, params={"username": username}) assert listed.status_code == 200 record = next(item for item in listed.json()["records"] if item["username"] == username) assert record["status"] == "active" def test_user_list_masks_mobile_and_hides_password(client: TestClient, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr("app.services.rbac.encrypt_text", lambda value: "cipher-text") headers = _staff_headers() username = _suffix("mobile") mobile = "13812348000" password = "Abcdef12" created = client.post( "/api/admin/users", headers=headers, json={ "username": username, "password": password, "displayName": "脱敏", "mobile": mobile, "roleIds": [], }, ) assert created.status_code == 200 created_body = created.json() assert created_body["mobileMask"] == mask_mobile(mobile) == "138****8000" _no_password(created_body) assert mobile not in created.text assert password not in created.text assert "cipher-text" not in created.text listed = client.get("/api/admin/users", headers=headers, params={"username": username}) assert listed.status_code == 200 body = listed.json() record = next(item for item in body["records"] if item["username"] == username) assert record["mobileMask"] == "138****8000" assert set(record) >= { "id", "username", "displayName", "mobileMask", "status", "roleIds", "createdAt", "updatedAt", } _no_password(body) assert mobile not in listed.text assert password not in listed.text assert "cipher-text" not in listed.text assert not _committed(username) def test_login_hides_reason(client: TestClient, monkeypatch: pytest.MonkeyPatch): failures: list[int] = [] clears: list[int] = [] def consume(captcha_id: str, code: str) -> bool: return code == "ok" def record(db, user) -> None: failures.append(user.id) def clear(db, user) -> None: clears.append(user.id) user.failed_login_count = 0 user.locked_until = None locked = {"value": False} monkeypatch.setattr("app.services.rbac.consume_captcha", consume) monkeypatch.setattr("app.services.rbac.record_failure", record) monkeypatch.setattr("app.services.rbac.clear_failures", clear) monkeypatch.setattr("app.services.rbac.is_currently_locked", lambda user: locked["value"]) headers = _staff_headers() username = _suffix("login") password = "Abcdef12" created = client.post( "/api/admin/users", headers=headers, json={"username": username, "password": password, "displayName": "登录", "roleIds": []}, ) assert created.status_code == 200 missing_user = client.post( "/api/admin/auth/login", json={"username": _suffix("none"), "password": password, "captchaId": "c1", "captchaCode": "ok"}, ) assert missing_user.status_code == 401 assert missing_user.json()["detail"] == LOGIN_FAILURE assert failures == [] bad_captcha = client.post( "/api/admin/auth/login", json={"username": username, "password": password, "captchaId": "c1", "captchaCode": "bad"}, ) assert bad_captcha.status_code == 401 assert bad_captcha.json()["detail"] == LOGIN_FAILURE assert len(failures) == 1 bad_password = client.post( "/api/admin/auth/login", json={"username": username, "password": "Wrongpass1", "captchaId": "c1", "captchaCode": "ok"}, ) assert bad_password.status_code == 401 assert bad_password.json()["detail"] == LOGIN_FAILURE assert len(failures) == 2 locked["value"] = True locked_response = client.post( "/api/admin/auth/login", json={"username": username, "password": password, "captchaId": "c1", "captchaCode": "ok"}, ) assert locked_response.status_code == 401 assert locked_response.json()["detail"] == LOGIN_FAILURE assert len(failures) == 2 locked["value"] = False success = client.post( "/api/admin/auth/login", json={"username": username, "password": password, "captchaId": "c1", "captchaCode": "ok"}, ) assert success.status_code == 200 payload = success.json() assert payload["user"]["username"] == username assert payload["user"]["displayName"] == "登录" assert isinstance(payload["menus"], list) assert isinstance(payload["permissions"], list) assert payload["token"] _no_password(payload) assert clears == [created.json()["id"]] def test_login_menu_excludes_buttons(client: TestClient, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr("app.services.rbac.consume_captcha", lambda captcha_id, code: True) monkeypatch.setattr("app.services.rbac.is_currently_locked", lambda user: False) monkeypatch.setattr("app.services.rbac.record_failure", lambda db, user: None) monkeypatch.setattr("app.services.rbac.clear_failures", lambda db, user: None) headers = _staff_headers() mark = uuid.uuid4().hex[:8] directory = client.post( "/api/admin/resources", headers=headers, json={ "type": "directory", "name": f"Dir{mark}", "title": "目录", "path": f"/t-{mark}", "sort": 2, "visible": True, "status": "active", }, ) assert directory.status_code == 200 directory_id = directory.json()["id"] later = client.post( "/api/admin/resources", headers=headers, json={ "parentId": directory_id, "type": "menu", "name": f"Later{mark}", "title": "后", "path": f"/t-{mark}/later", "component": f"/t-{mark}/later", "sort": 2, "status": "active", }, ) earlier = client.post( "/api/admin/resources", headers=headers, json={ "parentId": directory_id, "type": "menu", "name": f"Early{mark}", "title": "先", "path": f"/t-{mark}/early", "component": f"/t-{mark}/early", "sort": 1, "status": "active", }, ) hidden = client.post( "/api/admin/resources", headers=headers, json={ "parentId": directory_id, "type": "menu", "name": f"Off{mark}", "title": "停用页", "path": f"/t-{mark}/off", "component": f"/t-{mark}/off", "sort": 3, "status": "disabled", }, ) button_z = client.post( "/api/admin/resources", headers=headers, json={ "parentId": earlier.json()["id"], "type": "button", "name": f"b:z:{mark}", "title": "乙", "permission": f"b:z:{mark}", "visible": False, "status": "active", }, ) button_a = client.post( "/api/admin/resources", headers=headers, json={ "parentId": earlier.json()["id"], "type": "button", "name": f"b:a:{mark}", "title": "甲", "permission": f"b:a:{mark}", "visible": False, "status": "active", }, ) assert later.status_code == earlier.status_code == hidden.status_code == 200 assert button_z.status_code == button_a.status_code == 200 role = client.post( "/api/admin/roles", headers=headers, json={"code": f"R_{mark}", "name": "菜单角色", "description": ""}, ) assert role.status_code == 200 role_id = role.json()["id"] disabled_role = client.post( "/api/admin/roles", headers=headers, json={"code": f"R_OFF_{mark}", "name": "停用角色"}, ) assert disabled_role.status_code == 200 resource_ids = [ directory_id, later.json()["id"], earlier.json()["id"], button_a.json()["id"], button_z.json()["id"], ] replaced = client.put( f"/api/admin/roles/{role_id}/resources", headers=headers, json={"resourceIds": resource_ids}, ) assert replaced.status_code == 200 client.put( f"/api/admin/roles/{disabled_role.json()['id']}", headers=headers, json={"name": "停用角色", "description": "", "status": "disabled"}, ) extra_button = client.post( "/api/admin/resources", headers=headers, json={ "parentId": earlier.json()["id"], "type": "button", "name": f"b:off:{mark}", "title": "停用角色按钮", "permission": f"b:off:{mark}", "visible": False, "status": "active", }, ) client.put( f"/api/admin/roles/{disabled_role.json()['id']}/resources", headers=headers, json={"resourceIds": [extra_button.json()["id"]]}, ) username = _suffix("menu") created = client.post( "/api/admin/users", headers=headers, json={ "username": username, "password": "Abcdef12", "displayName": "菜单", "roleIds": [role_id, disabled_role.json()["id"]], }, ) assert created.status_code == 200 logged = client.post( "/api/admin/auth/login", json={"username": username, "password": "Abcdef12", "captchaId": "c", "captchaCode": "1"}, ) assert logged.status_code == 200 body = logged.json() nodes = list(_walk(body["menus"])) assert [node["type"] for node in nodes if node["type"] == "button"] == [] titles = [node["title"] for node in nodes] assert "停用页" not in titles children = body["menus"][0]["children"] assert [item["sort"] for item in children] == [1, 2] assert body["permissions"] == [f"b:a:{mark}", f"b:z:{mark}"] assert f"b:off:{mark}" not in body["permissions"] def test_delete_resource_with_children(client: TestClient): headers = _staff_headers() mark = uuid.uuid4().hex[:8] parent = client.post( "/api/admin/resources", headers=headers, json={"type": "directory", "name": f"P{mark}", "title": "父", "sort": 1, "status": "active"}, ) assert parent.status_code == 200 child = client.post( "/api/admin/resources", headers=headers, json={ "parentId": parent.json()["id"], "type": "menu", "name": f"C{mark}", "title": "子", "path": f"/c-{mark}", "component": f"/c-{mark}", "status": "active", }, ) assert child.status_code == 200 denied = client.delete(f"/api/admin/resources/{parent.json()['id']}", headers=headers) assert denied.status_code == 400 assert denied.json()["detail"] == "存在子节点，不能删除" removed = client.delete(f"/api/admin/resources/{child.json()['id']}", headers=headers) assert removed.status_code == 200 removed_parent = client.delete(f"/api/admin/resources/{parent.json()['id']}", headers=headers) assert removed_parent.status_code == 200 def test_preserve_last_full_admin(client: TestClient): connection = client.app.state.connection db = Session( bind=connection, autoflush=False, expire_on_commit=False, join_transaction_mode="create_savepoint", ) nested = db.begin_nested() try: db.execute(update(AdminUser).values(status="disabled")) db.flush() db.expire_all() assert count_full_admins(db) == 0 def make_user(tag: str) -> AdminUser: now = utcnow() role = AdminRole(code=_suffix("RF"), name=tag, status="active", created_at=now, updated_at=now) user = AdminUser( username=_suffix(tag), password_hash="x", display_name=tag, mobile_mask="", status="active", failed_login_count=0, created_at=now, updated_at=now, ) db.add_all([role, user]) db.flush() for resource_id in db.scalars(select(AdminResource.id).where(AdminResource.status == "active")).all(): db.add(AdminRoleResource(role_id=role.id, resource_id=resource_id)) db.add(AdminUserRole(user_id=user.id, role_id=role.id)) db.flush() return user first = make_user("fulla") second = make_user("fullb") db.expire_all() assert count_full_admins(db) == 2 def disable_first() -> None: row = db.get(AdminUser, first.id) assert row is not None row.status = "disabled" preserve_full_admin(db, disable_first) db.expire_all() assert count_full_admins(db) == 1 def disable_last() -> None: row = db.get(AdminUser, second.id) assert row is not None row.status = "disabled" with pytest.raises(HTTPException) as caught: preserve_full_admin(db, disable_last) assert caught.value.detail == KEEP_FULL_ADMIN finally: nested.rollback() db.close() def test_seed_superuser_is_idempotent(client: TestClient): connection = client.app.state.connection db = Session( bind=connection, autoflush=False, expire_on_commit=False, join_transaction_mode="create_savepoint", ) username = _suffix("boot") try: seed_superuser(db, username, "Seedpass1") seed_superuser(db, username, "Seedpass1") db.commit() users = db.scalars(select(AdminUser).where(AdminUser.username == username)).all() assert len(users) == 1 role = db.scalar(select(AdminRole).where(AdminRole.code == "R_SUPER")) assert role is not None assert role.name == "超级管理员" directory = db.scalar( select(AdminResource).where(AdminResource.parent_id.is_(None), AdminResource.name == "System") ) assert directory is not None assert directory.path == "/system" assert directory.icon == "ri:settings-3-line" page = db.scalar(select(AdminResource).where(AdminResource.parent_id == directory.id, AdminResource.name == "User")) assert page is not None assert page.component == "/system/user" button = db.scalar( select(AdminResource).where(AdminResource.parent_id == page.id, AdminResource.name == "user:add") ) assert button is not None assert button.type == "button" assert button.permission == "user:add" assert button.visible is False linked = db.get(AdminUserRole, (users[0].id, role.id)) assert linked is not None finally: db.close() assert not _committed(username)