AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP.Tests/S0/Dim/S0DimContractTests.cs

using System;
using System.IO;
using System.Linq;
using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.DataPlatform.S0Dim;
using Xunit;

namespace Admin.NET.Plugin.AiDOP.Tests.S0.Dim;

/// <summary>
/// S0 主数据维度的**声明与 SQL 生成**契约测试（纯函数，不接触数据库）。
///
/// 这些断言把两类曾经真实踩过的坑钉在编译/测试期：
/// ① LocationMaster 是 dual-column 复刻表，snake_case 一侧是**填了假值的死列**
///    （rec_id 恒 0 / domain_code 恒空串 / is_active 恒 0），取错不报错、静默返回垃圾；
/// ② 租户隔离必须由 SQL 谓词保证，不能依赖调用方自觉。
/// </summary>
public class S0DimContractTests
{
    private const StringComparison Ord = StringComparison.Ordinal;

    [Fact]
    public void Catalog_should_be_self_consistent()
    {
        S0DimCatalog.ValidateAll();
        // 断言内容而非数量：数量断言每加一个维度就要改一次，且改完什么也没守住。
        // 这里守的是「目录里恰好是这些维度、每个都能被 Find 命中、源系统统一」。
        Assert.Equal(
            [
                // C1
                "WORK_CENTER", "DEPARTMENT", "LOCATION", "LOCATION_SHELF",
                "EMPLOYEE", "LINE", "CUSTOMER", "PERSON_SKILL",
                // C2
                "LABEL_TYPE", "SOP_FILE_TYPE", "QUALITY_BASE_TYPE",
                "SAMPLING_SCHEME", "INSPECTION_ITEM", "INSPECTION_METHOD",
                "INSPECTION_FREQUENCY", "INSPECTION_INSTRUMENT",
                // C1 扩展
                "ITEM",
                // C3
                "EMP_WORK_DUTY", "LINE_POST", "LINE_POST_SKILL",
                // C3-A/C/B 物料桥（全部父 = dim_item）
                "ITEM_OP_LINE", "ITEM_ROUTING_OP",
                "ITEM_OP_CONDITION", "ITEM_OP_IMAGE",
                "ITEM_BOM", "ITEM_SUBSTITUTE"
            ],
            S0DimCatalog.All.Select(d => d.Key).ToArray());
        Assert.All(S0DimCatalog.All, d => Assert.Same(d, S0DimCatalog.Find(d.Key)));
        Assert.All(S0DimCatalog.All, d => Assert.Equal(S0DimCatalog.LocalSourceSystem, d.SourceSystem));
    }

    [Fact]
    public void Catalog_find_should_be_case_insensitive_and_null_safe()
    {
        Assert.Same(S0DimCatalog.WorkCenter, S0DimCatalog.Find("work_center"));
        Assert.Same(S0DimCatalog.LocationShelf, S0DimCatalog.Find(" LOCATION_SHELF "));
        Assert.Null(S0DimCatalog.Find("NOT_A_KEY"));
        Assert.Null(S0DimCatalog.Find(null));
    }

    [Fact]
    public void Every_definition_must_declare_explicit_json_path()
    {
        foreach (var def in S0DimCatalog.All)
            foreach (var col in def.Columns.Where(c => c.Kind != S0DimValueKind.TenantIdColumn))
                Assert.False(string.IsNullOrWhiteSpace(col.JsonPath),
                    $"{def.Key}.{col.TargetColumn} 必须显式声明 JsonPath");
    }

    [Fact]
    public void Definition_without_json_path_should_be_rejected()
    {
        var bad = S0DimCatalog.WorkCenter with
        {
            Columns =
            [
                new S0DimColumn("tenant_id", S0DimValueKind.TenantIdColumn),
                new S0DimColumn("domain_code", S0DimValueKind.Str, null, Required: true),
                new S0DimColumn("work_center_code", S0DimValueKind.Str, "WorkCtr", Required: true)
            ]
        };
        var ex = Assert.Throws<InvalidOperationException>(() => bad.Validate());
        Assert.Contains("JsonPath", ex.Message, Ord);
    }

    [Fact]
    public void Business_key_order_must_match_source_biz_key_expr()
    {
        // dim 业务键顺序与 mdp_entity.biz_key_expr 不一致时，stg.source_biz_key 无法与 dim 业务键对账
        var bad = S0DimCatalog.WorkCenter with { SourceBizKeyColumns = ["WorkCtr", "Domain"] };
        var ex = Assert.Throws<InvalidOperationException>(() => bad.Validate());
        Assert.Contains("业务键顺序不一致", ex.Message, Ord);
    }

    [Fact]
    public void Business_key_columns_must_be_required()
    {
        var cols = S0DimCatalog.WorkCenter.Columns
            .Select(c => c.TargetColumn == "domain_code" ? c with { Required = false } : c).ToList();
        var bad = S0DimCatalog.WorkCenter with { Columns = cols };
        Assert.Throws<InvalidOperationException>(() => bad.Validate());
    }

    [Fact]
    public void Business_key_must_start_with_tenant_id()
    {
        var bad = S0DimCatalog.WorkCenter with
        {
            BusinessKeyColumns = ["domain_code", "work_center_code"],
            SourceBizKeyColumns = ["WorkCtr"]
        };
        Assert.Throws<InvalidOperationException>(() => bad.Validate());
    }

    [Fact]
    public void Identifiers_must_be_whitelisted()
    {
        var bad = S0DimCatalog.WorkCenter with { DimTable = "dim_work_center; DROP TABLE x" };
        Assert.Throws<InvalidOperationException>(() => bad.Validate());
    }

    // ────────────────────────────────────────────────────────────────
    // LocationMaster 影子列黑名单（🔴 本仓最容易静默出错的一处）
    // ────────────────────────────────────────────────────────────────

    public static TheoryData<string> LocationShadowPaths() =>
        ["$.domain_code", "$.rec_id", "$.is_active", "$.physical_address", "$.create_user"];

    [Theory]
    [MemberData(nameof(LocationShadowPaths))]
    public void Location_insert_sql_must_never_read_shadow_columns(string shadowPath)
    {
        var sql = S0DimSqlBuilder.BuildInsertSql(S0DimCatalog.Location);
        Assert.DoesNotContain(shadowPath, sql, Ord);
    }

    /// <summary>
    /// LocationMaster 影子列的**裸列名**（不带 <c>$.</c> 前缀）。
    ///
    /// <para>源侧校验和 / 业务键 SQL 把 <c>JsonPath</c> 当作**源表物理列名**用
    /// （<c>S0DimSqlBuilder.BuildSourceChecksumSql</c>：<c>a.`Domain`</c>），**整条语句里没有任何 <c>$.</c>**。
    /// 因此对这两个生成器断言 <c>DoesNotContain("$.domain_code")</c> 是**恒真**的，零保护 ——
    /// 必须断言反引号形式。</para>
    /// </summary>
    public static TheoryData<string> LocationShadowColumns() =>
        ["domain_code", "rec_id", "is_active", "physical_address", "create_user"];

    [Theory]
    [MemberData(nameof(LocationShadowColumns))]
    public void Location_source_side_sql_must_never_read_shadow_columns(string shadowColumn)
    {
        // 这是本批**唯一能静默通过全部对账**的缺陷类：
        // JsonPath 若误写成影子列，dim 侧读 staging 的 $.is_active（=0），
        // 源侧校验和读源表的 a.`is_active`（同样 =0）→ 两侧自洽 → checksum 相等 → PASS，
        // 而真实的 IsActive=1 被静默翻成 0。影子列存 falsy 而非 NULL，不会抛任何异常。
        var needle = $"{S0DimSqlBuilder.SrcAlias}.`{shadowColumn}`";
        Assert.DoesNotContain(needle, S0DimSqlBuilder.BuildSourceChecksumSql(S0DimCatalog.Location), Ord);
        Assert.DoesNotContain(needle, S0DimSqlBuilder.SourceBizKeySetSql(S0DimCatalog.Location), Ord);
    }

