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

using System.Globalization;
using System.Text.Json.Nodes;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Browser;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;

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

/// <summary>
///     The <c>Sql</c> driver over <b>ODBC</b> (lmxopcua#753) — <c>OdbcDialect</c>'s ANSI profile through
///     <c>System.Data.Odbc</c>, the platform driver manager and a vendor ODBC driver, against the PostgreSQL
///     fixture's seed (<c>Docker/postgres</c>). What only a live ODBC stack can prove: positional <c>?</c>
///     binding in planner/catalog order, <c>FETCH FIRST</c> paging, the <c>INFORMATION_SCHEMA</c> catalog through
///     a driver, and the named-query transaction over ODBC's manual-commit mode.
///     <para><b>Env-gated on <see cref="ConnectionStringEnvVar"/></b> — a full ODBC connection string, e.g.
///     <c>Driver={PostgreSQL Unicode};Server=…;Port=…;Database=sqlpollfixture;Uid=otopcua;Pwd=…</c>. A macOS dev
///     box normally has no ODBC driver installed, so these skip there; they run inside a Linux container with
///     <c>unixodbc</c> + <c>odbc-postgresql</c> (see <c>Docker/odbc/run-odbc-gate.sh</c>).</para>
/// </summary>
public sealed class OdbcReadTests
{
    /// <summary>The full ODBC connection string. Absent ⇒ skip.</summary>
    public const string ConnectionStringEnvVar = "SQL_ODBC_CONNECTION_STRING";

    private static readonly string? ConnectionString =
        Environment.GetEnvironmentVariable(ConnectionStringEnvVar) is { Length: > 0 } cs ? cs : null;

    [Fact]
    public async Task InitializeAndRead_overOdbc_keyValueWideRowAndTopByTimestamp()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(null,
            Tag("Kv/Speed", new JsonObject
            {
                ["model"] = "KeyValue", ["table"] = "sqlpoll.tag_values", ["keyColumn"] = "tag_name",
                ["keyValue"] = "Line1.Speed", ["valueColumn"] = "num_value", ["timestampColumn"] = "sample_ts",
            }),
            Tag("Kv/Pressure", new JsonObject
            {
                ["model"] = "KeyValue", ["table"] = "sqlpoll.tag_values", ["keyColumn"] = "tag_name",
                ["keyValue"] = "Line1.Pressure", ["valueColumn"] = "num_value",
            }),
            Tag("Wide/Newest", new JsonObject
            {
                ["model"] = "WideRow", ["table"] = "sqlpoll.latest_status", ["columnName"] = "oven_temp",
                ["rowSelector"] = new JsonObject { ["topByTimestamp"] = "sample_ts" },
            }),
            Tag("Wide/Station7", new JsonObject
            {
                ["model"] = "WideRow", ["table"] = "sqlpoll.latest_status", ["columnName"] = "pressure",
                ["rowSelector"] = new JsonObject { ["whereColumn"] = "station_id", ["whereValue"] = 7 },
            }));

        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
        var snapshots = await driver.ReadAsync(
            ["Kv/Speed", "Kv/Pressure", "Wide/Newest", "Wide/Station7"], TestContext.Current.CancellationToken);

