{{ message }}
-
-
Notifications
You must be signed in to change notification settings - Fork 11
Expand file tree
/
Copy pathTestRunner.cs
More file actions
1366 lines (1261 loc) · 67 KB
/
Copy pathTestRunner.cs
File metadata and controls
1366 lines (1261 loc) · 67 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Security.Claims;
using System.Text;
using System.Xml.Linq;
using Microsoft.Extensions.Logging;
using Npgsql;
using NpgsqlRest;
namespace NpgsqlRestClient.Testing;
/// <summary>
/// SQL test runner (`--test`). Discovers *.test.sql files and runs each on its own non-pooled connection,
/// invoking endpoints in-process (via <see cref="RoutineInvoker"/> on the test's ambient connection) and
/// asserting on the captured response. See scrap/test-runner-plan for the full design.
/// </summary>
public sealed class TestRunner
{
// Report styling, matched to Serilog's Code theme. There is exactly ONE red style and ONE green style:
// AnsiFail is byte-identical to the chip the Code theme uses for its ERR/FTL level — red-197 text on a
// dark-grey-238 background (NOT white-on-red-196, which is the Literate theme's chip) — and it is used
// for the FAIL/ERROR labels AND all failure text, so they can never look different. AnsiOk is the same
// grammar with green-47, the exact color-cube mirror of red-197 (#ff005f ↔ #00ff5f), used for the PASS
// label and the all-green summary. Never 16-color ConsoleColor.Red (renders orange in some themes).
// Indentation is written OUTSIDE the style so the grey block starts at the text. Escapes are emitted
// only on a real terminal (see Out).
public const string AnsiFail = "\x1b[38;5;197m\x1b[48;5;238m";
// Channel sentinel written by the database poller (not a real path — NUL is invalid in paths).
private const string DbChangeSentinel = "\u0000database";
private const string AnsiOk = "\x1b[38;5;47m\x1b[48;5;238m";
private const string AnsiPlain = "\x1b[0m";
public const int ExitPass = 0;
public const int ExitFailures = 1;
public const int ExitErrors = 2;
public const int ExitConfig = 3;
public const int ExitNoTests = 4;
// Per-async-flow ambient connection: each parallel test file sets this so the endpoint pipeline
// (NpgsqlRestOptions.AmbientConnectionAccessor) runs on that file's connection/transaction.
private static readonly AsyncLocal<NpgsqlConnection?> Ambient = new();
private readonly NpgsqlRestOptions _rest;
private readonly TestRunnerOptions _opt;
private readonly string _connString;
// The debug mirror writes on its own POOLED connections (unlike test connections, which are
// deliberately non-pooled) — autocommit, outside any test transaction.
private readonly string _debugConnString;
private bool _debugTableWarned;
private readonly IReadOnlyDictionary<string, string> _named;
private readonly ILogger? _log;
private readonly bool _logNotices;
private readonly Out _out = new();
private volatile bool _failFast;
// Endpoints actually invoked during the run (canonical "METHOD path" keys) — the coverage numerator.
private readonly ConcurrentDictionary<string, byte> _invoked = new(StringComparer.OrdinalIgnoreCase);
// Lifetime cleanup: from Setup onward, SIGINT/SIGTERM and process exit all funnel into a run-once
// Teardown, executed SYNCHRONOUSLY in the signal handler — not after the run loop unwinds. A parent
// process (bun/npm run, a shell) that kills its children right after forwarding Ctrl+C would otherwise
// win the race and leak the test database. ProcessExit additionally covers hard exits such as
// SqlFileSource ErrorMode=Exit calling Environment.Exit(1). (A parent that SIGKILLs instantly is not
// survivable — this shrinks the window to the teardown itself.)
private readonly CancellationTokenSource _stop = new();
private readonly object _teardownLock = new();
private Task? _teardownTask;
private int _signalCount;
private System.Runtime.InteropServices.PosixSignalRegistration? _sigInt;
private System.Runtime.InteropServices.PosixSignalRegistration? _sigTerm;
private enum Outcome { Pass, Fail, Error }
// Result of executing one step. Emit=false marks an arrange/act step that is not a reported test (it
// only becomes a result when it errors). Name/Message describe the assertion when Emit=true.
private readonly record struct StepResult(bool Emit, Outcome Outcome, string? Message, string? Name);
// One reported test = one assertion: a boolean-returning SELECT, or a DO block (passes unless it
// raises, e.g. via ASSERT). Errors from any step (SQL or HTTP) are also recorded here so they surface
// and count.
private sealed class AssertionResult
{
public required string Name { get; init; }
public int? Line { get; init; }
public Outcome Outcome { get; init; }
public string? Message { get; init; }
public string? Sql { get; init; }
/// <summary>Full path of the included file this assertion came from (via \i/\ir); null = the test file.</summary>
public string? SourceFile { get; init; }
}
// A test file groups its assertions; the file's outcome is the worst of them (Error > Fail > Pass).
// Execution is fail-fast: assertions after the first failure/error in a file are not run (so not listed).
private sealed class FileResult
{
public required string File { get; init; }
public long ElapsedMs { get; set; }
public List<AssertionResult> Assertions { get; } = [];
public List<string> Notices { get; set; } = [];
public Outcome Outcome =>
Assertions.Any(a => a.Outcome == Outcome.Error) ? Outcome.Error :
Assertions.Any(a => a.Outcome == Outcome.Fail) ? Outcome.Fail :
Outcome.Pass;
}
public TestRunner(NpgsqlRestOptions rest, TestRunnerOptions opt, string baseConnectionString, ILogger? logger,
bool logConnectionNotices = false, IReadOnlyDictionary<string, string>? namedConnections = null)
{
_rest = rest;
_opt = opt;
// Always non-pooled: a fresh physical session per test (no temp-table / GUC / prepared-statement carryover).
// In test mode baseConnectionString is already the test connection (TestRunner.ConnectionName, when set).
_connString = new NpgsqlConnectionStringBuilder(baseConnectionString) { Pooling = false }.ConnectionString;
_debugConnString = baseConnectionString;
// Named ConnectionStrings entries for Setup/Teardown steps that target another connection (e.g. an
// "Admin" maintenance connection that runs create/drop database).
_named = namedConnections ?? new Dictionary<string, string>();
_log = logger;
_logNotices = logConnectionNotices;
// Wire the ambient accessor once; null when no test flow is active → zero effect on discovery/normal ops.
_rest.AmbientConnectionAccessor = () => Ambient.Value;
}
// Registers the run-once cleanup triggers. First SIGINT/SIGTERM: cancel everything, then run Teardown
// right here on the signal thread (ctx.Cancel keeps the process alive for it); a second signal force-
// quits. ProcessExit is the safety net for Environment.Exit paths and no-ops after a normal teardown.
private void RegisterLifetimeCleanup()
{
void OnSignal(System.Runtime.InteropServices.PosixSignalContext ctx)
{
if (Interlocked.Increment(ref _signalCount) > 1)
{
return; // second signal: leave ctx.Cancel=false → default handling kills the process
}
ctx.Cancel = true;
_stop.Cancel();
try { TeardownOnceAsync().GetAwaiter().GetResult(); } catch { /* best-effort */ }
}
_sigInt = System.Runtime.InteropServices.PosixSignalRegistration.Create(
System.Runtime.InteropServices.PosixSignal.SIGINT, OnSignal);
_sigTerm = System.Runtime.InteropServices.PosixSignalRegistration.Create(
System.Runtime.InteropServices.PosixSignal.SIGTERM, OnSignal);
AppDomain.CurrentDomain.ProcessExit += (s, e) =>
{
try { TeardownOnceAsync().GetAwaiter().GetResult(); } catch { /* best-effort */ }
};
}
// All teardown paths (signal, process exit, RunAsync/RunWatchAsync finally, Setup failure) funnel here.
// The first caller starts it; every caller gets the SAME task — so a concurrent caller (e.g. the run
// loop's finally racing the signal thread) awaits the in-flight teardown instead of letting the process
// exit while databases are still being dropped.
private Task TeardownOnceAsync()
{
lock (_teardownLock)
{
_teardownTask ??= TeardownAsync(CancellationToken.None);
return _teardownTask;
}
}
/// <summary>Runs before endpoint discovery: collision check + Setup steps. Returns false on failure (logged, teardown attempted).</summary>
public async Task<bool> SetupAsync(CancellationToken ct = default)
{
RegisterLifetimeCleanup();
try
{
// Response-table names are validated and created per HTTP block (no pre-run permanent-table
// collision check needed: writes are pg_temp-qualified and CREATE TEMP TABLE (no IF NOT EXISTS)
// fails loudly on a real duplicate).
_log?.LogDebug("setup: {Count} step(s)", _opt.Setup.Count);
// Strict declared order: each step runs in the order it is written (no Command-first grouping).
// To run something first, write it first. Each Sql/SqlFile step runs on its own connection
// (ConnectionName, else the test connection); Command steps run via the OS shell.
foreach (var step in _opt.Setup)
{
await RunStepAsync(step, "setup", ct);
}
return true;
}
catch (Exception ex)
{
_out.LineAnsi($"Test runner setup failed: {ex.Message}", AnsiFail);
_log?.LogError(ex, "Test runner setup failed");
await TeardownOnceAsync();
return false;
}
}
// Runs one Setup/Teardown step: a shell Command, or a Sql/SqlFile batch on its resolved connection.
// `phase` ("setup" | "teardown") is only used to label the step's log lines.
private async Task RunStepAsync(TestSetupStep step, string phase, CancellationToken ct)
{
if (step.Enabled is false)
{
// Disabled steps are ignored wherever referenced — the default config ships example steps
// ("Enabled": false) that users flip on instead of typing them from scratch.
_log?.LogDebug("{Phase} step {Step}: skipped (Enabled is false)", phase, step.Name ?? step.SqlFile ?? step.Command ?? "(inline)");
return;
}
if (step.IsCommand)
{
await RunCommandAsync(step, phase, ct);
return;
}
var connStr = ResolveStepConnection(step);
await using var conn = new NpgsqlConnection(connStr);
await conn.OpenAsync(ct);
await RunSqlStepAsync(conn, step, phase, ct);
}
// A step's connection: its explicit ConnectionName (a ConnectionStrings entry), else the test connection.
private string ResolveStepConnection(TestSetupStep step)
{
if (string.IsNullOrWhiteSpace(step.ConnectionName))
{
return _connString;
}
if (_named.TryGetValue(step.ConnectionName, out var cs))
{
return cs;
}
throw new InvalidOperationException(
$"TestRunner Setup/Teardown step references ConnectionName '{step.ConnectionName}', which has no matching entry under 'ConnectionStrings'.");
}
/// <summary>Runs after endpoint discovery: executes the test files and (always) Teardown. Returns the process exit code.</summary>
public async Task<int> RunAsync(RoutineEndpoint[] endpoints, CancellationToken ct = default)
{
using var linked = CancellationTokenSource.CreateLinkedTokenSource(_stop.Token, ct);
try
{
return await ExecuteDiscoveredFilesAsync(endpoints, BuildEndpointLookup(endpoints), only: null, linked.Token);
}
catch (OperationCanceledException)
{
// Interrupted (Ctrl+C/SIGTERM) — the signal handler has already run Teardown synchronously.
return ExitErrors;
}
finally
{
await TeardownOnceAsync();
}
}
// Discovers (or takes `only`), filters, runs, reports; returns the exit code. NO teardown — the caller
// owns that (RunAsync tears down in finally; watch mode tears down once, on exit).
private async Task<int> ExecuteDiscoveredFilesAsync(RoutineEndpoint[] endpoints, IReadOnlyDictionary<string, RoutineEndpoint> lookup, List<string>? only, CancellationToken ct)
{
_failFast = false; // reset per run (watch mode reuses the runner across iterations)
if (only is null) _invoked.Clear(); // coverage counts per full run (partial watch reruns don't report)
await EnsureDebugTableAsync(ct); // debug mirror holds the LAST run — recreated per run
var files = only ?? DiscoverFiles();
if (only is null)
{
_log?.LogDebug("discovered {Count} test file(s) matching {Pattern}", files.Count, _opt.FilePattern);
}
if (!string.IsNullOrWhiteSpace(_opt.Filter))
{
int discovered = files.Count;
files = files.Where(f => MatchesFilter(Path.GetRelativePath(Environment.CurrentDirectory, f), _opt.Filter!)).ToList();
_log?.LogDebug("filter {Filter} matched {Count} of {Discovered} discovered file(s)", _opt.Filter, files.Count, discovered);
if (files.Count == 0)
{
_out.Line($"Test runner: filter '{_opt.Filter}' matched none of the {discovered} discovered test file(s).", ConsoleColor.Yellow);
return _opt.AllowEmpty ? ExitPass : ExitNoTests;
}
}
if (_opt.Tags.Count > 0 || _opt.ExcludeTags.Count > 0)
{
int before = files.Count;
var include = new HashSet<string>(_opt.Tags, StringComparer.OrdinalIgnoreCase);
var exclude = new HashSet<string>(_opt.ExcludeTags, StringComparer.OrdinalIgnoreCase);
var selected = new List<string>(files.Count);
foreach (var f in files)
{
List<string> tags;
try
{
tags = TestFileHeader.Parse(await File.ReadAllTextAsync(f, ct), f).Tags;
}
catch
{
// Unreadable header (e.g. a broken include) — keep the file so the run surfaces the error.
selected.Add(f);
continue;
}
if (MatchesTags(tags, include, exclude)) selected.Add(f);
}
files = selected;
_log?.LogDebug("tag filter (tag: {Tags}; exclude: {Exclude}) matched {Count} of {Before} file(s)",
string.Join(',', _opt.Tags), string.Join(',', _opt.ExcludeTags), files.Count, before);
if (files.Count == 0)
{
_out.Line($"Test runner: tag filter matched none of the {before} test file(s).", ConsoleColor.Yellow);
return _opt.AllowEmpty ? ExitPass : ExitNoTests;
}
}
if (files.Count == 0)
{
_out.Line("Test runner: no test files matched FilePattern.", ConsoleColor.Yellow);
return _opt.AllowEmpty ? ExitPass : ExitNoTests;
}
var results = new ConcurrentBag<FileResult>();
int dop = _opt.MaxParallelism > 0 ? _opt.MaxParallelism : Environment.ProcessorCount;
// Console header stays a clean human report (just the file count, like other test runners). The
// parallelism is a diagnostic → log channel at Debug, not the always-on console line.
_out.Line($"NpgsqlRest test runner — {files.Count} file(s)", ConsoleColor.Cyan);
_log?.LogDebug("running {Count} test file(s) at degree of parallelism {Parallelism}", files.Count, dop);
await Parallel.ForEachAsync(
files,
new ParallelOptions { MaxDegreeOfParallelism = dop, CancellationToken = ct },
async (file, _) =>
{
if (_failFast) return; // stop scheduling new work after the first failure (in-flight finish)
var r = await RunFileAsync(file, lookup, ct);
results.Add(r);
if (_opt.FailFast && r.Outcome != Outcome.Pass) _failFast = true;
});
var ordered = results.OrderBy(r => r.File, StringComparer.Ordinal).ToList();
ReportConsole(ordered);
if (!string.IsNullOrWhiteSpace(_opt.JUnitOutput))
{
WriteJUnit(ordered, _opt.JUnitOutput!);
}
int exit = ordered.Any(r => r.Outcome == Outcome.Error) ? ExitErrors
: ordered.Any(r => r.Outcome == Outcome.Fail) ? ExitFailures
: ExitPass;
// Coverage reports on complete runs (never on watch partial reruns). Default (Coverage=null):
// report unless the run was narrowed by Filter/Tag — a deliberately partial run would just nag
// about endpoints its subset never touches. Coverage=true always reports, false never; a set
// CoverageThreshold always reports (and gates) — that's an explicit ask for enforcement.
bool narrowed = !string.IsNullOrWhiteSpace(_opt.Filter) || _opt.Tags.Count > 0 || _opt.ExcludeTags.Count > 0;
bool showCoverage = only is null
&& (_opt.CoverageThreshold is not null || _opt.Coverage == true || (_opt.Coverage is null && !narrowed));
if (showCoverage)
{
exit = ReportCoverage(endpoints, exit);
}
return exit;
}
// Endpoint-coverage summary: exercised N of M TESTABLE endpoints + the untested list. Kinds the runner
// rejects (SSE, upload, login/logout, outbound proxy) are excluded from the ratio and counted apart.
// When CoverageThreshold is set and coverage falls below it, an otherwise-passing run fails with exit 2.
private int ReportCoverage(RoutineEndpoint[] endpoints, int exit)
{
var testable = new List<RoutineEndpoint>(endpoints.Length);
int untestable = 0;
foreach (var ep in endpoints)
{
bool outboundProxy = ep.IsProxy && (ep.ProxyHost ?? _rest.ProxyOptions?.Host) is { } host && !host.StartsWith('/');
if (ep.SseEventsPath is not null || ep.SsePublishEnabled || ep.Upload || ep.Login || ep.Logout || outboundProxy)
{
untestable++;
continue;
}
testable.Add(ep);
}
var untested = testable.Where(ep => !_invoked.ContainsKey($"{ep.Method} {ep.Path}")).ToList();
int covered = testable.Count - untested.Count;
int pct = testable.Count == 0 ? 100 : covered * 100 / testable.Count;
bool gateFailed = _opt.CoverageThreshold is int threshold && pct < threshold;
var line = $"\nendpoint coverage: {covered}/{testable.Count} ({pct}%)"
+ (untestable > 0 ? $" — {untestable} endpoint(s) excluded (not testable in test mode)" : "");
if (gateFailed) { _out.NL(); _out.LineAnsi(line.TrimStart('\n'), AnsiFail); }
else _out.Line(line, untested.Count == 0 ? ConsoleColor.Green : ConsoleColor.Yellow);
foreach (var ep in untested)
{
_out.Line($" untested: {ep.Method} {ep.Path}", ConsoleColor.Yellow);
}
_log?.LogDebug("endpoint coverage {Covered}/{Testable} ({Pct}%), {Untestable} excluded", covered, testable.Count, pct, untestable);
if (gateFailed)
{
_out.LineChip(" ", AnsiPlain, $"coverage {pct}% is below the CoverageThreshold ({_opt.CoverageThreshold}%)", AnsiFail);
if (exit == ExitPass) exit = ExitErrors;
}
return exit;
}
/// <summary>
/// Tag filter: with include tags set, the file must carry at least one of them; a file carrying any
/// exclude tag is skipped (exclude wins). Sets must be case-insensitive (built once per run, so each
/// file is a single pass with O(1) membership checks).
/// </summary>
public static bool MatchesTags(IReadOnlyList<string> fileTags, IReadOnlySet<string> include, IReadOnlySet<string> exclude)
{
bool included = include.Count == 0;
foreach (var tag in fileTags)
{
if (exclude.Count > 0 && exclude.Contains(tag))
{
return false;
}
if (!included && include.Contains(tag))
{
included = true;
}
}
return included;
}
/// <summary>
/// Watch mode (interactive/dev-only): runs everything once, then re-runs on file changes until
/// Ctrl+C. A changed TEST file re-runs just that file; a changed ENDPOINT file (matching
/// <paramref name="endpointPattern"/>, when the SQL file source is enabled) triggers an in-process
/// endpoint REBUILD via <paramref name="rebuildEndpoints"/> followed by a full rerun; any other
/// changed .sql under the test tree (e.g. a \ir fixture — its dependents are unknown) re-runs
/// everything. Teardown runs once, on exit; the exit code is always 0 on a graceful stop (watch is
/// not for CI gating).
/// </summary>
public async Task<int> RunWatchAsync(
RoutineEndpoint[] endpoints,
string? endpointPattern = null,
Func<RoutineEndpoint[]>? rebuildEndpoints = null,
CancellationToken ct = default)
{
var lookup = BuildEndpointLookup(endpoints);
// SIGINT/SIGTERM handling is registered runner-wide (see RegisterLifetimeCleanup): the FIRST signal
// cancels this token and runs Teardown synchronously on the signal thread — so cleanup wins even
// when a parent process (bun/npm run) kills its children right after forwarding Ctrl+C.
using var stop = CancellationTokenSource.CreateLinkedTokenSource(_stop.Token, ct);
try
{
await ExecuteDiscoveredFilesAsync(endpoints, lookup, only: null, stop.Token);
var testBase = GetWatchBaseDir(_opt.FilePattern);
if (testBase is null)
{
_out.Line("Test runner: cannot watch — FilePattern has no existing base directory.", ConsoleColor.Yellow);
return ExitConfig;
}
var testBaseFull = Path.GetFullPath(testBase);
var endpointBaseFull = rebuildEndpoints is not null && !string.IsNullOrWhiteSpace(endpointPattern)
? GetWatchBaseDir(endpointPattern!) is { } eb ? Path.GetFullPath(eb) : null
: null;
var changes = System.Threading.Channels.Channel.CreateUnbounded<string>();
void OnFsEvent(object s, FileSystemEventArgs e) => changes.Writer.TryWrite(e.FullPath);
FileSystemWatcher NewWatcher(string dir)
{
var w = new FileSystemWatcher(dir, "*.sql")
{
IncludeSubdirectories = true,
NotifyFilter = NotifyFilters.LastWrite | NotifyFilters.FileName | NotifyFilters.Size,
};
w.Changed += OnFsEvent;
w.Created += OnFsEvent;
w.Renamed += (s, e) => changes.Writer.TryWrite(e.FullPath);
w.EnableRaisingEvents = true;
return w;
}
using var watcher = NewWatcher(testBaseFull);
// Watch the endpoint source tree too (unless it is the same/contained directory — the
// co-located layout — where the first watcher already covers it).
bool separateEndpointTree = endpointBaseFull is not null
&& !testBaseFull.StartsWith(endpointBaseFull, StringComparison.Ordinal)
&& !endpointBaseFull.StartsWith(testBaseFull, StringComparison.Ordinal);
using var endpointWatcher = separateEndpointTree ? NewWatcher(endpointBaseFull!) : null;
// Database polling: routine changes (create/replace/drop/comment, signature types) have no
// file to watch — a poller on the test connection runs the routine discovery query hashed
// server-side and feeds a sentinel into the same channel, triggering a rebuild + full rerun.
WatchDbPoller? dbPoller = null;
var routineSources = _rest.EndpointSources.OfType<RoutineSource>().ToArray();
if (_opt.DatabasePollingInterval > TimeSpan.Zero && routineSources.Length > 0)
{
dbPoller = new WatchDbPoller(_connString, _opt.DatabasePollingInterval,
onChange: () => changes.Writer.TryWrite(DbChangeSentinel), _log, routineSources);
_ = dbPoller.RunAsync(stop.Token);
}
var watching = Path.GetRelativePath(Environment.CurrentDirectory, testBaseFull)
+ (endpointBaseFull is not null && endpointBaseFull != testBaseFull
? $" + {Path.GetRelativePath(Environment.CurrentDirectory, endpointBaseFull)} (endpoints)"
: "")
+ (dbPoller is not null ? $" + database (poll {_opt.DatabasePollingInterval.TotalSeconds:0.#}s)" : "");
_out.Line($"\nwatching {watching} for changes — Ctrl+C to stop", ConsoleColor.Cyan);
while (!stop.IsCancellationRequested)
{
var changed = (await NextChangeBatchAsync(changes.Reader, stop.Token))?.Distinct().ToList();
if (changed is null) break; // cancelled
// Debounced batch, classified per file: test file → rerun it alone; endpoint source →
// rebuild endpoints + full rerun; anything else under the test tree (fixture/profile
// include — dependents unknown) → full rerun; anything else → ignore.
var testFiles = new List<string>();
bool runAll = false;
bool rebuild = false;
foreach (var path in changed)
{
if (string.Equals(path, DbChangeSentinel, StringComparison.Ordinal))
{
// Routines changed in the database: re-describe (rebuild) and re-run everything.
if (rebuildEndpoints is not null) rebuild = true; else runAll = true;
continue;
}
var rel = Path.GetRelativePath(Environment.CurrentDirectory, path).Replace('\\', '/');
if (MatchesCwdRelativePattern(rel, _opt.FilePattern))
{
if (File.Exists(path)) testFiles.Add(Path.GetFullPath(path));
}
else if (endpointPattern is not null && rebuildEndpoints is not null
&& MatchesCwdRelativePattern(rel, endpointPattern))
{
rebuild = true;
}
else if (Path.GetFullPath(path).StartsWith(testBaseFull, StringComparison.Ordinal))
{
runAll = true;
}
}
if (!rebuild && !runAll && testFiles.Count == 0) continue; // e.g. only deletions
var trigger = string.Equals(changed[0], DbChangeSentinel, StringComparison.Ordinal)
? "database"
: Path.GetRelativePath(Environment.CurrentDirectory, changed[0]);
_out.Line($"\n— {DateTime.Now:HH:mm:ss} change detected ({trigger}{(changed.Count > 1 ? $" +{changed.Count - 1}" : "")}) —", ConsoleColor.Cyan);
if (rebuild)
{
// In-process endpoint rebuild: UseNpgsqlRest re-reads the sources, re-describes against
// the test database, and atomically swaps the endpoint registry. On failure the old
// endpoints stay live. The delta report makes a dropped endpoint (= a broken SQL file,
// skipped by ErrorMode=Skip) visible even when the NpgsqlRest log channel is muted.
try
{
var fresh = rebuildEndpoints!();
ReportEndpointDelta(endpoints, fresh);
endpoints = fresh;
lookup = BuildEndpointLookup(fresh);
}
catch (Exception ex)
{
_out.LineAnsi($"endpoint rebuild failed: {ex.Message} — keeping the previous endpoints", AnsiFail);
_log?.LogError(ex, "endpoint rebuild failed");
}
runAll = true;
}
await ExecuteDiscoveredFilesAsync(endpoints, lookup, only: runAll ? null : testFiles.Distinct().ToList(), stop.Token);
// The rerun itself may have changed the database (committed fixtures, per-file steps) —
// re-baseline so self-inflicted changes never trigger another rerun.
dbPoller?.Rebaseline();
_out.Line($"\nwatching — Ctrl+C to stop", ConsoleColor.Cyan);
}
return ExitPass;
}
catch (OperationCanceledException)
{
return ExitPass;
}
finally
{
// Normally a no-op: the signal handler already ran Teardown; covers non-signal exits.
await TeardownOnceAsync();
}
}
// A cwd-relative path against a cwd-relative glob (shared with the server watch supervisor).
private static bool MatchesCwdRelativePattern(string relativePath, string pattern)
=> WatchUtils.MatchesCwdRelativePattern(relativePath, pattern);
// Rebuild delta: how many endpoints now, plus every added/removed "METHOD path" — a removed endpoint
// usually means its SQL file failed to parse/describe and was skipped.
private void ReportEndpointDelta(RoutineEndpoint[] before, RoutineEndpoint[] after)
{
static HashSet<string> Keys(RoutineEndpoint[] eps) =>
new(eps.Select(e => $"{e.Method} {e.Path}"), StringComparer.OrdinalIgnoreCase);
var beforeKeys = Keys(before);
var afterKeys = Keys(after);
var added = afterKeys.Except(beforeKeys).OrderBy(k => k, StringComparer.Ordinal).ToList();
var removed = beforeKeys.Except(afterKeys).OrderBy(k => k, StringComparer.Ordinal).ToList();
_out.Line($"endpoints rebuilt: {after.Length} (was {before.Length})",
removed.Count > 0 ? ConsoleColor.Yellow : ConsoleColor.Cyan);
foreach (var k in added)
{
_out.Line($" + {k}", ConsoleColor.Green);
}
foreach (var k in removed)
{
_out.Line($" - {k} (endpoint dropped — check its SQL file for errors)", ConsoleColor.Yellow);
}
_log?.LogDebug("endpoints rebuilt: {After} (was {Before}); +{Added} -{Removed}",
after.Length, before.Length, added.Count, removed.Count);
}
// First change (blocking) + quiet-period coalescing (shared with the server watch supervisor).
private static Task<List<string>?> NextChangeBatchAsync(System.Threading.Channels.ChannelReader<string> reader, CancellationToken ct)
=> WatchUtils.NextChangeBatchAsync(reader, ct);
// The deepest fixed (wildcard-free) directory of the FilePattern (shared with the supervisor).
private static string? GetWatchBaseDir(string pattern)
=> WatchUtils.GetWatchBaseDir(pattern);
private async Task<FileResult> RunFileAsync(string file, IReadOnlyDictionary<string, RoutineEndpoint> lookup, CancellationToken runCt)
{
var sw = Stopwatch.StartNew();
var result = new FileResult { File = file };
var notices = new List<string>();
NpgsqlConnection? conn = null;
var contextName = Path.GetRelativePath(Environment.CurrentDirectory, file);
using var perTestCts = CancellationTokenSource.CreateLinkedTokenSource(runCt);
if (_opt.PerTestTimeout > TimeSpan.Zero) perTestCts.CancelAfter(_opt.PerTestTimeout);
var ct = perTestCts.Token;
int timeoutSecs = _opt.PerTestTimeout > TimeSpan.Zero ? (int)Math.Ceiling(_opt.PerTestTimeout.TotalSeconds) : 0;
// Header annotations (leading -- comments): per-file setup/teardown step names + connection override.
string content;
TestFileHeader header;
try
{
content = await File.ReadAllTextAsync(file, ct);
header = TestFileHeader.Parse(content, file);
}
catch (Exception ex)
{
result.Assertions.Add(new AssertionResult { Name = "file read error", Outcome = Outcome.Error, Message = ex.Message });
result.ElapsedMs = sw.ElapsedMilliseconds;
return result;
}
// `-- @connection Name` runs this file (and its in-process endpoint calls) on a named connection —
// e.g. a per-test database that a `-- @setup` step below creates from a template.
string connString = _connString;
if (header.ConnectionName is not null)
{
if (_named.TryGetValue(header.ConnectionName, out var cs))
{
connString = cs;
}
else
{
result.Assertions.Add(new AssertionResult
{
Name = $"-- @connection {header.ConnectionName}",
Outcome = Outcome.Error,
Message = $"'{header.ConnectionName}' has no matching entry under 'ConnectionStrings'",
});
result.ElapsedMs = sw.ElapsedMilliseconds;
return result;
}
}
// Per-file setup steps — BEFORE the file's connection opens (a step may create the very database
// the connection targets). Each runs like a global step: own connection, committed work.
bool setupOk = true;
foreach (var name in header.Setup)
{
if (!_opt.Steps.TryGetValue(name, out var st))
{
result.Assertions.Add(new AssertionResult
{
Name = $"-- @setup {name}",
Outcome = Outcome.Error,
Message = $"unknown step '{name}' — define it under TestRunner:Steps",
});
setupOk = false;
break;
}
try
{
_log?.LogDebug("{File}: @setup {Step}", contextName, name);
await RunStepAsync(st, "setup", ct);
}
catch (Exception ex)
{
result.Assertions.Add(new AssertionResult { Name = $"-- @setup {name}", Outcome = Outcome.Error, Message = ex.Message });
setupOk = false;
break;
}
}
if (setupOk)
try
{
conn = new NpgsqlConnection(connString);
conn.Notice += (_, e) =>
{
// Route notices through the test runner's own log channel, tagged with this test file. The
// log level follows the notice severity (info/notice => Information, warning => Warning, …);
// the channel's MinimalLevel governs visibility. Also capture per-test for the console report.
if (_logNotices && _log is not null)
NpgsqlRestLogger.LogConnectionNotice(_log, e.Notice, _rest.LogConnectionNoticeEventsMode, contextName);
lock (notices) notices.Add(e.Notice.MessageText);
};
await conn.OpenAsync(ct);
Ambient.Value = conn;
// Parse + expand \i/\ir includes — paste semantics: included statements AND HTTP blocks execute
// in place (HTTP blocks participate in _response_{n} numbering exactly as if pasted); included
// steps carry SourceFile so failures are attributed to the included file and line.
var steps = TestFileLoader.LoadSteps(content, file);
int httpTotal = steps.Count(s => s is HttpStep);
int httpOrdinal = 0;
_log?.LogDebug("parsed {File}: {Steps} step(s), {Http} HTTP block(s)", contextName, steps.Count, httpTotal);
foreach (var step in steps)
{
StepResult sr;
try
{
if (step is SqlStep sql)
{
sr = await ExecuteSqlStepAsync(conn, sql, timeoutSecs, ct);
}
else if (step is HttpStep http)
{
httpOrdinal++;
sr = await InvokeHttpStepAsync(conn, http, ResolveResponseTable(http, httpOrdinal, httpTotal), lookup, contextName, ct);
}
else
{
continue;
}
}
catch (OperationCanceledException) { throw; } // timeout — outer handler
catch (PostgresException pg) when (pg.SqlState == "57014") { throw; } // timeout — outer handler
catch (Exception ex)
{
// Attribute the failure to THIS step (line number, name) instead of a file-level
// "execution error" — e.g. a duplicate `# @response` table (42P07) or an invalid name.
sr = new StepResult(true, Outcome.Error,
ex is PostgresException pgEx ? $"{pgEx.SqlState}: {pgEx.MessageText}" : ex.Message, null);
}
if (sr.Emit)
{
result.Assertions.Add(new AssertionResult
{
Name = sr.Name ?? DefaultAssertionName(step),
Line = step.LineNumber,
Outcome = sr.Outcome,
Message = sr.Message,
Sql = step is SqlStep s ? s.Text : null,
SourceFile = step.SourceFile,
});
}
// Fail-fast within a file: a failed/errored step (a failing DO-block assert aborts the PG
// transaction anyway) stops execution, so later assertions in this file are not run.
if (sr.Outcome is Outcome.Fail or Outcome.Error) break;
}
}
catch (OperationCanceledException) when (perTestCts.IsCancellationRequested && !runCt.IsCancellationRequested)
{
result.Assertions.Add(TimeoutAssertion());
}
catch (PostgresException pg) when (pg.SqlState == "57014")
{
result.Assertions.Add(TimeoutAssertion());
}
catch (Exception ex)
{
result.Assertions.Add(new AssertionResult { Name = "execution error", Outcome = Outcome.Error, Message = ex.Message });
}
finally
{
// No explicit ROLLBACK: the connection is non-pooled, so DisposeAsync physically closes it and
// the server aborts any still-open transaction — that is the safety net for a test that didn't
// roll back itself. An explicit ROLLBACK here would also raise PG warning 25P01
// ("there is no transaction in progress") for the common case where the test already rolled
// back or never opened a transaction.
Ambient.Value = null;
if (conn is not null) await conn.DisposeAsync();
}
// Per-file teardown — ALWAYS (best-effort), even when setup or the body failed, and AFTER the
// file's connection is closed (so e.g. `drop database ... with (force)` succeeds). Uses the run
// token, not the per-test one, so a timed-out test still gets its teardown.
foreach (var name in header.Teardown)
{
if (!_opt.Steps.TryGetValue(name, out var st))
{
_log?.LogWarning("{File}: @teardown references unknown step '{Step}' — define it under TestRunner:Steps", contextName, name);
continue;
}
try
{
_log?.LogDebug("{File}: @teardown {Step}", contextName, name);
await RunStepAsync(st, "teardown", runCt);
}
catch (Exception ex)
{
_log?.LogWarning(ex, "{File}: @teardown step '{Step}' failed", contextName, name);
}
}
result.ElapsedMs = sw.ElapsedMilliseconds;
result.Notices = notices;
_log?.LogDebug("{File}: {Outcome} ({Count} assertion(s), {Ms}ms)",
contextName, result.Outcome, result.Assertions.Count, result.ElapsedMs);
return result;
AssertionResult TimeoutAssertion() => new()
{
Name = "timed out",
Outcome = Outcome.Error,
Message = $"timed out after {_opt.PerTestTimeout.TotalSeconds:0}s",
};
}
private async Task<StepResult> ExecuteSqlStepAsync(
NpgsqlConnection conn, SqlStep step, int timeoutSecs, CancellationToken ct)
{
_log?.LogTrace("sql (line {Line}): {Sql}", step.LineNumber, step.Text.Trim());
try
{
await using var cmd = new NpgsqlCommand(step.Text, conn);
if (timeoutSecs > 0) cmd.CommandTimeout = timeoutSecs;
await using var reader = await cmd.ExecuteReaderAsync(ct);
// Boolean first column => assertion (first row only; 0 rows = pass). The optional second column
// is the human-readable assertion name/message.
if (reader.FieldCount >= 1 && reader.GetFieldType(0) == typeof(bool))
{
string? desc = null;
bool pass = true;
if (await reader.ReadAsync(ct))
{
pass = !reader.IsDBNull(0) && reader.GetBoolean(0);
desc = reader.FieldCount >= 2 && !reader.IsDBNull(1) ? reader.GetValue(1)?.ToString() : null;
}
return pass
? new StepResult(true, Outcome.Pass, null, desc)
: new StepResult(true, Outcome.Fail, desc ?? $"assertion failed: {FirstLine(step.Text)}", desc);
}
// A DO block produces no result set; it counts as an assertion that passes unless it raised
// (e.g. ASSERT => P0004, caught below). Any other statement is arrange/act, not a reported test.
return step.IsDoBlock
? new StepResult(true, Outcome.Pass, null, null)
: new StepResult(false, Outcome.Pass, null, null);
}
catch (PostgresException pg)
{
if (pg.SqlState == "P0004") return new StepResult(true, Outcome.Fail, pg.MessageText, null); // assert
if (pg.SqlState == "57014") throw; // timeout — let RunFileAsync classify
return new StepResult(true, Outcome.Error, $"{pg.SqlState}: {pg.MessageText}", null);
}
}
private async Task<StepResult> InvokeHttpStepAsync(
NpgsqlConnection conn, HttpStep step, string responseTable, IReadOnlyDictionary<string, RoutineEndpoint> lookup, string testFile, CancellationToken ct)
{
var httpName = $"{step.Method} {step.Path}";
var pathOnly = StripQuery(step.Path);
if (lookup.TryGetValue($"{step.Method} {pathOnly}", out var ep))
{
if (ep.SseEventsPath is not null || ep.SsePublishEnabled)
return new StepResult(true, Outcome.Error, $"SSE endpoints are not supported in test mode: {httpName}", httpName);
if (ep.Upload)
return new StepResult(true, Outcome.Error, $"upload endpoints are not supported in test mode: {httpName}", httpName);
if (ep.Login || ep.Logout)
return new StepResult(true, Outcome.Error, $"login/logout endpoints are not supported in test mode (inject the principal with `# @claim` instead): {httpName}", httpName);
if (ep.IsProxy)
{
var host = ep.ProxyHost ?? _rest.ProxyOptions?.Host;
if (host is not null && !host.StartsWith('/'))
return new StepResult(true, Outcome.Error, $"outbound proxy/HTTP-type endpoints are disallowed in test mode: {httpName}", httpName);
}
}
else
{
// Not fatal — the request still runs (a test may assert the 404 deliberately) — but the most
// common cause is a path typo or a missing UrlPathPrefix (default "/api"), so surface a warning.
_log?.LogWarning("no endpoint matches {Method} {Path} — the response will be a 404; check the path (including UrlPathPrefix, default \"/api\")",
step.Method, pathOnly);
}
var user = BuildPrincipal(step.Claims);
Dictionary<string, string>? headers = null;
string? contentType = null;
if (step.Headers.Count > 0)
{
headers = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
foreach (var (name, value) in step.Headers)
{
headers[name] = value;
if (string.Equals(name, "Content-Type", StringComparison.OrdinalIgnoreCase)) contentType = value;
}
}
Ambient.Value = conn; // ensure this file's connection is used by the endpoint pipeline
_log?.LogTrace("http {Method} {Path} (line {Line})", step.Method, step.Path, step.LineNumber);
var response = await RoutineInvoker.InvokeAsync(step.Method, step.Path, headers, step.Body, contentType, user, ct);
if (ep is not null)
{
_invoked[$"{ep.Method} {ep.Path}"] = 1; // endpoint-coverage numerator (canonical key)
}
_log?.LogTrace("http {Method} {Path} → {Status}, captured into temp table \"{Table}\"",
step.Method, step.Path, response.StatusCode, responseTable);
await WriteResponseAsync(conn, responseTable, response, ct);
if (string.IsNullOrWhiteSpace(_opt.ResponseTempTable.DebugTable) is false)
{
await MirrorResponseAsync(testFile, responseTable, step, response, ct);
}
// An HTTP block is an act step: it captures the response into the temp table and produces no test
// of its own — the assertions are the boolean SELECTs that follow it. It only surfaces as a result
// when it errors (unsupported endpoint kind above, or an exception attributed by the step loop).
return new StepResult(false, Outcome.Pass, null, null);
}
private async Task WriteResponseAsync(NpgsqlConnection conn, string table, RoutineInvokeResult response, CancellationToken ct)
{
if (!ValidateIdentifier(table))
throw new InvalidOperationException($"invalid response table name '{table}'");
var c = _opt.ResponseTempTable.Columns;
var cols = new List<(string Name, string Type, object? Value, bool Jsonb)>();
if (!string.IsNullOrWhiteSpace(c.Status)) cols.Add((c.Status!, "int", response.StatusCode, false));
if (!string.IsNullOrWhiteSpace(c.Body)) cols.Add((c.Body!, "text", (object?)response.Body ?? DBNull.Value, false));
if (!string.IsNullOrWhiteSpace(c.ContentType)) cols.Add((c.ContentType!, "text", (object?)response.ContentType ?? DBNull.Value, false));
if (!string.IsNullOrWhiteSpace(c.Headers)) cols.Add((c.Headers!, "jsonb", (object?)response.Headers ?? DBNull.Value, true));
if (!string.IsNullOrWhiteSpace(c.IsSuccess)) cols.Add((c.IsSuccess!, "boolean", response.IsSuccess, false));
if (cols.Count == 0)
throw new InvalidOperationException("ResponseTempTable.Columns has no columns enabled.");
foreach (var col in cols)
{
if (!ValidateIdentifier(col.Name))
throw new InvalidOperationException($"invalid response column name '{col.Name}'");
}
// CREATE TEMP TABLE (no IF NOT EXISTS): each HTTP block gets its own fresh table, so a name clash
// (e.g. a duplicate `# @response` name) fails the test loudly. TEMP forces pg_temp; the INSERT is
// pg_temp-qualified so it can never touch a permanent table. (Client runs with Npgsql SQL-rewriting
// disabled → one statement per command, positional params.)
var defs = string.Join(", ", cols.Select(col => $"\"{col.Name}\" {col.Type}"));
await using (var create = new NpgsqlCommand($"create temp table \"{table}\" ({defs})", conn))
{
await create.ExecuteNonQueryAsync(ct);
}
var colList = string.Join(", ", cols.Select(col => $"\"{col.Name}\""));
var valList = string.Join(", ", cols.Select((col, i) => col.Jsonb ? $"${i + 1}::jsonb" : $"${i + 1}"));
await using var ins = new NpgsqlCommand($"insert into pg_temp.\"{table}\" ({colList}) values ({valList})", conn);
foreach (var col in cols)
{
ins.Parameters.AddWithValue(col.Value ?? DBNull.Value);
}
await ins.ExecuteNonQueryAsync(ct);
}
/// <summary>
/// Debug mirror (ResponseTempTable.DebugTable): recreate the PERMANENT mirror table at the start of a
/// run and print the do-not-use-in-CI warning once. Runs on its own autocommit connection, so mirrored
/// rows survive test rollbacks and connection closes — that is the whole point.
/// </summary>
private async Task EnsureDebugTableAsync(CancellationToken ct)
{
var table = _opt.ResponseTempTable.DebugTable;
if (string.IsNullOrWhiteSpace(table))
{
return;
}
if (!ValidateIdentifier(table))
{
throw new InvalidOperationException($"invalid ResponseTempTable.DebugTable name '{table}'");
}
if (_debugTableWarned is false)
{
_debugTableWarned = true;
_out.Line($"WARNING: ResponseTempTable.DebugTable is enabled — every captured response is written to the PERMANENT table \"{table}\" (debugging aid; do not enable in CI)", ConsoleColor.Yellow);
_log?.LogWarning("ResponseTempTable.DebugTable enabled: mirroring responses into permanent table {Table}", table);
}
await using var conn = new NpgsqlConnection(_debugConnString);
await conn.OpenAsync(ct);
await using var cmd = conn.CreateCommand();
cmd.CommandText = $"""
drop table if exists "{table}";
create table "{table}" (
captured_at timestamptz not null default now(),
test_file text not null,
block text not null,
method text not null,
path text not null,
status int,
body text,
content_type text,
headers jsonb,
is_success boolean
)
You can’t perform that action at this time.