    [Theory]
    [InlineData("Domain")]
    [InlineData("IsActive")]
    [InlineData("PhysicalAddress")]
    [InlineData("UpdateTime")]
    public void Location_source_checksum_must_read_real_columns(string realColumn)
    {
        // 正向断言：光有 DoesNotContain 不够 —— 列被整个删掉时它也通过。
        Assert.Contains($"{S0DimSqlBuilder.SrcAlias}.`{realColumn}`",
            S0DimSqlBuilder.BuildSourceChecksumSql(S0DimCatalog.Location), Ord);
    }

    [Fact]
    public void Insert_and_source_checksum_must_read_the_same_declared_path()
    {
        // 通用护栏（不限 Location）：dim 物化走 staging 的 $.JsonPath，
        // 源侧校验和走源表的 a.`JsonPath`。两者必须来自**同一个**声明，
        // 否则「写进 dim 的值」与「用来验证 dim 的值」根本不是一回事。
        foreach (var def in S0DimCatalog.All)
        {
            var insert = S0DimSqlBuilder.BuildInsertSql(def);
            var srcChk = S0DimSqlBuilder.BuildSourceChecksumSql(def);
            foreach (var c in def.Columns)
            {
                if (c.Kind == S0DimValueKind.TenantIdColumn) continue;
                Assert.Contains($"$.{c.JsonPath}", insert, Ord);
                Assert.Contains($"{S0DimSqlBuilder.SrcAlias}.`{c.JsonPath}`", srcChk, Ord);
            }
        }
    }

    [Fact]
    public void Source_and_dim_checksum_must_have_identical_segment_count()
    {
        // 位置对称性：两侧 CONCAT_WS 的段数/顺序错位会产出**完全自洽的错值**。
        // 段数相等是最容易机械校验的必要条件（顺序由 ChecksumColumns 同一函数保证）。
        foreach (var def in S0DimCatalog.All)
        {
            var src = S0DimSqlBuilder.BuildSourceChecksumSql(def);
            var dim = S0DimSqlBuilder.BuildDimChecksumSql(def);
            Assert.Equal(Occurrences(src, "IFNULL("), Occurrences(dim, "IFNULL("));
            Assert.Equal(Occurrences(src, "CONCAT_WS("), Occurrences(dim, "CONCAT_WS("));
        }
    }

    private static int Occurrences(string haystack, string needle)
    {
        var n = 0;
        for (var i = haystack.IndexOf(needle, Ord); i >= 0; i = haystack.IndexOf(needle, i + needle.Length, Ord))
            n++;
        return n;
    }

    [Fact]
    public void Required_string_columns_must_also_reject_empty_string()
    {
        // IS NOT NULL 挡不住空串，而 CONCAT_WS 会保留空段：
        // Domain='' 会拼出 '#LOC01' 这种「形似合法」的业务键，且不会被 blank_cnt 探针捕获
        // （它只捕获整串全空）。故 Required 的 Str 列必须同时有 <> '' 守卫。
        foreach (var def in S0DimCatalog.All)
        {
            var sql = S0DimSqlBuilder.BuildInsertSql(def);
            foreach (var c in def.Columns)
            {
                if (c.Kind != S0DimValueKind.Str || !c.Required) continue;
                // 必须逐列绑定：整串级的 Assert.Contains("<> ''") 在「5 个 Required 列只有 1 个带守卫」时照样通过。
                var expr = MdpJsonSql.Str(S0DimSqlBuilder.StgAlias, c.JsonPath!);
                Assert.Contains($"({expr}) IS NOT NULL", sql, Ord);
                Assert.Contains($"({expr}) <> ''", sql, Ord);
            }
        }
    }

    [Fact]
    public void Every_business_key_component_must_be_required_and_of_key_safe_kind()
    {
        // 守住上一条测试的前提：若将来把某个业务键分量改成非 Required，
        // 空串守卫会连同 NULL 守卫一起消失，而上一条测试仍会通过（它只遍历 Required 列）。
        foreach (var def in S0DimCatalog.All)
        {
            foreach (var bk in def.BusinessKeyColumns.Skip(1))
            {
                var col = def.Columns.First(x => x.TargetColumn == bk);
                Assert.True(col.Required, $"[{def.Key}] 业务键分量 {bk} 必须 Required");
                Assert.True(col.Kind is S0DimValueKind.Str or S0DimValueKind.Int,
                    $"[{def.Key}] 业务键分量 {bk} 的 Kind={col.Kind} 不适合做键" +
                    "（Dec 有精度格式化差异、DateTimeSec 截秒、BoolTrue 基数过低）");
            }
        }
    }

    public static TheoryData<string> LocationRealPaths() =>
        ["$.Domain", "$.IsActive", "$.PhysicalAddress", "$.location", "$.descr", "$.typed", "$.storer", "$.UpdateTime"];

    [Theory]
    [MemberData(nameof(LocationRealPaths))]
    public void Location_insert_sql_must_read_real_columns(string realPath)
    {
        var sql = S0DimSqlBuilder.BuildInsertSql(S0DimCatalog.Location);
        Assert.Contains(realPath, sql, Ord);
    }

    [Fact]
    public void Location_shadow_blacklist_must_not_be_applied_to_shelf()
    {
        // LocationShelfMaster 本身就是 snake_case、无影子列：
        // 这里的 $.domain_code / $.update_time 是**正确**的物理列名，不适用父表的禁用规则
        var sql = S0DimSqlBuilder.BuildInsertSql(S0DimCatalog.LocationShelf);
        Assert.Contains("$.domain_code", sql, Ord);
        Assert.Contains("$.location", sql, Ord);
        Assert.Contains("$.inv_shelf", sql, Ord);
        Assert.Contains("$.update_time", sql, Ord);
        Assert.DoesNotContain("$.Domain'", sql, Ord);
        Assert.DoesNotContain("$.InvShelf", sql, Ord);
    }

    // ────────────────────────────────────────────────────────────────
    // 租户隔离：所有生成的 SQL 都必须带租户谓词
    // ────────────────────────────────────────────────────────────────

    [Fact]
    public void Insert_sql_must_filter_tenant_source_table_and_batch()
    {
        foreach (var def in S0DimCatalog.All)
        {
            var sql = S0DimSqlBuilder.BuildInsertSql(def);
            Assert.Contains("s.`tenant_id` = @TenantId", sql, Ord);
            Assert.Contains("s.`source_system` = @SourceSystem", sql, Ord);
            Assert.Contains("s.`source_table` = @SourceTable", sql, Ord);
            Assert.Contains("s.`sync_batch_id` = @BatchId", sql, Ord);
        }
    }

    [Fact]
    public void Insert_sql_must_not_use_on_duplicate_key_update()
    {
        // 源侧业务键重复必须撞 dim 唯一键并回滚（transform fail），绝不 last-wins
        foreach (var def in S0DimCatalog.All)
            Assert.DoesNotContain("ON DUPLICATE KEY", S0DimSqlBuilder.BuildInsertSql(def), StringComparison.OrdinalIgnoreCase);
    }

    [Fact]
    public void Purge_sql_must_carry_all_three_predicates()
    {
        foreach (var def in S0DimCatalog.All)
        {
            var sql = S0DimSqlBuilder.BuildPurgeStagingSql(def);
            Assert.Contains("s.`tenant_id` = @TenantId", sql, Ord);
            Assert.Contains("s.`source_system` = @SourceSystem", sql, Ord);
            Assert.Contains("s.`source_table` = @SourceTable", sql, Ord);
            // purge 不限批次：本轮之前的残留也要清掉
            Assert.DoesNotContain("sync_batch_id", sql, Ord);
        }
    }

    [Fact]
    public void All_count_and_set_sql_must_be_tenant_scoped()
    {
        foreach (var def in S0DimCatalog.All)
        {
            Assert.Contains("@TenantId", S0DimSqlBuilder.BuildSourceCountSql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.BuildStagingCountSql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.BuildDimCountSql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.BuildDimBatchImpuritySql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.SourceBizKeySetSql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.StagingBizKeySetSql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.DimBizKeySetSql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.BuildSourceChecksumSql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.BuildDimChecksumSql(def), Ord);
            Assert.Contains("@TenantId", S0DimSqlBuilder.BuildDimDuplicateBizKeySql(def), Ord);
        }
    }

