zkxw/backend/tests/test_market_mobile_menu_sql.py

import re
from pathlib import Path


MIGRATION = (
    Path(__file__).resolve().parents[2]
    / "docs/sql/migrations/2026-10-10-market-mobile-menus.sql"
)

ROLES = {
    "specialist": "市场专员",
    "manager": "地区经理",
}
MODULES = {
    "presign": "预签约订单",
    "protocol": "协议单",
    "payment-chase": "催缴信息",
    "payment": "缴费信息",
    "sample": "样本信息",
    "freight": "运费信息",
}


def _sql() -> str:
    return MIGRATION.read_text(encoding="utf-8")


def _compact(sql: str) -> str:
    return re.sub(r"\s+", " ", sql)


def _without_comments(sql: str) -> str:
    return re.sub(r"--[^\n]*", "", sql)


def _insert_statements(sql: str) -> list[str]:
    return [
        _compact(statement)
        for statement in _without_comments(sql).split(";")
        if re.search(r"\bINSERT\s+INTO\s+sys_menu\b", statement, re.IGNORECASE)
    ]


def test_market_mobile_menu_migration_defines_two_empty_path_directories():
    sql = _compact(_without_comments(_sql()))

    for directory_name in ROLES.values():
        assert re.search(
            rf"'{directory_name}'\s*,\s*root\.id\s*,\s*'M'\s*,\s*''",
            sql,
        )


def test_market_mobile_menu_migration_defines_exactly_twelve_fixed_leaves():
    sql = _without_comments(_sql())
    expected_paths = {
        f"/pages/market/webview/index?entry={role}-{module}"
        for role in ROLES
        for module in MODULES
    }
    expected_perms = {
        f"mini:market:{role}:{module}"
        for role in ROLES
        for module in MODULES
    }

    paths = re.findall(
        r"['\"](/pages/market/webview/index\?entry=[a-z-]+)['\"]", sql
    )
    perms = re.findall(r"['\"](mini:market:[a-z-]+:[a-z-]+)['\"]", sql)

    assert len(paths) == 12
    assert set(paths) == expected_paths
    assert len(perms) == 12
    assert set(perms) == expected_perms
    for role in ROLES:
        for module in MODULES:
            path = f"/pages/market/webview/index?entry={role}-{module}"
            permission = f"mini:market:{role}:{module}"
            assert re.search(
                rf"'{re.escape(path)}'\s*(?:AS\s+path)?\s*,\s*"
                rf"'{re.escape(permission)}'\s*(?:AS\s+perms)?",
                sql,
                re.IGNORECASE,
            )
    assert re.search(
        r"entry\.menu_name\s*,\s*directory\.id\s*,\s*'C'\s*,\s*entry\.path",
        _compact(sql),
    )


def test_market_mobile_menu_migration_is_idempotent_and_tenant_aware():
    sql = _sql()
    compact = _compact(_without_comments(sql))
    inserts = _insert_statements(sql)

    assert len(inserts) == 3
    assert all(re.search(r"\bNOT\s+EXISTS\b", item, re.IGNORECASE) for item in inserts)
    assert re.search(r"root\.parent_id\s*=\s*'0'", compact)
    assert re.search(r"root\.menu_name\s*=\s*'移动端菜单'", compact)
    assert re.search(r"root\.del_flag\s*=\s*0", compact)
    assert compact.count("root.tenant_no") >= len(inserts)
    assert "'zkxw'" not in sql
    assert re.search(r"existing\.parent_id\s*=\s*directory\.id", compact)
    assert re.search(r"existing\.path\s*=\s*entry\.path", compact)
    assert re.search(r"existing\.menu_type\s*=\s*'C'", compact)


def test_each_insert_uses_the_complete_active_natural_key_without_bare_id_or():
    specialist, manager, leaves = _insert_statements(_sql())

    for directory, name in ((specialist, "市场专员"), (manager, "地区经理")):
        assert "existing.id" not in directory
        assert re.search(r"existing\.tenant_no\s*=\s*root\.tenant_no", directory)
        assert re.search(r"existing\.parent_id\s*=\s*root\.id", directory)
        assert re.search(r"existing\.menu_name\s*=\s*'" + name + r"'", directory)
        assert re.search(r"existing\.menu_type\s*=\s*'M'", directory)
        assert re.search(r"existing\.path\s*=\s*''", directory)
        assert re.search(r"existing\.del_flag\s*=\s*0", directory)

    assert "existing.id" not in leaves
    assert re.search(r"existing\.tenant_no\s*=\s*root\.tenant_no", leaves)
    assert re.search(r"existing\.parent_id\s*=\s*directory\.id", leaves)
    assert re.search(r"existing\.menu_name\s*=\s*entry\.menu_name", leaves)
    assert re.search(r"existing\.menu_type\s*=\s*'C'", leaves)
    assert re.search(r"existing\.path\s*=\s*entry\.path", leaves)
    assert re.search(r"existing\.perms\s*=\s*entry\.perms", leaves)
    assert re.search(r"existing\.del_flag\s*=\s*0", leaves)


def test_ids_and_directory_join_are_scoped_to_the_expected_tree_location():
    specialist, manager, leaves = _insert_statements(_sql())

    for directory in (specialist, manager):
        id_expression = re.search(
            r"\bSELECT\s+(MD5\(.*?\))\s*,\s*'(?:市场专员|地区经理)'",
            directory,
        ).group(1)
        assert "root.tenant_no" in id_expression
        assert "root.id" in id_expression

    leaf_id_expression = re.search(
        r"\bSELECT\s+(MD5\(.*?\))\s*,\s*entry\.menu_name", leaves
    ).group(1)
    assert "root.tenant_no" in leaf_id_expression
    assert "root.id" in leaf_id_expression
    assert "directory.id" in leaf_id_expression
    assert "entry.path" in leaf_id_expression
    assert re.search(r"directory\.path\s*=\s*''", leaves)


def test_same_id_wrong_or_soft_deleted_rows_are_repaired_instead_of_blocking_insert():
    specialist, manager, leaves = _insert_statements(_sql())

    for statement in (specialist, manager, leaves):
        assert re.search(r"\bON\s+DUPLICATE\s+KEY\s+UPDATE\b", statement)
        repair = re.split(
            r"\bON\s+DUPLICATE\s+KEY\s+UPDATE\b", statement, maxsplit=1
        )[1]
        for field in (
            "tenant_no",
            "parent_id",
            "menu_name",
            "menu_type",
            "path",
            "perms",
            "del_flag",
        ):
            assert re.search(
                rf"\b{field}\s*=\s*VALUES\s*\(\s*{field}\s*\)", repair
            ), field


def test_market_mobile_menu_migration_does_not_grant_roles_automatically():
    assert "sys_role_menu" not in _without_comments(_sql())