lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlServerReadTests.cs

using System.Data;
using System.Data.Common;
using System.Globalization;
using System.Text.Json.Nodes;
using Microsoft.Data.SqlClient;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

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

/// <summary>
///     The <c>Sql</c> driver against a <b>real SQL Server</b>, over the <c>Microsoft.Data.SqlClient</c>
///     provider it actually ships. Everything else in this driver's test estate runs on SQLite, which
///     means three things have never been exercised anywhere until this suite: the shipped ADO.NET
///     provider, T-SQL's own syntax (bracket-quoted two-part names, <c>SELECT TOP 1</c>), and the
///     <c>INFORMATION_SCHEMA</c> catalog SQL that <c>SqlServerDialect</c> carries but that no offline test
///     can reach — the SQLite dialect answers those questions with <c>pragma_table_info</c> instead.
///     <para><b>Env-gated; skipping is the normal outcome.</b> See <see cref="SqlPollServerFixture"/> for
///     the gate. With it unset nothing here opens a socket, so the suite is instant and green offline —
///     which is the point: a live test that goes red on a laptop teaches people to ignore red.</para>
///     <para><b>Out of scope, deliberately:</b> the frozen-database / <c>BadTimeout</c> gate. That needs a
///     <c>docker pause</c>-able container this suite must never own, because the server it points at is
///     the rig's shared central SQL Server. It is a separate task with its own dedicated mssql
///     container.</para>
/// </summary>
[Collection(SqlPollServerCollection.Name)]
public sealed class SqlServerReadTests
{
    private readonly SqlPollServerFixture _sql;

    /// <summary>Initializes a new instance of the <see cref="SqlServerReadTests"/> class.</summary>
    /// <param name="sql">The seeded live-server fixture (collection-scoped).</param>
    public SqlServerReadTests(SqlPollServerFixture sql) => _sql = sql;

    // ---- lifecycle ----

    /// <summary>
    ///     Initialize's liveness check (<c>SELECT 1</c> over a real connection) succeeds and the driver
    ///     reports Healthy with the endpoint described credential-free.
    /// </summary>
    [Fact]
    public async Task InitializeAsync_realSqlServer_reportsHealthyAndDescribesTheEndpoint()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(KeyValue("Sql/Line1/Speed", SqlPollServerFixture.PresentKey));

        var health = driver.GetHealth();
        health.State.ShouldBe(DriverState.Healthy);
        health.LastSuccessfulRead.ShouldNotBeNull();

        // The endpoint description is what every log line and the Admin UI show. It must name the
        // database and must not carry the password that reached the provider.
        driver.Endpoint.ShouldContain(SqlPollServerFixture.FixtureDatabase);
        driver.Endpoint.ShouldNotContain("Password");