    [Fact]
    public void Bool_columns_must_carry_recognizability_guard()
    {
        // MdpJsonSql.BoolTrue 对 null / 不可识别值静默返回 0；必须有守卫把这类行挡在 dim 之外。
        // 遍历全部维度并逐列绑定：只测 WorkCenter 会漏掉 Location / Department 的 BoolTrue 列。
        var covered = 0;
        foreach (var def in S0DimCatalog.All)
        {
            var sql = S0DimSqlBuilder.BuildInsertSql(def);
            foreach (var c in def.Columns)
            {
                if (c.Kind != S0DimValueKind.BoolTrue) continue;
                var raw = MdpJsonSql.Raw(S0DimSqlBuilder.StgAlias, c.JsonPath!);
                Assert.Contains($"LOWER({raw}) IN ('1','0','true','false')", sql, Ord);
                covered++;
            }
        }
        Assert.True(covered > 0, "目录里一个 BoolTrue 列都没有，本测试已失去意义");
    }

    [Fact]
    public void Checksum_must_use_exact_integer_summation_not_double()
    {
        // 回归守卫（2026-09-07 Batch 2 实测缺陷）：CONV() 返回字符串，SUM(字符串) 在 MySQL 里按 DOUBLE 累加，
        // 只有约 16 位有效数字；两侧扫描顺序不同会产生不同舍入，导致完全相同的数据被误判为 checksum 不一致
        // （租户 838257186181189：source=…980000 / dim=…980700，count 相同）。
        // 必须先 CAST 成 UNSIGNED，SUM 才走 DECIMAL 精确累加、与顺序无关。
        foreach (var def in S0DimCatalog.All)
        {
            foreach (var sql in new[] { S0DimSqlBuilder.BuildSourceChecksumSql(def), S0DimSqlBuilder.BuildDimChecksumSql(def) })
            {
                Assert.Contains("SUM(CAST(CONV(", sql, Ord);
                Assert.Contains("AS UNSIGNED))", sql, Ord);
                Assert.DoesNotContain("SUM(CONV(", sql, Ord);
            }
        }
    }

    [Fact]
    public void Shelf_definition_declares_parent_and_location_declares_mirror_unique()
    {
        Assert.Equal("dim_location", S0DimCatalog.LocationShelf.ParentDimTable);
        Assert.Equal(["domain_code", "location_code"], S0DimCatalog.LocationShelf.ParentKeyColumns);
        Assert.Equal(["tenant_id", "location_code"], S0DimCatalog.Location.MirrorUniqueColumns);

        var orphanSql = S0DimSqlBuilder.BuildOrphanChildSql(S0DimCatalog.LocationShelf, 10);
        Assert.Contains("dim_location", orphanSql, Ord);
        Assert.Contains("@TenantId", orphanSql, Ord);
    }

    [Fact]
    public void Work_center_dim_must_not_invent_company_or_factory_columns()
    {
        // 源 WorkCtrMaster 没有 company_ref_id / factory_ref_id，dim 不得凭空补
        var targets = S0DimCatalog.WorkCenter.Columns.Select(c => c.TargetColumn).ToList();
        Assert.DoesNotContain("company_id", targets);
        Assert.DoesNotContain("factory_id", targets);
    }

    [Fact]
    public void Shelf_dim_must_not_invent_is_active()
    {
        // 源 LocationShelfMaster 没有 IsActive 列，不发明状态语义
        var targets = S0DimCatalog.LocationShelf.Columns.Select(c => c.TargetColumn).ToList();
        Assert.DoesNotContain("is_active", targets);
    }

    [Fact]
    public void Two_definitions_may_share_staging_but_never_share_dim()
    {
        Assert.Equal(S0DimCatalog.Location.StagingTable, S0DimCatalog.LocationShelf.StagingTable);
        Assert.NotEqual(S0DimCatalog.Location.SourceTable, S0DimCatalog.LocationShelf.SourceTable);
        // 与维度数量解耦：新增维度不该让这条断言失效，它要守的是「dim 表不共用」这个不变量
        Assert.Equal(S0DimCatalog.All.Count, S0DimCatalog.All.Select(d => d.DimTable).Distinct().Count());
    }

    // ────────────────────────────────────────────────────────────────
    // Phase 2 · Batch 4：EmployeeMaster / LineMaster
    // ────────────────────────────────────────────────────────────────

    [Fact]
    public void Employee_identity_must_be_tenant_domain_employee()
    {
        var d = S0DimCatalog.Employee;
        Assert.Equal(["tenant_id", "employee_code"], d.BusinessKeyColumns);
        Assert.Equal(["Employee"], d.SourceBizKeyColumns);
        // legacy scope 不得混进身份
        foreach (var legacy in new[] { "domain_code", "company_id", "factory_id" })
            Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
        Assert.Equal("EmployeeMaster", d.SourceTable);
        Assert.Equal("mdp_stg_s0_employee", d.StagingTable);
        Assert.Equal("dim_employee", d.DimTable);
    }

    [Fact]
    public void Employee_sys_user_id_must_be_attribute_only_never_business_key()
    {
        // 源实测 1105 行中仅 7 行非空（0.6%）：把账号 id 当员工身份会让 99.4% 的员工无法标识。
        var d = S0DimCatalog.Employee;
        Assert.DoesNotContain("sys_user_id", d.BusinessKeyColumns);
        Assert.DoesNotContain("sys_user_id", d.SourceBizKeyColumns);
        Assert.DoesNotContain("sys_user_id", d.ParentKeyColumns!);

        var col = d.Columns.Single(c => c.TargetColumn == "sys_user_id");
        Assert.False(col.Required);      // 必须可空，否则 1098 行会被 RowGuards 过滤掉
        Assert.Equal("sys_user_id", col.JsonPath);
    }

    [Fact]
    public void Employee_parent_is_department_by_domain_and_department_code()
    {
        var d = S0DimCatalog.Employee;
        Assert.Equal("dim_department", d.ParentDimTable);
        Assert.Equal(["domain_code", "department_code"], d.ParentKeyColumns);
        // 父键分量不得 Required：源侧 292 行未分配部门，Required 会把它们整行过滤掉
        Assert.False(d.Columns.Single(c => c.TargetColumn == "department_code").Required);
        // 依赖顺序：父必须排在子之前，否则孤儿判定时父维度还没就绪
        Assert.True(S0DimCatalog.All.ToList().IndexOf(S0DimCatalog.Department)
                  < S0DimCatalog.All.ToList().IndexOf(d));
    }

    [Fact]
    public void Orphan_sql_must_not_report_unassigned_parent_key_as_orphan()
    {
        // 「未分配部门」(department_code IS NULL) 不是孤儿，「部门码找不到」才是。
        // <=> 是 NULL 安全等值，而 dim_department.department_code 为 NOT NULL，
        // 不加 IS NOT NULL 过滤会把 292 行未分配全部误报成孤儿，淹没 97 个真悬挂。
        foreach (var sql in new[]
                 {
                     S0DimSqlBuilder.BuildOrphanChildSql(S0DimCatalog.Employee, 10),
                     S0DimSqlBuilder.BuildOrphanChildCountSql(S0DimCatalog.Employee)
                 })
        {
            Assert.Contains("d.`domain_code` IS NOT NULL", sql, Ord);
            Assert.Contains("d.`department_code` IS NOT NULL", sql, Ord);
            Assert.Contains("p.`id` IS NULL", sql, Ord);
            Assert.Contains("d.`tenant_id` = @TenantId", sql, Ord);
        }
    }

    [Fact]
    public void Orphan_count_sql_must_not_be_truncated_by_sample_limit()
    {
        // 告警数字必须来自不截断的 COUNT，而不是被 OrphanSampleLimit 砍过的样本条数。
        var count = S0DimSqlBuilder.BuildOrphanChildCountSql(S0DimCatalog.Employee);
        Assert.Contains("SELECT COUNT(*)", count, Ord);
        Assert.DoesNotContain("LIMIT", count, Ord);
    }

