diff --git a/src/Dexpace.Sdk.Core/Pipeline/HttpPipeline.cs b/src/Dexpace.Sdk.Core/Pipeline/HttpPipeline.cs new file mode 100644 index 0000000..9f4fe56 --- /dev/null +++ b/src/Dexpace.Sdk.Core/Pipeline/HttpPipeline.cs @@ -0,0 +1,84 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +using Dexpace.Sdk.Core.Client; +using Dexpace.Sdk.Core.Configuration; +using Dexpace.Sdk.Core.Errors; +using Dexpace.Sdk.Core.Http.Request; +using Dexpace.Sdk.Core.Http.Response; + +namespace Dexpace.Sdk.Core.Pipeline; + +/// +/// The entry point for sending an HTTP request through the configured policy chain. +/// +/// +/// +/// Instances are created exclusively by . The pipeline is +/// immutable after construction: the sorted policy array and transport are captured at build time. +/// +/// +/// Sync bridge. blocks the calling thread by driving the async chain +/// synchronously via GetAwaiter().GetResult(). Callers on a thread pool should prefer +/// to avoid thread starvation. +/// +/// +public sealed class HttpPipeline +{ + private readonly HttpPipelinePolicy[] _policies; + private readonly IAsyncHttpClient _transport; + + internal HttpPipeline(HttpPipelinePolicy[] policies, IAsyncHttpClient transport) + { + _policies = policies; + _transport = transport; + } + + /// + /// Asynchronously sends through the pipeline and returns the + /// response produced by the terminal transport. + /// + /// The request to send. + /// Client options that apply to this call. + /// An optional token to cancel the call. + /// + /// A that completes with the once + /// the pipeline chain has finished. + /// + /// + /// No policy or the transport produced a by the time the chain + /// completed (i.e. the pipeline was short-circuited without setting a response). + /// + public async ValueTask SendAsync( + Request request, + DexpaceClientOptions options, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + ArgumentNullException.ThrowIfNull(options); + + var context = new PipelineContext(request, options, cancellationToken); + await new PipelineRunner(_policies, 0, _transport).RunAsync(context).ConfigureAwait(false); + + return context.Response + ?? throw new PipelineAbortedException( + "The pipeline completed without producing a response."); + } + + /// + /// Synchronously sends through the pipeline and returns the + /// response. Blocks the calling thread until the async chain completes. + /// + /// The request to send. + /// Client options that apply to this call. + /// An optional token to cancel the call. + /// The produced by the pipeline. + /// + /// The pipeline completed without producing a response. + /// + public Response Send( + Request request, + DexpaceClientOptions options, + CancellationToken cancellationToken = default) => + SendAsync(request, options, cancellationToken).AsTask().GetAwaiter().GetResult(); +} diff --git a/src/Dexpace.Sdk.Core/Pipeline/HttpPipelinePolicy.cs b/src/Dexpace.Sdk.Core/Pipeline/HttpPipelinePolicy.cs new file mode 100644 index 0000000..695c184 --- /dev/null +++ b/src/Dexpace.Sdk.Core/Pipeline/HttpPipelinePolicy.cs @@ -0,0 +1,46 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +namespace Dexpace.Sdk.Core.Pipeline; + +/// +/// Base class for every policy in the HTTP pipeline. +/// +/// +/// +/// A policy participates in the request/response lifecycle by implementing +/// . Before calling next.RunAsync, a policy may mutate +/// ; after the call returns, it may inspect or replace +/// . +/// +/// +/// Re-entrancy. is a value type, so a policy may call +/// next.RunAsync more than once — this is how redirect and retry policies work. +/// +/// +/// Async-only in v1. There is no synchronous Process override on this base class. +/// The sync entry point on the pipeline drives the async chain via a blocking +/// bridge; concrete policy subclasses are only required to implement the async path. +/// +/// +public abstract class HttpPipelinePolicy +{ + /// + /// The stage at which this policy is inserted in the pipeline. + /// + public abstract PipelineStage Stage { get; } + + /// + /// Asynchronously participates in processing the request/response. + /// + /// + /// The mutable context carrying the current , + /// , and ancillary state for this call. + /// + /// + /// The continuation that runs the remaining policies and eventually invokes the transport. + /// Call this to forward the request; omit the call to short-circuit the chain. + /// + /// A that completes when the policy has finished. + public abstract ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation); +} diff --git a/src/Dexpace.Sdk.Core/Pipeline/PipelineBuilder.cs b/src/Dexpace.Sdk.Core/Pipeline/PipelineBuilder.cs new file mode 100644 index 0000000..5df44ab --- /dev/null +++ b/src/Dexpace.Sdk.Core/Pipeline/PipelineBuilder.cs @@ -0,0 +1,161 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +using Dexpace.Sdk.Core.Client; + +namespace Dexpace.Sdk.Core.Pipeline; + +/// +/// Builds an from an ordered set of +/// instances and a terminal transport. +/// +/// +/// +/// Policies are kept in an internal list. appends to that list; +/// InsertBefore<T>, InsertAfter<T>, Replace<T>, and +/// Remove<T> operate relative to the first policy of runtime type T. +/// +/// +/// performs a stable sort by +/// (preserving list order within a stage) and then validates pillar-stage cardinality: +/// stages marked as pillar admit exactly one policy. A violation throws +/// with an actionable message naming the offending stage. +/// +/// +public sealed class PipelineBuilder +{ + private readonly List _list = []; + + /// + /// Appends to the internal list. The stage-based sort happens at + /// time, not here. + /// + /// The policy to add. + /// This builder (fluent interface). + public PipelineBuilder Add(HttpPipelinePolicy policy) + { + ArgumentNullException.ThrowIfNull(policy); + _list.Add(policy); + return this; + } + + /// + /// Inserts immediately before the first policy of runtime type + /// in the current list. + /// + /// The type to search for. + /// The policy to insert. + /// This builder (fluent interface). + /// + /// No policy of type is present in the list. + /// + public PipelineBuilder InsertBefore(HttpPipelinePolicy policy) + where T : HttpPipelinePolicy + { + ArgumentNullException.ThrowIfNull(policy); + var index = FindFirst(); + _list.Insert(index, policy); + return this; + } + + /// + /// Inserts immediately after the first policy of runtime type + /// in the current list. + /// + /// The type to search for. + /// The policy to insert. + /// This builder (fluent interface). + /// + /// No policy of type is present in the list. + /// + public PipelineBuilder InsertAfter(HttpPipelinePolicy policy) + where T : HttpPipelinePolicy + { + ArgumentNullException.ThrowIfNull(policy); + var index = FindFirst(); + _list.Insert(index + 1, policy); + return this; + } + + /// + /// Replaces the first policy of runtime type with + /// . + /// + /// The type to replace. + /// The replacement policy. + /// This builder (fluent interface). + /// + /// No policy of type is present in the list. + /// + public PipelineBuilder Replace(HttpPipelinePolicy policy) + where T : HttpPipelinePolicy + { + ArgumentNullException.ThrowIfNull(policy); + var index = FindFirst(); + _list[index] = policy; + return this; + } + + /// + /// Removes every policy of runtime type from the list. + /// If none are present, this is a no-op. + /// + /// The type to remove. + /// This builder (fluent interface). + public PipelineBuilder Remove() + where T : HttpPipelinePolicy + { + _list.RemoveAll(p => p is T); + return this; + } + + /// + /// Stable-sorts the registered policies by , validates + /// pillar-stage cardinality, and constructs the with the given + /// as the terminal. + /// + /// + /// The terminal transport; invoked after all policies have run. + /// + /// A fully configured . + /// + /// A pillar stage contains more than one policy. + /// + public HttpPipeline Build(IAsyncHttpClient transport) + { + ArgumentNullException.ThrowIfNull(transport); + + // Stable sort by Stage value + HttpPipelinePolicy[] sorted = [.. _list.OrderBy(p => (int)p.Stage)]; + + // Validate pillar cardinality + foreach (var stage in PipelineStageHelper.PillarStages) + { + var count = sorted.Count(p => p.Stage == stage); + if (count > 1) + { + throw new InvalidOperationException( + $"Pipeline stage '{stage}' is a pillar stage and may contain at most one policy, " + + $"but {count} policies were registered for it. " + + $"Remove the duplicate or use a non-pillar stage."); + } + } + + return new HttpPipeline(sorted, transport); + } + + // Returns the index of the first policy of type T, or throws. + private int FindFirst() where T : HttpPipelinePolicy + { + for (var i = 0; i < _list.Count; i++) + { + if (_list[i] is T) + { + return i; + } + } + + throw new InvalidOperationException( + $"No policy of type '{typeof(T).Name}' is registered in this builder."); + } +} diff --git a/src/Dexpace.Sdk.Core/Pipeline/PipelineRunner.cs b/src/Dexpace.Sdk.Core/Pipeline/PipelineRunner.cs new file mode 100644 index 0000000..72d61d3 --- /dev/null +++ b/src/Dexpace.Sdk.Core/Pipeline/PipelineRunner.cs @@ -0,0 +1,63 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +using Dexpace.Sdk.Core.Client; + +namespace Dexpace.Sdk.Core.Pipeline; + +/// +/// The "next" continuation passed to each call. +/// Advances the policy index and ultimately invokes the transport. +/// +/// +/// +/// is a readonly struct so it carries zero allocation per +/// policy hop. A policy may call more than once (e.g. retry, redirect) +/// because the runner is immutable — each call re-advances from the same index with its own +/// in-flight state. +/// +/// +/// Callers must not retain or share a beyond the duration of +/// . +/// +/// +public readonly struct PipelineRunner +{ + private readonly HttpPipelinePolicy[] _policies; + private readonly int _index; + private readonly IAsyncHttpClient _transport; + + /// + /// Initializes a runner. Called by the pipeline entry point and recursively by + /// . + /// + /// The ordered (sorted-by-stage) policy array. + /// The index of the next policy to invoke. + /// The terminal transport invoked when all policies have run. + internal PipelineRunner(HttpPipelinePolicy[] policies, int index, IAsyncHttpClient transport) + { + _policies = policies; + _index = index; + _transport = transport; + } + + /// + /// Runs the remainder of the pipeline starting at the current index, then invokes the + /// transport if no earlier policy short-circuited. + /// + /// The mutable context for the current call. + /// A that completes when the pipeline tail has run. + public async ValueTask RunAsync(PipelineContext context) + { + if (_index >= _policies.Length) + { + context.Response = await _transport + .ExecuteAsync(context.Request, context.CancellationToken) + .ConfigureAwait(false); + return; + } + + var next = new PipelineRunner(_policies, _index + 1, _transport); + await _policies[_index].ProcessAsync(context, next).ConfigureAwait(false); + } +} diff --git a/src/Dexpace.Sdk.Core/Pipeline/PipelineStage.cs b/src/Dexpace.Sdk.Core/Pipeline/PipelineStage.cs new file mode 100644 index 0000000..8f0ccef --- /dev/null +++ b/src/Dexpace.Sdk.Core/Pipeline/PipelineStage.cs @@ -0,0 +1,70 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +namespace Dexpace.Sdk.Core.Pipeline; + +/// +/// Identifies where in the pipeline chain a policy is inserted. +/// Policies execute in ascending numeric order (outermost first on the way in; +/// innermost first on the way out). +/// +/// +/// +/// Numbers are sparse to leave room for future stages without breaking existing values. +/// +/// +/// Pillar stages — , , , +/// , and — admit exactly one policy each. +/// Adding a second policy to a pillar stage is a configuration error detected at +/// pipeline build time. +/// +/// +/// Non-pillar stages — and — may hold +/// multiple policies, which execute in the order they were registered. +/// +/// +public enum PipelineStage +{ + /// + /// Outermost stage. Runs once per logical operation — opens the operation span and applies + /// the overall deadline. Pillar: at most one policy. + /// + Operation = 100, + + /// + /// Redirect-following stage. Runs outside the retry loop so each hop triggers a full retry + /// sequence. Pillar: at most one policy. + /// + Redirect = 200, + + /// + /// Per-call stage (non-pillar). Policies here run once per logical call, above the retry + /// boundary — suitable for stable cross-attempt concerns such as idempotency keys and + /// client identity headers. + /// + PerCall = 250, + + /// + /// Retry stage. Wraps everything below it so that each retry attempt re-executes all + /// inner stages. Pillar: at most one policy. + /// + Retry = 300, + + /// + /// Per-attempt stage (non-pillar). Policies here run on every attempt inside the retry + /// loop — suitable for per-attempt concerns such as a fresh Date header. + /// + PerAttempt = 400, + + /// + /// Auth stage. Placed inside the retry loop so a token refresh applies to the next + /// retry attempt. Pillar: at most one policy. + /// + Auth = 500, + + /// + /// Diagnostics stage. Closest to the transport wire; wraps the per-attempt span, + /// metrics, and structured log events. Pillar: at most one policy. + /// + Diagnostics = 600, +} diff --git a/src/Dexpace.Sdk.Core/Pipeline/PipelineStageHelper.cs b/src/Dexpace.Sdk.Core/Pipeline/PipelineStageHelper.cs new file mode 100644 index 0000000..78d3f4d --- /dev/null +++ b/src/Dexpace.Sdk.Core/Pipeline/PipelineStageHelper.cs @@ -0,0 +1,37 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +namespace Dexpace.Sdk.Core.Pipeline; + +/// +/// Internal helpers for pillar classification. +/// +internal static class PipelineStageHelper +{ + /// + /// Returns when is a pillar stage that + /// admits at most one policy. + /// + internal static bool IsPillar(PipelineStage stage) => stage switch + { + PipelineStage.Operation => true, + PipelineStage.Redirect => true, + PipelineStage.Retry => true, + PipelineStage.Auth => true, + PipelineStage.Diagnostics => true, + _ => false, + }; + + /// + /// The set of all pillar stages, used for cardinality validation during + /// . + /// + internal static readonly PipelineStage[] PillarStages = + [ + PipelineStage.Operation, + PipelineStage.Redirect, + PipelineStage.Retry, + PipelineStage.Auth, + PipelineStage.Diagnostics, + ]; +} diff --git a/tests/Dexpace.Sdk.Core.Tests/Pipeline/HttpPipelineTests.cs b/tests/Dexpace.Sdk.Core.Tests/Pipeline/HttpPipelineTests.cs new file mode 100644 index 0000000..ce4da1f --- /dev/null +++ b/tests/Dexpace.Sdk.Core.Tests/Pipeline/HttpPipelineTests.cs @@ -0,0 +1,79 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +using Dexpace.Sdk.Core.Client; +using Dexpace.Sdk.Core.Configuration; +using Dexpace.Sdk.Core.Http.Request; +using Dexpace.Sdk.Core.Http.Response; +using Dexpace.Sdk.Core.Pipeline; +using Xunit; + +namespace Dexpace.Sdk.Core.Tests.Pipeline; + +public class HttpPipelineTests +{ + private static Request MakeRequest() => + Request.Get("https://api.example.com/v1/resource"); + + private static DexpaceClientOptions MakeOptions() => new(); + + private sealed class CannedTransport(Response canned) : IAsyncHttpClient + { + public Task ExecuteAsync(Request request, CancellationToken cancellationToken = default) => + Task.FromResult(canned); + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } + + [Fact] + public async Task SendAsync_ReturnsTransportResponse() + { + var expected = new Response(Status.Ok); + var pipeline = new PipelineBuilder().Build(new CannedTransport(expected)); + + var actual = await pipeline.SendAsync(MakeRequest(), MakeOptions()); + + Assert.Same(expected, actual); + } + + [Fact] + public void Send_ReturnsTransportResponse() + { + var expected = new Response(Status.Ok); + var pipeline = new PipelineBuilder().Build(new CannedTransport(expected)); + + var actual = pipeline.Send(MakeRequest(), MakeOptions()); + + Assert.Same(expected, actual); + } + + [Fact] + public async Task SendAsync_WithPolicies_PoliciesInvokedAndResponseReturned() + { + var log = new List(); + var expected = new Response(Status.Ok); + + var pipeline = new PipelineBuilder() + .Add(new LoggingPolicy("a", PipelineStage.Operation, log)) + .Add(new LoggingPolicy("b", PipelineStage.PerAttempt, log)) + .Build(new CannedTransport(expected)); + + var actual = await pipeline.SendAsync(MakeRequest(), MakeOptions()); + + Assert.Same(expected, actual); + Assert.Equal(["a:in", "b:in", "b:out", "a:out"], log); + } + + private sealed class LoggingPolicy(string name, PipelineStage stage, List log) + : HttpPipelinePolicy + { + public override PipelineStage Stage => stage; + + public override async ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation) + { + log.Add($"{name}:in"); + await continuation.RunAsync(context); + log.Add($"{name}:out"); + } + } +} diff --git a/tests/Dexpace.Sdk.Core.Tests/Pipeline/PipelineBuilderTests.cs b/tests/Dexpace.Sdk.Core.Tests/Pipeline/PipelineBuilderTests.cs new file mode 100644 index 0000000..a20c88c --- /dev/null +++ b/tests/Dexpace.Sdk.Core.Tests/Pipeline/PipelineBuilderTests.cs @@ -0,0 +1,315 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +using Dexpace.Sdk.Core.Client; +using Dexpace.Sdk.Core.Configuration; +using Dexpace.Sdk.Core.Http.Request; +using Dexpace.Sdk.Core.Http.Response; +using Dexpace.Sdk.Core.Pipeline; +using Xunit; + +namespace Dexpace.Sdk.Core.Tests.Pipeline; + +public class PipelineBuilderTests +{ + // --------------------------------------------------------------------------- + // Concrete test policy stubs + // --------------------------------------------------------------------------- + + /// + /// Pass-through stub for cardinality / type-not-found tests. + /// + private abstract class StubPolicy(PipelineStage stage) : HttpPipelinePolicy + { + public override PipelineStage Stage => stage; + + public override ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation) => + continuation.RunAsync(context); + } + + private sealed class OperationStub() : StubPolicy(PipelineStage.Operation); + private sealed class RedirectStub() : StubPolicy(PipelineStage.Redirect); + private sealed class RetryStub() : StubPolicy(PipelineStage.Retry); + private sealed class AuthStub() : StubPolicy(PipelineStage.Auth); + private sealed class DiagnosticsStub() : StubPolicy(PipelineStage.Diagnostics); + private sealed class PerCallStubA() : StubPolicy(PipelineStage.PerCall); + private sealed class PerAttemptStub() : StubPolicy(PipelineStage.PerAttempt); + + /// + /// Recording stub: appends "name:in" before and "name:out" after the continuation. + /// Used by execution-log tests to verify actual invocation order. + /// + private sealed class RecordingPolicy(string name, PipelineStage stage, List log) + : HttpPipelinePolicy + { + public override PipelineStage Stage => stage; + + public override async ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation) + { + log.Add($"{name}:in"); + await continuation.RunAsync(context).ConfigureAwait(false); + log.Add($"{name}:out"); + } + } + + /// Two distinct PerCall recording types for InsertAfter/InsertBefore/Replace/Remove tests. + private sealed class RecordingPerCallA(List log) : HttpPipelinePolicy + { + public override PipelineStage Stage => PipelineStage.PerCall; + + public override async ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation) + { + log.Add("A:in"); + await continuation.RunAsync(context).ConfigureAwait(false); + log.Add("A:out"); + } + } + + private sealed class RecordingPerCallA2(List log) : HttpPipelinePolicy + { + public override PipelineStage Stage => PipelineStage.PerCall; + + public override async ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation) + { + log.Add("A2:in"); + await continuation.RunAsync(context).ConfigureAwait(false); + log.Add("A2:out"); + } + } + + private sealed class RecordingPerCallB(List log) : HttpPipelinePolicy + { + public override PipelineStage Stage => PipelineStage.PerCall; + + public override async ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation) + { + log.Add("B:in"); + await continuation.RunAsync(context).ConfigureAwait(false); + log.Add("B:out"); + } + } + + // --------------------------------------------------------------------------- + // Fakes / helpers + // --------------------------------------------------------------------------- + + private sealed class FakeTransport : IAsyncHttpClient + { + public Task ExecuteAsync(Request request, CancellationToken cancellationToken = default) => + Task.FromResult(new Response(Status.Ok)); + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } + + private static FakeTransport MakeTransport() => new(); + + private static DexpaceClientOptions MakeOptions() => new(); + + private static Request MakeRequest() => Request.Get("https://api.example.com/v1/resource"); + + // --------------------------------------------------------------------------- + // Tests: Stage sort + // --------------------------------------------------------------------------- + + /// + /// Policies added in reverse stage order must execute in ascending stage order + /// (Operation=100, Redirect=200, Diagnostics=600). + /// + [Fact] + public async Task Add_StageSortedExecution_PoliciesRunInStageOrder() + { + var log = new List(); + + // Added in reverse stage order: Diagnostics, Redirect, Operation + var pipeline = new PipelineBuilder() + .Add(new RecordingPolicy("diag", PipelineStage.Diagnostics, log)) + .Add(new RecordingPolicy("redirect", PipelineStage.Redirect, log)) + .Add(new RecordingPolicy("op", PipelineStage.Operation, log)) + .Build(MakeTransport()); + + await pipeline.SendAsync(MakeRequest(), MakeOptions()); + + // Build stable-sorts by stage (ascending), so execution order is: + // op (100) → redirect (200) → diag (600), then unwind. + Assert.Equal( + ["op:in", "redirect:in", "diag:in", "diag:out", "redirect:out", "op:out"], + log); + } + + /// + /// Multiple policies in the same non-pillar stage must not throw at Build time. + /// + [Fact] + public void Add_MultiplePoliciesInSameNonPillarStage_DoesNotThrow() + { + var pipeline = new PipelineBuilder() + .Add(new PerCallStubA()) + .Add(new PerCallStubA()) + .Build(MakeTransport()); + + Assert.NotNull(pipeline); + } + + /// + /// Two policies in a pillar stage must cause Build to throw with the stage name in the message. + /// + [Fact] + public void Build_TwoPoliciesInPillarStage_Throws() + { + var ex = Assert.Throws(() => + new PipelineBuilder() + .Add(new RetryStub()) + .Add(new RetryStub()) + .Build(MakeTransport())); + + Assert.Contains("Retry", ex.Message, StringComparison.OrdinalIgnoreCase); + } + + // --------------------------------------------------------------------------- + // Tests: InsertAfter within a stage + // --------------------------------------------------------------------------- + + /// + /// InsertAfter<A>(B) where A and B share the same stage (PerCall) must produce + /// execution order [A, B] — Build preserves within-stage list order after the sort. + /// + [Fact] + public async Task InsertAfter_SameStage_InsertsAfterAnchor() + { + var log = new List(); + + var pipeline = new PipelineBuilder() + .Add(new RecordingPerCallA(log)) + .InsertAfter(new RecordingPerCallB(log)) // list: [A, B] + .Build(MakeTransport()); + + await pipeline.SendAsync(MakeRequest(), MakeOptions()); + + Assert.Equal(["A:in", "B:in", "B:out", "A:out"], log); + } + + /// + /// InsertAfter when the anchor type is absent must throw with the type name in the message. + /// + [Fact] + public void InsertAfter_TypeNotPresent_Throws() + { + var ex = Assert.Throws(() => + new PipelineBuilder() + .InsertAfter(new PerCallStubA())); + + Assert.Contains("RetryStub", ex.Message, StringComparison.OrdinalIgnoreCase); + } + + // --------------------------------------------------------------------------- + // Tests: InsertBefore within a stage + // --------------------------------------------------------------------------- + + /// + /// InsertBefore<A>(B) where A and B share the same stage (PerCall) must produce + /// execution order [B, A] — Build preserves within-stage list order after the sort. + /// + [Fact] + public async Task InsertBefore_SameStage_InsertsBeforeAnchor() + { + var log = new List(); + + var pipeline = new PipelineBuilder() + .Add(new RecordingPerCallA(log)) + .InsertBefore(new RecordingPerCallB(log)) // list: [B, A] + .Build(MakeTransport()); + + await pipeline.SendAsync(MakeRequest(), MakeOptions()); + + Assert.Equal(["B:in", "A:in", "A:out", "B:out"], log); + } + + /// + /// InsertBefore when the anchor type is absent must throw with the type name in the message. + /// + [Fact] + public void InsertBefore_TypeNotPresent_Throws() + { + var ex = Assert.Throws(() => + new PipelineBuilder() + .InsertBefore(new PerCallStubA())); + + Assert.Contains("RetryStub", ex.Message, StringComparison.OrdinalIgnoreCase); + } + + // --------------------------------------------------------------------------- + // Tests: Replace + // --------------------------------------------------------------------------- + + /// + /// Replace<A>(A2) where A2 is a distinct PerCall type must put A2 in the execution + /// log and exclude A. + /// + [Fact] + public async Task Replace_SubstitutesPolicy_LogContainsReplacementNotOriginal() + { + var log = new List(); + + var pipeline = new PipelineBuilder() + .Add(new RecordingPerCallA(log)) + .Replace(new RecordingPerCallA2(log)) // A swapped for A2 + .Build(MakeTransport()); + + await pipeline.SendAsync(MakeRequest(), MakeOptions()); + + Assert.Contains("A2:in", log); + Assert.DoesNotContain("A:in", log); + } + + /// + /// Replace when the target type is absent must throw with the type name in the message. + /// + [Fact] + public void Replace_TypeNotPresent_Throws() + { + var ex = Assert.Throws(() => + new PipelineBuilder() + .Replace(new RetryStub())); + + Assert.Contains("RetryStub", ex.Message, StringComparison.OrdinalIgnoreCase); + } + + // --------------------------------------------------------------------------- + // Tests: Remove + // --------------------------------------------------------------------------- + + /// + /// Remove<A> with both A and B present must keep B in the execution log and drop A. + /// + [Fact] + public async Task Remove_RemovesMatchingType_OtherPolicyStillRuns() + { + var log = new List(); + + var pipeline = new PipelineBuilder() + .Add(new RecordingPerCallA(log)) + .Add(new RecordingPerCallB(log)) + .Remove() + .Build(MakeTransport()); + + await pipeline.SendAsync(MakeRequest(), MakeOptions()); + + Assert.Contains("B:in", log); + Assert.DoesNotContain("A:in", log); + } + + // --------------------------------------------------------------------------- + // Tests: Edge cases + // --------------------------------------------------------------------------- + + /// + /// An empty builder must produce a valid pipeline that the transport can service. + /// + [Fact] + public async Task Build_EmptyPipeline_TransportResponds() + { + var pipeline = new PipelineBuilder().Build(MakeTransport()); + var response = await pipeline.SendAsync(MakeRequest(), MakeOptions()); + Assert.NotNull(response); + } +} diff --git a/tests/Dexpace.Sdk.Core.Tests/Pipeline/PipelineRunnerTests.cs b/tests/Dexpace.Sdk.Core.Tests/Pipeline/PipelineRunnerTests.cs new file mode 100644 index 0000000..59e50ca --- /dev/null +++ b/tests/Dexpace.Sdk.Core.Tests/Pipeline/PipelineRunnerTests.cs @@ -0,0 +1,106 @@ +// Copyright (c) 2026 dexpace and Omar Aljarrah. +// Licensed under the MIT License. See LICENSE in the repository root for details. + +using Dexpace.Sdk.Core.Client; +using Dexpace.Sdk.Core.Configuration; +using Dexpace.Sdk.Core.Http.Common; +using Dexpace.Sdk.Core.Http.Request; +using Dexpace.Sdk.Core.Http.Response; +using Dexpace.Sdk.Core.Pipeline; +using Xunit; + +namespace Dexpace.Sdk.Core.Tests.Pipeline; + +public class PipelineRunnerTests +{ + private static Request MakeRequest() => + Request.Get("https://api.example.com/v1/resource"); + + private static PipelineContext MakeContext() => + new(MakeRequest(), new DexpaceClientOptions()); + + // --------------------------------------------------------------------------- + // Test fakes + // --------------------------------------------------------------------------- + + private sealed class RecordingPolicy(string name, PipelineStage stage, List log) + : HttpPipelinePolicy + { + public override PipelineStage Stage => stage; + + public override async ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation) + { + log.Add($"{name}:in"); + await continuation.RunAsync(context).ConfigureAwait(false); + log.Add($"{name}:out"); + } + } + + private sealed class FakeTransport : IAsyncHttpClient + { + private readonly Response _canned; + + public FakeTransport(Response? canned = null) => + _canned = canned ?? new Response(Status.Ok); + + public int InvocationCount { get; private set; } + + public Task ExecuteAsync(Request request, CancellationToken cancellationToken = default) + { + InvocationCount++; + return Task.FromResult(_canned); + } + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } + + // --------------------------------------------------------------------------- + // Tests + // --------------------------------------------------------------------------- + + [Fact] + public async Task ExecutionOrder_PoliciesRunInStageOrderInAndReversedOut_TransportInvokedOnce() + { + var log = new List(); + var transport = new FakeTransport(); + + // a = Operation (100), b = PerAttempt (400) — stage ordering: a before b + var policies = new HttpPipelinePolicy[] + { + new RecordingPolicy("a", PipelineStage.Operation, log), + new RecordingPolicy("b", PipelineStage.PerAttempt, log), + }; + + var runner = new PipelineRunner(policies, 0, transport); + var context = MakeContext(); + await runner.RunAsync(context); + + Assert.Equal(["a:in", "b:in", "b:out", "a:out"], log); + Assert.Equal(1, transport.InvocationCount); + } + + [Fact] + public async Task Reentrancy_PolicyCallingNextTwice_TransportInvokedTwice() + { + var transport = new FakeTransport(); + var doubleCallPolicy = new DoubleDipPolicy(); + var policies = new HttpPipelinePolicy[] { doubleCallPolicy }; + + var runner = new PipelineRunner(policies, 0, transport); + var context = MakeContext(); + await runner.RunAsync(context); + + Assert.Equal(2, transport.InvocationCount); + } + + private sealed class DoubleDipPolicy : HttpPipelinePolicy + { + public override PipelineStage Stage => PipelineStage.PerAttempt; + + public override async ValueTask ProcessAsync(PipelineContext context, PipelineRunner continuation) + { + await continuation.RunAsync(context).ConfigureAwait(false); + await continuation.RunAsync(context).ConfigureAwait(false); + } + } +}