        driver.GetHostStatuses().ShouldHaveSingleItem().State.ShouldBe(HostState.Running);
    }

    // ---- key-value model ----

    /// <summary>The plan's headline case: a seeded key-value row round-trips as a Good snapshot.</summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_roundTripsSeededKeyValue()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(KeyValue("Sql/Line1/Speed", SqlPollServerFixture.PresentKey));

        var snapshots = await driver.ReadAsync(["Sql/Line1/Speed"], TestContext.Current.CancellationToken);

        var snapshot = snapshots.ShouldHaveSingleItem();
        snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);
        Convert.ToDouble(snapshot.Value).ShouldBe(SqlPollServerFixture.PresentValue);
        // datetime2 comes back Unspecified; the reader stamps it UTC rather than reinterpreting it
        // through the host's zone, which is the behaviour this asserts against a real column type.
        snapshot.SourceTimestampUtc.ShouldBe(SqlPollServerFixture.PresentKeyTimestampUtc);
    }

    /// <summary>
    ///     Two keys on the same table fold into ONE query (<c>… WHERE [tag_name] IN (@k0, @k1)</c>) and are
    ///     sliced back to the right tags. Against a real server this also proves the parameter binding and
    ///     the key-column stringification survive a genuine <c>nvarchar</c> key.
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_foldsTwoKeysIntoOneGroupAndSlicesBackCorrectly()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(
            KeyValue("Sql/Line1/Speed", SqlPollServerFixture.PresentKey),
            KeyValue("Sql/Line1/Pressure", SqlPollServerFixture.SecondPresentKey));

        var snapshots = await driver.ReadAsync(
            ["Sql/Line1/Speed", "Sql/Line1/Pressure"], TestContext.Current.CancellationToken);

        snapshots.Count.ShouldBe(2);
        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
        Convert.ToDouble(snapshots[0].Value).ShouldBe(SqlPollServerFixture.PresentValue);
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.Good);
        Convert.ToDouble(snapshots[1].Value).ShouldBe(SqlPollServerFixture.SecondPresentValue);
    }

    /// <summary>
    ///     A present row whose value cell is a real T-SQL <c>NULL</c> is Uncertain, not Bad and not
    ///     BadNoData — the row WAS read.
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_nullValueCellIsUncertain()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(KeyValue("Sql/Line1/Temp", SqlPollServerFixture.NullValueKey));

        var snapshot = (await driver.ReadAsync(["Sql/Line1/Temp"], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem();

        snapshot.StatusCode.ShouldBe(SqlStatusCodes.Uncertain);
        snapshot.Value.ShouldBeNull();
        // The row exists, so its timestamp is still readable — that is exactly what distinguishes this
        // from the absent-key case below.
        snapshot.SourceTimestampUtc.ShouldNotBeNull();
    }

    /// <summary><c>nullIsBad</c> re-codes the same NULL cell as Bad, and only that cell.</summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_nullValueCellIsBadWhenNullIsBadIsSet()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(
            nullIsBad: true, KeyValue("Sql/Line1/Temp", SqlPollServerFixture.NullValueKey));

        var snapshot = (await driver.ReadAsync(["Sql/Line1/Temp"], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem();

        snapshot.StatusCode.ShouldBe(SqlStatusCodes.Bad);
    }

    /// <summary>A key the table has no row for is BadNoData with no source timestamp.</summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_absentKeyIsBadNoData()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(KeyValue("Sql/Line1/Missing", SqlPollServerFixture.AbsentKey));

        var snapshot = (await driver.ReadAsync(["Sql/Line1/Missing"], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem();

        snapshot.StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
        snapshot.SourceTimestampUtc.ShouldBeNull();
    }

    // ---- wide-row model ----

    /// <summary>
    ///     Two columns of one wide row, selected by a <c>whereColumn/whereValue</c> pair, come back from a
    ///     single query — including the bound selector value coercing from authored text to a real
    ///     <c>int</c> column server-side.
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_wideRowSelectorReadsEveryColumnFromOneRow()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(
            WideRowWhere("Sql/Station7/OvenTemp", "oven_temp", SqlPollServerFixture.PresentStation),
            WideRowWhere("Sql/Station7/Pressure", "pressure", SqlPollServerFixture.PresentStation),
            WideRowWhere("Sql/Station404/OvenTemp", "oven_temp", SqlPollServerFixture.AbsentStation));

        var snapshots = await driver.ReadAsync(
            ["Sql/Station7/OvenTemp", "Sql/Station7/Pressure", "Sql/Station404/OvenTemp"],
            TestContext.Current.CancellationToken);

        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
        Convert.ToDouble(snapshots[0].Value).ShouldBe(SqlPollServerFixture.PresentStationOvenTemp);
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.Good);
        Convert.ToDouble(snapshots[1].Value).ShouldBe(SqlPollServerFixture.PresentStationPressure);
        // A selector matching no row is BadNoData, not an error — and it does not disturb its siblings.
        snapshots[2].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
    }

    /// <summary>
    ///     The <c>topByTimestamp</c> selector resolves to the newest row.
    ///     <para><b>This is the T-SQL-specific leg.</b> The planner emits the row limit through the
    ///     dialect's <c>SingleRowLimitPrefix</c> — <c>SELECT TOP 1 …</c> for T-SQL, where every offline
    ///     test exercises SQLite's trailing <c>LIMIT 1</c> instead. The seed makes the newest row not the
    ///     first-inserted one, so losing either the <c>TOP 1</c> or the <c>ORDER BY … DESC</c> yields a
    ///     different, wrong value rather than a coincidentally-right one.</para>
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_topByTimestampSelectsTheNewestRow()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(WideRowTop("Sql/Newest/OvenTemp", "oven_temp"));

        var snapshot = (await driver.ReadAsync(["Sql/Newest/OvenTemp"], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem();

        snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);
        Convert.ToDouble(snapshot.Value).ShouldBe(SqlPollServerFixture.NewestStationOvenTemp);
    }

    /// <summary>A view reads exactly like the table it wraps — the shape operators are told to author.</summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_readsThroughAView()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(
            WideRowWhere(
                "Sql/View/OvenTemp", "oven_temp", SqlPollServerFixture.PresentStation,
                table: SqlPollServerFixture.WideRowView));

        var snapshot = (await driver.ReadAsync(["Sql/View/OvenTemp"], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem();

        snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);
        Convert.ToDouble(snapshot.Value).ShouldBe(SqlPollServerFixture.PresentStationOvenTemp);
    }

    // ---- type mapping against real T-SQL column types ----

    /// <summary>
    ///     With no authored <c>type</c>, each tag's driver type is inferred from the provider's own
    ///     <c>GetDataTypeName</c> through <c>SqlServerDialect.MapColumnType</c>. This is the only test in
    ///     the estate where that map is checked against the strings SQL Server <em>actually</em> returns —
    ///     everywhere else it is fed a hand-written list, i.e. checked against itself.
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_infersDriverTypesFromRealTsqlColumnTypes()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(
            TypeProbe("Sql/Probe/Bit", "bit_col"),
            TypeProbe("Sql/Probe/Int", "int_col"),
            TypeProbe("Sql/Probe/BigInt", "bigint_col"),
            TypeProbe("Sql/Probe/Real", "real_col"),
            TypeProbe("Sql/Probe/Float", "float_col"),
            TypeProbe("Sql/Probe/Decimal", "decimal_col"),
            TypeProbe("Sql/Probe/NVarChar", "nvarchar_col"),
            TypeProbe("Sql/Probe/DateTime2", "datetime2_col"));

        var snapshots = await driver.ReadAsync(
            [
                "Sql/Probe/Bit", "Sql/Probe/Int", "Sql/Probe/BigInt", "Sql/Probe/Real",
                "Sql/Probe/Float", "Sql/Probe/Decimal", "Sql/Probe/NVarChar", "Sql/Probe/DateTime2",
            ],
            TestContext.Current.CancellationToken);

        foreach (var snapshot in snapshots) snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);

        // bit → Boolean, not "1"; the CLR type is what the address space declared for the node.
        snapshots[0].Value.ShouldBeOfType<bool>().ShouldBe(SqlPollServerFixture.ProbeBit);
        snapshots[1].Value.ShouldBeOfType<int>().ShouldBe(SqlPollServerFixture.ProbeInt);
        // bigint stays Int64: the seeded value is past double's exact-integer range, so a slip to
        // Float64 would come back as a different number rather than as a type-only difference.
        snapshots[2].Value.ShouldBeOfType<long>().ShouldBe(SqlPollServerFixture.ProbeBigInt);
        snapshots[3].Value.ShouldBeOfType<float>().ShouldBe(SqlPollServerFixture.ProbeReal);
        snapshots[4].Value.ShouldBeOfType<double>().ShouldBe(SqlPollServerFixture.ProbeFloat);
        // decimal collapses to Float64 — the documented v1 precision caveat, pinned here so a future
        // change to that policy cannot land silently.
        snapshots[5].Value.ShouldBeOfType<double>()
            .ShouldBe((double)SqlPollServerFixture.ProbeDecimal, tolerance: 1e-9);
        snapshots[6].Value.ShouldBeOfType<string>().ShouldBe(SqlPollServerFixture.ProbeNVarChar);
        snapshots[7].Value.ShouldBeOfType<DateTime>().ShouldBe(SqlPollServerFixture.ProbeDateTimeUtc);
    }

    /// <summary>
    ///     An authored <c>type</c> wins over the inferred column type, and the coercion runs over the
    ///     provider's real CLR cell rather than over a test double.
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_declaredTypeOverridesTheInferredColumnType()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(
            TypeProbe("Sql/Probe/IntAsString", "int_col", DriverDataType.String),
            TypeProbe("Sql/Probe/BitAsInt32", "bit_col", DriverDataType.Int32),
            TypeProbe("Sql/Probe/RealAsFloat64", "real_col", DriverDataType.Float64));

        var snapshots = await driver.ReadAsync(
            ["Sql/Probe/IntAsString", "Sql/Probe/BitAsInt32", "Sql/Probe/RealAsFloat64"],
            TestContext.Current.CancellationToken);

        foreach (var snapshot in snapshots) snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);
        snapshots[0].Value.ShouldBeOfType<string>()
            .ShouldBe(SqlPollServerFixture.ProbeInt.ToString(CultureInfo.InvariantCulture));
        snapshots[1].Value.ShouldBeOfType<int>().ShouldBe(1);
        snapshots[2].Value.ShouldBeOfType<double>().ShouldBe(SqlPollServerFixture.ProbeReal);
    }

    /// <summary>
    ///     A cell the declared type cannot hold is BadTypeMismatch — Bad quality on that tag alone,
    ///     never a thrown read. <c>int.MaxValue</c> into an <c>Int16</c> is a genuine provider-level
    ///     <c>OverflowException</c>, not a synthesised one.
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_uncoercibleCellIsBadTypeMismatchAndSpares_itsSiblings()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(
            TypeProbe("Sql/Probe/IntAsInt16", "int_col", DriverDataType.Int16),
            TypeProbe("Sql/Probe/Float", "float_col"));

        var snapshots = await driver.ReadAsync(
            ["Sql/Probe/IntAsInt16", "Sql/Probe/Float"], TestContext.Current.CancellationToken);

        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadTypeMismatch);
        snapshots[0].Value.ShouldBeNull();
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.Good);
    }

    // ---- cancellation + connection lifecycle ----

    /// <summary>
    ///     A cancelled poll propagates <see cref="OperationCanceledException"/> rather than publishing
    ///     Bad-quality snapshots: the engine asked the poll to stop and nobody is waiting for values. The
    ///     asymmetry with a deadline breach (which DOES publish BadTimeout) is deliberate.
    ///     <para><b>What this proves, precisely.</b> The token is <em>already</em> cancelled before
    ///     <c>ReadAsync</c> is called, so the cancellation is observed at the reader's first checkpoint —
    ///     <c>SemaphoreSlim.WaitAsync(_, token)</c> — <b>before any <c>SqlConnection</c> opens</b>. It
    ///     therefore does <b>not</b> exercise <c>Microsoft.Data.SqlClient</c>'s in-flight cancellation of a
    ///     running command; that is left to the blackhole gate's deadline path
    ///     (<c>SqlBlackholeTimeoutTests</c>). What it pins is the driver-shell contract that a cancelled
    ///     token surfaces as an <see cref="OperationCanceledException"/> and is <b>never</b> swallowed into
    ///     a Bad snapshot or misread as an unreachable-database verdict — an offline-equivalent guarantee,
    ///     run here over the real provider only to keep it beside its siblings.</para>
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_cancelledTokenPropagatesRatherThanBadCoding()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(KeyValue("Sql/Line1/Speed", SqlPollServerFixture.PresentKey));

        using var cts = new CancellationTokenSource();
        await cts.CancelAsync();

        // ThrowsAnyAsync, not ThrowsAsync: the provider may surface the derived TaskCanceledException, and
        // which of the two arrives is not a contract this driver makes.
        await Assert.ThrowsAnyAsync<OperationCanceledException>(
            async () => await driver.ReadAsync(["Sql/Line1/Speed"], cts.Token));

        // Cancellation is teardown, not a database verdict: health must be untouched.
        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
    }

    /// <summary>
    ///     The reader opens and disposes a connection per poll, which is only sustainable because ADO.NET
    ///     pools them. After many polls the server should still hold a <b>small, non-zero</b> number of
    ///     sessions for this driver's <c>Application Name</c>: non-zero proves the disposed connections
    ///     went back to a pool instead of being torn down, and small proves the poll loop is not leaking
    ///     one per pass. Counting per application name is what makes this safe on a server shared with the
    ///     whole dev rig.
    /// </summary>
    [Fact]
    public async Task ReadAsync_realSqlServer_repeatedPollsReusePooledConnections()
    {
        SkipUnlessLive();

        const int polls = 25;
        await using var driver = await StartAsync(
            KeyValue("Sql/Line1/Speed", SqlPollServerFixture.PresentKey),
            KeyValue("Sql/Line1/Pressure", SqlPollServerFixture.SecondPresentKey));

        for (var i = 0; i < polls; i++)
        {
            var snapshots = await driver.ReadAsync(
                ["Sql/Line1/Speed", "Sql/Line1/Pressure"], TestContext.Current.CancellationToken);
            foreach (var snapshot in snapshots) snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);
        }

        var sessions = await CountDriverSessionsAsync();
        sessions.ShouldBeGreaterThan(0, "the disposed connections should still be pooled open server-side");
        sessions.ShouldBeLessThan(polls, "a poll loop that leaked a connection per pass would show ~25");
    }

    // ---- INFORMATION_SCHEMA catalog (the browse path's SQL) ----

    /// <summary>The dialect's schema list — real <c>INFORMATION_SCHEMA.TABLES</c> — sees the seeded schema.</summary>
    [Fact]
    public async Task Catalog_listSchemasSql_returnsTheSeededSchema()
    {
        SkipUnlessLive();

        var schemas = await QueryCatalogAsync(
            new SqlServerDialect().ListSchemasSql,
            _ => { },
            reader => reader.GetString(reader.GetOrdinal("TABLE_SCHEMA")));

        schemas.ShouldContain(SqlPollServerFixture.Schema);
    }

    /// <summary>
    ///     The dialect's table list returns the seeded relations for the bound schema, and flags the view
    ///     as <c>VIEW</c> — the <c>TABLE_TYPE</c> the browse session decorates its label from.
    /// </summary>
    [Fact]
    public async Task Catalog_listTablesSql_returnsSeededTablesAndMarksTheView()
    {
        SkipUnlessLive();

        var rows = await QueryCatalogAsync(
            new SqlServerDialect().ListTablesSql,
            command => Bind(command, "@schema", SqlPollServerFixture.Schema),
            reader => (
                Name: reader.GetString(reader.GetOrdinal("TABLE_NAME")),
                Type: reader.GetString(reader.GetOrdinal("TABLE_TYPE"))));

        var byName = rows.ToDictionary(r => r.Name, r => r.Type, StringComparer.Ordinal);
        byName.ShouldContainKeyAndValue(SqlPollServerFixture.KeyValueTable, "BASE TABLE");
        byName.ShouldContainKeyAndValue(SqlPollServerFixture.WideRowTable, "BASE TABLE");
        byName.ShouldContainKeyAndValue(SqlPollServerFixture.TypeProbeTable, "BASE TABLE");
        byName.ShouldContainKeyAndValue(SqlPollServerFixture.WideRowView, "VIEW");
    }

    /// <summary>
    ///     The dialect's column list returns the key-value table's columns <b>in ordinal order</b> (the
    ///     order the browse tree renders), with <c>DATA_TYPE</c> values the dialect's map recognises.
    /// </summary>
    [Fact]
    public async Task Catalog_listColumnsSql_returnsSeededColumnsInOrdinalOrder()
    {
        SkipUnlessLive();

        var columns = await ReadCatalogColumnsAsync(SqlPollServerFixture.KeyValueTable);

        columns.Select(c => c.Name).ToArray().ShouldBe(new[]
        {
            SqlPollServerFixture.KeyColumn,
            SqlPollServerFixture.ValueColumn,
            SqlPollServerFixture.TimestampColumn,
        });

        var dialect = new SqlServerDialect();
        dialect.MapColumnType(columns[0].DataType).ShouldBe(DriverDataType.String);    // nvarchar
        dialect.MapColumnType(columns[1].DataType).ShouldBe(DriverDataType.Float64);   // float
        dialect.MapColumnType(columns[2].DataType).ShouldBe(DriverDataType.DateTime);  // datetime2

        // IS_NULLABLE is the third projected column and the browse reads it; prove it is really there.
        columns[0].IsNullable.ShouldBe("NO");
        columns[1].IsNullable.ShouldBe("YES");
    }

    /// <summary>
    ///     Every T-SQL family the type-probe table declares maps to the driver type
    ///     <c>SqlServerDialect.MapColumnType</c> promises — read from the catalog's own
    ///     <c>DATA_TYPE</c> strings, which is the exact input the browse side-panel feeds it.
    /// </summary>
    [Fact]
    public async Task Catalog_listColumnsSql_typeProbeDataTypesMapAcrossTheTsqlFamilies()
    {
        SkipUnlessLive();

        var columns = await ReadCatalogColumnsAsync(SqlPollServerFixture.TypeProbeTable);
        var byName = columns.ToDictionary(c => c.Name, c => c.DataType, StringComparer.Ordinal);
        var dialect = new SqlServerDialect();

        dialect.MapColumnType(byName["bit_col"]).ShouldBe(DriverDataType.Boolean);
        dialect.MapColumnType(byName["int_col"]).ShouldBe(DriverDataType.Int32);
        dialect.MapColumnType(byName["bigint_col"]).ShouldBe(DriverDataType.Int64);
        dialect.MapColumnType(byName["real_col"]).ShouldBe(DriverDataType.Float32);
        dialect.MapColumnType(byName["float_col"]).ShouldBe(DriverDataType.Float64);
        dialect.MapColumnType(byName["decimal_col"]).ShouldBe(DriverDataType.Float64);
        dialect.MapColumnType(byName["nvarchar_col"]).ShouldBe(DriverDataType.String);
        dialect.MapColumnType(byName["datetime2_col"]).ShouldBe(DriverDataType.DateTime);
    }

    // ---- helpers ----

    // ---- design §8.1 catalog gate (Gitea #496), against a real INFORMATION_SCHEMA ----

    /// <summary>
    ///     The gate's live proof: an authored column that does not exist is refused by the allow-list at
    ///     Initialize, so it never reaches a query — and its neighbour on the same table keeps reading.
    /// </summary>
    /// <remarks>
    ///     Only a real SQL Server exercises the real <c>SELECT SCHEMA_NAME()</c> + <c>INFORMATION_SCHEMA</c>
    ///     path the loader depends on; the SQLite-backed suites prove the logic but not that T-SQL's catalog
    ///     answers the shape the loader expects.
    /// </remarks>
    [Fact]
    public async Task CatalogGate_realSqlServer_rejectsAnUnknownColumn_andSparesItsNeighbour()
    {
        SkipUnlessLive();

        var bogus = Tag("Sql/Line1/Bogus", new JsonObject
        {
            ["driver"] = "Sql",
            ["model"] = "KeyValue",
            ["table"] = SqlPollServerFixture.Qualified(SqlPollServerFixture.KeyValueTable),
            ["keyColumn"] = SqlPollServerFixture.KeyColumn,
            ["keyValue"] = SqlPollServerFixture.PresentKey,
            ["valueColumn"] = "no_such_column",
            ["timestampColumn"] = SqlPollServerFixture.TimestampColumn,
        });

        await using var driver = await StartAsync(
            KeyValue("Sql/Line1/Speed", SqlPollServerFixture.PresentKey), bogus);

        // An authoring typo is not a database fault: the driver stays Healthy.
        driver.GetHealth().State.ShouldBe(DriverState.Healthy);

        var snapshots = await driver.ReadAsync(
            ["Sql/Line1/Speed", "Sql/Line1/Bogus"], TestContext.Current.CancellationToken);

        SqlStatusCodes.IsGood(snapshots[0].StatusCode).ShouldBeTrue();
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
    }

    /// <summary>
    ///     T-SQL object names are case-insensitive under the default collation, so a case-variant tag has
    ///     always been valid and must keep working — the gate substitutes the catalog's spelling rather than
    ///     rejecting the operator's.
    /// </summary>
    [Fact]
    public async Task CatalogGate_realSqlServer_acceptsACaseVariantIdentifierAndStillReads()
    {
        SkipUnlessLive();

        var upper = Tag("Sql/Line1/Speed", new JsonObject
        {
            ["driver"] = "Sql",
            ["model"] = "KeyValue",
            ["table"] = SqlPollServerFixture.Qualified(SqlPollServerFixture.KeyValueTable).ToUpperInvariant(),
            ["keyColumn"] = SqlPollServerFixture.KeyColumn.ToUpperInvariant(),
            ["keyValue"] = SqlPollServerFixture.PresentKey,
            ["valueColumn"] = SqlPollServerFixture.ValueColumn.ToUpperInvariant(),
            ["timestampColumn"] = SqlPollServerFixture.TimestampColumn.ToUpperInvariant(),
        });

        await using var driver = await StartAsync(upper);

        var snapshot = (await driver.ReadAsync(["Sql/Line1/Speed"], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem();
        SqlStatusCodes.IsGood(snapshot.StatusCode).ShouldBeTrue();
    }

    /// <summary>
    ///     A cross-database name cannot be validated from this connection's catalog, so it is refused rather
    ///     than quoted into a query — the documented v1 limitation, asserted so it stays deliberate.
    /// </summary>
    [Fact]
    public async Task CatalogGate_realSqlServer_rejectsAThreePartName()
    {
        SkipUnlessLive();

        var threePart = Tag("Sql/Line1/Remote", new JsonObject
        {
            ["driver"] = "Sql",
            ["model"] = "KeyValue",
            ["table"] = $"otherdb.{SqlPollServerFixture.Qualified(SqlPollServerFixture.KeyValueTable)}",
            ["keyColumn"] = SqlPollServerFixture.KeyColumn,
            ["keyValue"] = SqlPollServerFixture.PresentKey,
            ["valueColumn"] = SqlPollServerFixture.ValueColumn,
        });

        await using var driver = await StartAsync(threePart);

        (await driver.ReadAsync(["Sql/Line1/Remote"], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem().StatusCode.ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
    }

    /// <summary>Skips the calling test when the live-server gate is not configured or not reachable.</summary>
    private void SkipUnlessLive()
    {
        if (_sql.SkipReason is not null) Assert.Skip(_sql.SkipReason);
    }

    /// <summary>
    ///     Builds and initializes a driver over the seeded database.
    ///     <para><b>There is no <c>ForProduction</c> / <c>ForTest</c> hatch on <see cref="SqlDriver"/></b>
    ///     — its single public constructor takes the resolved connection string, the dialect and an
    ///     optional factory, and leaving the factory unset is precisely what makes this suite exercise
    ///     <c>SqlClientFactory.Instance</c>, the provider the driver ships.</para>
    /// </summary>
    private Task<SqlDriver> StartAsync(params RawTagEntry[] tags) => StartAsync(nullIsBad: false, tags);

    /// <inheritdoc cref="StartAsync(RawTagEntry[])"/>
    private async Task<SqlDriver> StartAsync(bool nullIsBad, params RawTagEntry[] tags)
    {
        var options = new SqlDriverOptions
        {
            RawTags = tags,
            NullIsBad = nullIsBad,
            CommandTimeout = TimeSpan.FromSeconds(10),
            OperationTimeout = TimeSpan.FromSeconds(15),
        };

        var driver = new SqlDriver(
            options,
            driverInstanceId: "sql-integration",
            dialect: new SqlServerDialect(),
            connectionString: _sql.ConnectionString);

        try
        {
            await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
        }
        catch
        {
            await driver.DisposeAsync();
            throw;
        }

        return driver;
    }

    /// <summary>A key-value tag over the seeded EAV table.</summary>
    private static RawTagEntry KeyValue(string rawPath, string key) => Tag(rawPath, new JsonObject
    {
        ["driver"] = "Sql",
        ["model"] = "KeyValue",
        ["table"] = SqlPollServerFixture.Qualified(SqlPollServerFixture.KeyValueTable),
        ["keyColumn"] = SqlPollServerFixture.KeyColumn,
        ["keyValue"] = key,
        ["valueColumn"] = SqlPollServerFixture.ValueColumn,
        ["timestampColumn"] = SqlPollServerFixture.TimestampColumn,
    });

    /// <summary>A wide-row tag selected by a <c>whereColumn</c>/<c>whereValue</c> pair.</summary>
    private static RawTagEntry WideRowWhere(
        string rawPath, string column, string station, string? table = null) => Tag(rawPath, new JsonObject
    {
        ["driver"] = "Sql",
        ["model"] = "WideRow",
        ["table"] = SqlPollServerFixture.Qualified(table ?? SqlPollServerFixture.WideRowTable),
        ["columnName"] = column,
        ["timestampColumn"] = SqlPollServerFixture.TimestampColumn,
        ["rowSelector"] = new JsonObject
        {
            ["whereColumn"] = SqlPollServerFixture.WideRowSelectorColumn,
            ["whereValue"] = station,
        },
    });

    /// <summary>A wide-row tag selected by newest timestamp.</summary>
    private static RawTagEntry WideRowTop(string rawPath, string column) => Tag(rawPath, new JsonObject
    {
        ["driver"] = "Sql",
        ["model"] = "WideRow",
        ["table"] = SqlPollServerFixture.Qualified(SqlPollServerFixture.WideRowTable),
        ["columnName"] = column,
        ["timestampColumn"] = SqlPollServerFixture.TimestampColumn,
        ["rowSelector"] = new JsonObject
        {
            ["topByTimestamp"] = SqlPollServerFixture.TimestampColumn,
        },
    });

    /// <summary>A wide-row tag over the single-row type-probe table, optionally with a declared type.</summary>
    private static RawTagEntry TypeProbe(string rawPath, string column, DriverDataType? declared = null)
    {
        var config = new JsonObject
        {
            ["driver"] = "Sql",
            ["model"] = "WideRow",
            ["table"] = SqlPollServerFixture.Qualified(SqlPollServerFixture.TypeProbeTable),
            ["columnName"] = column,
            ["timestampColumn"] = SqlPollServerFixture.TimestampColumn,
            ["rowSelector"] = new JsonObject
            {
                ["whereColumn"] = SqlPollServerFixture.TypeProbeSelectorColumn,
                ["whereValue"] = SqlPollServerFixture.TypeProbeRow,
            },
        };

        // Absent "type" ⇒ the driver infers from the result set's column metadata, which is the whole
        // point of the inference test; present ⇒ it overrides.
        if (declared is not null) config["type"] = declared.Value.ToString();
        return Tag(rawPath, config);
    }

    /// <summary>
    ///     Wraps an authored <c>TagConfig</c> object as a raw tag entry.
    ///     <para>Composed as a <see cref="JsonObject"/> rather than as an interpolated string literal on
    ///     purpose: these blobs are brace-dense and nested, and a hand-written literal that loses a brace
    ///     produces a config the parser silently rejects — which surfaces as <c>BadNodeIdUnknown</c>, i.e.
    ///     as a plausible-looking test failure about the driver rather than about the test.</para>
    /// </summary>
    private static RawTagEntry Tag(string rawPath, JsonObject config) =>
        new(rawPath, config.ToJsonString(), WriteIdempotent: false);

    /// <summary>Runs one of the dialect's catalog statements against the seeded database.</summary>
    private async Task<List<T>> QueryCatalogAsync<T>(
        string sql, Action<DbCommand> bind, Func<DbDataReader, T> project)
    {
        var connection = await _sql.OpenInspectionConnectionAsync();
        await using (connection.ConfigureAwait(false))
        {
            var command = connection.CreateCommand();
            await using (command.ConfigureAwait(false))
            {
                command.CommandText = sql;
                bind(command);

                var results = new List<T>();
                var reader = await command.ExecuteReaderAsync(TestContext.Current.CancellationToken);
                await using (reader.ConfigureAwait(false))
                {
                    while (await reader.ReadAsync(TestContext.Current.CancellationToken))
                        results.Add(project(reader));
                }

                return results;
            }
        }
    }

    /// <summary>Reads one seeded table's catalog columns through the dialect's <c>ListColumnsSql</c>.</summary>
    private Task<List<(string Name, string DataType, string IsNullable)>> ReadCatalogColumnsAsync(string table) =>
        QueryCatalogAsync(
            new SqlServerDialect().ListColumnsSql,
            command =>
            {
                Bind(command, "@schema", SqlPollServerFixture.Schema);
                Bind(command, "@table", table);
            },
            reader => (
                Name: reader.GetString(reader.GetOrdinal("COLUMN_NAME")),
                DataType: reader.GetString(reader.GetOrdinal("DATA_TYPE")),
                IsNullable: reader.GetString(reader.GetOrdinal("IS_NULLABLE"))));

    /// <summary>
    ///     Counts the server-side sessions carrying the driver's application name. Skips — rather than
    ///     failing — when the configured login cannot read the DMV, since <c>VIEW SERVER STATE</c> is a
    ///     property of the credentials the operator supplied and not of the code under test.
    /// </summary>
    private async Task<int> CountDriverSessionsAsync()
    {
        try
        {
            var counts = await QueryCatalogAsync(
                "SELECT COUNT(*) AS n FROM sys.dm_exec_sessions WHERE program_name = @app",
                command => Bind(command, "@app", SqlPollServerFixture.DriverApplicationName),
                reader => reader.GetInt32(reader.GetOrdinal("n")));
            return counts[0];
        }
        catch (SqlException ex)
        {
            Assert.Skip(
                "The configured login cannot read sys.dm_exec_sessions (VIEW SERVER STATE), so connection " +
                $"pooling cannot be observed from here: {ex.Message}");
            throw;   // unreachable; Assert.Skip throws.
        }
    }

    /// <summary>Binds one string catalog parameter, exactly as the browse session does.</summary>
    private static void Bind(DbCommand command, string name, string value)
    {
        var parameter = command.CreateParameter();
        parameter.ParameterName = name;
        parameter.DbType = DbType.String;
        parameter.Value = value;
        command.Parameters.Add(parameter);
    }
}