AiDOPWarehouse/tools/integration-simulator/tests/test_mock_db.py

"""模拟 MySQL（DB_SYNC 对端）守卫：DDL 生成、类型推断、库名硬阻断、配置清单。

需要真实 MySQL 的用例在未设置 AIDOP_SIM_MYSQL_PASSWORD 时跳过，并明确标注“未实测”，
不得把跳过当作通过。
"""
from __future__ import annotations

import os

import pytest

from catalog import sample_adapters as sa
from providers import mock_db

RUN_ID = "20260930120000-abc123"


def test_blocks_aidop_databases():
    for bad in ("aidopdev", "aidop_uat", "aidop_prod", "aidop", "admin", "mdp_std"):
        with pytest.raises(mock_db.BlockedDatabaseError):
            mock_db.assert_sim_database(bad)


def test_allows_simulator_schema():
    mock_db.assert_sim_database(mock_db.SIM_DB_DEFAULT)
    mock_db.assert_sim_database("aidop_integration_sim_uat")


def test_rejects_empty_database_name():
    with pytest.raises(mock_db.BlockedDatabaseError):
        mock_db.assert_sim_database("")


def test_conn_params_never_carry_aidop_db(monkeypatch):
    monkeypatch.setenv("AIDOP_SIM_MYSQL_DATABASE", "aidopdev")
    with pytest.raises(mock_db.BlockedDatabaseError):
        mock_db.mysql_conn_params()


def test_type_inference():
    assert mock_db.infer_mysql_type([1, 2, None]) == "BIGINT"
    assert mock_db.infer_mysql_type([1.5]) == "DECIMAL(18,6)"
    assert mock_db.infer_mysql_type([True]) == "TINYINT(1)"
    assert mock_db.infer_mysql_type(["abc"]).startswith("VARCHAR(")
    assert mock_db.infer_mysql_type([None]) == "VARCHAR(191)"


def test_ddl_has_business_unique_key_and_increment_index():
    rows = sa.adapt_db_rows(sa.find_object("ITEM"), RUN_ID)
    ddl = mock_db.generate_ddl("sim_item", rows, ["bizKey"], "sourceUpdatedAt")
    assert ddl.startswith("CREATE TABLE IF NOT EXISTS `sim_item`")
    assert "UNIQUE KEY `uk_sim_item` (`bizKey`)" in ddl
    assert "KEY `ix_sim_item_incr` (`sourceUpdatedAt`)" in ddl
    assert "ENGINE=InnoDB" in ddl


def test_ddl_rejects_empty_rows():
    with pytest.raises(ValueError):
        mock_db.generate_ddl("sim_x", [], ["bizKey"])


def test_upsert_sql_is_idempotent():
    sql = mock_db._insert_sql("sim_item", ["bizKey", "sourceUpdatedAt"])
    assert "ON DUPLICATE KEY UPDATE" in sql
    assert sql.count("%s") == 2


def test_aidop_config_uses_dedicated_test_sources():
    cfg = mock_db.generate_aidop_config(sa.find_object("ITEM"))
    codes = {s["source_code"] for s in cfg["sources"]}
    assert codes == {"SIM_MODE3_DB", "SIM_MODE3_API", "SIM_MODE3_INBOUND"}
    assert cfg["systemCode"] == "SIM_MODE3"
    db_entity = next(e for e in cfg["entities"] if e["source_code"] == "SIM_MODE3_DB")
    assert db_entity["source_table_name"] == "sim_item"
    assert db_entity["incr_column"] == "sourceUpdatedAt"


def test_aidop_config_omits_inbound_when_unsupported():
    cfg = mock_db.generate_aidop_config(sa.find_object("SALES_ORDER_HEAD"))
    assert "SIM_MODE3_INBOUND" not in {s["source_code"] for s in cfg["sources"]}


requires_mysql = pytest.mark.skipif(
    not os.environ.get("AIDOP_SIM_MYSQL_PASSWORD"),
    reason="未设置 AIDOP_SIM_MYSQL_PASSWORD：真实 MySQL 用例未执行（不代表通过）")


@requires_mysql
def test_live_schema_load_and_increment():
    obj = sa.find_object("ITEM")
    spec = obj["dbSync"]
    rows = sa.adapt_db_rows(obj, RUN_ID)
    mock_db.init_schema({spec["table"]: {"rows": rows, "businessKey": spec["businessKey"],
                                         "incrementColumn": spec["incrementColumn"]}})
    mock_db.reset_table(spec["table"])
    assert mock_db.upsert_rows(spec["table"], rows) == len(rows)
    # 重复装载不得产生重复业务行
    mock_db.upsert_rows(spec["table"], rows)
    stats = mock_db.table_stats(spec["table"], spec["incrementColumn"])
    assert stats["rowCount"] == len(rows)
    assert stats["maxIncrement"]
    # 更新并推进增量列
    affected = mock_db.update_rows(spec["table"], {"itemName": "更新后"},
                                   {"bizKey": rows[0]["bizKey"]}, spec["incrementColumn"],
                                   "2026-09-30T23:59:59")
    assert affected == 1
    assert mock_db.table_stats(spec["table"], spec["incrementColumn"])["maxIncrement"] >= "2026-09-30"