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;
///
/// The Sql driver over ODBC (lmxopcua#753) — OdbcDialect's ANSI profile through
/// System.Data.Odbc, the platform driver manager and a vendor ODBC driver, against the PostgreSQL
/// fixture's seed (Docker/postgres). What only a live ODBC stack can prove: positional ?
/// binding in planner/catalog order, FETCH FIRST paging, the INFORMATION_SCHEMA catalog through
/// a driver, and the named-query transaction over ODBC's manual-commit mode.
/// Env-gated on — a full ODBC connection string, e.g.
/// Driver={PostgreSQL Unicode};Server=…;Port=…;Database=sqlpollfixture;Uid=otopcua;Pwd=…. A macOS dev
/// box normally has no ODBC driver installed, so these skip there; they run inside a Linux container with
/// unixodbc + odbc-postgresql (see Docker/odbc/run-odbc-gate.sh).
///
public sealed class OdbcReadTests
{
/// The full ODBC connection string. Absent ⇒ skip.
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"]);
}
///
/// lmxopcua#781: the two write modes Odbc serves, through positional ? 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.
///
[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 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.");
}
/// Builds the driver through the production factory with provider: Odbc.
private static Task StartAsync(string? queriesJson, params RawTagEntry[] tags)
=> StartAsync(queriesJson, allowWrites: false, tags);
/// Builds the driver through the production factory with provider: Odbc.
private static async Task 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);
}
}