Files
Joseph Doherty 29ac56006d feat: implement ETL pipeline redesign and ConfigManager improvements
- Add pipeline registry with JSON-based configuration and hot-reload support
- Implement manual sync request feature with API, client UI, and database
- Improve ConfigManager: connection string dropdown in pipeline editor,
  step delete/reorder functionality, and fix JSON parsing for ConnectionStrings
2026-01-22 17:48:33 -05:00

435 lines
13 KiB
C#

using JdeScoping.DataSync.Configuration;
using JdeScoping.DataSync.Options;
using JdeScoping.DataSync.Services;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using NSubstitute;
using Shouldly;
using System.Text.Json;
using Xunit;
using MsOptions = Microsoft.Extensions.Options.Options;
namespace JdeScoping.DataSync.Tests.Services;
public class PipelineRegistryTests : IDisposable
{
private readonly string _testDirectory;
private readonly IPipelineValidator _validator;
private readonly IOptions<DataSyncOptions> _options;
private readonly ILogger<PipelineRegistry> _logger;
private readonly IHostEnvironment _environment;
public PipelineRegistryTests()
{
_testDirectory = Path.Combine(Path.GetTempPath(), $"PipelineRegistryTests_{Guid.NewGuid():N}");
Directory.CreateDirectory(_testDirectory);
_validator = new PipelineValidator();
_logger = Substitute.For<ILogger<PipelineRegistry>>();
_environment = Substitute.For<IHostEnvironment>();
_environment.ContentRootPath.Returns(_testDirectory);
_options = MsOptions.Create(new DataSyncOptions
{
PipelinesDirectory = "Pipelines",
StrictPipelineValidation = false
});
// Create the Pipelines subdirectory
Directory.CreateDirectory(Path.Combine(_testDirectory, "Pipelines"));
}
public void Dispose()
{
if (Directory.Exists(_testDirectory))
{
Directory.Delete(_testDirectory, recursive: true);
}
}
private string PipelinesDir => Path.Combine(_testDirectory, "Pipelines");
#region Loading Tests
[Fact]
public async Task LoadPipelines_ValidDirectory_LoadsAll()
{
// Arrange
CreatePipelineFile("Pipeline1", true);
CreatePipelineFile("Pipeline2", true);
CreatePipelineFile("Pipeline3", true);
var registry = CreateRegistry();
// Act
var result = await registry.ReloadAsync();
// Assert
result.Success.ShouldBeTrue();
result.PipelinesLoaded.ShouldBe(3);
registry.GetAllPipelines().Count.ShouldBe(3);
}
[Fact]
public async Task LoadPipelines_EmptyDirectory_ReturnsEmpty()
{
// Arrange - empty directory (Pipelines subdirectory is already created but empty)
var registry = CreateRegistry();
// Act
var result = await registry.ReloadAsync();
// Assert
result.Success.ShouldBeTrue();
result.PipelinesLoaded.ShouldBe(0);
registry.GetAllPipelines().ShouldBeEmpty();
}
[Fact]
public async Task LoadPipelines_OnlyLoadsJsonFiles()
{
// Arrange
CreatePipelineFile("ValidPipeline", true);
File.WriteAllText(Path.Combine(PipelinesDir, "readme.txt"), "Some text");
File.WriteAllText(Path.Combine(PipelinesDir, "config.xml"), "<xml/>");
var registry = CreateRegistry();
// Act
var result = await registry.ReloadAsync();
// Assert
result.PipelinesLoaded.ShouldBe(1);
}
#endregion
#region Error Handling Tests
[Fact]
public async Task LoadPipelines_InvalidJson_ReturnsError()
{
// Arrange
File.WriteAllText(Path.Combine(PipelinesDir, "pipeline.Invalid.json"), "{ invalid json }");
var registry = CreateRegistry();
// Act
var result = await registry.ReloadAsync();
// Assert
result.Errors.ShouldContain(e => e.ErrorType.Contains("parse", StringComparison.OrdinalIgnoreCase));
}
[Fact]
public async Task LoadPipelines_DuplicateNames_ReturnsError()
{
// Arrange - two files with same pipeline name
CreatePipelineFile("DuplicateName", true, "pipeline.First.json");
CreatePipelineFile("DuplicateName", true, "pipeline.Second.json");
var registry = CreateRegistry();
// Act
var result = await registry.ReloadAsync();
// Assert
result.Errors.ShouldContain(e => e.ErrorType.Contains("validation", StringComparison.OrdinalIgnoreCase)
&& e.Messages.Any(m => m.Contains("Duplicate", StringComparison.OrdinalIgnoreCase)));
}
#endregion
#region Retrieval Tests
[Fact]
public async Task GetPipeline_ByName_ReturnsCorrect()
{
// Arrange
CreatePipelineFile("TestPipeline", true);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act
var pipeline = registry.GetPipeline("TestPipeline");
// Assert
pipeline.ShouldNotBeNull();
pipeline.Name.ShouldBe("TestPipeline");
}
[Fact]
public async Task GetPipeline_CaseInsensitive_ReturnsCorrect()
{
// Arrange
CreatePipelineFile("MyPipeline", true);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act
var pipeline = registry.GetPipeline("mypipeline");
// Assert
pipeline.ShouldNotBeNull();
pipeline.Name.ShouldBe("MyPipeline");
}
[Fact]
public async Task GetPipeline_NotFound_ReturnsNull()
{
// Arrange
CreatePipelineFile("ExistingPipeline", true);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act
var pipeline = registry.GetPipeline("NonExistentPipeline");
// Assert
pipeline.ShouldBeNull();
}
[Fact]
public async Task GetEnabledPipelines_OnlyReturnsEnabled()
{
// Arrange
CreatePipelineFile("EnabledPipeline1", true);
CreatePipelineFile("EnabledPipeline2", true);
CreatePipelineFile("DisabledPipeline", false);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act
var enabledPipelines = registry.GetEnabledPipelines();
// Assert
enabledPipelines.Count.ShouldBe(2);
enabledPipelines.ShouldAllBe(p => p.IsEnabled);
}
[Fact]
public async Task GetAllPipelines_IncludesDisabled()
{
// Arrange
CreatePipelineFile("EnabledPipeline", true);
CreatePipelineFile("DisabledPipeline", false);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act
var allPipelines = registry.GetAllPipelines();
// Assert
allPipelines.Count.ShouldBe(2);
allPipelines.ShouldContain(p => !p.IsEnabled);
}
#endregion
#region Reload Tests
[Fact]
public async Task ReloadAsync_UpdatesSnapshot()
{
// Arrange
CreatePipelineFile("Pipeline1", true);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Add another pipeline
CreatePipelineFile("Pipeline2", true);
// Act
await registry.ReloadAsync();
// Assert
registry.GetAllPipelines().Count.ShouldBe(2);
}
[Fact]
public async Task ReloadAsync_IncrementsVersion()
{
// Arrange
CreatePipelineFile("Pipeline1", true);
var registry = CreateRegistry();
// Act
await registry.ReloadAsync();
var version1 = registry.Version;
await registry.ReloadAsync();
var version2 = registry.Version;
// Assert
version2.ShouldBe(version1 + 1);
}
[Fact]
public async Task ReloadAsync_UpdatesLastLoadedAt()
{
// Arrange
CreatePipelineFile("Pipeline1", true);
var registry = CreateRegistry();
// Act
var before = DateTime.UtcNow;
await registry.ReloadAsync();
var after = DateTime.UtcNow;
// Assert
registry.LastLoadedAt.ShouldNotBeNull();
registry.LastLoadedAt.Value.ShouldBeGreaterThanOrEqualTo(before.AddSeconds(-1));
registry.LastLoadedAt.Value.ShouldBeLessThanOrEqualTo(after.AddSeconds(1));
}
#endregion
#region IsValidPipelineAndSyncType Tests
[Fact]
public async Task IsValidPipelineAndSyncType_ValidCombination_ReturnsTrue()
{
// Arrange
CreatePipelineFile("TestPipeline", true, massSyncInterval: 1440, dailySyncInterval: 60);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act & Assert
registry.IsValidPipelineAndSyncType("TestPipeline", "mass").ShouldBeTrue();
registry.IsValidPipelineAndSyncType("TestPipeline", "daily").ShouldBeTrue();
}
[Fact]
public async Task IsValidPipelineAndSyncType_UnsupportedSyncType_ReturnsFalse()
{
// Arrange - only mass sync
CreatePipelineFile("TestPipeline", true, massSyncInterval: 1440);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act & Assert
registry.IsValidPipelineAndSyncType("TestPipeline", "hourly").ShouldBeFalse();
}
[Fact]
public async Task IsValidPipelineAndSyncType_UnknownPipeline_ReturnsFalse()
{
// Arrange
CreatePipelineFile("TestPipeline", true);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act & Assert
registry.IsValidPipelineAndSyncType("UnknownPipeline", "mass").ShouldBeFalse();
}
[Fact]
public async Task IsValidPipelineAndSyncType_DisabledPipeline_ReturnsFalse()
{
// Arrange
CreatePipelineFile("DisabledPipeline", false, massSyncInterval: 1440);
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act & Assert
registry.IsValidPipelineAndSyncType("DisabledPipeline", "mass").ShouldBeFalse();
}
#endregion
#region Thread Safety Tests
[Fact]
public async Task ReadOperations_ThreadSafe()
{
// Arrange
for (int i = 0; i < 10; i++)
{
CreatePipelineFile($"Pipeline{i}", true);
}
var registry = CreateRegistry();
await registry.ReloadAsync();
// Act - multiple concurrent reads
var tasks = Enumerable.Range(0, 100).Select(_ => Task.Run(() =>
{
var all = registry.GetAllPipelines();
var enabled = registry.GetEnabledPipelines();
var specific = registry.GetPipeline("Pipeline5");
return all.Count + enabled.Count + (specific != null ? 1 : 0);
}));
// Assert - no exceptions
var results = await Task.WhenAll(tasks);
results.ShouldAllBe(r => r > 0);
}
[Fact]
public async Task ConcurrentReloads_Serialized()
{
// Arrange
CreatePipelineFile("Pipeline1", true);
var registry = CreateRegistry();
// Act - multiple concurrent reloads
var tasks = Enumerable.Range(0, 10).Select(_ => registry.ReloadAsync());
// Assert - no exceptions and final state is valid
await Task.WhenAll(tasks);
registry.GetAllPipelines().Count.ShouldBe(1);
}
#endregion
#region Helper Methods
private PipelineRegistry CreateRegistry() =>
new PipelineRegistry(_options, _validator, _logger, _environment);
private void CreatePipelineFile(
string name,
bool isEnabled,
string? fileName = null,
int? massSyncInterval = 1440,
int? dailySyncInterval = null,
int? hourlySyncInterval = null)
{
var pipeline = new
{
name,
isEnabled,
isManualOnly = !massSyncInterval.HasValue && !dailySyncInterval.HasValue && !hourlySyncInterval.HasValue && !isEnabled,
massSyncIntervalMinutes = massSyncInterval,
dailySyncIntervalMinutes = dailySyncInterval,
hourlySyncIntervalMinutes = hourlySyncInterval,
source = new
{
connection = "jde",
query = "SELECT * FROM TestTable"
},
destination = new
{
table = $"{name}_Table",
matchColumns = new[] { "Id" }
}
};
var json = JsonSerializer.Serialize(pipeline, new JsonSerializerOptions
{
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
WriteIndented = true,
DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull
});
var actualFileName = fileName ?? $"pipeline.{name}.json";
var filePath = Path.Combine(PipelinesDir, actualFileName);
File.WriteAllText(filePath, json);
}
#endregion
}