722 lines
23 KiB
Python
722 lines
23 KiB
Python
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)
|