    [Fact]
    public void Line_identity_must_be_tenant_and_line_only_no_legacy_scope()
    {
        // 按「当前业务证据优先」重判：CRUD 无查重、consumer 只用 (TenantId, Line)、
        // 且每租户在 LineMaster 只有 1 个 Domain/company/factory（租户内常量，零区分力）。
        // 旧物理 UK 含 company+factory，属 legacy scope，不予继承。
        var d = S0DimCatalog.Line;
        Assert.Equal(["tenant_id", "line_code"], d.BusinessKeyColumns);
        Assert.Equal(["Line"], d.SourceBizKeyColumns);
        foreach (var legacy in new[] { "domain_code", "company_id", "factory_id" })
            Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
        Assert.Null(d.ParentDimTable);
    }

    [Fact]
    public void Line_must_read_real_describe_column_not_shadow()
    {
        // Describe 169/169 填充；line_describe 0/169——本仓第三例恒空影子列。
        var sql = S0DimSqlBuilder.BuildInsertSql(S0DimCatalog.Line);
        Assert.Contains("$.Describe", sql, Ord);
        Assert.DoesNotContain("$.line_describe", sql, Ord);
        Assert.Contains($"{S0DimSqlBuilder.SrcAlias}.`Describe`",
            S0DimSqlBuilder.BuildSourceChecksumSql(S0DimCatalog.Line), Ord);
        Assert.DoesNotContain($"{S0DimSqlBuilder.SrcAlias}.`line_describe`",
            S0DimSqlBuilder.BuildSourceChecksumSql(S0DimCatalog.Line), Ord);
    }

    [Theory]
    [InlineData("EMPLOYEE")]
    [InlineData("LINE")]
    [InlineData("CUSTOMER")]
    [InlineData("PERSON_SKILL")]
    public void New_dims_must_carry_tenant_predicate_and_no_upsert(string key)
    {
        var d = S0DimCatalog.Find(key)!;
        var insert = S0DimSqlBuilder.BuildInsertSql(d);
        Assert.Contains("s.`tenant_id` = @TenantId", insert, Ord);
        Assert.Contains("s.`source_system` = @SourceSystem", insert, Ord);
        Assert.Contains("s.`source_table` = @SourceTable", insert, Ord);
        Assert.Contains("s.`sync_batch_id` = @BatchId", insert, Ord);
        Assert.DoesNotContain("ON DUPLICATE KEY", insert, Ord);

        var purge = S0DimSqlBuilder.BuildPurgeStagingSql(d);
        Assert.Contains("s.`tenant_id` = @TenantId", purge, Ord);
        Assert.DoesNotContain("sync_batch_id", purge, Ord);
    }

    [Fact]
    public void PullContext_must_never_set_FactoryId_or_CompanyId()
    {
        // I-015：MdpStagingWriter 只识别 factory_id / FactoryId 两个列名。
        // LineMaster / EmployeeMaster 用的是 factory_ref_id → 解析为 null →
        // 一旦 ctx.FactoryId > 0，比较基准退化成 1，整批行被 `return 0` 静默丢弃。
        // 因此 S0 的 pullCtx 必须只按租户取范围，company/factory 只能是源侧身份字段。
        var src = File.ReadAllText(FindSourceFile("S0DimMaterializer.cs"));
        var ctxBlock = src[src.IndexOf("new MdpPullContext", Ord)..];
        ctxBlock = ctxBlock[..ctxBlock.IndexOf("};", Ord)];

        Assert.DoesNotContain("FactoryId", ctxBlock, Ord);
        Assert.DoesNotContain("CompanyId", ctxBlock, Ord);
        Assert.Contains("TenantId = tenantId", ctxBlock, Ord);
        Assert.Contains("RequireMatchingSourceTenant = true", ctxBlock, Ord);
    }

    private static string FindSourceFile(string fileName)
    {
        var dir = new DirectoryInfo(AppContext.BaseDirectory);
        while (dir != null)
        {
            var candidate = Path.Combine(dir.FullName,
                "Admin.NET.Plugin.AiDOP", "DataPlatform", "S0Dim", fileName);
            if (File.Exists(candidate)) return candidate;
            dir = dir.Parent;
        }
        throw new FileNotFoundException($"未找到 {fileName}");
    }

    // ────────────────────────────────────────────────────────────────
    // Phase 2 · Batch 5：CustMaster / PersonSkill
    // ────────────────────────────────────────────────────────────────

    [Theory]
    [InlineData("CUSTOMER", "customer_code", "Cust")]
    [InlineData("PERSON_SKILL", "skill_code", "Code")]
    public void Batch5_identity_must_be_tenant_plus_code_only(string key, string dimCol, string srcCol)
    {
        // 身份按当前业务证据判定，不继承旧物理唯一键：
        //  · Customer   CRUD 查重 = tenant + FactoryRefId(=f(tenant)) + Cust ⇒ 等价 tenant + Cust
        //  · PersonSkill 现有消费方已按 (租户, Code) 解析，并把同租户跨厂重码当歧义报错
        var d = S0DimCatalog.Find(key)!;
        Assert.Equal(["tenant_id", dimCol], d.BusinessKeyColumns);
        Assert.Equal([srcCol], d.SourceBizKeyColumns);
        foreach (var legacy in new[] { "domain_code", "company_id", "factory_id" })
            Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
    }

    [Fact]
    public void PersonSkill_must_not_declare_fake_parent_to_employee_or_skill_bridge()
    {
        // EmpSkills.SkillNo 与 PersonSkill.Code 同租户实测 650 行中 632 行对不上，
        // 不能因为都叫 Skill 就建立虚假父子关系。
        Assert.Null(S0DimCatalog.PersonSkill.ParentDimTable);
        Assert.Null(S0DimCatalog.PersonSkill.ParentKeyColumns);
    }

    [Fact]
    public void PersonSkill_watermark_must_map_updated_at_not_created_at()
    {
        // UpdatedAt 是维护时间（源 5/46 非空 = 只有改过的行才有值）；
        // 不得用 CreatedAt 冒充「最近更新」，那会让每行都显得刚被维护过。
        var col = S0DimCatalog.PersonSkill.Columns.Single(c => c.TargetColumn == "source_updated_at");
        Assert.Equal("UpdatedAt", col.JsonPath);
        Assert.False(col.Required);
        Assert.DoesNotContain("CreatedAt",
            S0DimCatalog.PersonSkill.Columns.Select(c => c.JsonPath ?? ""));
    }

    // ────────────────────────────────────────────────────────────────
    // C2 / C3：本轮实测出的框架地雷，钉在测试里
    // ────────────────────────────────────────────────────────────────

    [Fact]
    public void Qms_status_columns_must_be_str_never_booltrue()
    {
        // 实测值域含 `已审核` / `已启用` / `A` / `ENABLED` / `INVALID_STATUS`。
        // 用 BoolTrue 会命中 RowGuards 的可辨识性守卫：4 张表整表归零，
        // 而 qms_inspectionfreq.FENABLE 恰好存活 8/13 —— 部分静默丢失，最难发现。
        foreach (var key in new[] { "SAMPLING_SCHEME", "INSPECTION_ITEM", "INSPECTION_METHOD",
                                    "INSPECTION_FREQUENCY", "INSPECTION_INSTRUMENT" })
        {
            var d = S0DimCatalog.Find(key)!;
            foreach (var name in new[] { "status_code", "enable_code" })
            {
                var col = d.Columns.FirstOrDefault(c => c.TargetColumn == name);
                if (col is null) continue;
                Assert.Equal(S0DimValueKind.Str, col.Kind);
            }
        }
    }

    [Fact]
    public void InspectionFrequency_remark_must_map_FREMARK_not_FCOMMENT()
    {
        // 兄弟表都叫 FCOMMENT，唯独本表叫 FREMARK。照抄会取到不存在的列。
        var col = S0DimCatalog.InspectionFrequency.Columns.Single(c => c.TargetColumn == "remark");
        Assert.Equal("FREMARK", col.JsonPath);
    }

