From 76f1d2f8afa9ff011d87155cb6ddf1bda9a9ada1 Mon Sep 17 00:00:00 2001 From: StuartFerguson Date: Tue, 20 Oct 2026 14:58:14 +0100 Subject: [PATCH 1/3] add the subscription service to the docker helper --- .../GenericSteps.cs | 4 +- .../CatchupServiceClient.cs | 168 ++++++++++++++++++ Shared.IntegrationTesting/ContainerType.cs | 4 +- Shared.IntegrationTesting/DockerPorts.cs | 2 + .../TestContainers/BaseDockerHelper.cs | 110 ++++++++---- .../TestContainers/DockerHelper.cs | 1 + Shared.sln | 4 +- 7 files changed, 253 insertions(+), 40 deletions(-) create mode 100644 Shared.IntegrationTesting/CatchupServiceClient.cs diff --git a/Shared.IntegrationTesting.Tests/GenericSteps.cs b/Shared.IntegrationTesting.Tests/GenericSteps.cs index 6c7c554..905b985 100644 --- a/Shared.IntegrationTesting.Tests/GenericSteps.cs +++ b/Shared.IntegrationTesting.Tests/GenericSteps.cs @@ -33,7 +33,7 @@ public GenericSteps(ScenarioContext scenarioContext, TestingContext testingConte [BeforeScenario()] public async Task StartSystem() { // Initialise a logger - String scenarioName = this.ScenarioContext.ScenarioInfo.Title.Replace(" ", ""); + String scenarioName = this.ScenarioContext.ScenarioInfo.Title.Replace(" ", String.Empty); NlogLogger logger = new(); logger.Initialise(LogManager.GetLogger(scenarioName), scenarioName); LogManager.AddHiddenAssembly(typeof(NlogLogger).Assembly); @@ -45,7 +45,7 @@ public async Task StartSystem() { DockerServices.CallbackHandler | DockerServices.FileProcessor | DockerServices.TestHost | DockerServices.TransactionProcessor | DockerServices.TransactionProcessorAcl | DockerServices.ConfigurationHost | - DockerServices.EstateManagementUI | DockerServices.EstateReporting; + DockerServices.EstateManagementUI | DockerServices.EstateReporting | DockerServices.SubscriptionService; this.TestingContext.Logger = logger; this.TestingContext.Logger.LogInformation("About to Start Containers for Scenario Run"); diff --git a/Shared.IntegrationTesting/CatchupServiceClient.cs b/Shared.IntegrationTesting/CatchupServiceClient.cs new file mode 100644 index 0000000..371c459 --- /dev/null +++ b/Shared.IntegrationTesting/CatchupServiceClient.cs @@ -0,0 +1,168 @@ +using System; +using System.Collections.Generic; +using System.Net; +using System.Net.Http; +using System.Net.Http.Json; +using System.Text; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; + +namespace Shared.IntegrationTesting +{ + public sealed record SubscriptionConfigurationRequest + { + public string? SubscriptionId { get; init; } + public string? SecondaryIndexName { get; init; } + public string? EndpointUrl { get; init; } + public string? Tag { get; init; } + public int TimeoutSeconds { get; init; } = 30; + public int RetryMaxAttempts { get; init; } = 3; + public int RetryDelaySeconds { get; init; } = 1; + public int CheckpointBatchSize { get; init; } = 100; + public bool ContinueOnParked { get; init; } + public AuthenticationConfiguration? Authentication { get; init; } + public bool SoftDeleteParked { get; init; } = true; + } + + public interface ISubscriptionServiceClient { + Task>> GetEndpointsAsync(CancellationToken cancellationToken = default); + Task> CreateEndpointAsync(EndpointDefinition endpoint, CancellationToken cancellationToken = default); + Task>> GetSubscriptionConfigurationsAsync(CancellationToken cancellationToken = default); + Task> CreateSubscriptionConfigurationAsync(SubscriptionConfigurationRequest request, CancellationToken cancellationToken = default); + Task StartSubscriptionAsync(string subscriptionId, CancellationToken cancellationToken = default); + Task> CreateIndexAsync(string indexName, string payload, CancellationToken cancellationToken = default); + Task> GetIndexAsync(String indexName,CancellationToken cancellationToken = default); + } + + public record SubscriptionApiResponse(HttpStatusCode StatusCode, string? Content, string? ContentType) + { + public bool IsSuccessStatusCode => (int)StatusCode is >= 200 and < 300; + } + + public sealed record SubscriptionApiResponse(HttpStatusCode StatusCode, T? Value, string? Content, string? ContentType) + : SubscriptionApiResponse(StatusCode, Content, ContentType); + + public sealed class SubscriptionServiceClient(Func urlResolver, HttpClient httpClient) : ISubscriptionServiceClient { + private readonly Func UrlResolver = urlResolver; + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + + public Task> GetIndexAsync(String indexName, CancellationToken cancellationToken = default) => + this.SendRawAsync(HttpMethod.Get, $"indexes/{Escape(indexName)}", null, cancellationToken); + + public Task>> GetEndpointsAsync(CancellationToken cancellationToken = default) => this.SendJsonAsync>(HttpMethod.Get, "endpoints", null, cancellationToken); + + public Task> CreateEndpointAsync(EndpointDefinition endpoint, CancellationToken cancellationToken = default) => this.SendJsonAsync(HttpMethod.Post, "endpoints", endpoint, cancellationToken); + + public Task>> GetSubscriptionConfigurationsAsync(CancellationToken cancellationToken = default) => this.SendJsonAsync>(HttpMethod.Get, "subscriptions/config", null, cancellationToken); + + public Task> CreateSubscriptionConfigurationAsync(SubscriptionConfigurationRequest request, CancellationToken cancellationToken = default) => this.SendJsonAsync(HttpMethod.Post, "subscriptions/config", request, cancellationToken); + + public Task StartSubscriptionAsync(string subscriptionId, CancellationToken cancellationToken = default) => + this.SendAsync(HttpMethod.Post, $"subscriptions/config/{Escape(subscriptionId)}/start", null, cancellationToken); + + public Task> CreateIndexAsync(string indexName, string payload, CancellationToken cancellationToken = default) => + this.SendRawAsync(HttpMethod.Post, $"indexes/{Escape(indexName)}", payload, cancellationToken); + + private async Task> SendJsonAsync(HttpMethod method, string path, object? body, CancellationToken cancellationToken) + { + var response = await this.SendAsync(method, path, body, cancellationToken); + T? value = default; + if (response.IsSuccessStatusCode && !string.IsNullOrWhiteSpace(response.Content)) + value = JsonSerializer.Deserialize(response.Content, JsonOptions); + + return new(response.StatusCode, value, response.Content, response.ContentType); + } + + private async Task> SendRawAsync(HttpMethod method, string path, string? body, CancellationToken cancellationToken) + { + var response = await this.SendAsync(method, path, body is null ? null : new StringContent(body, Encoding.UTF8, "application/json"), cancellationToken); + return new(response.StatusCode, response.Content, response.Content, response.ContentType); + } + + private async Task SendAsync(HttpMethod method, string path, object? body, CancellationToken cancellationToken) + { + using var content = body is null ? null : JsonContent.Create(body, options: JsonOptions); + return await this.SendAsync(method, path, content, cancellationToken); + } + + private async Task SendAsync(HttpMethod method, string path, HttpContent? content, CancellationToken cancellationToken) { + var baseUrl = this.UrlResolver("CatchupService"); + + string fullUrl = $"{baseUrl}/{path}"; + + + using var request = new HttpRequestMessage(method, fullUrl) { Content = content }; + using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken); + var responseContent = await response.Content.ReadAsStringAsync(cancellationToken); + return new(response.StatusCode, string.IsNullOrWhiteSpace(responseContent) ? null : responseContent, response.Content.Headers.ContentType?.ToString()); + } + + private static string Escape(string value) => Uri.EscapeDataString(value); + + private static void AddQuery(ICollection query, string name, string? value) + { + if (!string.IsNullOrWhiteSpace(value)) + query.Add($"{Escape(name)}={Escape(value)}"); + } + + private static string AddQueryString(string path, IReadOnlyCollection query) => + query.Count == 0 ? path : $"{path}?{string.Join('&', query)}"; + } + + public sealed record SubscriptionDefinition( + string SubscriptionId, + string SecondaryIndexName, + string EndpointUrl, + string Tag, + TimeoutSettings Timeout, + RetrySettings Retry, + CheckpointSettings Checkpoint, + bool ContinueOnParked = false, + AuthenticationConfiguration? Authentication = null) + { + public int? EndpointId { get; init; } + public bool Enabled { get; init; } = true; + public SubscriptionOperationalState OperationalState { get; init; } = SubscriptionOperationalState.Healthy; + public string? OperationalReason { get; init; } + public bool SoftDeleteParked { get; init; } = true; + + public Uri Endpoint => new(EndpointUrl, UriKind.Absolute); + } + + public sealed record RetrySettings(int MaxAttempts, TimeSpan Delay) + { + public static RetrySettings Default { get; } = new(3, TimeSpan.FromSeconds(1)); + } + + public sealed record CheckpointSettings(int BatchSize) + { + public static CheckpointSettings Default { get; } = new(100); + } + + public sealed record TimeoutSettings(TimeSpan RequestTimeout) + { + public static TimeoutSettings Default { get; } = new(TimeSpan.FromSeconds(30)); + } + + public sealed record AuthenticationConfiguration(string? Scheme, IReadOnlyDictionary Parameters) + { + public static AuthenticationConfiguration Empty { get; } = new(null, new Dictionary()); + } + + + public enum SubscriptionOperationalState + { + Healthy = 0, + Stopped = 1, + Faulted = 2 + } + + public sealed record EndpointDefinition( + string Name, + string Url, + AuthenticationConfiguration? Authentication = null) + { + public Uri Uri => new(Url, UriKind.Absolute); + } +} diff --git a/Shared.IntegrationTesting/ContainerType.cs b/Shared.IntegrationTesting/ContainerType.cs index 21875e0..1d8b925 100644 --- a/Shared.IntegrationTesting/ContainerType.cs +++ b/Shared.IntegrationTesting/ContainerType.cs @@ -16,6 +16,7 @@ public enum ContainerType ConfigurationHost, EstateManagementUI, EstateReporting, + SubscriptionService, NotSet } @@ -34,5 +35,6 @@ public enum DockerServices TransactionProcessorAcl = 256, ConfigurationHost = 512, EstateManagementUI = 1024, - EstateReporting = 2048 + EstateReporting = 2048, + SubscriptionService = 4096, } diff --git a/Shared.IntegrationTesting/DockerPorts.cs b/Shared.IntegrationTesting/DockerPorts.cs index 151689e..99ebca5 100644 --- a/Shared.IntegrationTesting/DockerPorts.cs +++ b/Shared.IntegrationTesting/DockerPorts.cs @@ -25,5 +25,7 @@ public static class DockerPorts public static readonly Int32 EstateManagementUIDockerPort = 5004; public static readonly Int32 EstateReportingDockerPort = 5011; + + public static readonly Int32 SubscriptionServiceDockerPort = 8080; public static readonly Int32 SqlServerDockerPort = 1433; } \ No newline at end of file diff --git a/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs b/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs index 368607b..a3156af 100644 --- a/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs +++ b/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs @@ -50,9 +50,13 @@ public abstract class BaseDockerHelper{ protected String CallbackHandlerContainerName; protected String EstateReportingContainerName; + protected String SubscriptionServiceContainerName; + protected Int32 CallbackHandlerPort; protected Int32 EstateReportingPort; + protected Int32 SubscriptionServicePort; + protected (String clientId, String clientSecret) ClientDetails; protected List<(DockerServices, IContainer)> Containers; @@ -139,16 +143,13 @@ protected BaseDockerHelper(Boolean skipHealthChecks=false) { this.ImageDetails.Add(ContainerType.ConfigurationHost, ("stuartferguson/mobileconfiguration:master", true)); this.ImageDetails.Add(ContainerType.EstateManagementUI, ("stuartferguson/estatemanagementui:master", true)); this.ImageDetails.Add(ContainerType.EstateReporting, ("stuartferguson/estatereportingapi:master", true)); + this.ImageDetails.Add(ContainerType.SubscriptionService, ("stuartferguson/subscriptionservice:master", true)); this.HostPorts = new Dictionary(); } - - public Boolean IsSecureKurrentDb { get; protected set; } - - public virtual Dictionary GetAdditionalVariables(ContainerType containerType){ Dictionary result = new(); @@ -193,6 +194,7 @@ protected virtual IReadOnlyList> GetStartupGroups( List> startupGroups = [ [DockerServices.SqlServer], [DockerServices.KurrentDb], + [DockerServices.SubscriptionService], [ DockerServices.MessagingService, DockerServices.SecurityService, @@ -205,7 +207,7 @@ protected virtual IReadOnlyList> GetStartupGroups( DockerServices.EstateReporting ], [ - DockerServices.EstateManagementUI + DockerServices.EstateManagementUI, ] ]; @@ -232,6 +234,7 @@ public static DockerClient GetDockerHost() { ContainerType.SqlServer => this.SqlServerPort, ContainerType.EstateManagementUI => this.EstateManagementUiPort, ContainerType.EstateReporting => this.EstateReportingPort, + ContainerType.SubscriptionService => this.SubscriptionServicePort, _ when key == ContainerType.KurrentDb && this.IsSecureKurrentDb => this.KurrentDbSecureHttpPort, _ when key == ContainerType.KurrentDb && this.IsSecureKurrentDb == false => this.KurrentDbHttpPort, _ => null @@ -304,6 +307,7 @@ public virtual void SetupContainerNames(){ this.ConfigHostContainerName = $"mobileconfighost{this.TestId:N}"; this.EstateManagementUiContainerName = $"estateadministrationui{this.TestId:N}"; this.EstateReportingContainerName = $"estatereporting{this.TestId:N}"; + this.SubscriptionServiceContainerName= $"subscriptionservice{this.TestId:N}"; } protected Int32 EstateManagementUiPort; @@ -354,8 +358,6 @@ public virtual ContainerBuilder SetupKurrentDbContainer(){ (String imageName, Boolean useLatest) imageDetails = this.GetImageDetails(ContainerType.KurrentDb).Data; this.Trace($"About to Start Kurrent Db Container using image [{imageDetails.imageName}]"); - - kurrentDbContainer = kurrentDbContainer.WithName(this.KurrentDbContainerName) // similar to WithName() .WithImage(imageDetails.imageName) .WithEnvironment(environmentVariables) @@ -449,6 +451,24 @@ protected virtual ContainerBuilder SetupEstateManagementUiContainer() return containerBuilder; } + public virtual ContainerBuilder SetupSubscriptionServiceContainer() { + this.Trace("About to Start Subscription Service Container"); + + Dictionary environmentVariables = new(); + environmentVariables.Add("ConnectionStrings:KurrentDb", this.GenerateKurrentDbConnectionString()); + environmentVariables.Add("ConnectionStrings:SubscriptionServiceDb", this.SetConnectionString("SubscriptionService", this.UseSecureSqlServerDatabase)); + + (String imageName, Boolean useLatest) imageDetails = this.GetImageDetails(ContainerType.SubscriptionService).Data; + + ContainerBuilder subscriptionServiceContainer = new ContainerBuilder().WithName(this.SubscriptionServiceContainerName) + .WithImage(imageDetails.imageName) + .WithEnvironment(environmentVariables) + .MountHostFolder(this.HostTraceFolder) + .WithPortBinding(DockerPorts.SubscriptionServiceDockerPort, true); + + return subscriptionServiceContainer; + } + public virtual ContainerBuilder SetupConfigHostContainer() { this.Trace("About to Start Config Host Container"); @@ -494,7 +514,6 @@ public virtual ContainerBuilder SetupMessagingServiceContainer(){ messagingServiceContainer = messagingServiceContainer .WithPortBinding(DockerPorts.MessagingServiceDockerPort, true); - //.WithWaitStrategy(Wait.ForUnixContainer().UntilInternalTcpPortIsAvailable(DockerPorts.MessagingServiceDockerPort)); return messagingServiceContainer; } @@ -533,17 +552,14 @@ public virtual ContainerBuilder SetupSecurityServiceContainer() environmentVariables.Add("ConnectionStrings:AuthenticationDbContext", this.SetConnectionString($"Authentication-{this.TestId}", this.UseSecureSqlServerDatabase)); Dictionary additionalEnvironmentVariables = this.GetAdditionalVariables(ContainerType.SecurityService); - if (additionalEnvironmentVariables != null) - { - foreach (KeyValuePair additionalEnvironmentVariable in additionalEnvironmentVariables) - { + if (additionalEnvironmentVariables != null) { + foreach (KeyValuePair additionalEnvironmentVariable in additionalEnvironmentVariables) { environmentVariables.Add(additionalEnvironmentVariable.Key, additionalEnvironmentVariable.Value); } } SimpleResults.Result<(String imageName, Boolean useLatest)> imageDetailsResult = this.GetImageDetails(ContainerType.SecurityService); - if (imageDetailsResult.IsFailed) - { + if (imageDetailsResult.IsFailed) { throw new Exception($"Image details not found for {ContainerType.SecurityService}"); } @@ -554,12 +570,10 @@ public virtual ContainerBuilder SetupSecurityServiceContainer() .MountHostFolder(this.IntegrationTestCertificate.CertificateDirectory, this.IntegrationTestCertificate.ContainerCertificateDirectory); Int32? hostPort = this.GetHostPort(ContainerType.SecurityService); - if (hostPort is null || hostPort <= 0) - { + if (hostPort is null or <= 0) { securityServiceContainer = securityServiceContainer.WithPortBinding(DockerPorts.SecurityServiceDockerPort, true); } - else - { + else { securityServiceContainer = securityServiceContainer.WithPortBinding(DockerPorts.SecurityServiceDockerPort, hostPort.Value); } @@ -581,19 +595,6 @@ public virtual ContainerBuilder ConfigureSqlContainer() return containerService; } - //public virtual async Task SetupSqlServerContainer(INetwork networkService){ - // if (this.SqlCredentials == default) - // throw new ArgumentNullException("Sql Credentials have not been set"); - - // IContainer databaseServerContainer = await this.StartContainer2(this.ConfigureSqlContainer, - // new List{ - // networkService - // }, - // DockerServices.SqlServer); - - // return databaseServerContainer; - //} - public virtual ContainerBuilder SetupTestHostContainer(){ this.Trace("About to Start Test Hosts Container"); @@ -712,10 +713,9 @@ public virtual ContainerBuilder SetupTransactionProcessorContainer(){ return transactionProcessorContainer; } - public abstract Task StartContainersForScenarioRun( - String scenarioName, - DockerServices dockerServices, - CancellationToken cancellationToken); + public abstract Task StartContainersForScenarioRun(String scenarioName, + DockerServices dockerServices, + CancellationToken cancellationToken); public abstract Task StopContainersForScenarioRun(DockerServices sharedDockerServices); @@ -778,6 +778,44 @@ protected virtual async Task CreatePersistentSubscription((String streamName, St this.Trace($"Subscription Group [{subscription.groupName}] Stream [{subscription.streamName}] created"); } + public ISubscriptionServiceClient SubscriptionServiceClient; + + protected virtual async Task CreateSubscription(EndpointDefinition endpointDefinition, String indexName, String tag, CancellationToken cancellationToken) { + // Check if the endpoint exists already + SubscriptionApiResponse> existingEndpoints = await this.SubscriptionServiceClient.GetEndpointsAsync(cancellationToken); + + if (existingEndpoints.Value.SingleOrDefault(s => s.Url == endpointDefinition.Url) == null) { + // Need to create a new endpoint here + SubscriptionApiResponse result = await this.SubscriptionServiceClient.CreateEndpointAsync(endpointDefinition, cancellationToken); + if (result.IsSuccessStatusCode == false) + throw new Exception($"Error creating endpoint name {endpointDefinition.Name} Status Code = {result.StatusCode}"); + } + + String subscriptionId = $"{indexName}_{endpointDefinition.Name}_{tag}"; + + SubscriptionApiResponse> subscriptionConfigList = await this.SubscriptionServiceClient.GetSubscriptionConfigurationsAsync(cancellationToken); + + if (subscriptionConfigList.Value.SingleOrDefault(x => x.SubscriptionId == subscriptionId) == null) { + SubscriptionConfigurationRequest subscriptionConfigurationRequest = new(); + subscriptionConfigurationRequest = subscriptionConfigurationRequest with { + ContinueOnParked = true, + CheckpointBatchSize = 100, + EndpointUrl = endpointDefinition.Url, + RetryDelaySeconds = 5, + RetryMaxAttempts = 0, + SecondaryIndexName = indexName, + SoftDeleteParked = true, + SubscriptionId = subscriptionId, + Tag = tag, + TimeoutSeconds = 30, + }; + + SubscriptionApiResponse result = await this.SubscriptionServiceClient.CreateSubscriptionConfigurationAsync(subscriptionConfigurationRequest, cancellationToken); + if (result.IsSuccessStatusCode == false) + throw new Exception($"Error subscription configuration id {subscriptionConfigurationRequest.SubscriptionId} Status Code = {result.StatusCode}"); + } + } + protected async Task DoSqlServerHealthCheck( IContainer containerService, CancellationToken cancellationToken){ @@ -841,7 +879,7 @@ await Retry.For(async cancellationToken => using HttpResponseMessage infoResponse = await KurrentDbHealthCheckClient.GetAsync(url, cancellationToken).ConfigureAwait(false); infoResponse.StatusCode.ShouldBe(HttpStatusCode.OK); - String infoData = await infoResponse.Content.ReadAsStringAsync().ConfigureAwait(false); + String infoData = await infoResponse.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); this.Trace(infoData); }, cancellationToken); diff --git a/Shared.IntegrationTesting/TestContainers/DockerHelper.cs b/Shared.IntegrationTesting/TestContainers/DockerHelper.cs index 7c62047..722360b 100644 --- a/Shared.IntegrationTesting/TestContainers/DockerHelper.cs +++ b/Shared.IntegrationTesting/TestContainers/DockerHelper.cs @@ -131,6 +131,7 @@ async Task StartWithTrace(Func DockerServices.ConfigurationHost => this.SetupConfigHostContainer, DockerServices.EstateManagementUI => this.SetupEstateManagementUiContainer, DockerServices.EstateReporting => this.SetupEstateReportingContainer, + DockerServices.SubscriptionService => this.SetupSubscriptionServiceContainer, _ => throw new InvalidOperationException($"No startup plan exists for docker service [{service}]") }; diff --git a/Shared.sln b/Shared.sln index cec953f..d0a9b41 100644 --- a/Shared.sln +++ b/Shared.sln @@ -1,4 +1,3 @@ - Microsoft Visual Studio Solution File, Format Version 12.00 # Visual Studio Version 18 VisualStudioVersion = 18.0.11205.157 @@ -139,6 +138,9 @@ Global {EA875346-AE13-41F2-AFF0-93F470EA536E} = {1B484B1C-212E-4CF5-B666-213E9546E11C} {5A8811D2-46B8-4E92-E6DF-B482185EA211} = {1B484B1C-212E-4CF5-B666-213E9546E11C} {9A7AF486-C3B6-463D-812E-DD6D5517C7F3} = {1B484B1C-212E-4CF5-B666-213E9546E11C} + {95B8412E-6764-4115-BB49-2EC3674AF2B3} = {1B484B1C-212E-4CF5-B666-213E9546E11C} + {BF7D54EF-C5C8-422E-86FB-73A371DA18E1} = {C641A0E0-A26E-4B16-89E2-DD6E33509C6E} + {705E5A37-38A8-4F4B-9594-180B90B1A2BC} = {C641A0E0-A26E-4B16-89E2-DD6E33509C6E} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {34032CEC-68CD-41EA-8707-D450689AEA46} From c4000e58ec51388e34e566980c93ad5f4b089e59 Mon Sep 17 00:00:00 2001 From: StuartFerguson Date: Tue, 20 Oct 2026 15:12:15 +0100 Subject: [PATCH 2/3] some tweaks to the code for stability --- .../TestContainers/BaseDockerHelper.cs | 20 ++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs b/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs index a3156af..f76b07d 100644 --- a/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs +++ b/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs @@ -780,15 +780,27 @@ protected virtual async Task CreatePersistentSubscription((String streamName, St public ISubscriptionServiceClient SubscriptionServiceClient; - protected virtual async Task CreateSubscription(EndpointDefinition endpointDefinition, String indexName, String tag, CancellationToken cancellationToken) { + protected virtual async Task CreateSubscription(EndpointDefinition endpointDefinition, String indexName, String tag, CancellationToken cancellationToken) { + if (endpointDefinition == null) { + return Result.Failure("Endpoint definition must not be null"); + } + + if (this.SubscriptionServiceClient == null) { + return Result.Failure("SubscriptionServiceClient needs initialised"); + } + // Check if the endpoint exists already SubscriptionApiResponse> existingEndpoints = await this.SubscriptionServiceClient.GetEndpointsAsync(cancellationToken); + if (existingEndpoints.IsSuccessStatusCode == false) { + return Result.Failure($"Error getting existing endpoints Status Code = {existingEndpoints.StatusCode}"); + } + if (existingEndpoints.Value.SingleOrDefault(s => s.Url == endpointDefinition.Url) == null) { // Need to create a new endpoint here SubscriptionApiResponse result = await this.SubscriptionServiceClient.CreateEndpointAsync(endpointDefinition, cancellationToken); if (result.IsSuccessStatusCode == false) - throw new Exception($"Error creating endpoint name {endpointDefinition.Name} Status Code = {result.StatusCode}"); + return Result.Failure($"Error creating endpoint name {endpointDefinition.Name} Status Code = {result.StatusCode}"); } String subscriptionId = $"{indexName}_{endpointDefinition.Name}_{tag}"; @@ -812,8 +824,10 @@ protected virtual async Task CreateSubscription(EndpointDefinition endpointDefin SubscriptionApiResponse result = await this.SubscriptionServiceClient.CreateSubscriptionConfigurationAsync(subscriptionConfigurationRequest, cancellationToken); if (result.IsSuccessStatusCode == false) - throw new Exception($"Error subscription configuration id {subscriptionConfigurationRequest.SubscriptionId} Status Code = {result.StatusCode}"); + return Result.Failure($"Error subscription configuration id {subscriptionConfigurationRequest.SubscriptionId} Status Code = {result.StatusCode}"); } + + return Result.Success(); } protected async Task DoSqlServerHealthCheck( From a2d9c3344d8a3726c9017f39d02be826624f22b4 Mon Sep 17 00:00:00 2001 From: StuartFerguson Date: Tue, 20 Oct 2026 15:28:12 +0100 Subject: [PATCH 3/3] Capture subscription service host port --- .../DockerHelperStartupPlanTests.cs | 1 + Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs | 4 ++++ 2 files changed, 5 insertions(+) diff --git a/Shared.IntegrationTesting.Tests/DockerHelperStartupPlanTests.cs b/Shared.IntegrationTesting.Tests/DockerHelperStartupPlanTests.cs index b55cd1c..5b7da4a 100644 --- a/Shared.IntegrationTesting.Tests/DockerHelperStartupPlanTests.cs +++ b/Shared.IntegrationTesting.Tests/DockerHelperStartupPlanTests.cs @@ -122,6 +122,7 @@ public void StartupGroups_place_foundation_services_first_and_estate_management_ startupGroups.Select(group => group.ToArray()).ShouldBe([ [DockerServices.SqlServer], [DockerServices.KurrentDb], + [DockerServices.SubscriptionService], [ DockerServices.MessagingService, DockerServices.SecurityService, diff --git a/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs b/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs index f76b07d..69174b2 100644 --- a/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs +++ b/Shared.IntegrationTesting/TestContainers/BaseDockerHelper.cs @@ -1201,6 +1201,7 @@ await Retry.For(async cancellationToken => DockerServices.ConfigurationHost => ContainerType.ConfigurationHost, DockerServices.EstateManagementUI => ContainerType.EstateManagementUI, DockerServices.EstateReporting => ContainerType.EstateReporting, + DockerServices.SubscriptionService => ContainerType.SubscriptionService, _ => ContainerType.NotSet }; @@ -1317,6 +1318,9 @@ UInt16 GetPort(Int32 dockerPort) => case ContainerType.EstateReporting: EstateReportingPort = GetPort(DockerPorts.EstateReportingDockerPort); break; + case ContainerType.SubscriptionService: + SubscriptionServicePort = GetPort(DockerPorts.SubscriptionServiceDockerPort); + break; } }