Phase 8: Production readiness — failover tests, security hardening, sandboxing, deployment docs
- WP-1-3: Central/site failover + dual-node recovery tests (17 tests) - WP-4: Performance testing framework for target scale (7 tests) - WP-5: Security hardening (LDAPS, JWT key length, no secrets in logs) (11 tests) - WP-6: Script sandboxing adversarial tests (28 tests, all forbidden APIs) - WP-7: Recovery drill test scaffolds (5 tests) - WP-8: Observability validation (structured logs, correlation IDs, metrics) (6 tests) - WP-9: Message contract compatibility (forward/backward compat) (18 tests) - WP-10: Deployment packaging (installation guide, production checklist, topology) - WP-11: Operational runbooks (failover, troubleshooting, maintenance) 92 new tests, all passing. Zero warnings.
This commit is contained in:
Joseph Doherty
215
tests/ScadaLink.IntegrationTests/SiteFailoverTests.cs
Normal file
215
tests/ScadaLink.IntegrationTests/SiteFailoverTests.cs
Normal file
@@ -0,0 +1,215 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ScadaLink.Commons.Messages.Health;
|
||||
using ScadaLink.Commons.Messages.Streaming;
|
||||
using ScadaLink.Commons.Types.Enums;
|
||||
using ScadaLink.StoreAndForward;
|
||||
|
||||
namespace ScadaLink.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// WP-2 (Phase 8): Full-system failover testing — Site.
|
||||
/// Verifies S&F buffer takeover, DCL reconnection structure, alarm re-evaluation,
|
||||
/// and script trigger resumption after site failover.
|
||||
/// </summary>
|
||||
public class SiteFailoverTests
|
||||
{
|
||||
[Trait("Category", "Integration")]
|
||||
[Fact]
|
||||
public async Task StoreAndForward_BufferSurvivesRestart_MessagesRetained()
|
||||
{
|
||||
// Simulates site failover: messages buffered in SQLite survive process restart.
|
||||
// The standby node picks up the same SQLite file and retries pending messages.
|
||||
var dbPath = Path.Combine(Path.GetTempPath(), $"sf_failover_{Guid.NewGuid():N}.db");
|
||||
var connStr = $"Data Source={dbPath}";
|
||||
|
||||
try
|
||||
{
|
||||
// Phase 1: Buffer messages on "primary" node
|
||||
var storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await storage.InitializeAsync();
|
||||
|
||||
var message = new StoreAndForwardMessage
|
||||
{
|
||||
Id = Guid.NewGuid().ToString("N"),
|
||||
Category = StoreAndForwardCategory.ExternalSystem,
|
||||
Target = "https://api.example.com/data",
|
||||
PayloadJson = """{"temperature":42.5}""",
|
||||
RetryCount = 2,
|
||||
MaxRetries = 50,
|
||||
RetryIntervalMs = 30000,
|
||||
CreatedAt = DateTimeOffset.UtcNow,
|
||||
Status = StoreAndForwardMessageStatus.Pending,
|
||||
OriginInstanceName = "pump-station-1"
|
||||
};
|
||||
|
||||
await storage.EnqueueAsync(message);
|
||||
|
||||
// Phase 2: "Standby" node opens the same database (simulating failover)
|
||||
var standbyStorage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await standbyStorage.InitializeAsync();
|
||||
|
||||
var pending = await standbyStorage.GetMessagesForRetryAsync();
|
||||
Assert.Single(pending);
|
||||
Assert.Equal(message.Id, pending[0].Id);
|
||||
Assert.Equal("pump-station-1", pending[0].OriginInstanceName);
|
||||
Assert.Equal(2, pending[0].RetryCount);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (File.Exists(dbPath))
|
||||
File.Delete(dbPath);
|
||||
}
|
||||
}
|
||||
|
||||
[Trait("Category", "Integration")]
|
||||
[Fact]
|
||||
public async Task StoreAndForward_ParkedMessages_SurviveFailover()
|
||||
{
|
||||
var dbPath = Path.Combine(Path.GetTempPath(), $"sf_parked_{Guid.NewGuid():N}.db");
|
||||
var connStr = $"Data Source={dbPath}";
|
||||
|
||||
try
|
||||
{
|
||||
var storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await storage.InitializeAsync();
|
||||
|
||||
var parkedMsg = new StoreAndForwardMessage
|
||||
{
|
||||
Id = Guid.NewGuid().ToString("N"),
|
||||
Category = StoreAndForwardCategory.Notification,
|
||||
Target = "alert-list",
|
||||
PayloadJson = """{"subject":"Critical alarm"}""",
|
||||
RetryCount = 50,
|
||||
MaxRetries = 50,
|
||||
RetryIntervalMs = 30000,
|
||||
CreatedAt = DateTimeOffset.UtcNow.AddHours(-1),
|
||||
LastAttemptAt = DateTimeOffset.UtcNow,
|
||||
Status = StoreAndForwardMessageStatus.Parked,
|
||||
LastError = "SMTP connection timeout",
|
||||
OriginInstanceName = "compressor-1"
|
||||
};
|
||||
|
||||
await storage.EnqueueAsync(parkedMsg);
|
||||
|
||||
// Standby opens same DB
|
||||
var standbyStorage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await standbyStorage.InitializeAsync();
|
||||
|
||||
var (parked, count) = await standbyStorage.GetParkedMessagesAsync();
|
||||
Assert.Equal(1, count);
|
||||
Assert.Equal("SMTP connection timeout", parked[0].LastError);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (File.Exists(dbPath))
|
||||
File.Delete(dbPath);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void AlarmReEvaluation_IncomingValue_TriggersNewState()
|
||||
{
|
||||
// Structural verification: AlarmStateChanged carries all data needed for
|
||||
// re-evaluation after failover. When DCL reconnects and pushes new values,
|
||||
// the Alarm Actor evaluates from the incoming value (not stale state).
|
||||
var alarmEvent = new AlarmStateChanged(
|
||||
"pump-station-1",
|
||||
"HighPressureAlarm",
|
||||
AlarmState.Active,
|
||||
1,
|
||||
DateTimeOffset.UtcNow);
|
||||
|
||||
Assert.Equal(AlarmState.Active, alarmEvent.State);
|
||||
Assert.Equal("pump-station-1", alarmEvent.InstanceUniqueName);
|
||||
|
||||
// After failover, a new value triggers re-evaluation
|
||||
var clearedEvent = new AlarmStateChanged(
|
||||
"pump-station-1",
|
||||
"HighPressureAlarm",
|
||||
AlarmState.Normal,
|
||||
1,
|
||||
DateTimeOffset.UtcNow.AddSeconds(5));
|
||||
|
||||
Assert.Equal(AlarmState.Normal, clearedEvent.State);
|
||||
Assert.True(clearedEvent.Timestamp > alarmEvent.Timestamp);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ScriptTriggerResumption_ValueChangeTriggersScript()
|
||||
{
|
||||
// Structural verification: AttributeValueChanged messages from DCL after reconnection
|
||||
// will be routed to Script Actors, which evaluate triggers based on incoming values.
|
||||
// No stale trigger state needed — triggers fire on new values.
|
||||
var valueChange = new AttributeValueChanged(
|
||||
"pump-station-1",
|
||||
"OPC:ns=2;s=Pressure",
|
||||
"Pressure",
|
||||
150.0,
|
||||
"Good",
|
||||
DateTimeOffset.UtcNow);
|
||||
|
||||
Assert.Equal("Pressure", valueChange.AttributeName);
|
||||
Assert.Equal("OPC:ns=2;s=Pressure", valueChange.AttributePath);
|
||||
Assert.Equal(150.0, valueChange.Value);
|
||||
Assert.Equal("Good", valueChange.Quality);
|
||||
}
|
||||
|
||||
[Trait("Category", "Integration")]
|
||||
[Fact]
|
||||
public async Task StoreAndForward_BufferDepth_ReportedAfterFailover()
|
||||
{
|
||||
var dbPath = Path.Combine(Path.GetTempPath(), $"sf_depth_{Guid.NewGuid():N}.db");
|
||||
var connStr = $"Data Source={dbPath}";
|
||||
|
||||
try
|
||||
{
|
||||
var storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await storage.InitializeAsync();
|
||||
|
||||
// Enqueue messages in different categories
|
||||
for (var i = 0; i < 5; i++)
|
||||
{
|
||||
await storage.EnqueueAsync(new StoreAndForwardMessage
|
||||
{
|
||||
Id = Guid.NewGuid().ToString("N"),
|
||||
Category = StoreAndForwardCategory.ExternalSystem,
|
||||
Target = "api",
|
||||
PayloadJson = "{}",
|
||||
MaxRetries = 50,
|
||||
RetryIntervalMs = 30000,
|
||||
CreatedAt = DateTimeOffset.UtcNow,
|
||||
Status = StoreAndForwardMessageStatus.Pending,
|
||||
});
|
||||
}
|
||||
|
||||
for (var i = 0; i < 3; i++)
|
||||
{
|
||||
await storage.EnqueueAsync(new StoreAndForwardMessage
|
||||
{
|
||||
Id = Guid.NewGuid().ToString("N"),
|
||||
Category = StoreAndForwardCategory.Notification,
|
||||
Target = "alerts",
|
||||
PayloadJson = "{}",
|
||||
MaxRetries = 50,
|
||||
RetryIntervalMs = 30000,
|
||||
CreatedAt = DateTimeOffset.UtcNow,
|
||||
Status = StoreAndForwardMessageStatus.Pending,
|
||||
});
|
||||
}
|
||||
|
||||
// After failover, standby reads buffer depths
|
||||
var standbyStorage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await standbyStorage.InitializeAsync();
|
||||
|
||||
var depths = await standbyStorage.GetBufferDepthByCategoryAsync();
|
||||
Assert.Equal(5, depths[StoreAndForwardCategory.ExternalSystem]);
|
||||
Assert.Equal(3, depths[StoreAndForwardCategory.Notification]);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (File.Exists(dbPath))
|
||||
File.Delete(dbPath);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user