using Xunit; using Npgsql; using NpgsqlTypes; using System.Data; using System.Data.Common; namespace KArtSell.Integration.Tests; /// /// DbUp Migration Validation Tests /// Covers: Fresh Install, Idempotency, Schema Validation, Trigger/Constraint Enforcement /// Following AGENTS.md v16.0: Necessity-driven, Complete evidence preservation, Reproducible /// public sealed class DbUpMigrationTests : IAsyncLifetime { private NpgsqlDataSource _dataSource = null!; private const string DefaultConnString = "Host=localhost;Port=5432;Database=kartsell_migration_test;Username=kartsell;Password=kartsell"; public async Task InitializeAsync() { var connString = TestDatabaseConnection.GetConnectionString(); // Create test database if needed var adminConnString = connString.Replace("kartsell_migration_test", "postgres"); await using var adminConn = new NpgsqlConnection(adminConnString); await adminConn.OpenAsync(); try { await using var cmd = adminConn.CreateCommand(); cmd.CommandText = "DROP DATABASE IF EXISTS kartsell_migration_test WITH (FORCE);"; await cmd.ExecuteNonQueryAsync(); } catch (PostgresException ex) when (ex.SqlState == "3D000") { /* DB doesn't exist */ } await using var createCmd = adminConn.CreateCommand(); createCmd.CommandText = "CREATE DATABASE kartsell_migration_test;"; await createCmd.ExecuteNonQueryAsync(); await adminConn.CloseAsync(); // Connect to test database _dataSource = new NpgsqlDataSourceBuilder(connString).Build(); // Apply prerequisite migrations (0000-0007) await ApplyPrerequisiteMigrationsAsync(); } public async Task DisposeAsync() { await _dataSource.DisposeAsync(); // Cleanup test database var adminConnString = TestDatabaseConnection.GetConnectionString(); adminConnString = adminConnString.Replace("kartsell_migration_test", "postgres"); await using var adminConn = new NpgsqlConnection(adminConnString); await adminConn.OpenAsync(); await using var dropCmd = adminConn.CreateCommand(); dropCmd.CommandText = "DROP DATABASE IF EXISTS kartsell_migration_test WITH (FORCE);"; await dropCmd.ExecuteNonQueryAsync(); await adminConn.CloseAsync(); } /// /// Prerequisite: Apply migrations 0000-0007 to prepare for testing 0008, 0009, 0010 /// private async Task ApplyPrerequisiteMigrationsAsync() { await using var connection = await _dataSource.OpenConnectionAsync(); // Create __dbup_schema_history table await using var createHistoryCmd = connection.CreateCommand(); createHistoryCmd.CommandText = """ CREATE TABLE IF NOT EXISTS __dbup_schema_history ( id SERIAL PRIMARY KEY, scriptname VARCHAR(255) NOT NULL, applied TIMESTAMP NOT NULL DEFAULT NOW() ); """; await createHistoryCmd.ExecuteNonQueryAsync(); // Apply building blocks schema (0000) await using var bbCmd = connection.CreateCommand(); bbCmd.CommandText = """ CREATE SCHEMA IF NOT EXISTS building_blocks; CREATE TABLE IF NOT EXISTS building_blocks.outbox_message ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), event_type VARCHAR(256) NOT NULL, payload JSONB NOT NULL, published BOOLEAN DEFAULT FALSE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP ); CREATE SCHEMA IF NOT EXISTS outbox; CREATE TABLE IF NOT EXISTS outbox.outbox ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), event_type VARCHAR(256) NOT NULL, payload JSONB NOT NULL, published BOOLEAN DEFAULT FALSE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP ); """; await bbCmd.ExecuteNonQueryAsync(); } /// /// Gate 1: Fresh Install - Migrations run in order with no errors /// [Fact] public async Task Migration0008_FreshInstall_CreatesValidShadowRunSchema() { // Act: Apply migration 0008 await ApplyMigration0008(); // Assert: Table exists with correct structure await using var connection = await _dataSource.OpenConnectionAsync(); var tableExists = await TableExistsAsync(connection, "model_operations", "shadow_run"); Assert.True(tableExists, "shadow_run table should exist"); var columns = await GetTableColumnsAsync(connection, "model_operations", "shadow_run"); Assert.Contains("run_id", columns); Assert.Contains("model_id", columns); Assert.Contains("status", columns); Assert.Contains("published_at", columns); Assert.Contains("metrics_json", columns); Assert.Contains("phase_analysis_json", columns); Assert.Contains("cost_analysis_json", columns); Assert.Contains("false_exit_analysis_json", columns); Assert.Contains("validation_gates_json", columns); } /// /// Gate 1: Fresh Install - All 3 migrations run in sequence /// [Fact] public async Task Migration0009_0010_FreshInstall_CreatesCompleteSchema() { // Act: Apply all three migrations await ApplyMigration0008(); await ApplyMigration0009(); await ApplyMigration0010(); await using var connection = await _dataSource.OpenConnectionAsync(); // Assert: All three tables exist Assert.True(await TableExistsAsync(connection, "model_operations", "shadow_run")); Assert.True(await TableExistsAsync(connection, "building_blocks", "inbox_message")); Assert.True(await TableExistsAsync(connection, "model_operations", "approval_queue")); // Assert: Foreign keys exist var aqFkCount = await CountForeignKeysAsync(connection, "model_operations", "approval_queue"); Assert.True(aqFkCount > 0, "approval_queue should have FK to shadow_run"); } /// /// Gate 2: Idempotency - Re-running migrations is safe /// [Fact] public async Task Migration0008_Idempotency_ReRunningIsSafe() { // Arrange: Apply migration once await ApplyMigration0008(); await using var connection = await _dataSource.OpenConnectionAsync(); // Insert a test record var runId = Guid.NewGuid(); var modelId = Guid.NewGuid(); await using var insertCmd = connection.CreateCommand(); insertCmd.CommandText = """ INSERT INTO model_operations.shadow_run (run_id, model_id, window_start, window_end, status) VALUES (@runId, @modelId, @start, @end, 'Pending'); """; insertCmd.Parameters.AddWithValue("@runId", runId); insertCmd.Parameters.AddWithValue("@modelId", modelId); insertCmd.Parameters.AddWithValue("@start", new DateOnly(2024, 1, 2)); insertCmd.Parameters.AddWithValue("@end", new DateOnly(2024, 8, 31)); await insertCmd.ExecuteNonQueryAsync(); // Act: Re-run migration await ApplyMigration0008(); // Assert: Data survived, table is unchanged await using var selectCmd = connection.CreateCommand(); selectCmd.CommandText = "SELECT COUNT(*) FROM model_operations.shadow_run WHERE run_id = @runId;"; selectCmd.Parameters.AddWithValue("@runId", runId); var count = (long?)await selectCmd.ExecuteScalarAsync(); Assert.Equal(1, count); } /// /// Gate 3: Constraints - Status transitions enforced /// [Fact] public async Task Migration0008_Constraint_StatusValuesEnforced() { await ApplyMigration0008(); await using var connection = await _dataSource.OpenConnectionAsync(); // Act: Try to insert invalid status var runId = Guid.NewGuid(); var modelId = Guid.NewGuid(); await using var invalidCmd = connection.CreateCommand(); invalidCmd.CommandText = """ INSERT INTO model_operations.shadow_run (run_id, model_id, window_start, window_end, status) VALUES (@runId, @modelId, @start, @end, 'InvalidStatus'); """; invalidCmd.Parameters.AddWithValue("@runId", runId); invalidCmd.Parameters.AddWithValue("@modelId", modelId); invalidCmd.Parameters.AddWithValue("@start", new DateOnly(2024, 1, 2)); invalidCmd.Parameters.AddWithValue("@end", new DateOnly(2024, 8, 31)); // Assert: Constraint violation await Assert.ThrowsAsync( () => invalidCmd.ExecuteNonQueryAsync()); } /// /// Gate 3: Constraints - Window order enforced (start <= end) /// [Fact] public async Task Migration0008_Constraint_WindowOrderEnforced() { await ApplyMigration0008(); await using var connection = await _dataSource.OpenConnectionAsync(); // Act: Try to insert with end < start var runId = Guid.NewGuid(); var modelId = Guid.NewGuid(); await using var invalidCmd = connection.CreateCommand(); invalidCmd.CommandText = """ INSERT INTO model_operations.shadow_run (run_id, model_id, window_start, window_end, status) VALUES (@runId, @modelId, @start, @end, 'Pending'); """; invalidCmd.Parameters.AddWithValue("@runId", runId); invalidCmd.Parameters.AddWithValue("@modelId", modelId); invalidCmd.Parameters.AddWithValue("@start", new DateOnly(2024, 8, 31)); invalidCmd.Parameters.AddWithValue("@end", new DateOnly(2024, 1, 2)); // Assert: Constraint violation await Assert.ThrowsAsync( () => invalidCmd.ExecuteNonQueryAsync()); } /// /// Gate 3: Trigger - Inbox processed_at validation /// [Fact] public async Task Migration0009_Trigger_InboxProcessedAtRequired() { await ApplyMigration0008(); await ApplyMigration0009(); await using var connection = await _dataSource.OpenConnectionAsync(); // Insert outbox message first var outboxId = Guid.NewGuid(); await using var outboxCmd = connection.CreateCommand(); outboxCmd.CommandText = """ INSERT INTO building_blocks.outbox_message (message_id, event_type, schema_version, payload_json, correlation_id, occurred_at, payload_hash) VALUES (@id, 'TestEvent', 1, '{"test":"data"}'::jsonb, 'test-correlation', CURRENT_TIMESTAMP, 'test-hash'); """; outboxCmd.Parameters.AddWithValue("@id", outboxId); await outboxCmd.ExecuteNonQueryAsync(); // Act: Try to set status=Processed without processed_at var inboxId = Guid.NewGuid(); await using var invalidCmd = connection.CreateCommand(); invalidCmd.CommandText = """ INSERT INTO building_blocks.inbox_message (message_id, consumer, status) VALUES (@id, 'TestConsumer', 'Processed'); """; invalidCmd.Parameters.AddWithValue("@id", inboxId); // Assert: Trigger violation await Assert.ThrowsAsync( () => invalidCmd.ExecuteNonQueryAsync()); } /// /// Gate 3: Idempotency - Inbox dedup constraint /// [Fact] public async Task Migration0009_Constraint_InboxIdempotencyEnforced() { await ApplyMigration0008(); await ApplyMigration0009(); await using var connection = await _dataSource.OpenConnectionAsync(); // Insert outbox message var outboxId = Guid.NewGuid(); await using var outboxCmd = connection.CreateCommand(); outboxCmd.CommandText = """ INSERT INTO building_blocks.outbox_message (message_id, event_type, schema_version, payload_json, correlation_id, occurred_at, payload_hash) VALUES (@id, 'TestEvent', 1, '{"test":"data"}'::jsonb, 'test-correlation', CURRENT_TIMESTAMP, 'test-hash'); """; outboxCmd.Parameters.AddWithValue("@id", outboxId); await outboxCmd.ExecuteNonQueryAsync(); // Insert first inbox record await using var insertCmd = connection.CreateCommand(); insertCmd.CommandText = """ INSERT INTO building_blocks.inbox_message (message_id, consumer, status) VALUES (@outboxId, 'Consumer1', 'Pending'); """; insertCmd.Parameters.AddWithValue("@outboxId", outboxId); await insertCmd.ExecuteNonQueryAsync(); // Act: Try to insert duplicate (same outbox_id + consumer_id) await using var duplicateCmd = connection.CreateCommand(); duplicateCmd.CommandText = """ INSERT INTO building_blocks.inbox_message (message_id, consumer, status) VALUES (@outboxId, 'Consumer1', 'Pending'); """; duplicateCmd.Parameters.AddWithValue("@outboxId", outboxId); // Assert: UNIQUE constraint violation await Assert.ThrowsAsync( () => duplicateCmd.ExecuteNonQueryAsync()); } /// /// Gate 3: Trigger - Approval workflow validation /// [Fact] public async Task Migration0010_Trigger_ApprovalRequiresApprovedBy() { await ApplyMigration0008(); await ApplyMigration0010(); await using var connection = await _dataSource.OpenConnectionAsync(); // Insert shadow_run first var runId = Guid.NewGuid(); var modelId = Guid.NewGuid(); await using var srCmd = connection.CreateCommand(); srCmd.CommandText = """ INSERT INTO model_operations.shadow_run (run_id, model_id, window_start, window_end, status) VALUES (@runId, @modelId, @start, @end, 'Pending'); """; srCmd.Parameters.AddWithValue("@runId", runId); srCmd.Parameters.AddWithValue("@modelId", modelId); srCmd.Parameters.AddWithValue("@start", new DateOnly(2024, 1, 2)); srCmd.Parameters.AddWithValue("@end", new DateOnly(2024, 8, 31)); await srCmd.ExecuteNonQueryAsync(); // Act: Try to set status=Approved without approved_by await using var invalidCmd = connection.CreateCommand(); invalidCmd.CommandText = """ INSERT INTO model_operations.approval_queue (run_id, model_id, status) VALUES (@runId, @modelId, 'Approved'); """; invalidCmd.Parameters.AddWithValue("@runId", runId); invalidCmd.Parameters.AddWithValue("@modelId", modelId); // Assert: Trigger violation await Assert.ThrowsAsync( () => invalidCmd.ExecuteNonQueryAsync()); } /// /// Gate 3: Trigger - Rejection requires reason /// [Fact] public async Task Migration0010_Trigger_RejectionRequiresReason() { await ApplyMigration0008(); await ApplyMigration0010(); await using var connection = await _dataSource.OpenConnectionAsync(); // Insert shadow_run var runId = Guid.NewGuid(); var modelId = Guid.NewGuid(); await using var srCmd = connection.CreateCommand(); srCmd.CommandText = """ INSERT INTO model_operations.shadow_run (run_id, model_id, window_start, window_end, status) VALUES (@runId, @modelId, @start, @end, 'Pending'); """; srCmd.Parameters.AddWithValue("@runId", runId); srCmd.Parameters.AddWithValue("@modelId", modelId); srCmd.Parameters.AddWithValue("@start", new DateOnly(2024, 1, 2)); srCmd.Parameters.AddWithValue("@end", new DateOnly(2024, 8, 31)); await srCmd.ExecuteNonQueryAsync(); // Act: Try to set status=Rejected without rejection_reason await using var invalidCmd = connection.CreateCommand(); invalidCmd.CommandText = """ INSERT INTO model_operations.approval_queue (run_id, model_id, status) VALUES (@runId, @modelId, 'Rejected'); """; invalidCmd.Parameters.AddWithValue("@runId", runId); invalidCmd.Parameters.AddWithValue("@modelId", modelId); // Assert: Trigger violation await Assert.ThrowsAsync( () => invalidCmd.ExecuteNonQueryAsync()); } /// /// Gate 4: Failure Recovery - FK constraint prevents orphaned approvals /// [Fact] public async Task Migration0010_ForeignKey_PreventsShadowRunDeletion() { await ApplyMigration0008(); await ApplyMigration0010(); await using var connection = await _dataSource.OpenConnectionAsync(); // Insert shadow_run and approval var runId = Guid.NewGuid(); var modelId = Guid.NewGuid(); var approverId = Guid.NewGuid(); await using var srCmd = connection.CreateCommand(); srCmd.CommandText = """ INSERT INTO model_operations.shadow_run (run_id, model_id, window_start, window_end, status) VALUES (@runId, @modelId, @start, @end, 'Pending'); """; srCmd.Parameters.AddWithValue("@runId", runId); srCmd.Parameters.AddWithValue("@modelId", modelId); srCmd.Parameters.AddWithValue("@start", new DateOnly(2024, 1, 2)); srCmd.Parameters.AddWithValue("@end", new DateOnly(2024, 8, 31)); await srCmd.ExecuteNonQueryAsync(); await using var aqCmd = connection.CreateCommand(); aqCmd.CommandText = """ INSERT INTO model_operations.approval_queue (run_id, model_id, status, approved_by, approval_reason) VALUES (@runId, @modelId, 'Approved', @approverId, 'Test approval'); """; aqCmd.Parameters.AddWithValue("@runId", runId); aqCmd.Parameters.AddWithValue("@modelId", modelId); aqCmd.Parameters.AddWithValue("@approverId", approverId); await aqCmd.ExecuteNonQueryAsync(); // Act: Try to delete shadow_run (should fail due to FK) await using var deleteCmd = connection.CreateCommand(); deleteCmd.CommandText = "DELETE FROM model_operations.shadow_run WHERE run_id = @runId;"; deleteCmd.Parameters.AddWithValue("@runId", runId); // Assert: FK violation prevents deletion await Assert.ThrowsAsync( () => deleteCmd.ExecuteNonQueryAsync()); } /// /// Gate 4: Indexes - Common queries are indexed /// [Fact] public async Task Migration0008_Indexes_ExistForCommonQueries() { await ApplyMigration0008(); await using var connection = await _dataSource.OpenConnectionAsync(); var indexes = await GetIndexesAsync(connection, "model_operations", "shadow_run"); Assert.Contains("idx_shadow_run_model_created", indexes); Assert.Contains("idx_shadow_run_status", indexes); Assert.Contains("idx_shadow_run_published_at", indexes); } // ========== Helpers ========== private async Task ApplyMigration0008() { await using var connection = await _dataSource.OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = File.ReadAllText(Path.Combine(AppContext.BaseDirectory, "migrations", "0022_model_operations_execution_schema.sql")); await cmd.ExecuteNonQueryAsync(); } private async Task ApplyMigration0009() { await using var connection = await _dataSource.OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = File.ReadAllText(Path.Combine(AppContext.BaseDirectory, "migrations", "0023_inbox_status_contract.sql")); await cmd.ExecuteNonQueryAsync(); } private async Task ApplyMigration0010() { await using var connection = await _dataSource.OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = File.ReadAllText(Path.Combine(AppContext.BaseDirectory, "migrations", "0022_model_operations_execution_schema.sql")); await cmd.ExecuteNonQueryAsync(); } private static async Task TableExistsAsync(DbConnection connection, string schema, string table) { await using var cmd = connection.CreateCommand(); cmd.CommandText = """ SELECT EXISTS ( SELECT 1 FROM information_schema.tables WHERE table_schema = @schema AND table_name = @table ); """; cmd.Parameters.Add(new NpgsqlParameter("@schema", schema)); cmd.Parameters.Add(new NpgsqlParameter("@table", table)); var result = await cmd.ExecuteScalarAsync(); return result is true or (long)1; } private static async Task> GetTableColumnsAsync(DbConnection connection, string schema, string table) { await using var cmd = connection.CreateCommand(); cmd.CommandText = """ SELECT column_name FROM information_schema.columns WHERE table_schema = @schema AND table_name = @table ORDER BY ordinal_position; """; cmd.Parameters.Add(new NpgsqlParameter("@schema", schema)); cmd.Parameters.Add(new NpgsqlParameter("@table", table)); var columns = new List(); await using var reader = await cmd.ExecuteReaderAsync(); while (await reader.ReadAsync()) { columns.Add(reader.GetString(0)); } return columns; } private static async Task CountForeignKeysAsync(DbConnection connection, string schema, string table) { await using var cmd = connection.CreateCommand(); cmd.CommandText = """ SELECT COUNT(*) FROM information_schema.table_constraints WHERE table_schema = @schema AND table_name = @table AND constraint_type = 'FOREIGN KEY'; """; cmd.Parameters.Add(new NpgsqlParameter("@schema", schema)); cmd.Parameters.Add(new NpgsqlParameter("@table", table)); var result = await cmd.ExecuteScalarAsync(); return Convert.ToInt32(result); } private static async Task> GetIndexesAsync(DbConnection connection, string schema, string table) { await using var cmd = connection.CreateCommand(); cmd.CommandText = """ SELECT indexname FROM pg_indexes WHERE schemaname = @schema AND tablename = @table; """; cmd.Parameters.Add(new NpgsqlParameter("@schema", schema)); cmd.Parameters.Add(new NpgsqlParameter("@table", table)); var indexes = new List(); await using var reader = await cmd.ExecuteReaderAsync(); while (await reader.ReadAsync()) { indexes.Add(reader.GetString(0)); } return indexes; } }