Joseph Doherty
a68593e714
SqlTagModel.Query (design §5.4) is now served end to end instead of refused. The driver config declares `queries` (name -> sql, params, maxRows); a tag names the query, the result `column` it reads and optionally a `rowKey` picking its row. - Parser/deploy gate: query + column required, rowKey all-or-nothing; Inspect explains each refusal. Table is not read for this model. - SqlReadOnlyStatement: single SELECT/WITH only, no data/schema-changing keyword outside literals/comments (refuses SELECT INTO, FOR UPDATE, DML CTEs). Shared by the factory and the AdminUI driver form (SqlNamedQueryRules.Describe), so the form refuses what the driver refuses. One bad entry fails the driver. - Parameters are always bound; the factory refuses a declared parameter whose dialect marker the statement never uses. - Reader: one execution per query per poll, inside a transaction that is always rolled back; result columns matched exact-then-unique-CI; no row -> BadNoData, ambiguous row / missing column / over maxRows -> BadConfigurationError. - Catalog gate passes Query tags through (nothing is quoted into SQL); a Query-only driver skips the catalog load. A tag naming an undeclared query is skipped with the reason on /hosts. - AdminUI: typed editor offers Query (query/column/rowKey); driver form edits `queries` as JSON and keeps the last valid set on a typo. (lmxopcua#753) Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
371 lines
18 KiB
C#
371 lines
18 KiB
C#
using System.Data;
|
|
using System.Data.Common;
|
|
using System.Globalization;
|
|
using System.Text.Json.Nodes;
|
|
using Microsoft.Data.Sqlite;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests;
|
|
|
|
/// <summary>
|
|
/// The named-query model (<see cref="SqlTagModel.Query"/>, design §5.4, lmxopcua#753): the tag blob parse,
|
|
/// the read-only statement guard, the factory's query validation, the planner, and the end-to-end read
|
|
/// through <see cref="SqlDriver"/> over SQLite — including the rolled-back transaction every named query
|
|
/// runs inside.
|
|
/// </summary>
|
|
public sealed class SqlNamedQueryTests
|
|
{
|
|
private const string DriverInstanceId = "sql-query";
|
|
|
|
// ---- tag blob parse + deploy-gate explanation ----
|
|
|
|
[Fact]
|
|
public void TryParse_aQueryTag_mapsQueryColumnRowKeyAndType()
|
|
{
|
|
const string json = """
|
|
{"driver":"Sql","model":"Query","query":"lineStatus","column":"speed",
|
|
"rowKey":{"column":"line","value":7},"timestampColumn":"ts","type":"Float64"}
|
|
""";
|
|
|
|
SqlEquipmentTagParser.TryParse(json, "raw/1", out var def).ShouldBeTrue();
|
|
|
|
def.Model.ShouldBe(SqlTagModel.Query);
|
|
def.Name.ShouldBe("raw/1");
|
|
def.QueryName.ShouldBe("lineStatus");
|
|
def.ColumnName.ShouldBe("speed");
|
|
def.RowKeyColumn.ShouldBe("line");
|
|
def.RowKeyValue.ShouldBe("7"); // a JSON number is captured as text, like keyValue
|
|
def.TimestampColumn.ShouldBe("ts");
|
|
def.DeclaredType.ShouldBe(DriverDataType.Float64);
|
|
def.Table.ShouldBeEmpty(); // a Query tag reads no table of its own
|
|
SqlTagDefinitionFactory.Inspect(json).ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public void TryParse_aQueryTagWithoutRowKey_isAccepted()
|
|
=> SqlEquipmentTagParser.TryParse(
|
|
"""{"model":"Query","query":"q","column":"c"}""", "raw", out _).ShouldBeTrue();
|
|
|
|
[Theory]
|
|
[InlineData("""{"model":"Query","column":"c"}""", "'query'")]
|
|
[InlineData("""{"model":"Query","query":"q"}""", "'column'")]
|
|
[InlineData("""{"model":"Query","query":"q","column":"c","rowKey":{"column":"k"}}""", "rowKey")]
|
|
[InlineData("""{"model":"Query","query":"q","column":"c","rowKey":{"value":"v"}}""", "rowKey")]
|
|
[InlineData("""{"model":"Query","query":"q","column":"c","rowKey":"k"}""", "rowKey")]
|
|
public void TryParse_anIncompleteQueryTag_isRejected_andTheGateSaysWhy(string json, string named)
|
|
{
|
|
SqlEquipmentTagParser.TryParse(json, "raw", out _).ShouldBeFalse();
|
|
SqlTagDefinitionFactory.Inspect(json).ShouldContain(w => w.Contains(named, StringComparison.Ordinal));
|
|
}
|
|
|
|
// ---- the read-only statement guard ----
|
|
|
|
[Theory]
|
|
[InlineData("SELECT a, b FROM t WHERE x = @x")]
|
|
[InlineData(" select a from t; ")]
|
|
[InlineData("WITH latest AS (SELECT a FROM t) SELECT a FROM latest")]
|
|
[InlineData("SELECT 'DELETE FROM t; DROP TABLE t' AS note FROM t")] // keywords inside a literal
|
|
[InlineData("SELECT a /* UPDATE t */ FROM t -- INSERT INTO t\n WHERE 1 = 1")] // …and inside comments
|
|
[InlineData("SELECT [update] FROM t")] // a quoted identifier
|
|
[InlineData("SELECT \"delete\" FROM t")]
|
|
public void ReadOnlyStatement_acceptsASingleRead(string sql)
|
|
=> SqlReadOnlyStatement.Check(sql, out _).ShouldBeNull();
|
|
|
|
[Theory]
|
|
[InlineData("UPDATE t SET a = 1")]
|
|
[InlineData("DELETE FROM t")]
|
|
[InlineData("INSERT INTO t VALUES (1)")]
|
|
[InlineData("EXEC dbo.DoThings")]
|
|
[InlineData("SELECT a INTO t2 FROM t")] // SELECT … INTO creates a table
|
|
[InlineData("SELECT a FROM t FOR UPDATE")] // takes write locks
|
|
[InlineData("WITH d AS (DELETE FROM t RETURNING a) SELECT a FROM d")] // data-modifying CTE
|
|
[InlineData("SELECT 1; DROP TABLE t")] // a second statement
|
|
[InlineData("SELECT 1; SELECT 2")]
|
|
[InlineData("SELECT 'unterminated FROM t")]
|
|
[InlineData("SELECT 1 /* never closed")]
|
|
[InlineData(" ")]
|
|
public void ReadOnlyStatement_refusesAnythingElse(string sql)
|
|
=> SqlReadOnlyStatement.Check(sql, out _).ShouldNotBeNull();
|
|
|
|
[Fact]
|
|
public void ReadOnlyStatement_stripsOneTrailingTerminator()
|
|
{
|
|
SqlReadOnlyStatement.Check(" SELECT a FROM t ; ", out var normalised).ShouldBeNull();
|
|
normalised.ShouldBe("SELECT a FROM t");
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData("SELECT a FROM t WHERE x = @line", "@line", true)]
|
|
[InlineData("SELECT a FROM t WHERE x = @lineId", "@line", false)] // a longer name is not the marker
|
|
[InlineData("SELECT '@line' FROM t", "@line", false)] // inside a literal
|
|
[InlineData("SELECT a FROM t WHERE x = :p_line", ":p_line", true)]
|
|
public void UsesMarker_findsOnlyARealParameterToken(string sql, string marker, bool expected)
|
|
=> SqlReadOnlyStatement.UsesMarker(sql, marker).ShouldBe(expected);
|
|
|
|
// ---- factory: named queries are validated, and one bad entry fails the driver ----
|
|
|
|
[Fact]
|
|
public void BuildOptions_typesAValidNamedQuery()
|
|
{
|
|
var options = SqlDriverFactoryExtensions.BuildOptions(
|
|
Dto("""{"q":{"sql":"SELECT a FROM t WHERE x = @x AND y = @y AND z = @z;","params":{"@x":7,"y":"s","z":1.5}}}"""),
|
|
"d1", new SqlServerDialect());
|
|
|
|
var query = options.Queries["q"];
|
|
query.Sql.ShouldBe("SELECT a FROM t WHERE x = @x AND y = @y AND z = @z");
|
|
query.ParameterNames.ShouldBe(["x", "y", "z"]);
|
|
query.ParameterValues.ShouldBe([7L, "s", 1.5]);
|
|
query.MaxRows.ShouldBe(SqlNamedQueryRules.DefaultMaxRows);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData("""{"q":{"sql":"DELETE FROM t"}}""", "single SELECT")]
|
|
[InlineData("""{"q":{"sql":"SELECT a FROM t","params":{"x":1}}}""", "'@x'")] // declared, never used
|
|
[InlineData("""{"q":{"sql":"SELECT a FROM t WHERE x = @x","params":{"x y":1}}}""", "plain identifier")]
|
|
[InlineData("""{"q":{"sql":"SELECT a FROM t WHERE x = @x","params":{"x":[1]}}}""", "string, number")]
|
|
[InlineData("""{"q":{"sql":"SELECT a FROM t","maxRows":0}}""", "maxRows")]
|
|
public void BuildOptions_anUnusableNamedQuery_failsTheDriver(string queries, string named)
|
|
{
|
|
var ex = Should.Throw<InvalidOperationException>(() =>
|
|
SqlDriverFactoryExtensions.BuildOptions(Dto(queries), "d1", new SqlServerDialect()));
|
|
ex.Message.ShouldContain("d1");
|
|
ex.Message.ShouldContain(named);
|
|
}
|
|
|
|
[Fact]
|
|
public void BuildOptions_anOracleQueryMustUseTheOracleMarker()
|
|
{
|
|
Should.Throw<InvalidOperationException>(() => SqlDriverFactoryExtensions.BuildOptions(
|
|
Dto("""{"q":{"sql":"SELECT a FROM t WHERE x = @x","params":{"x":1}}}"""), "d1", new OracleDialect()));
|
|
|
|
SqlDriverFactoryExtensions.BuildOptions(
|
|
Dto("""{"q":{"sql":"SELECT a FROM t WHERE x = :p_x","params":{"x":1}}}"""), "d1", new OracleDialect())
|
|
.Queries["q"].ParameterNames.ShouldBe(["x"]);
|
|
}
|
|
|
|
// ---- planner ----
|
|
|
|
[Fact]
|
|
public void Plan_foldsEveryTagOfOneQueryIntoOnePlanRunningTheQueryVerbatim()
|
|
{
|
|
var queries = Queries(new SqlNamedQuery("q", "SELECT a, b FROM t WHERE x = @x", ["x"], [7L], 50));
|
|
|
|
var plan = SqlGroupPlanner.Plan(
|
|
[
|
|
QueryTag("A", "q", "a"),
|
|
QueryTag("B", "q", "b", rowKeyColumn: "a", rowKeyValue: "1"),
|
|
], new SqlServerDialect(), queries).ShouldHaveSingleItem();
|
|
|
|
plan.Model.ShouldBe(SqlTagModel.Query);
|
|
plan.SqlText.ShouldBe("SELECT a, b FROM t WHERE x = @x");
|
|
plan.ParameterNames.ShouldBe(["@x"]);
|
|
plan.ParameterBindNames.ShouldBe(["@x"]);
|
|
plan.Parameters.ShouldBe([7L]);
|
|
plan.Members.Count.ShouldBe(2);
|
|
plan.RowCap.ShouldBe(50);
|
|
plan.SelectedColumns.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public void Plan_aQueryTagNamingAnUndeclaredQuery_throws()
|
|
=> Should.Throw<ArgumentException>(() =>
|
|
SqlGroupPlanner.Plan([QueryTag("A", "missing", "a")], new SqlServerDialect(), Queries()));
|
|
|
|
[Fact]
|
|
public void CatalogGate_passesQueryTagsThroughUntouched()
|
|
{
|
|
var tag = QueryTag("A", "q", "a");
|
|
var result = SqlCatalogGate.Apply([tag], SqlCatalog.Empty, new SqlServerDialect());
|
|
result.Accepted.ShouldHaveSingleItem().ShouldBeSameAs(tag);
|
|
result.Rejected.ShouldBeEmpty();
|
|
}
|
|
|
|
// ---- end to end over SQLite ----
|
|
|
|
[Fact]
|
|
public async Task Read_aRowKeyedQueryTag_readsItsRow_withBoundParameters()
|
|
{
|
|
using var fixture = new SqlitePollFixture();
|
|
await using var driver = await StartAsync(fixture,
|
|
"""
|
|
{"stations":{"sql":"SELECT station_id AS Station, oven_temp AS Temp, sample_ts AS Ts FROM LatestStatus WHERE pressure > @minPressure ORDER BY station_id",
|
|
"params":{"minPressure":1.3}}}
|
|
""",
|
|
Tag("Q/Newest", """{"model":"Query","query":"stations","column":"temp","rowKey":{"column":"station","value":8},"timestampColumn":"ts","type":"Float64"}"""),
|
|
Tag("Q/Present", """{"model":"Query","query":"stations","column":"Temp","rowKey":{"column":"Station","value":"7"},"type":"Float64"}"""));
|
|
|
|
var snapshots = await driver.ReadAsync(["Q/Newest", "Q/Present"], CancellationToken.None);
|
|
|
|
// The parameter filters station 7 out (pressure 1.2): proof @minPressure was bound, not ignored.
|
|
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
|
|
Convert.ToDouble(snapshots[0].Value, CultureInfo.InvariantCulture)
|
|
.ShouldBe(SqlitePollFixture.NewestStationOvenTemp);
|
|
snapshots[0].SourceTimestampUtc.ShouldBe(new DateTime(2026, 7, 24, 10, 0, 2, DateTimeKind.Utc));
|
|
snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Read_aSingleRowQuery_withoutRowKey_readsTheRow_andMoreRowsAreAmbiguous()
|
|
{
|
|
using var fixture = new SqlitePollFixture();
|
|
await using var driver = await StartAsync(fixture,
|
|
"""
|
|
{"one":{"sql":"SELECT MAX(num_value) AS top FROM TagValues"},
|
|
"many":{"sql":"SELECT num_value FROM TagValues"}}
|
|
""",
|
|
Tag("Q/One", """{"model":"Query","query":"one","column":"top","type":"Float64"}"""),
|
|
Tag("Q/Many", """{"model":"Query","query":"many","column":"num_value","type":"Float64"}"""));
|
|
|
|
var snapshots = await driver.ReadAsync(["Q/One", "Q/Many"], CancellationToken.None);
|
|
|
|
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
|
|
Convert.ToDouble(snapshots[0].Value, CultureInfo.InvariantCulture).ShouldBe(SqlitePollFixture.PresentValue);
|
|
// Three rows and no rowKey: never "last row wins" — the selector is ambiguous.
|
|
snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadConfigurationError);
|
|
// …and that refusal is the database answering, so the driver stays Healthy.
|
|
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Read_aColumnTheQueryDoesNotReturn_isAConfigurationError_notSomeoneElsesCell()
|
|
{
|
|
using var fixture = new SqlitePollFixture();
|
|
await using var driver = await StartAsync(fixture,
|
|
"""{"one":{"sql":"SELECT MAX(num_value) AS top FROM TagValues"}}""",
|
|
Tag("Q/Wrong", """{"model":"Query","query":"one","column":"nope"}"""));
|
|
|
|
(await driver.ReadAsync(["Q/Wrong"], CancellationToken.None))[0].StatusCode
|
|
.ShouldBe(SqlStatusCodes.BadConfigurationError);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Read_moreRowsThanMaxRows_refusesTheWholeQuery()
|
|
{
|
|
using var fixture = new SqlitePollFixture();
|
|
await using var driver = await StartAsync(fixture,
|
|
"""{"all":{"sql":"SELECT tag_name, num_value FROM TagValues","maxRows":2}}""",
|
|
Tag("Q/Speed", $$$"""{"model":"Query","query":"all","column":"num_value","rowKey":{"column":"tag_name","value":"{{{SqlitePollFixture.PresentKey}}}"}}"""));
|
|
|
|
(await driver.ReadAsync(["Q/Speed"], CancellationToken.None))[0].StatusCode
|
|
.ShouldBe(SqlStatusCodes.BadConfigurationError);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Initialize_aQueryTagNamingAnUndeclaredQuery_isSkippedWithTheReason()
|
|
{
|
|
using var fixture = new SqlitePollFixture();
|
|
await using var driver = await StartAsync(fixture,
|
|
"""{"one":{"sql":"SELECT 1 AS v"}}""",
|
|
Tag("Q/Ok", """{"model":"Query","query":"one","column":"v"}"""),
|
|
Tag("Q/Missing", """{"model":"Query","query":"nope","column":"v"}"""));
|
|
|
|
var skipped = ((ISkippedTagReporter)driver).GetSkippedTags().ShouldNotBeNull().Sample.ShouldHaveSingleItem();
|
|
skipped.RawPath.ShouldBe("Q/Missing");
|
|
skipped.Reason.ShouldContain("'nope'");
|
|
(await driver.ReadAsync(["Q/Missing", "Q/Ok"], CancellationToken.None))
|
|
.Select(s => s.StatusCode).ShouldBe([SqlStatusCodes.BadNodeIdUnknown, SqlStatusCodes.Good]);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The backstop behind the authoring guard: a named query runs inside a transaction that is always
|
|
/// rolled back. The probe connection performs a WRITE inside every transaction the driver opens —
|
|
/// standing in for a side-effecting function the lexical guard cannot see — so the table staying empty
|
|
/// after a read proves the work was rolled back, and the begin counter (the positive control) proves a
|
|
/// transaction was opened at all. A reader that committed would leave one row per poll.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Read_aNamedQuery_runsInsideATransactionThatIsRolledBack()
|
|
{
|
|
using var fixture = new SqlitePollFixture();
|
|
fixture.Execute("CREATE TABLE SideEffects (n INTEGER)");
|
|
var factory = new SideEffectFactory();
|
|
await using var driver = await StartAsync(fixture, factory,
|
|
"""{"one":{"sql":"SELECT MAX(num_value) AS top FROM TagValues"}}""",
|
|
Tag("Q/One", """{"model":"Query","query":"one","column":"top"}"""));
|
|
|
|
(await driver.ReadAsync(["Q/One"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
|
|
(await driver.ReadAsync(["Q/One"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
|
|
|
|
factory.TransactionsBegun.ShouldBe(2);
|
|
using var count = fixture.Connection.CreateCommand();
|
|
count.CommandText = "SELECT COUNT(*) FROM SideEffects";
|
|
Convert.ToInt64(count.ExecuteScalar(), CultureInfo.InvariantCulture).ShouldBe(0L);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Read_keyValueTags_doNotOpenATransaction()
|
|
{
|
|
using var fixture = new SqlitePollFixture();
|
|
fixture.Execute("CREATE TABLE SideEffects (n INTEGER)");
|
|
var factory = new SideEffectFactory();
|
|
await using var driver = await StartAsync(fixture, factory, queriesJson: null,
|
|
Tag("Kv", $$"""{"model":"KeyValue","table":"TagValues","keyColumn":"tag_name","keyValue":"{{SqlitePollFixture.PresentKey}}","valueColumn":"num_value"}"""));
|
|
|
|
(await driver.ReadAsync(["Kv"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
|
|
factory.TransactionsBegun.ShouldBe(0);
|
|
}
|
|
|
|
// ---- helpers ----
|
|
|
|
private static SqlDriverConfigDto Dto(string queriesJson) => System.Text.Json.JsonSerializer.Deserialize<SqlDriverConfigDto>(
|
|
new JsonObject { ["connectionStringRef"] = "x", ["queries"] = JsonNode.Parse(queriesJson) }.ToJsonString(),
|
|
SqlDriverFactoryExtensions.JsonOptionsForTest)!;
|
|
|
|
private static Dictionary<string, SqlNamedQuery> Queries(params SqlNamedQuery[] queries)
|
|
=> queries.ToDictionary(q => q.Name, StringComparer.Ordinal);
|
|
|
|
private static SqlTagDefinition QueryTag(
|
|
string name, string query, string column, string? rowKeyColumn = null, string? rowKeyValue = null)
|
|
=> new(name, SqlTagModel.Query, string.Empty, ColumnName: column, QueryName: query,
|
|
RowKeyColumn: rowKeyColumn, RowKeyValue: rowKeyValue);
|
|
|
|
private static RawTagEntry Tag(string rawPath, string json) => new(rawPath, json, WriteIdempotent: false);
|
|
|
|
private static Task<SqlDriver> StartAsync(SqlitePollFixture fixture, string? queriesJson, params RawTagEntry[] tags)
|
|
=> StartAsync(fixture, fixture.Factory, queriesJson, tags);
|
|
|
|
private static async Task<SqlDriver> StartAsync(
|
|
SqlitePollFixture fixture, DbProviderFactory factory, string? queriesJson, params RawTagEntry[] tags)
|
|
{
|
|
var dto = queriesJson is null ? new SqlDriverConfigDto() : Dto(queriesJson);
|
|
var built = SqlDriverFactoryExtensions.BuildOptions(dto, DriverInstanceId, new SqliteDialect());
|
|
var options = new SqlDriverOptions
|
|
{
|
|
RawTags = tags,
|
|
OperationTimeout = TimeSpan.FromSeconds(15),
|
|
CommandTimeout = TimeSpan.FromSeconds(10),
|
|
Queries = built.Queries,
|
|
};
|
|
var driver = new SqlDriver(options, DriverInstanceId, new SqliteDialect(), fixture.ConnectionString, factory);
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
return driver;
|
|
}
|
|
|
|
/// <summary>A SQLite factory whose connections write a row inside every transaction they begin.</summary>
|
|
private sealed class SideEffectFactory : DbProviderFactory
|
|
{
|
|
private int _begun;
|
|
|
|
public int TransactionsBegun => Volatile.Read(ref _begun);
|
|
|
|
public override DbConnection CreateConnection() => new SideEffectConnection(this);
|
|
|
|
private sealed class SideEffectConnection(SideEffectFactory owner) : SqliteConnection
|
|
{
|
|
protected override DbTransaction BeginDbTransaction(IsolationLevel isolationLevel)
|
|
{
|
|
Interlocked.Increment(ref owner._begun);
|
|
var transaction = BeginTransaction(isolationLevel);
|
|
using var write = CreateCommand();
|
|
write.Transaction = transaction;
|
|
write.CommandText = "INSERT INTO SideEffects (n) VALUES (1)";
|
|
write.ExecuteNonQuery();
|
|
return transaction;
|
|
}
|
|
}
|
|
}
|
|
}
|