        snapshots.Select(s => s.StatusCode).ShouldAllBe(code => code == SqlStatusCodes.Good);
        // Two keys in one IN (?, ?) list: each value landing on its own tag is the positional-binding proof.
        Number(snapshots[0]).ShouldBe(42.0);
        Number(snapshots[1]).ShouldBe(3.5);
        snapshots[0].SourceTimestampUtc.ShouldBe(new DateTime(2026, 7, 24, 10, 0, 0, DateTimeKind.Utc));
        Number(snapshots[2]).ShouldBe(210.25);   // ORDER BY … DESC FETCH FIRST 1 ROWS ONLY
        Number(snapshots[3]).ShouldBe(1.2);
    }

    [Fact]
    public async Task NamedQuery_overOdbc_bindsPositionalParametersInAuthoredOrder()
    {
        SkipUnlessLive();

        await using var driver = await StartAsync(
            """{"band":{"sql":"SELECT station_id AS station, oven_temp AS temp FROM sqlpoll.latest_status WHERE pressure > ? AND pressure < ?","params":{"lo":1.3,"hi":1.5}}}""",
            Tag("Q/Eight", new JsonObject
            {
                ["model"] = "Query", ["query"] = "band", ["column"] = "temp",
                ["rowKey"] = new JsonObject { ["column"] = "station", ["value"] = 8 },
            }),
            Tag("Q/Seven", new JsonObject
            {
                ["model"] = "Query", ["query"] = "band", ["column"] = "temp",
                ["rowKey"] = new JsonObject { ["column"] = "station", ["value"] = 7 },
            }));

        var snapshots = await driver.ReadAsync(["Q/Eight", "Q/Seven"], TestContext.Current.CancellationToken);
        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);   // 1.4 is inside (1.3, 1.5)
        Number(snapshots[0]).ShouldBe(210.25);
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);   // 1.2 is outside — the bounds were bound in order
    }

    [Fact]
    public async Task Browse_overOdbc_walksSchemaToColumns()
    {
        SkipUnlessLive();

        var config = new JsonObject
        {
            ["provider"] = nameof(SqlProvider.Odbc),
            ["connectionString"] = ConnectionString,
        }.ToJsonString();

        await using var session = await new SqlDriverBrowser().OpenAsync(config, TestContext.Current.CancellationToken);
        var schema = (await session.RootAsync(TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem("the fixture login can SELECT in exactly one schema");
        var tables = await session.ExpandAsync(schema.NodeId, TestContext.Current.CancellationToken);
        var columns = await session.ExpandAsync(
            tables.First(t => t.DisplayName == "tag_values").NodeId, TestContext.Current.CancellationToken);
        columns.Select(c => c.DisplayName).ShouldBe(["tag_name", "num_value", "sample_ts"]);
    }

    /// <summary>
    ///     lmxopcua#781: the two write modes Odbc serves, through positional <c>?</c> binding — an Update that also
    ///     stamps (value, stamp, key in marker order), an Insert (key, value), and its duplicate surfacing the
    ///     backend's SQLSTATE 23505 through the driver manager as BadEntryExists. Upsert is refused, not attempted.
    /// </summary>
    [Fact]
    public async Task Write_overOdbc_updateWithStamp_insert_andDuplicateKey()
    {
        SkipUnlessLive();
        var key = "O-" + Guid.NewGuid().ToString("N")[..12];
        static JsonObject Target(string keyValue, JsonObject write) => new()
        {
            ["model"] = "KeyValue", ["table"] = "sqlpoll.write_targets", ["keyColumn"] = "tag_name",
            ["keyValue"] = keyValue, ["valueColumn"] = "sp_value", ["timestampColumn"] = "stamp_ts",
            ["type"] = "Float64", ["writable"] = true, ["write"] = write,
        };

        await using var driver = await StartAsync(null, allowWrites: true,
            Tag("W/Update", Target("Present", new JsonObject { ["mode"] = "Update", ["stampTimestamp"] = true })),
            Tag("W/Insert", Target(key, new JsonObject { ["mode"] = "Insert" })),
            Tag("W/Upsert", Target(key, new JsonObject { ["mode"] = "Upsert" })));

        var value = 500.0 + Random.Shared.Next(1, 1000);
        (await WriteAsync(driver, "W/Update", value)).ShouldBe(SqlStatusCodes.Good);
        Number((await driver.ReadAsync(["W/Update"], TestContext.Current.CancellationToken))[0]).ShouldBe(value);

        (await WriteAsync(driver, "W/Insert", 1.25)).ShouldBe(SqlStatusCodes.Good);
        Number((await driver.ReadAsync(["W/Insert"], TestContext.Current.CancellationToken))[0]).ShouldBe(1.25);
        (await WriteAsync(driver, "W/Insert", 2.25)).ShouldBe(SqlStatusCodes.BadEntryExists);

        (await WriteAsync(driver, "W/Upsert", 3.25)).ShouldBe(SqlStatusCodes.BadNotSupported);
    }

    private static async Task<uint> WriteAsync(SqlDriver driver, string reference, object? value)
        => (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem().StatusCode;

    private static double Number(DataValueSnapshot snapshot) => Convert.ToDouble(snapshot.Value, CultureInfo.InvariantCulture);

    private static void SkipUnlessLive()
    {
        if (ConnectionString is null)
            Assert.Skip($"Set {ConnectionStringEnvVar} to a full ODBC connection string (and install the driver manager + a " +
                        "vendor ODBC driver) to run the Sql driver's ODBC gate. See Docker/odbc/run-odbc-gate.sh.");
    }

    /// <summary>Builds the driver through the production factory with <c>provider: Odbc</c>.</summary>
    private static Task<SqlDriver> StartAsync(string? queriesJson, params RawTagEntry[] tags)
        => StartAsync(queriesJson, allowWrites: false, tags);

    /// <summary>Builds the driver through the production factory with <c>provider: Odbc</c>.</summary>
    private static async Task<SqlDriver> StartAsync(string? queriesJson, bool allowWrites, params RawTagEntry[] tags)
    {
        var reference = $"OdbcIt{Guid.NewGuid():N}";
        Environment.SetEnvironmentVariable(SqlConnectionStringResolver.EnvironmentVariableFor(reference), ConnectionString);

        var rawTags = new JsonArray();
        foreach (var tag in tags)
            rawTags.Add(new JsonObject { ["rawPath"] = tag.RawPath, ["tagConfig"] = tag.TagConfig, ["writeIdempotent"] = false });

        var config = new JsonObject
        {
            ["provider"] = nameof(SqlProvider.Odbc),
            ["connectionStringRef"] = reference,
            ["rawTags"] = rawTags,
        };
        if (queriesJson is not null) config["queries"] = JsonNode.Parse(queriesJson);
        if (allowWrites) config["allowWrites"] = true;
        var json = config.ToJsonString();

        var driver = SqlDriverFactoryExtensions.CreateInstance("sql-odbc-integration", json);
        try
        {
            await driver.InitializeAsync(json, TestContext.Current.CancellationToken);
        }
        catch
        {
            await driver.DisposeAsync();
            throw;
        }

        return driver;
    }

    private static RawTagEntry Tag(string rawPath, JsonObject config)
    {
        config["driver"] = "Sql";
        return new RawTagEntry(rawPath, config.ToJsonString(), WriteIdempotent: false);
    }
}