lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlCatalogGateTests.cs

using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;

namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests;

/// <summary>
///     The design §8.1 catalog gate (Gitea #496), tested as a pure function over a hand-built
///     <see cref="SqlCatalog"/>. The end-to-end behaviour against a real catalog — and the proof that a
///     rejected tag keeps its node and publishes <c>BadNodeIdUnknown</c> — lives in
///     <see cref="SqlCatalogGateDriverTests"/>.
/// </summary>
public sealed class SqlCatalogGateTests
{
    private static readonly SqliteDialect Dialect = new();

    /// <summary>A catalog with one schema, two relations, and deliberately mixed-case column spellings.</summary>
    private static SqlCatalog Catalog(string defaultSchema = "dbo") => new(
        defaultSchema,
        ["dbo", "mes"],
        new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal)
        {
            ["dbo"] = ["TagValues", "LatestStatus"],
            ["mes"] = ["Orders"],
        },
        new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal)
        {
            ["dbo.TagValues"] = ["tag_name", "num_value", "sample_ts"],
            ["dbo.LatestStatus"] = ["station_id", "oven_temp", "pressure", "sample_ts"],
            ["mes.Orders"] = ["order_id", "qty"],
        });

    private static SqlTagDefinition KeyValueTag(
        string table = "dbo.TagValues",
        string keyColumn = "tag_name",
        string valueColumn = "num_value",
        string? timestampColumn = "sample_ts") =>
        new("plant/sql/Speed", SqlTagModel.KeyValue, table,
            KeyColumn: keyColumn, KeyValue: "Line1.Speed",
            ValueColumn: valueColumn, TimestampColumn: timestampColumn);

    private static SqlCatalogGateResult Apply(params SqlTagDefinition[] tags) =>
        SqlCatalogGate.Apply(tags, Catalog(), Dialect);

    [Fact]
    public void A_fully_resolvable_tag_is_accepted()
    {
        var result = Apply(KeyValueTag());

        result.Rejected.ShouldBeEmpty();
        result.Accepted.Count.ShouldBe(1);
    }

    /// <summary>
    ///     The heart of §8.1: what reaches the planner must be a string the catalog gave us, not the string an
    ///     operator typed. Authoring every identifier in the wrong case proves the substitution actually
    ///     happens rather than the input merely being waved through.
    /// </summary>
    [Fact]
    public void Accepted_identifiers_are_rewritten_to_the_catalogs_own_spelling()
    {
        var authored = KeyValueTag(
            table: "DBO.TAGVALUES", keyColumn: "TAG_NAME", valueColumn: "Num_Value", timestampColumn: "SAMPLE_TS");

        var accepted = Apply(authored).Accepted.ShouldHaveSingleItem();

        accepted.Table.ShouldBe("dbo.TagValues");
        accepted.KeyColumn.ShouldBe("tag_name");
        accepted.ValueColumn.ShouldBe("num_value");
        accepted.TimestampColumn.ShouldBe("sample_ts");
        // Identity and bound VALUES are untouched — the gate rewrites identifiers only.
        accepted.Name.ShouldBe(authored.Name);
        accepted.KeyValue.ShouldBe(authored.KeyValue);
    }

    [Fact]
    public void An_unqualified_table_resolves_in_the_default_schema()
    {
        var accepted = Apply(KeyValueTag(table: "TagValues")).Accepted.ShouldHaveSingleItem();

        accepted.Table.ShouldBe("dbo.TagValues");
    }

    /// <summary>
    ///     Guessing <c>dbo</c> would be a silent lie on an estate that maps service accounts to their own
    ///     default schema, so the gate must resolve an unqualified name in whatever schema the server reports.
    /// </summary>
    [Fact]
    public void An_unqualified_table_follows_a_non_dbo_default_schema()
    {
        var catalog = Catalog(defaultSchema: "mes");

        var result = SqlCatalogGate.Apply([KeyValueTag(table: "Orders", keyColumn: "order_id", valueColumn: "qty", timestampColumn: null)], catalog, Dialect);

        result.Accepted.ShouldHaveSingleItem().Table.ShouldBe("mes.Orders");
    }

    [Fact]
    public void An_unknown_table_rejects_the_tag()
    {
        var rejection = Apply(KeyValueTag(table: "dbo.NoSuchTable")).Rejected.ShouldHaveSingleItem();

        rejection.RawPath.ShouldBe("plant/sql/Speed");
        rejection.Field.ShouldBe(nameof(SqlTagDefinition.Table));
        rejection.Reason.ShouldContain("NoSuchTable");
    }

    [Fact]
    public void An_unknown_schema_rejects_the_tag()
    {
        var rejection = Apply(KeyValueTag(table: "nope.TagValues")).Rejected.ShouldHaveSingleItem();

        rejection.Field.ShouldBe(nameof(SqlTagDefinition.Table));
        rejection.Reason.ShouldContain("nope");
    }

    [Fact]
    public void An_unknown_column_rejects_the_tag_and_names_the_field()
    {
        var rejection = Apply(KeyValueTag(valueColumn: "no_such_column")).Rejected.ShouldHaveSingleItem();

        rejection.Field.ShouldBe(nameof(SqlTagDefinition.ValueColumn));
        rejection.Reason.ShouldContain("no_such_column");
        rejection.Reason.ShouldContain("dbo.TagValues");
    }

    /// <summary>
    ///     A table in the right catalog but the wrong schema must not resolve — otherwise the gate would
    ///     allow-list a column set from a relation the query will never read.
    /// </summary>
    [Fact]
    public void A_table_from_another_schema_does_not_resolve_unqualified()
    {
        var rejection = Apply(KeyValueTag(table: "Orders", keyColumn: "order_id", valueColumn: "qty", timestampColumn: null))
            .Rejected.ShouldHaveSingleItem();

        rejection.Field.ShouldBe(nameof(SqlTagDefinition.Table));
    }

    /// <summary>
    ///     A cross-database or linked-server name addresses a catalog this connection cannot enumerate, so it
    ///     cannot be allow-listed. Rejecting is the fail-closed answer, and the message says what to do instead.
    /// </summary>
    [Fact]
    public void A_three_part_name_is_rejected_with_an_actionable_message()
    {
        var rejection = Apply(KeyValueTag(table: "otherdb.dbo.TagValues")).Rejected.ShouldHaveSingleItem();

        rejection.Field.ShouldBe(nameof(SqlTagDefinition.Table));
        rejection.Reason.ShouldContain("3-part");
        rejection.Reason.ShouldContain("view");
    }

    /// <summary>
    ///     The injection shape the whole gate exists for: a hostile identifier must be refused by the
    ///     allow-list, not merely quoted into a query against a nonexistent object.
    /// </summary>
    [Theory]
    [InlineData("TagValues]; DROP TABLE TagValues--")]
    [InlineData("'; DROP TABLE TagValues--")]
    [InlineData("TagValues WHERE 1=1 OR 1=1")]
    public void A_hostile_table_name_is_rejected_by_the_allow_list(string table)
    {
        Apply(KeyValueTag(table: table)).Rejected.ShouldHaveSingleItem()
            .Field.ShouldBe(nameof(SqlTagDefinition.Table));
    }

    [Theory]
    [InlineData("num_value]; DROP TABLE TagValues--")]
    [InlineData("(SELECT password FROM users)")]
    public void A_hostile_column_name_is_rejected_by_the_allow_list(string column)
    {
        Apply(KeyValueTag(valueColumn: column)).Rejected.ShouldHaveSingleItem()
            .Field.ShouldBe(nameof(SqlTagDefinition.ValueColumn));
    }

    /// <summary>
    ///     A name carrying a control or Unicode format character cannot be safely rendered into a log line
    ///     (the Trojan-Source class, CVE-2021-42574), so the gate rejects it <em>without echoing it</em> — the
    ///     charset check runs before the catalog lookup precisely so no such string can reach a message.
    /// </summary>
    /// <remarks>
    ///     Driven against <see cref="SqlServerDialect"/>, not the test-only <see cref="SqliteDialect"/>: the
    ///     charset rules belong to the dialect (the gate delegates to
    ///     <see cref="ISqlDialect.QuoteIdentifier"/> rather than duplicating them), and SQLite's rules
    ///     deliberately stop at control characters. Asserting SQL Server's rules means using SQL Server's
    ///     dialect — the first draft of this test asserted them against SQLite's and failed for that reason.
    /// </remarks>
    [Theory]
    [InlineData("num\u0000value")]   // Cc — embedded NUL
    [InlineData("num\u0009value")]   // Cc — tab; truncates or corrupts a logged statement
    [InlineData("num\u202Evalue")]   // Cf — right-to-left override
    [InlineData("num\u200Bvalue")]   // Cf — zero-width space
    public void An_unrenderable_identifier_is_rejected_without_being_echoed(string column)
    {
        var result = SqlCatalogGate.Apply(
            [KeyValueTag(valueColumn: column)], Catalog(), new SqlServerDialect());

        var rejection = result.Rejected.ShouldHaveSingleItem();
        rejection.Field.ShouldBe(nameof(SqlTagDefinition.ValueColumn));
        rejection.Reason.ShouldContain("withheld");
        rejection.Reason.ShouldNotContain(column);
    }

    /// <summary>
    ///     An over-long name cannot name a real object and is likewise withheld, rather than pasting an
    ///     unbounded slab of operator input into a log line.
    /// </summary>
    [Fact]
    public void An_over_long_identifier_is_rejected_without_being_echoed()
    {
        var column = new string('x', SqlServerDialect.MaxIdentifierLength + 1);

        var result = SqlCatalogGate.Apply(
            [KeyValueTag(valueColumn: column)], Catalog(), new SqlServerDialect());

        result.Rejected.ShouldHaveSingleItem().Reason.ShouldContain("withheld");
    }

    /// <summary>
    ///     The complement, and the reason the charset check is not simply "withhold everything": a name that
    ///     IS safely renderable gets echoed, because an operator hunting a typo has to see what they wrote.
    /// </summary>
    [Fact]
    public void A_renderable_but_unknown_identifier_IS_echoed_so_the_typo_is_findable()
    {
        var rejection = Apply(KeyValueTag(valueColumn: "num_valeu")).Rejected.ShouldHaveSingleItem();

        rejection.Reason.ShouldContain("num_valeu");
        rejection.Reason.ShouldNotContain("withheld");
    }

    /// <summary>
    ///     On a case-sensitive collation a relation may legitimately carry both <c>Value</c> and <c>value</c>.
    ///     Picking one would publish a different column's data under the operator's node, so the only safe
    ///     answer is to refuse — but an EXACT match must still win, or a valid config would break.
    /// </summary>
    [Fact]
    public void An_ambiguous_case_insensitive_column_match_is_rejected_but_an_exact_match_still_wins()
    {
        var catalog = new SqlCatalog(
            "dbo",
            ["dbo"],
            new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal) { ["dbo"] = ["T"] },
            new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal)
            {
                ["dbo.T"] = ["k", "Value", "value"],
            });

        var ambiguous = new SqlTagDefinition(
            "p/Ambiguous", SqlTagModel.KeyValue, "dbo.T",
            KeyColumn: "k", KeyValue: "x", ValueColumn: "VALUE");
        SqlCatalogGate.Apply([ambiguous], catalog, Dialect).Rejected.ShouldHaveSingleItem()
            .Field.ShouldBe(nameof(SqlTagDefinition.ValueColumn));

        var exact = ambiguous with { Name = "p/Exact", ValueColumn = "value" };
        SqlCatalogGate.Apply([exact], catalog, Dialect).Accepted.ShouldHaveSingleItem()
            .ValueColumn.ShouldBe("value");
    }

    /// <summary>
    ///     One operator typo must not stop the other tags on the same database — the same rule the tag-table
    ///     build already follows for a malformed blob.
    /// </summary>
    [Fact]
    public void A_rejected_tag_does_not_take_its_healthy_neighbours_with_it()
    {
        var good = KeyValueTag() with { Name = "plant/sql/Good" };
        var bad = KeyValueTag(valueColumn: "typo") with { Name = "plant/sql/Bad" };

        var result = Apply(good, bad);

        result.Accepted.ShouldHaveSingleItem().Name.ShouldBe("plant/sql/Good");
        result.Rejected.ShouldHaveSingleItem().RawPath.ShouldBe("plant/sql/Bad");
    }

    /// <summary>Every identifier-bearing field is checked, not just the two the key-value model happens to use.</summary>
    [Fact]
    public void The_wide_row_models_identifier_fields_are_validated_too()
    {
        var selectorTypo = new SqlTagDefinition(
            "p/Oven", SqlTagModel.WideRow, "dbo.LatestStatus",
            ColumnName: "oven_temp", RowSelectorColumn: "no_such_selector", RowSelectorValue: "7");
        Apply(selectorTypo).Rejected.ShouldHaveSingleItem()
            .Field.ShouldBe(nameof(SqlTagDefinition.RowSelectorColumn));

        var orderTypo = new SqlTagDefinition(
            "p/Newest", SqlTagModel.WideRow, "dbo.LatestStatus",
            ColumnName: "oven_temp", RowSelectorTopByTimestamp: "no_such_ts");
        Apply(orderTypo).Rejected.ShouldHaveSingleItem()
            .Field.ShouldBe(nameof(SqlTagDefinition.RowSelectorTopByTimestamp));

        var columnTypo = new SqlTagDefinition(
            "p/Bad", SqlTagModel.WideRow, "dbo.LatestStatus",
            ColumnName: "no_such_column", RowSelectorColumn: "station_id", RowSelectorValue: "7");
        Apply(columnTypo).Rejected.ShouldHaveSingleItem()
            .Field.ShouldBe(nameof(SqlTagDefinition.ColumnName));

        var ok = new SqlTagDefinition(
            "p/Ok", SqlTagModel.WideRow, "dbo.LatestStatus",
            ColumnName: "OVEN_TEMP", RowSelectorColumn: "STATION_ID", RowSelectorValue: "7");
        var accepted = Apply(ok).Accepted.ShouldHaveSingleItem();
        accepted.ColumnName.ShouldBe("oven_temp");
        accepted.RowSelectorColumn.ShouldBe("station_id");
        accepted.RowSelectorValue.ShouldBe("7");   // a bound value, never canonicalized
    }

    /// <summary>An absent optional identifier is not a rejection — only a present, unresolvable one is.</summary>
    [Fact]
    public void An_absent_optional_timestamp_column_is_left_alone()
    {
        var accepted = Apply(KeyValueTag(timestampColumn: null)).Accepted.ShouldHaveSingleItem();

        accepted.TimestampColumn.ShouldBeNull();
    }
}