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(
            ["WORK_CENTER", "DEPARTMENT", "LOCATION", "LOCATION_SHELF", "EMPLOYEE", "LINE", "CUSTOMER", "PERSON_SKILL"],
            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 ?? ""));
    }

    [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);
        }
    }
}