    [Fact]
    public void SopFileType_must_not_declare_watermark_it_has_no_time_column()
    {
        // 源 ImageType 只有 5 列、无任何时间列。不得为"标准字段齐整"而回填。
        Assert.DoesNotContain("source_updated_at",
            S0DimCatalog.SopFileType.Columns.Select(c => c.TargetColumn));
    }

    [Fact]
    public void LinePostSkill_effective_date_must_read_StartDate_not_the_shadow_column()
    {
        // 🔴 目标列名 effective_date 与源表那个 0/261 的影子列同名，是本批最易取错的一处。
        var col = S0DimCatalog.LinePostSkill.Columns.Single(c => c.TargetColumn == "effective_date");
        Assert.Equal("StartDate", col.JsonPath);

        var insert = S0DimSqlBuilder.BuildInsertSql(S0DimCatalog.LinePostSkill);
        Assert.Contains("$.StartDate", insert, Ord);
        Assert.DoesNotContain("$.effective_date", insert, Ord);

        var srcChk = S0DimSqlBuilder.BuildSourceChecksumSql(S0DimCatalog.LinePostSkill);
        Assert.Contains($"{S0DimSqlBuilder.SrcAlias}.`StartDate`", srcChk, Ord);
        Assert.DoesNotContain($"{S0DimSqlBuilder.SrcAlias}.`effective_date`", srcChk, Ord);
        // 另三对发散型影子列同样禁用
        foreach (var shadow in new[] { "id", "line_skill_master_id", "person_skill_id" })
            Assert.DoesNotContain($"{S0DimSqlBuilder.SrcAlias}.`{shadow}`", srcChk, Ord);
    }

    [Fact]
    public void LinePostSkill_parent_must_use_natural_key_not_surrogate()
    {
        // 代理键已污染：LineSkillRecID 同租户孤儿 210/261，PersonSkillId 250/261 跨租户。
        var d = S0DimCatalog.LinePostSkill;
        Assert.Equal("dim_line_post", d.ParentDimTable);
        Assert.Equal(["line_code", "job_no"], d.ParentKeyColumns);
        // 父键必须与 dim_line_post 业务键（去 tenant）逐字一致
        Assert.Equal(S0DimCatalog.LinePost.BusinessKeyColumns.Skip(1), d.ParentKeyColumns);
        // legacy id 只能是普通属性，绝不入键
        foreach (var legacy in new[] { "legacy_line_skill_rec_id", "legacy_person_skill_id" })
            Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
    }

    [Fact]
    public void EmpWorkDuty_identity_must_keep_location()
    {
        // 实测 tenant+Employee+Duty 有 1 组重复（G00308 同 Duty 下 Location 1001 vs 1000）。
        // location 是真实区分维，不是 legacy scope，不可简化掉。
        var d = S0DimCatalog.EmpWorkDuty;
        Assert.Equal(["tenant_id", "employee_code", "duty", "location_code"], d.BusinessKeyColumns);
        Assert.Equal(["Employee", "Duty", "Location"], d.SourceBizKeyColumns);
        Assert.Equal("dim_employee", d.ParentDimTable);
        Assert.Equal(["employee_code"], d.ParentKeyColumns);
    }

    [Fact]
    public void LinePost_run_crew_precision_must_match_source_decimal_10_3()
    {
        // 对账两侧都走 CAST(expr AS CHAR)：scale 不同会让 '0.000' 与 '0.0' 不等 → 校验和 FAIL。
        var col = S0DimCatalog.LinePost.Columns.Single(c => c.TargetColumn == "run_crew");
        Assert.Equal(S0DimValueKind.Dec, col.Kind);
        Assert.Equal(10, col.Precision);
        Assert.Equal(3, col.Scale);
    }

    [Fact]
    public void C2_and_C3_identities_must_not_contain_legacy_scope()
    {
        foreach (var key in new[] { "LABEL_TYPE", "SOP_FILE_TYPE", "QUALITY_BASE_TYPE",
                                    "SAMPLING_SCHEME", "INSPECTION_ITEM", "INSPECTION_METHOD",
                                    "INSPECTION_FREQUENCY", "INSPECTION_INSTRUMENT",
                                    "LINE_POST", "LINE_POST_SKILL" })
        {
            var d = S0DimCatalog.Find(key)!;
            foreach (var legacy in new[] { "domain_code", "company_id", "factory_id" })
                Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
        }
    }

    // ════════════════════════════════════════════════════════════════════════════
    // 同库集合式贴源快路径（S0DimSameDbStagingLoader / BuildSameDbStagingInsertSql）
    //
    // 这些断言钉死的是「快路径与旧逐行 MdpStagingWriter 的语义等价」。每一条对应一个
    // 在 aidopdev 上实测过的差异点；改动这些表达式前先看注释里的实测值。
    // ════════════════════════════════════════════════════════════════════════════

    /// <summary>构造一列源列描述，便于各用例复用。</summary>
    private static S0DimSourceColumn Col(string name, string dataType, string? columnType = null, bool nullable = true)
        => new(name, dataType, columnType ?? dataType, nullable);

    private static readonly IReadOnlySet<string> StgCols =
        new HashSet<string>(StringComparer.OrdinalIgnoreCase)
        { "tenant_id", "factory_id", "source_system", "source_table", "source_row_id",
          "source_biz_key", "raw_data", "sync_batch_id", "sync_time", "process_status", "create_time", "update_time" };

    private static string SameDbSql(params S0DimSourceColumn[] cols) =>
        S0DimSqlBuilder.BuildSameDbStagingInsertSql(
            S0DimCatalog.Item, cols, StgCols, ["RecID"]);

    [Fact]
    public void SameDb_unsigned_integer_json_type_label_is_downstream_irrelevant()
    {
        // 已知差异（如实登记，非缺陷）：bit(1) 与 `tinyint(1) unsigned zerofill` 在
        // 旧路径的 raw_data 里 JSON_TYPE = INTEGER，快路径则是 UNSIGNED INTEGER。
        //
        // 为什么不去消除它：downstream 只经 MdpJsonSql 读值，而它一律走
        // JSON_UNQUOTE(JSON_EXTRACT(...)) 再 CAST —— 两种标签下取出的文本完全相同。
        // 实证：同一批 16150 行真实数据，旧路径与快路径产出的 dim 校验和**逐位相同**
        // （9215906568012255809101），即该标签差异对 dim 契约零影响。
        //
        // 本断言钉死「MdpJsonSql 不依赖 JSON_TYPE」这一前提 —— 一旦有人改成按类型分支，
        // 上述等价性即不再成立，此测试会失败并把人引到这段注释。
        foreach (var expr in new[]
                 {
                     MdpJsonSql.Str("s", "X"), MdpJsonSql.Int("s", "X"),
                     MdpJsonSql.Dec("s", "X", 18, 5), MdpJsonSql.DateTimeSec("s", "X"),
                     MdpJsonSql.BoolTrue("s", "X"), MdpJsonSql.Raw("s", "X")
                 })
            Assert.DoesNotContain("JSON_TYPE", expr, Ord);
    }

    // ════════════════════════════════════════════════════════════════════════════
    // C3-A 物料桥（ProdLineDetail / RoutingOpDetail）
    // 每条断言都对应一个实测过的取错点；改这些映射前先看 S0DimCatalog 里的类型注释。
    // ════════════════════════════════════════════════════════════════════════════

    /// <summary>全部 6 张 C3 物料桥 —— 通则断言对每一张都必须成立。</summary>
    public static TheoryData<string> C3aItemBridges() =>
    [
        "ITEM_OP_LINE", "ITEM_ROUTING_OP",
        "ITEM_OP_CONDITION", "ITEM_OP_IMAGE",
        "ITEM_BOM", "ITEM_SUBSTITUTE"
    ];

