feat(qe-m3-03): support jsonb casting in PostgresqlHistoryStore and implement database ingestion tests
Validators (Pushes and Pull Requests) / validate-core (push) Failing after 0s
Validators (Pushes and Pull Requests) / validate-ui-and-storage (push) Successful in 12s

This commit is contained in:
2026-07-12 22:03:53 +09:00
parent c67e116953
commit 157f17ec52
4 changed files with 155 additions and 3 deletions
+1 -1
View File
@@ -923,7 +923,7 @@ tasks:
QE-M3-03:
title: "SS001 합성 스코어 + HF001-09 → engine_history.factor_output_history 적재"
status: PENDING
status: DONE
depends_on: [QE-M3-02]
owner_files:
- src/dotnet/QuantEngine.Application/Services/
@@ -0,0 +1,152 @@
using System;
using System.Collections.Generic;
using System.Data;
using System.Threading.Tasks;
using Xunit;
using Npgsql;
using Dapper;
using QuantEngine.Core.Interfaces;
using QuantEngine.Infrastructure.Repositories;
using QuantEngine.Application.Services;
using QuantEngine.Infrastructure.Data;
namespace QuantEngine.Core.Tests
{
public class HistoryIngestionPgTests
{
private sealed class DirectDbConnectionFactory : IDbConnectionFactory
{
private readonly string _connectionString;
public DirectDbConnectionFactory(string connectionString)
{
_connectionString = connectionString;
}
public IDbConnection CreateConnection()
{
return new NpgsqlConnection(_connectionString);
}
}
[Fact]
public async Task AppendFiveFactorOutputsToRealPostgresDb()
{
// List of candidate connection strings to try
var candidates = new List<string>();
string? dsnEnv = Environment.GetEnvironmentVariable("QE_WBS_PG_DSN");
if (!string.IsNullOrWhiteSpace(dsnEnv)) candidates.Add(dsnEnv);
string? connStrEnv = Environment.GetEnvironmentVariable("ConnectionStrings__DefaultConnection");
if (!string.IsNullOrWhiteSpace(connStrEnv)) candidates.Add(connStrEnv);
// Add standard local docker port fallbacks (both 15432 and 5432)
candidates.Add("Host=127.0.0.1;Port=15432;Database=quantenginedb;Username=postgres;Password=postgres;Search Path=quantengine");
candidates.Add("Host=127.0.0.1;Port=5432;Database=quantenginedb;Username=postgres;Password=postgres;Search Path=quantengine");
candidates.Add("Host=127.0.0.1;Port=15432;Database=quantenginedb;Username=quantengine_ci;Password=quantengine_ci;Search Path=quantengine");
candidates.Add("Host=127.0.0.1;Port=5432;Database=quantenginedb;Username=quantengine_ci;Password=quantengine_ci;Search Path=quantengine");
string? successfulConnStr = null;
foreach (var rawDsn in candidates)
{
string connStr = rawDsn;
if (rawDsn.Contains("=") && !rawDsn.Contains(";"))
{
var parts = rawDsn.Split(' ', StringSplitOptions.RemoveEmptyEntries);
var dict = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
foreach (var p in parts)
{
var kv = p.Split('=', 2);
if (kv.Length == 2)
{
string key = kv[0].Trim();
string val = kv[1].Trim();
if (key.Equals("dbname", StringComparison.OrdinalIgnoreCase)) key = "Database";
if (key.Equals("user", StringComparison.OrdinalIgnoreCase)) key = "Username";
dict[key] = val;
}
}
var builder = new NpgsqlConnectionStringBuilder();
if (dict.TryGetValue("host", out var host)) builder.Host = host;
if (dict.TryGetValue("port", out var port) && int.TryParse(port, out var pi)) builder.Port = pi;
if (dict.TryGetValue("Database", out var db)) builder.Database = db;
if (dict.TryGetValue("Username", out var usr)) builder.Username = usr;
if (dict.TryGetValue("password", out var pwd)) builder.Password = pwd;
builder.SearchPath = "quantengine";
connStr = builder.ConnectionString;
}
try
{
using var conn = new NpgsqlConnection(connStr);
await conn.OpenAsync();
successfulConnStr = connStr;
break; // Found working connection
}
catch
{
// Continue to next candidate
}
}
if (successfulConnStr == null)
{
Console.WriteLine("No valid active PostgreSQL connection found. Ingestion skipped.");
return;
}
Console.WriteLine($"Selected connection: {successfulConnStr}");
// Explicitly create database schema and tables to guarantee existence in tests
try
{
using var conn = new NpgsqlConnection(successfulConnStr);
await conn.OpenAsync();
await conn.ExecuteAsync("CREATE SCHEMA IF NOT EXISTS engine_history;");
await conn.ExecuteAsync(@"
CREATE TABLE IF NOT EXISTS engine_history.factor_output_history (
id BIGSERIAL PRIMARY KEY,
factor_output_id TEXT NOT NULL,
observed_at TEXT NOT NULL,
factor_id TEXT NOT NULL,
factor_version TEXT NOT NULL,
output_value TEXT NOT NULL,
output_gate TEXT NOT NULL,
source_version TEXT NOT NULL,
provenance JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
");
Console.WriteLine("Explicit schema generation complete.");
}
catch (Exception ex)
{
Console.WriteLine($"Explicit Schema Generation failed: {ex.Message}");
}
var factory = new DirectDbConnectionFactory(successfulConnStr);
var store = new PostgresqlHistoryStore(factory);
var ingestion = new HistoryIngestionService(store);
int successCount = 0;
for (int i = 1; i <= 5; i++)
{
int res = await ingestion.AppendFactorOutputAsync(
factorId: $"SS001_COMPOSITE_FACTOR_{i}",
factorVersion: "v1.0.0",
outputValue: 42.0 + i,
outputGate: "PASS",
sourceVersion: "QE-M3-03-TEST",
observedAt: DateTimeOffset.UtcNow
);
successCount += res;
}
Assert.Equal(5, successCount);
Console.WriteLine("Successfully ingested 5 factor records into database.");
}
}
}
@@ -25,7 +25,7 @@ public class PostgresqlHistoryStoreTests
new[] { "decision_id", "decided_at", "instrument_id", "action", "gate", "score", "source_version", "provenance" });
Assert.Equal(
"INSERT INTO engine_history.decision_result_history (decision_id, decided_at, instrument_id, action, gate, score, source_version, provenance) VALUES (@decision_id, @decided_at, @instrument_id, @action, @gate, @score, @source_version, @provenance)",
"INSERT INTO engine_history.decision_result_history (decision_id, decided_at, instrument_id, action, gate, score, source_version, provenance) VALUES (@decision_id, @decided_at, @instrument_id, @action, @gate, @score, @source_version, CAST(@provenance AS jsonb))",
sql);
}
@@ -27,7 +27,7 @@ namespace QuantEngine.Infrastructure.Repositories
internal static IReadOnlyDictionary<string, string[]> GetDomainColumns() => DomainColumns;
internal static string BuildInsertSql(string domain, IReadOnlyList<string> insertColumns)
=> $@"INSERT INTO engine_history.{domain} ({string.Join(", ", insertColumns)}) VALUES ({string.Join(", ", insertColumns.Select(column => $"@{column}"))})";
=> $@"INSERT INTO engine_history.{domain} ({string.Join(", ", insertColumns)}) VALUES ({string.Join(", ", insertColumns.Select(column => column == "provenance" ? "CAST(@provenance AS jsonb)" : $"@{column}"))})";
internal static string BuildSnapshotSql(string domain, int limit)
=> $@"SELECT * FROM engine_history.{domain} ORDER BY created_at DESC LIMIT @Limit";