Files
jdescopingtool/NEW/tests/JdeScoping.DataSync.Tests/Services/EtlPipelineFactoryTests.cs
T
Joseph Doherty 795c15df56 feat(datasync): add EtlPipelineFactory with JSON config support
- Implement IEtlPipelineFactory with ForTable() method returning a builder
- Load pipeline config from JSON file path (from PipelineOptions)
- Parse config using System.Text.Json with PropertyNameCaseInsensitive
- Builder supports WithMode() and WithMinimumDate() fluent methods
- Create DbQuerySource for source with ParameterFormatConverter for JDE dates
- Create DbBulkMergeDestination or DbBulkImportDestination based on sync mode
- Mass mode defaults to bulkImport, incremental defaults to bulkMerge
- Support destination override in sync mode config
- Execute pre/post scripts from config (prePurge, reIndex, custom scripts)
- Validate config: require mass and incremental modes, reject runtime params
- Add comprehensive tests for factory, builder, and config validation
2026-01-06 13:45:36 -05:00

811 lines
28 KiB
C#

using JdeScoping.DataAccess.Interfaces;
using JdeScoping.DataSync.Configuration;
using JdeScoping.DataSync.Contracts;
using JdeScoping.DataSync.Etl.Pipeline;
using JdeScoping.DataSync.Services;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using NSubstitute;
using Shouldly;
namespace JdeScoping.DataSync.Tests.Services;
public class EtlPipelineFactoryTests
{
private readonly IDbConnectionFactory _connectionFactory;
private readonly ILogger<EtlPipeline> _logger;
public EtlPipelineFactoryTests()
{
_connectionFactory = Substitute.For<IDbConnectionFactory>();
_logger = NullLogger<EtlPipeline>.Instance;
}
#region ForTable Tests
[Fact]
public void ForTable_WithValidTable_ReturnsBuilder()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act
var builder = factory.ForTable("TestTable");
// Assert
builder.ShouldNotBeNull();
builder.ShouldBeAssignableTo<IEtlPipelineBuilder>();
}
[Fact]
public void ForTable_WithUnknownTable_ThrowsInvalidOperationException()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act & Assert
var ex = Should.Throw<InvalidOperationException>(() => factory.ForTable("NonExistentTable"));
ex.Message.ShouldContain("No pipeline configured for table: NonExistentTable");
ex.Message.ShouldContain("TestTable"); // Should list available tables
}
[Fact]
public void ForTable_WithNullTableName_ThrowsArgumentException()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act & Assert
Should.Throw<ArgumentException>(() => factory.ForTable(null!));
}
[Fact]
public void ForTable_WithEmptyTableName_ThrowsArgumentException()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act & Assert
Should.Throw<ArgumentException>(() => factory.ForTable(""));
}
#endregion
#region Builder WithMode Tests
[Fact]
public void Builder_WithMassMode_BuildsPipeline()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Mass)
.Build();
// Assert
pipeline.ShouldNotBeNull();
pipeline.PipelineName.ShouldBe("TestTable");
}
[Fact]
public void Builder_WithIncrementalMode_BuildsPipeline()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build();
// Assert
pipeline.ShouldNotBeNull();
pipeline.PipelineName.ShouldBe("TestTable");
}
[Fact]
public void Builder_DefaultMode_IsIncremental()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act - don't call WithMode()
var pipeline = factory.ForTable("TestTable")
.Build();
// Assert - should work because incremental mode is defined
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_WithUndefinedSyncMode_ThrowsInvalidOperationException()
{
// Arrange - config with only mass mode
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true)
// No incremental mode defined
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
// Act & Assert - validation fails at factory creation
var ex = Should.Throw<InvalidOperationException>(() => CreateFactory(config));
ex.Message.ShouldContain("missing required 'incremental' sync mode");
}
#endregion
#region Builder WithMinimumDate Tests
[Fact]
public void Builder_WithMinimumDate_OverridesConfigOffset()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
var customDate = new DateTime(2024, 1, 1, 0, 0, 0, DateTimeKind.Utc);
// Act - should not throw even though we're overriding
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.WithMinimumDate(customDate)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_WithNullMinimumDate_UsesConfigOffset()
{
// Arrange
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act - null minDt means use config offset
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.WithMinimumDate(null)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
#endregion
#region Config Validation Tests
[Fact]
public void Validate_ConfigMissingMassMode_ThrowsInvalidOperationException()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
// Missing mass mode
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
// Act & Assert
var ex = Should.Throw<InvalidOperationException>(() => CreateFactory(config));
ex.Message.ShouldContain("missing required 'mass' sync mode");
}
[Fact]
public void Validate_ConfigWithRuntimeParameter_ThrowsNotSupportedException()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test WHERE Id = @Id",
new Dictionary<string, ParameterConfig>
{
["id"] = new ParameterConfig("@Id", null, "runtime", null)
}),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00"),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
// Act & Assert
var ex = Should.Throw<NotSupportedException>(() => CreateFactory(config));
ex.Message.ShouldContain("runtime parameter source is not yet supported");
}
#endregion
#region Destination Type Tests
[Fact]
public void Builder_MassMode_DefaultsToBulkImport()
{
// Arrange - no destination override
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act - should use bulkImport for mass mode
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Mass)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_IncrementalMode_DefaultsToBulkMerge()
{
// Arrange - no destination override
var config = CreateValidConfig();
var factory = CreateFactory(config);
// Act - should use bulkMerge for incremental mode
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_ModeWithDestinationOverride_UsesOverride()
{
// Arrange - mass mode with bulkMerge override
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00",
Destination: new DestinationOverride("bulkMerge", null, null)),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act - mass mode should use bulkMerge due to override
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Mass)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_BulkMergeWithoutMatchColumns_ThrowsInvalidOperationException()
{
// Arrange - bulkMerge needs matchColumns
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", null, null), // No matchColumns!
null,
null)
});
var factory = CreateFactory(config);
// Act & Assert
var ex = Should.Throw<InvalidOperationException>(() =>
factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental) // Uses bulkMerge
.Build());
ex.Message.ShouldContain("matchColumns required for bulkMerge");
}
#endregion
#region Parameter Tests
[Fact]
public void Builder_WithOffsetParameter_CreatesSource()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test WHERE UpdateDt >= @MinDt",
new Dictionary<string, ParameterConfig>
{
["minDt"] = new ParameterConfig("@MinDt", null, "offset", null)
}),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_WithJdeJulianParameter_CreatesSource()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("jde", "SELECT * FROM Test WHERE UPMJ >= :dateUpdated",
new Dictionary<string, ParameterConfig>
{
["minDt"] = new ParameterConfig(":dateUpdated", "jdeJulian", "offset", null)
}),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_WithStaticParameter_UsesConfiguredValue()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test WHERE Status = @Status",
new Dictionary<string, ParameterConfig>
{
["status"] = new ParameterConfig("@Status", null, "static", "Active")
}),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_WithStaticParameterNoValue_ThrowsInvalidOperationException()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test WHERE Status = @Status",
new Dictionary<string, ParameterConfig>
{
["status"] = new ParameterConfig("@Status", null, "static", null) // No value!
}),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act & Assert
var ex = Should.Throw<InvalidOperationException>(() =>
factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build());
ex.Message.ShouldContain("Static parameter '@Status' requires a value");
}
#endregion
#region Script Tests
[Fact]
public void Builder_WithPrePurge_AddsTruncateScript()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Mass)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_WithReIndex_AddsRebuildScript()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true, ReIndex: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Mass)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_WithPreScripts_AddsConfiguredScripts()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
["EXEC sp_BeforeSync"],
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Mass)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
[Fact]
public void Builder_WithPostScripts_AddsConfiguredScripts()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
["UPDATE TestTable SET ProcessedFlag = 1 WHERE ProcessedFlag IS NULL"])
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
#endregion
#region Connection Type Tests
[Theory]
[InlineData("jde")]
[InlineData("cms")]
[InlineData("lotfinder")]
public void Builder_WithValidConnectionType_BuildsPipeline(string connectionType)
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig(connectionType, "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Mass)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
#endregion
#region Settings Tests
[Fact]
public void Factory_WithNullSettings_UsesDefaults()
{
// Arrange - null settings should use defaults
var config = new PipelinesRoot(
null, // Null settings
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
#endregion
#region MinDtOffset Format Tests
[Fact]
public void Builder_WithInvalidMinDtOffsetFormat_ThrowsInvalidOperationException()
{
// Arrange
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true),
["incremental"] = new SyncModeConfig("not-a-valid-timespan") // Invalid!
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act & Assert
var ex = Should.Throw<InvalidOperationException>(() =>
factory.ForTable("TestTable")
.WithMode(SyncMode.Incremental)
.Build());
ex.Message.ShouldContain("Invalid minDtOffset format");
}
[Fact]
public void Builder_WithNullMinDtOffset_DoesNotThrow()
{
// Arrange - null offset should be valid (no date filtering)
var config = new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test", null),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig(null, PrePurge: true), // Null offset
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
var factory = CreateFactory(config);
// Act
var pipeline = factory.ForTable("TestTable")
.WithMode(SyncMode.Mass)
.Build();
// Assert
pipeline.ShouldNotBeNull();
}
#endregion
#region Helper Methods
private PipelinesRoot CreateValidConfig()
{
return new PipelinesRoot(
new PipelineSettings("UTC"),
new Dictionary<string, PipelineConfig>
{
["TestTable"] = new PipelineConfig(
new SourceConfig("lotfinder", "SELECT * FROM Test WHERE UpdateDt >= @MinDt",
new Dictionary<string, ParameterConfig>
{
["minDt"] = new ParameterConfig("@MinDt", null, "offset", null)
}),
new Dictionary<string, SyncModeConfig>
{
["mass"] = new SyncModeConfig("-365.00:00:00", PrePurge: true, ReIndex: true),
["incremental"] = new SyncModeConfig("-1.00:00:00")
},
null,
new DestinationConfig("TestTable", ["Id"], null),
null,
null)
});
}
private EtlPipelineFactory CreateFactory(PipelinesRoot config)
{
return new EtlPipelineFactory(_connectionFactory, config, _logger);
}
#endregion
}