    [Theory][MemberData(nameof(C3aItemBridges))]
    public void C3a_bridges_must_parent_dim_item_by_natural_key(string key)
    {
        // 父连接走 p.<col> <=> d.<col> **同名列** join（S0DimSqlBuilder.OrphanFromWhere），
        // 而 dim_item 的身份列叫 item_code —— 子表若命名为 item_num，Validate() 会过、
        // 运行时 join 报未知列。
        var d = S0DimCatalog.All.Single(x => x.Key == key);
        Assert.Equal("dim_item", d.ParentDimTable);
        Assert.Equal(["item_code"], d.ParentKeyColumns);
        Assert.Contains("item_code", d.BusinessKeyColumns);
        // 绝不能用 dim 代理 id 建关系：每轮 FULL REPLACE 后 dim_item.id 都会变
        Assert.DoesNotContain("dim_item_id", d.Columns.Select(c => c.TargetColumn));
    }

    [Theory][MemberData(nameof(C3aItemBridges))]
    public void C3a_identities_must_not_contain_legacy_scope(string key)
    {
        var d = S0DimCatalog.All.Single(x => x.Key == key);
        foreach (var legacy in new[] { "domain_code", "company_id", "factory_id", "work_center_code", "route_code" })
            Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
        Assert.Equal("tenant_id", d.BusinessKeyColumns[0]);
    }

    [Fact]
    public void ItemOpLine_must_read_real_equipment_column_not_the_dead_shadow()
    {
        // 真列 InternalEquipmentCode 有 2430/10646 行有值；
        // 同表 internal_equipment_type_code 是 0/10646 的死影子列，取错会全 NULL 且不报错。
        var paths = S0DimCatalog.ItemOpLine.Columns.Select(c => c.JsonPath ?? "").ToList();
        Assert.Contains("InternalEquipmentCode", paths);
        Assert.DoesNotContain("internal_equipment_type_code", paths);
        var sql = S0DimSqlBuilder.BuildInsertSql(S0DimCatalog.ItemOpLine);
        Assert.DoesNotContain("internal_equipment_type_code", sql, Ord);
    }

    [Fact]
    public void ItemOpLine_run_crew_must_be_string_not_decimal()
    {
        // 🔴 ProdLineDetail.RunCrew 是 varchar(8)；LinePost 的同名列源自 LineSkillMaster
        // 是 decimal(10,3)。同名不同型，照抄会当场类型出错。
        Assert.Equal(S0DimValueKind.Str,
            S0DimCatalog.ItemOpLine.Columns.Single(c => c.TargetColumn == "run_crew").Kind);
        Assert.Equal(S0DimValueKind.Dec,
            S0DimCatalog.LinePost.Columns.Single(c => c.TargetColumn == "run_crew").Kind);
    }

    [Fact]
    public void ItemOpLine_line_code_must_stay_in_identity()
    {
        // 去掉 Line 后 tenant+Part+Op 有 197 组 / 394 行重复 —— 不可逆的键收缩
        Assert.Equal(["tenant_id", "item_code", "operation_no", "line_code"],
            S0DimCatalog.ItemOpLine.BusinessKeyColumns);
        Assert.Equal(["Part", "Op", "Line"], S0DimCatalog.ItemOpLine.SourceBizKeyColumns);
    }

    [Fact]
    public void ItemRoutingOp_watermark_must_read_UpdatedAt_not_UpdateTime()
    {
        // 🔴 本表时间列叫 UpdatedAt/CreatedAt。写成全仓通用的 UpdateTime 不会报错，
        // JSON 键不存在 → JSON_EXTRACT 返回 SQL NULL → 该列恒 NULL，静默失真。
        Assert.Equal("UpdatedAt",
            S0DimCatalog.ItemRoutingOp.Columns.Single(c => c.TargetColumn == "source_updated_at").JsonPath);
        Assert.DoesNotContain("UpdateTime", S0DimCatalog.ItemRoutingOp.Columns.Select(c => c.JsonPath ?? ""));
    }

    public static TheoryData<string> RoutingOpForbiddenColumns() =>
        // 六组新旧列近重复 + 两类不可物化的布尔。取错得到的不是 NULL 而是「看起来合法的错值」。
        ["RoutingCode", "Descr", "MilestoneOperation", "MilestoneOp", "IsActive", "IsEnabled", "UpdateTime"];

    [Theory][MemberData(nameof(RoutingOpForbiddenColumns))]
    public void ItemRoutingOp_must_never_read_legacy_or_unmaterializable_columns(string forbidden)
    {
        Assert.DoesNotContain(forbidden, S0DimCatalog.ItemRoutingOp.Columns.Select(c => c.JsonPath ?? ""));
    }

    [Fact]
    public void ItemRoutingOp_std_op_must_be_attribute_with_no_parent()
    {
        // StdOp 对 StdOpMaster 孤儿 37112/38746，CRUD 不写、UI 无下拉、码空间不同构。
        // 它与 item 孤儿（仅 44）是两回事，勿混淆。框架只能挂一个父，父是 dim_item。
        var d = S0DimCatalog.ItemRoutingOp;
        Assert.Contains("StdOp", d.Columns.Select(c => c.JsonPath ?? ""));
        Assert.DoesNotContain("std_op", d.BusinessKeyColumns);
        Assert.Equal("dim_item", d.ParentDimTable);
    }

    [Fact]
    public void ItemRoutingOp_decimal_scales_must_match_source_exactly()
    {
        // UDeci1..5 源 decimal(18,5)，RunTime 源 decimal(18,6) —— 不得统一，
        // 精度不一致会让 CAST AS CHAR 后的 checksum 两侧不等。
        var d = S0DimCatalog.ItemRoutingOp;
        foreach (var c in new[] { "udeci1", "udeci2", "udeci3", "udeci4", "udeci5" })
        {
            var col = d.Columns.Single(x => x.TargetColumn == c);
            Assert.Equal((18, 5), (col.Precision, col.Scale));
        }
        var rt = d.Columns.Single(x => x.TargetColumn == "run_time");
        Assert.Equal((18, 6), (rt.Precision, rt.Scale));
    }

    [Fact]
    public void ItemOpLine_decimal_scales_must_all_be_18_6()
    {
        foreach (var c in new[] { "rate", "setup_time", "overlap_time", "sequence_no", "standard_staff_count" })
        {
            var col = S0DimCatalog.ItemOpLine.Columns.Single(x => x.TargetColumn == c);
            Assert.Equal((18, 6), (col.Precision, col.Scale));
        }
    }

    [Fact]
    public void ItemRoutingOp_domain_must_not_be_required()
    {
        // RecID=106435 那行 TEST 数据的 Domain 是 NULL；设成 Required 会让该行被 RowGuards
        // 整行滤掉 → dim_count < stg_count → 对账 FAIL。
        Assert.False(S0DimCatalog.ItemRoutingOp.Columns.Single(c => c.TargetColumn == "domain_code").Required);
    }

    [Fact]
    public void ItemOpCondition_is_main_must_be_booltrue_attribute_never_key()
    {
        // 源 tinyint(1) 在 raw_data 里是 JSON BOOLEAN。Int 会把 true/false 双双静默压成 0
        // （比 bit(1) 地雷更毒：不报错、不为 NULL）；Str 会得到 'true' 而源侧 checksum 是 '1'。
        var d = S0DimCatalog.ItemOpCondition;
        Assert.Equal(S0DimValueKind.BoolTrue, d.Columns.Single(c => c.TargetColumn == "is_main").Kind);
        Assert.DoesNotContain("is_main", d.BusinessKeyColumns);
        Assert.DoesNotContain("IsMain", d.SourceBizKeyColumns);
    }

    [Fact]
    public void ItemOpCondition_element_code_must_not_be_confused_with_item_code()
    {
        // $.ItemCode 是「要素编码」，$.ItemNum 是「物料编码」，实测两列 0 行相等、前者 5/10 为空。
        var d = S0DimCatalog.ItemOpCondition;
        Assert.Equal("ItemCode", d.Columns.Single(c => c.TargetColumn == "element_code").JsonPath);
        Assert.Equal("ItemNum",  d.Columns.Single(c => c.TargetColumn == "item_code").JsonPath);
    }

