74 lines
5.0 KiB
Python
74 lines
5.0 KiB
Python
"""在 sw8/sw9 用专用二级账号创建临时三级账号，核对层级和角色后清理。"""
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
from sw_write_verify import Rehearsal
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--runtime", type=Path, required=True)
|
|
parser.add_argument("--config", type=Path, required=True)
|
|
parser.add_argument("--old-url", default="http://127.0.0.1:18080")
|
|
parser.add_argument("--new-url", default="http://127.0.0.1:19002")
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]):
|
|
raise ValueError("测试凭据必须位于仓库外")
|
|
run = Rehearsal(args)
|
|
username = "mig_recheck_" + run.suffix
|
|
password = (args.runtime / "synthetic-test-password.txt").read_text(encoding="utf-8").strip()
|
|
source_roles = run.query("SELECT role_id FROM sw8.sys_user_role WHERE user_id=%s", (run.owner("legacy", 3),))
|
|
expected_roles = run.query("SELECT sys_authority_authority_id id FROM sw9.sys_user_authority WHERE sys_user_id=%s", (run.owner("new", 3),))
|
|
created_sides = set()
|
|
try:
|
|
run.check("测试角色存在", len(source_roles) == 1 and len(expected_roles) == 1)
|
|
for side, database, table, column in (("legacy", "sw8", "sys_user", "user_name"), ("new", "sw9", "sys_users", "username")):
|
|
run.check("临时账号不存在 " + side, not run.query(f"SELECT 1 FROM {database}.{table} WHERE {column}=%s", (username,)))
|
|
payload = {"userName": username, "password": password, "nickName": "迁移复核临时账号", "realName": "迁移复核临时账号"}
|
|
run.call("legacy", "testlevel2", "POST", "/system/user", {**payload, "roleIds": [str(source_roles[0]["role_id"])]})
|
|
created_sides.add("sw8")
|
|
run.call("new", "testlevel2", "POST", "/sw/business", {**payload, "parentId": 0, "level": 1, "authorityId": 888})
|
|
created_sides.add("sw9")
|
|
old = run.query("SELECT user_id,level,parent_id,id_path,status FROM sw8.sys_user WHERE user_name=%s", (username,))[0]
|
|
new = run.query("SELECT id,level,parent_id,id_path,business_status,authority_id,enable FROM sw9.sys_users WHERE username=%s", (username,))[0]
|
|
run.check("新旧创建三级且挂靠当前二级", old["level"] == new["level"] == 3 and str(old["parent_id"]) == str(run.owner("legacy", 2)) and new["parent_id"] == run.owner("new", 2))
|
|
path = ",".join(run.mapping.get(part, part) for part in str(old["id_path"]).split(","))
|
|
run.check("祖先路径按用户映射一致", path == new["id_path"])
|
|
run.check("默认状态正常且角色不能由请求提权", str(old["status"]) == str(new["business_status"]) == "0" and new["enable"] == 1 and new["authority_id"] == expected_roles[0]["id"] and new["authority_id"] != 888)
|
|
old_roles = run.query("SELECT role_id FROM sw8.sys_user_role WHERE user_id=%s", (old["user_id"],))
|
|
new_roles = run.query("SELECT sys_authority_authority_id id FROM sw9.sys_user_authority WHERE sys_user_id=%s", (new["id"],))
|
|
run.check("关联角色正确", old_roles == source_roles and new_roles == expected_roles)
|
|
values = []
|
|
for database, user_id in (("sw8", old["user_id"]), ("sw9", new["id"])):
|
|
values.append(run.query(f"SELECT include_sub_promote,idcard_dedup FROM {database}.sys_user_preference WHERE user_id=%s", (user_id,)))
|
|
run.check("默认统计偏好一致", values[0] == values[1] == [{"include_sub_promote": 1, "idcard_dedup": 1}])
|
|
finally:
|
|
# 只清理本次唯一名称创建的临时账号及其角色/偏好，不触碰原有测试链。
|
|
for database, table, id_col, name_col in (("sw8", "sys_user", "user_id", "user_name"), ("sw9", "sys_users", "id", "username")):
|
|
if database not in created_sides:
|
|
continue
|
|
rows = run.query(f"SELECT {id_col} id FROM {database}.{table} WHERE {name_col}=%s", (username,))
|
|
for row in rows:
|
|
user_id = row["id"]
|
|
run.db.begin()
|
|
try:
|
|
run.query(f"DELETE FROM {database}.sys_user_preference WHERE user_id=%s", (user_id,))
|
|
if database == "sw8":
|
|
for relation in ("sys_user_role", "sys_user_post"):
|
|
run.query(f"DELETE FROM sw8.{relation} WHERE user_id=%s", (user_id,))
|
|
else:
|
|
run.query("DELETE FROM sw9.sys_user_authority WHERE sys_user_id=%s", (user_id,))
|
|
run.query(f"DELETE FROM {database}.{table} WHERE {id_col}=%s AND {name_col}=%s", (user_id, username))
|
|
run.db.commit()
|
|
except Exception:
|
|
run.db.rollback()
|
|
raise
|
|
run.check("临时账号清理完成 " + database, not run.query(f"SELECT 1 FROM {database}.{table} WHERE {name_col}=%s", (username,)))
|
|
run.db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|