zkxw/backend/tests/test_market_payment_indexes_sql.py

from pathlib import Path


SQL_PATH = (
    Path(__file__).resolve().parents[2]
    / 'docs/sql/migrations/2026-10-11-market-payment-indexes.sql'
)


def test_market_payment_index_migration_is_mysql57_idempotent_and_repairs_names():
    sql = SQL_PATH.read_text(encoding='utf-8')

    assert 'CREATE INDEX IF NOT EXISTS' not in sql.upper()
    assert sql.count('information_schema.statistics') >= 2
    assert sql.count('GROUP_CONCAT(column_name ORDER BY seq_in_index') >= 2
    assert sql.count('PREPARE stmt FROM @ddl') >= 2
    assert sql.count('MIN(non_unique)') >= 2
    assert sql.count('MAX(non_unique)') >= 2
    assert sql.count("SUM(CASE WHEN sub_part IS NOT NULL THEN 1 ELSE 0 END)") >= 2
    assert sql.count('COUNT(DISTINCT index_type)') >= 2
    assert sql.count('MIN(index_type)') >= 2
    assert sql.count("@index_non_unique_min = 1") >= 2
    assert sql.count("@index_non_unique_max = 1") >= 2
    assert sql.count("@index_prefix_columns = 0") >= 2
    assert sql.count("@index_type_count = 1") >= 2
    assert sql.count("@index_type = 'BTREE'") >= 2
    assert 'DROP INDEX idx_biz_protocol_payment_schedule_market_scope' in sql
    assert 'DROP INDEX idx_biz_protocol_payment_transfer_market_scope' in sql


def test_market_payment_index_migration_has_exact_required_column_orders():
    sql = SQL_PATH.read_text(encoding='utf-8')

    assert "'tenant_no,sample_payment_id,del_flag'" in sql
    assert "'tenant_no,schedule_id,del_flag'" in sql
    assert (
        'idx_biz_protocol_payment_schedule_market_scope '
        '(tenant_no, sample_payment_id, del_flag)'
    ) in sql
    assert (
        'idx_biz_protocol_payment_transfer_market_scope '
        '(tenant_no, schedule_id, del_flag)'
    ) in sql