    [Fact]
    public void ItemOpImage_must_not_declare_watermark_because_source_has_no_time_column()
    {
        // QualitySegmentImage 11 列里没有任何时间列（实测）。为字段齐整回填会得到恒 NULL 的假列。
        Assert.DoesNotContain("source_updated_at",
            S0DimCatalog.ItemOpImage.Columns.Select(c => c.TargetColumn));
        // 对照：ItemOpCondition 的源有 UpdateTime，故必须有
        Assert.Contains("source_updated_at",
            S0DimCatalog.ItemOpCondition.Columns.Select(c => c.TargetColumn));
    }

    [Fact]
    public void ItemBom_legacy_surrogate_ids_must_stay_out_of_identity()
    {
        // 🔴 实测跨租户泄漏向量：ComponentMaterialId → ItemMaster.RecID
        // 同租户失败 170 / 忽略租户失败 0，而那 170 行恰恰全挂到别的租户；
        // ParentMaterialId 同款（同租户失败 2060）。自然键则 0/62036 零孤儿且 61866/61866 一致。
        var d = S0DimCatalog.ItemBom;
        foreach (var legacy in new[] { "legacy_parent_item_id", "legacy_component_item_id", "company_id", "factory_id" })
            Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
        Assert.Equal(["ParentItem", "ComponentItem"], d.SourceBizKeyColumns);
        // 但必须留痕，供 legacy 对账
        Assert.Contains("legacy_component_item_id", d.Columns.Select(c => c.TargetColumn));
    }

    [Fact]
    public void ItemBom_must_not_materialize_nullable_bit_or_zero_information_columns()
    {
        // IsActive 是 bit(1) 且含 1 行 NULL —— Int 让对账必挂、BoolTrue 会滤掉那行。
        // IsEnabled/SequenceNum/StructureType distinct=1；LineOp/Remark/StartEff/EndEff 全库 0 填充。
        var paths = S0DimCatalog.ItemBom.Columns.Select(c => c.JsonPath ?? "").ToList();
        foreach (var forbidden in new[]
                 { "IsActive", "IsEnabled", "SequenceNum", "StructureType", "LineOp", "Remark", "StartEff", "EndEff" })
            Assert.DoesNotContain(forbidden, paths);
    }

    [Fact]
    public void ItemSubstitute_all_three_item_refs_must_target_dim_item_naturally()
    {
        // 三个 item 引用全部统一到 dim_item 的自然键；框架单父，取被替代的 ItemNum 作 item_code。
        var d = S0DimCatalog.ItemSubstitute;
        var paths = d.Columns.ToDictionary(c => c.TargetColumn, c => c.JsonPath);
        Assert.Equal("ParentItem",     paths["parent_item_code"]);
        Assert.Equal("ItemNum",        paths["item_code"]);
        Assert.Equal("SubstituteItem", paths["substitute_item_code"]);
        Assert.Equal("dim_item", d.ParentDimTable);
        Assert.Equal(["item_code"], d.ParentKeyColumns);
    }

    [Fact]
    public void ItemSubstitute_parent_item_must_stay_in_identity()
    {
        // ParentItem 与 ItemNum 实测 16 行无一相等（distinct 9 vs 14），是真实不同信息。
        // 全表仅 16 行，「去掉后仍唯一」证明不了可去 —— 去掉即不可逆的键收缩。
        Assert.Contains("parent_item_code", S0DimCatalog.ItemSubstitute.BusinessKeyColumns);
        Assert.Equal(["ParentItem", "ItemNum", "SubstituteItem"], S0DimCatalog.ItemSubstitute.SourceBizKeyColumns);
    }

    [Fact]
    public void SameDb_source_column_must_be_sqlsugar_mappable()
    {
        // 🔴 SqlSugar 的 IDataReaderEntityBuilder.CreateBuilder 走反射构造，
        // 对只有主构造函数的 positional record 会抛 ArgumentNullException (Parameter 'con')。
        // 该异常会被 S0DimSameDbStagingLoader 吞掉并静默回退逐行路径 ——
        // **表现为「变慢」而不是「报错」**，实测踩过一次。此断言让回归变成编译期/单测期可见。
        var t = typeof(S0DimSourceColumn);
        Assert.NotNull(t.GetConstructor(Type.EmptyTypes));
        foreach (var prop in new[] { "Name", "DataType", "ColumnType", "IsNullable" })
        {
            var p = t.GetProperty(prop);
            Assert.NotNull(p);
            Assert.True(p!.CanWrite, $"{prop} 必须可写，否则 SqlSugar 无法回填");
        }
    }

    [Fact]
    public void SameDb_tinyint1_must_emit_json_boolean_not_integer()
    {
        // 旧路径：MySqlConnector 的 TreatTinyAsBoolean 把 tinyint(1) 变成 CLR bool
        // → JsonSerializer 产出 JSON BOOLEAN（实测 mdp_stg_s0_item.$.IsActive = BOOLEAN "true"）。
        // JSON_OBJECT 直出会是 INTEGER 1 —— 必须用 CAST(... AS JSON) 纠正。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("IsActive", "tinyint", "tinyint(1)"));
        Assert.Contains("CAST(IF(a.`IsActive` IS NULL, NULL, IF(a.`IsActive` <> 0, 'true', 'false')) AS JSON)", sql, Ord);
    }

    [Fact]
    public void SameDb_zerofill_tinyint1_must_stay_integer()
    {
        // 🔴 TreatTinyAsBoolean 对 `tinyint(1) unsigned zerofill` **不生效**
        // （实测 $.enable_bom_version = INTEGER 0，而 $.IsActive = BOOLEAN）。
        // 判定必须比对 COLUMN_TYPE 全串而非 DATA_TYPE，否则这一列会被错误地转成布尔。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false),
                            Col("enable_bom_version", "tinyint", "tinyint(1) unsigned zerofill"));
        Assert.DoesNotContain("`enable_bom_version` IS NULL, NULL, IF(", sql, Ord);
        Assert.Contains("'enable_bom_version', a.`enable_bom_version`", sql, Ord);
    }

    [Fact]
    public void SameDb_bit_must_cast_unsigned_never_char()
    {
        // bit(1) 在旧路径是 INTEGER（实测 $.SuppWarranty = INTEGER 0）。
        // 若直出，JSON_OBJECT 会得到裸字节；CAST AS CHAR 更糟（0x01 而非 '1'，见 S0DimValueKind.Int 注释）。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("SuppWarranty", "bit", "bit(1)"));
        Assert.Contains("'SuppWarranty', CAST(a.`SuppWarranty` AS UNSIGNED)", sql, Ord);
    }

    [Fact]
    public void SameDb_decimal_must_emit_double_not_decimal()
    {
        // 旧路径 decimal → CLR decimal → JSON DOUBLE «0.0»；JSON_OBJECT 直出是 DECIMAL «0.00000»。
        // JSON_UNQUOTE 后字符串不同 → staging 侧任何按原文比对都会差异。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("NetWeight", "decimal", "decimal(18,5)"));
        Assert.Contains("'NetWeight', CAST(a.`NetWeight` AS DOUBLE)", sql, Ord);
    }

    [Fact]
    public void SameDb_datetime_must_emit_mysql_literal_string()
    {
        // 旧路径用 MysqlLiteralDateTimeConverter 写成 STRING "yyyy-MM-dd HH:mm:ss.ffffff"
        // （实测 $.UpdateTime = STRING "2023-12-18 13:57:30.000000"）；JSON_OBJECT 直出是 JSON DATETIME。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("UpdateTime", "datetime"));
        Assert.Contains("'UpdateTime', DATE_FORMAT(a.`UpdateTime`, '%Y-%m-%d %H:%i:%s.%f')", sql, Ord);
    }

    [Fact]
    public void SameDb_string_columns_must_pass_through_untouched()
    {
        // 空串必须保持空串（不能被当成 NULL）；字符类直出即 JSON STRING，与旧路径一致。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("Descr", "varchar", "varchar(200)"));
        Assert.Contains("'Descr', a.`Descr`", sql, Ord);
    }

    [Fact]
    public void SameDb_biz_key_must_fall_back_to_row_id_when_any_part_is_null_or_blank()
    {
        // 🔴 MdpStagingWriter.BuildBizKey：任一分量 null/空串 → 整个 biz key 作废 → 回落 source_row_id。
        // 直接写 CONCAT_WS 会**跳过** NULL 分量而产出「看起来合法」的短键，与旧路径不等价，
        // 且要到下一轮四向集合比对才炸。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("ItemNum", "varchar", "varchar(100)"));
        Assert.Contains("CASE WHEN (a.`ItemNum` IS NOT NULL AND CAST(a.`ItemNum` AS CHAR) <> '')", sql, Ord);
        Assert.Contains("ELSE IF(a.`RecID` IS NULL OR TRIM(CAST(a.`RecID` AS CHAR)) = '', NULL, CAST(a.`RecID` AS CHAR)) END", sql, Ord);
    }

    [Fact]
    public void SameDb_row_id_candidate_order_must_match_shared_pull_executor()
    {
        // 顺序与 MdpDbPullExecutor.ResolveSourceRowId 逐字一致；错序会让同一源行在两条路径下
        // 得到不同 source_row_id，贴源唯一键随之漂移。
        Assert.Equal(
            ["RecID", "RecId", "recid", "Recid", "id", "Id", "ID", "noid", "billno", "BillNo", "djbh", "Djbh"],
            S0DimSqlBuilder.SourceRowIdCandidates);
    }

    [Fact]
    public void SameDb_sql_must_never_contain_nondeterministic_uuid()
    {
        // 旧路径在全部候选都空白时兜底 Guid.NewGuid()。快路径不复刻该分支，
        // 而是由 Loader 要求「首个候选是 NOT NULL 整数」后判定适用，从源头排除。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("ItemNum", "varchar", "varchar(100)"));
        Assert.DoesNotContain("UUID(", sql, StringComparison.OrdinalIgnoreCase);
    }

    [Fact]
    public void SameDb_sql_must_carry_structural_tenant_gate()
    {
        // 租户硬门必须结构性存在，且绝不允许出现 ctx 兜底常量
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("ItemNum", "varchar", "varchar(100)"));
        Assert.Contains("WHERE a.`tenant_id` = @TenantId", sql, Ord);
        Assert.Contains("`tenant_id`", sql, Ord);
        Assert.DoesNotContain("tenant_id = 0", sql, Ord);
        Assert.DoesNotContain("IFNULL(a.`tenant_id`", sql, Ord);
    }

    [Fact]
    public void SameDb_supported_types_must_be_an_allowlist_excluding_binary_and_json()
    {
        // 白名单而非黑名单：未知类型一律回退旧路径，不猜其序列化形态。
        // blob 在旧路径是 byte[]→base64，JSON_OBJECT 会直出原始字节，静默失真。
        foreach (var ok in new[] { "varchar", "bigint", "datetime", "tinyint", "decimal", "int", "bit" })
            Assert.Contains(ok, S0DimSqlBuilder.SameDbSupportedDataTypes);
        foreach (var no in new[] { "blob", "varbinary", "binary", "json", "enum", "set", "time", "geometry" })
            Assert.DoesNotContain(no, S0DimSqlBuilder.SameDbSupportedDataTypes);
    }

    [Fact]
    public void SameDb_must_preserve_full_semantics_with_upsert()
    {
        // 旧 MdpStagingWriter 是 upsert；保留 ON DUPLICATE KEY UPDATE 才能在源侧出现重复
        // source_row_id 这类异常数据下两条路径同构，而不是一条报错一条 last-wins。
        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("ItemNum", "varchar", "varchar(100)"));
        Assert.Contains("ON DUPLICATE KEY UPDATE", sql, Ord);
        Assert.Contains("`process_status`='PENDING'", sql, Ord);
    }

    [Fact]
    public void AntiJoin_must_not_use_nested_loop_left_join_shape()
    {
        // 🔴 LEFT JOIN ... ON y.bk <=> x.bk 在「右侧是 CONCAT_WS 计算列」的方向上退化成 nested loop：
        // 16150 行的 stg\src 实测 130.3s（rows=32,762,720），撞穿 CommandTimeOut=30；
        // 而对称的 src\stg 只要 0.5s —— 缺陷只在一半方向显形，小表永远发现不了。
        var sql = S0DimSqlBuilder.BuildAntiJoinCountSql("SELECT 1 AS bk", "SELECT 2 AS bk");
        Assert.DoesNotContain("LEFT JOIN", sql, Ord);
        Assert.Contains("UNION ALL", sql, Ord);
        Assert.Contains("GROUP BY u.bk", sql, Ord);
        // 必须按左行计数（SUM(l_cnt)）而非去重键数，才与旧 COUNT(*) 语义一致
        Assert.Contains("SUM(CASE WHEN g.r_cnt = 0 THEN g.l_cnt ELSE 0 END)", sql, Ord);
    }

    [Fact]
    public void Item_identity_must_be_tenant_and_item_code_only()
    {
        // 旧 UK uk_ItemMaster_factory_item 含 factory，不予继承：factory_ref_id 连 f(tenant) 都不是
        // （838257186181189 有 3 个取值含 0），且 8010 被三个租户共用。
        var d = S0DimCatalog.Item;
        Assert.Equal(["tenant_id", "item_code"], d.BusinessKeyColumns);
        Assert.Equal(["ItemNum"], d.SourceBizKeyColumns);
        foreach (var legacy in new[] { "domain_code", "company_id", "factory_id", "legacy_source_id" })
            Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
        Assert.Null(d.ParentDimTable);
    }

    [Fact]
    public void Item_legacy_source_id_is_technical_attribute_only()
    {
        // 供 C3 的 ComponentMaterialId 这类 legacy 代理键做对账解析；绝不是身份。
        // 也绝不能用 dim 的代理 id 建外键 —— 每轮 FULL REPLACE 后 dim_item.id 都会变。
        var col = S0DimCatalog.Item.Columns.Single(c => c.TargetColumn == "legacy_source_id");
        Assert.Equal("RecID", col.JsonPath);
        Assert.Equal(S0DimValueKind.Int, col.Kind);
    }

    [Fact]
    public void Item_must_not_materialize_bit_or_nullable_bool_or_shadow_columns()
    {
        // SuppWarranty 是 bit(1) 且含 2 行 NULL —— Int 会让对账必挂、BoolTrue 会滤掉那 2 行。
        // DEAControlled 是可空 tinyint(1)，同样会被 BoolTrue 守卫滤行。
        // expire_alarm_day 是 0 填充的影子列，真列为 ExpireAlarmDays。
        var paths = S0DimCatalog.Item.Columns.Select(c => c.JsonPath ?? "").ToList();
        foreach (var forbidden in new[] { "SuppWarranty", "DEAControlled", "expire_alarm_day" })
            Assert.DoesNotContain(forbidden, paths);
        Assert.Contains("ExpireAlarmDays", paths);
    }

    [Fact]
    public void Item_must_precede_c3_definitions_in_catalog_order()
    {
        // 后续 6 张 C3 都要挂 dim_item 作父，父必须先就绪
        var keys = S0DimCatalog.All.Select(d => d.Key).ToList();
        var item = keys.IndexOf("ITEM");
        Assert.True(item >= 0);
        foreach (var child in new[] { "EMP_WORK_DUTY", "LINE_POST", "LINE_POST_SKILL" })
            Assert.True(item < keys.IndexOf(child), $"ITEM 必须排在 {child} 之前");
    }

    [Fact]
    public void Batch4_must_not_reuse_dopdemorq_staging_tables()
    {
        // mdp_stg_employee / mdp_stg_line 属 DOPDEMORQ（当前 0 行），不是 S0 的 authority
        foreach (var d in S0DimCatalog.All)
        {
            Assert.NotEqual("mdp_stg_employee", d.StagingTable);
            Assert.NotEqual("mdp_stg_line", d.StagingTable);
            Assert.NotEqual("mdp_stg_location", d.StagingTable);
            Assert.NotEqual("mdp_stg_customer", d.StagingTable);
        }
    }
}