30629636973

Committed 31 Jul 2026 12:10PM UTC coverage: 83.079% (+0.8%) from 82.321%

Build # 30629636973

Build Type

Pull #81

github

Committed by

web-flow

Commit Message

Merge 108bf070f into be7540808

Pull Request Pull Request #81: Add a durable skipped job state

Coverage Stats

125 of 133 new or added lines in 10 files covered. (93.98%)

5 existing lines in 3 files now uncovered.

7473 of 8995 relevant lines covered (83.08%)

2.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.39

/src/Immediate.Jobs.Testing/JobTestHarness.cs

using System.Globalization;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Time.Testing;

namespace Immediate.Jobs.Testing;

/// <summary>
/// Hosts Immediate.Jobs with in-memory storage and a controllable clock without starting background threads.
/// </summary>
public sealed class JobTestHarness : IAsyncDisposable, IDisposable
{
        private readonly ServiceProvider _serviceProvider;
        private readonly IJobGraphStorage _graphStorage;
        private bool _disposed;

        /// <summary>Creates a harness at the Unix epoch.</summary>
        /// <param name="configureServices">
        /// Registers generated job definitions and the services used by their handlers. Calling the generated
        /// <c>AddImmediateJobs</c> method here is supported; the harness clock and in-memory provider remain authoritative.
        /// </param>
        public JobTestHarness(Action<IServiceCollection>? configureServices = null)
                : this(DateTimeOffset.UnixEpoch, configureServices)
        {
        }

        /// <summary>Creates a harness at a specified UTC instant.</summary>
        /// <param name="start">The initial UTC time exposed by the controllable clock.</param>
        /// <param name="configureServices">
        /// Registers generated job definitions and the services used by their handlers. Calling the generated
        /// <c>AddImmediateJobs</c> method here is supported; the harness clock and in-memory provider remain authoritative.
        /// </param>
        public JobTestHarness(
                DateTimeOffset start,
                Action<IServiceCollection>? configureServices = null
        )
        {
                TimeProvider = new(start);
                var services = new ServiceCollection();
                _ = services.AddSingleton<TimeProvider>(TimeProvider);
                _ = services.AddSingleton(TimeProvider);
                _ = services.AddSingleton(typeof(ILogger<>), typeof(NullLogger<>));
                _ = services.AddImmediateJobsCore(options =>
                {
                        _ = options.UseInMemory();
                        options.MaxParallelJobs = 1;
                });
                configureServices?.Invoke(services);
                _serviceProvider = services.BuildServiceProvider(new ServiceProviderOptions
                {
                        ValidateScopes = true,
                        ValidateOnBuild = true,
                });

                Services = _serviceProvider;
                Storage = _serviceProvider.GetRequiredService<IJobStorage>();
                _graphStorage = Storage as IJobGraphStorage
                        ?? throw new NotSupportedException(
                                "Batches & continuations require a graph-capable storage provider (a SQL database). " +
                                "The configured provider implements the queue capability only."
                        );
                Batches = new JobBatchScheduler(
                        Storage,
                        TimeProvider,
                        _serviceProvider.GetRequiredService<IIdGenerator>()
                );
                Scheduler = _serviceProvider.GetRequiredService<JobSchedulerService>();
        }

        /// <summary>The controllable clock used by schedulers, storage, retries, and cron materialization.</summary>
        /// <value>The harness clock.</value>
        public FakeTimeProvider TimeProvider { get; }

        /// <summary>The harness service provider.</summary>
        /// <value>The service provider created for the harness.</value>
        public IServiceProvider Services { get; }

        /// <summary>The in-memory durable-state abstraction.</summary>
        /// <value>The storage provider used by the harness.</value>
        public IJobStorage Storage { get; }

        /// <summary>Builds atomic batches against the harness storage and fake clock.</summary>
        /// <value>The batch scheduler configured for the harness.</value>
        public IJobBatchScheduler Batches { get; }

        /// <summary>The production scheduler runner hosted by the harness.</summary>
        /// <value>The scheduler service configured for the harness.</value>
        public JobSchedulerService Scheduler { get; }

        /// <summary>Runs every invocation currently due and returns when the due queue is empty.</summary>
        /// <param name="cancellationToken">A token that can cancel draining.</param>
        /// <returns>A task that completes when no currently due work remains.</returns>
        public ValueTask DrainAsync(CancellationToken cancellationToken = default) =>
                Scheduler.DrainAsync(cancellationToken);

        /// <summary>Advances fake time and then runs every invocation that became due.</summary>
        /// <param name="amount">The amount by which to advance the clock.</param>
        /// <param name="cancellationToken">A token that can cancel draining.</param>
        /// <returns>A task that completes when no newly due work remains.</returns>
        public async ValueTask AdvanceTimeAndDrainAsync(
                TimeSpan amount,
                CancellationToken cancellationToken = default
        )
        {
                if (amount < TimeSpan.Zero)
                        throw new ArgumentOutOfRangeException(nameof(amount), "Fake time cannot move backwards.");

                TimeProvider.Advance(amount);
                await DrainAsync(cancellationToken).ConfigureAwait(false);
        }

        /// <summary>Advances fake time to an absolute instant and drains newly due work.</summary>
        /// <param name="instant">The absolute time to which the clock is advanced.</param>
        /// <param name="cancellationToken">A token that can cancel draining.</param>
        /// <returns>A task that completes when no newly due work remains.</returns>
        public ValueTask AdvanceTimeAndDrainAsync(
                DateTimeOffset instant,
                CancellationToken cancellationToken = default
        )
        {
                var amount = instant - TimeProvider.GetUtcNow();
                if (amount < TimeSpan.Zero)
                        throw new ArgumentOutOfRangeException(nameof(instant), "Fake time cannot move backwards.");
                return AdvanceTimeAndDrainAsync(amount, cancellationToken);
        }

        /// <summary>Returns persisted jobs matching a monitoring query.</summary>
        /// <param name="query">The optional monitoring query. When omitted, up to 1,000 jobs are returned.</param>
        /// <param name="cancellationToken">A token that can cancel the query.</param>
        /// <returns>The persisted jobs that match the query.</returns>
        public ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(
                JobQuery? query = null,
                CancellationToken cancellationToken = default
        ) => Storage.QueryJobsAsync(query ?? new() { Take = 1000 }, cancellationToken);

        /// <summary>Finds an invocation by identifier, or throws a test assertion exception.</summary>
        /// <param name="jobId">The invocation identifier.</param>
        /// <param name="cancellationToken">A token that can cancel the query.</param>
        /// <returns>The persisted job record.</returns>
        public async ValueTask<JobRecord> GetJobAsync(string jobId, CancellationToken cancellationToken = default)
        {
                var jobs = await Storage.QueryJobsAsync(new() { Take = 1000 }, cancellationToken).ConfigureAwait(false);
                return jobs.FirstOrDefault(job => string.Equals(job.Id, jobId, StringComparison.Ordinal))
                        ?? throw new JobTestAssertionException($"Expected job '{jobId}' to have been enqueued, but it was not found.");
        }

        /// <summary>Finds an invocation returned by a typed scheduler call.</summary>
        /// <param name="job">The handle returned by the scheduler.</param>
        /// <param name="cancellationToken">A token that can cancel the query.</param>
        /// <returns>The persisted job record.</returns>
        public ValueTask<JobRecord> GetJobAsync(JobHandle job, CancellationToken cancellationToken = default) =>
                GetJobAsync(job.Id, cancellationToken);

        /// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
        /// <typeparam name="TPayload">The expected payload type.</typeparam>
        /// <param name="jobId">The invocation identifier.</param>
        /// <param name="expectedState">The expected durable state, or <see langword="null"/> to accept any state.</param>
        /// <param name="cancellationToken">A token that can cancel the query.</param>
        /// <returns>The durable record paired with its deserialized payload.</returns>
        public async ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
                string jobId,
                JobState? expectedState = null,
                CancellationToken cancellationToken = default
        )
        {
                var job = await GetJobAsync(jobId, cancellationToken).ConfigureAwait(false);
                if (expectedState is { } state && job.State != state)
                {
                        throw new JobTestAssertionException(
                                $"Expected job '{jobId}' to be {state}, but it was {job.State}."
                        );
                }

                TPayload payload;
                try
                {
                        payload = _serviceProvider.GetRequiredService<IJobSerializer>().Deserialize<TPayload>(job.Payload);
                }
                catch (Exception exception)
                {
                        throw new JobTestAssertionException(
                                $"Job '{jobId}' did not contain a valid {typeof(TPayload).FullName} payload: {exception.Message}"
                        );
                }

                return new(job, payload);
        }

        /// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
        /// <typeparam name="TPayload">The expected payload type.</typeparam>
        /// <param name="job">The handle returned by the scheduler.</param>
        /// <param name="expectedState">The expected durable state, or <see langword="null"/> to accept any state.</param>
        /// <param name="cancellationToken">A token that can cancel the query.</param>
        /// <returns>The durable record paired with its deserialized payload.</returns>
        public ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
                JobHandle job,
                JobState? expectedState = null,
                CancellationToken cancellationToken = default
        ) => AssertEnqueuedAsync<TPayload>(job.Id, expectedState, cancellationToken);

        /// <summary>Asserts that a committed batch and exactly the expected number of members are visible together.</summary>
        /// <param name="batch">The committed batch handle.</param>
        /// <param name="expectedMembers">The expected number of committed batch members.</param>
        /// <param name="cancellationToken">A token that can cancel the assertion query.</param>
        /// <returns>A task that completes when the assertion succeeds.</returns>
        public async ValueTask AssertBatchCommittedAtomicallyAsync(
                BatchHandle batch,
                int expectedMembers,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(batch);
                var status = await _graphStorage.GetBatchStatusAsync(batch.Id, cancellationToken).ConfigureAwait(false)
                        ?? throw new JobTestAssertionException($"Expected batch '{batch.Id}' to be committed, but it was not found.");
                var members = await _graphStorage.QueryBatchMembersAsync(
                        batch.Id,
                        new() { Take = Math.Max(1, expectedMembers + 1) },
                        cancellationToken
                ).ConfigureAwait(false);
                if (status.Total != expectedMembers || members.Count != expectedMembers)
                {
                        throw new JobTestAssertionException(
                                string.Create(CultureInfo.InvariantCulture, $"Expected batch '{batch.Id}' to contain {expectedMembers} jobs, but its header reports {status.Total} and {members.Count} members were found.")
                        );
                }
        }

        /// <summary>Asserts that the child has a persisted dependency on the supplied parent.</summary>
        /// <param name="parent">The expected parent invocation.</param>
        /// <param name="child">The expected child invocation.</param>
        /// <param name="cancellationToken">A token that can cancel the assertion query.</param>
        /// <returns>A task that completes when the assertion succeeds.</returns>
        public async ValueTask AssertContinuationReleasedAfterAsync(
                JobHandle parent,
                JobHandle child,
                CancellationToken cancellationToken = default
        )
        {
                var childStatus = await Storage.GetJobStatusAsync(child.Id, cancellationToken).ConfigureAwait(false)
                        ?? throw new JobTestAssertionException($"Expected continuation '{child.Id}', but it was not found.");
                if (!childStatus.DependsOn.Any(edge => string.Equals(edge.ParentJobId, parent.Id, StringComparison.Ordinal)))
                {
                        throw new JobTestAssertionException(
                                $"Expected job '{child.Id}' to depend on '{parent.Id}', but no such edge was persisted."
                        );
                }

                if (childStatus.State is JobState.Cancelled or JobState.Skipped)
                        throw new JobTestAssertionException($"Expected continuation '{child.Id}' to be released, but it was {childStatus.State}.");
                if (childStatus.State == JobState.AwaitingContinuation)
                        throw new JobTestAssertionException($"Expected continuation '{child.Id}' to be released, but it is still waiting.");
        }

        /// <summary>Asserts that every supplied invocation was skipped by a dependency cascade.</summary>
        /// <param name="subtree">The invocations expected to be cascade-skipped.</param>
        /// <param name="cancellationToken">A token that can cancel the assertion query.</param>
        /// <returns>A task that completes when the assertion succeeds.</returns>
        public async ValueTask AssertCascadeSkippedAsync(
                IReadOnlyCollection<JobHandle> subtree,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(subtree);
                foreach (var handle in subtree)
                {
                        var job = await GetJobAsync(handle, cancellationToken).ConfigureAwait(false);
                        if (job.State != JobState.Skipped)
                                throw new JobTestAssertionException($"Expected job '{handle.Id}' to be cascade-skipped, but it was {job.State}.");
                }
        }

        /// <summary>Compatibility alias for <see cref="AssertCascadeSkippedAsync"/>.</summary>
        /// <param name="subtree">The invocations expected to be cascade-skipped.</param>
        /// <param name="cancellationToken">A token that can cancel the assertion query.</param>
        /// <returns>A task that completes when the assertion succeeds.</returns>
        public ValueTask AssertCascadeCancelledAsync(
                IReadOnlyCollection<JobHandle> subtree,
                CancellationToken cancellationToken = default
        ) => AssertCascadeSkippedAsync(subtree, cancellationToken);

        /// <summary>Runs one generated invoker, including its compile-time behavior pipeline, outside durable state.</summary>
        /// <typeparam name="TPayload">The payload type accepted by the generated invoker.</typeparam>
        /// <param name="definition">The generated job definition to invoke.</param>
        /// <param name="payload">The payload supplied to the invoker.</param>
        /// <param name="cancellationToken">A token that can cancel invocation.</param>
        /// <returns>A task that completes when the behavior pipeline finishes.</returns>
        public async ValueTask RunThroughPipelineAsync<TPayload>(
                JobDefinition definition,
                TPayload payload,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(definition);
                var now = TimeProvider.GetUtcNow();
                var record = new JobRecord
                {
                        Id = _serviceProvider.GetRequiredService<IIdGenerator>().CreateId(IdKind.Job),
                        JobName = definition.Name,
                        QueueName = definition.Queue.Name,
                        Payload = _serviceProvider.GetRequiredService<IJobSerializer>().Serialize(payload),
                        State = JobState.Active,
                        DueAt = now,
                        CreatedAt = now,
                        Attempt = 1,
                };
                await using var scope = _serviceProvider.CreateAsyncScope();
                await definition.Invoker.InvokeAsync(
                        scope.ServiceProvider,
                        new(record, definition, cancellationToken)
                ).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public void Dispose()
        {
                if (_disposed)
                        return;
                _disposed = true;
                _serviceProvider.Dispose();
        }

        /// <inheritdoc />
        public async ValueTask DisposeAsync()
        {
                if (_disposed)
                        return;
                _disposed = true;
                await _serviceProvider.DisposeAsync().ConfigureAwait(false);
        }
}

/// <summary>A durable invocation paired with its strongly typed deserialized payload.</summary>
/// <typeparam name="TPayload">The deserialized payload type.</typeparam>
/// <param name="Record">The persisted job record.</param>
/// <param name="Payload">The deserialized payload.</param>
public sealed record EnqueuedJob<TPayload>(JobRecord Record, TPayload Payload);

1	using System.Globalization;
2	using Microsoft.Extensions.DependencyInjection;
3	using Microsoft.Extensions.Logging;
4	using Microsoft.Extensions.Logging.Abstractions;
5	using Microsoft.Extensions.Time.Testing;
6
7	namespace Immediate.Jobs.Testing;
8
9	/// <summary>
10	/// Hosts Immediate.Jobs with in-memory storage and a controllable clock without starting background threads.
11	/// </summary>
12	public sealed class JobTestHarness : IAsyncDisposable, IDisposable
13	{
14	private readonly ServiceProvider _serviceProvider;
15	private readonly IJobGraphStorage _graphStorage;
16	private bool _disposed;
17
18	/// <summary>Creates a harness at the Unix epoch.</summary>
19	/// <param name="configureServices">
20	/// Registers generated job definitions and the services used by their handlers. Calling the generated
21	/// <c>AddImmediateJobs</c> method here is supported; the harness clock and in-memory provider remain authoritative.
22	/// </param>
23	public JobTestHarness(Action<IServiceCollection>? configureServices = null)
24	: this(DateTimeOffset.UnixEpoch, configureServices)	4✔
25	{
26	}	4✔
27
28	/// <summary>Creates a harness at a specified UTC instant.</summary>
29	/// <param name="start">The initial UTC time exposed by the controllable clock.</param>
30	/// <param name="configureServices">
31	/// Registers generated job definitions and the services used by their handlers. Calling the generated
32	/// <c>AddImmediateJobs</c> method here is supported; the harness clock and in-memory provider remain authoritative.
33	/// </param>
34	public JobTestHarness(	4✔
35	DateTimeOffset start,	4✔
36	Action<IServiceCollection>? configureServices = null	4✔
37	)	4✔
38	{
39	TimeProvider = new(start);	4✔
40	var services = new ServiceCollection();	4✔
41	_ = services.AddSingleton<TimeProvider>(TimeProvider);	4✔
42	_ = services.AddSingleton(TimeProvider);	4✔
43	_ = services.AddSingleton(typeof(ILogger<>), typeof(NullLogger<>));	4✔
44	_ = services.AddImmediateJobsCore(options =>	4✔
45	{	4✔
46	_ = options.UseInMemory();	4✔
47	options.MaxParallelJobs = 1;	4✔
48	});	4✔
49	configureServices?.Invoke(services);	4✔
50	_serviceProvider = services.BuildServiceProvider(new ServiceProviderOptions	4✔
51	{	4✔
52	ValidateScopes = true,	4✔
53	ValidateOnBuild = true,	4✔
54	});	4✔
55
56	Services = _serviceProvider;	4✔
57	Storage = _serviceProvider.GetRequiredService<IJobStorage>();	4✔
58	_graphStorage = Storage as IJobGraphStorage	4✔
59	?? throw new NotSupportedException(	4✔
60	"Batches & continuations require a graph-capable storage provider (a SQL database). " +	4✔
61	"The configured provider implements the queue capability only."	4✔
62	);	4✔
63	Batches = new JobBatchScheduler(	4✔
64	Storage,	4✔
65	TimeProvider,	4✔
66	_serviceProvider.GetRequiredService<IIdGenerator>()	4✔
67	);	4✔
68	Scheduler = _serviceProvider.GetRequiredService<JobSchedulerService>();	4✔
69	}	4✔
70
71	/// <summary>The controllable clock used by schedulers, storage, retries, and cron materialization.</summary>
72	/// <value>The harness clock.</value>
73	public FakeTimeProvider TimeProvider { get; }
74
75	/// <summary>The harness service provider.</summary>
76	/// <value>The service provider created for the harness.</value>
77	public IServiceProvider Services { get; }
78
79	/// <summary>The in-memory durable-state abstraction.</summary>
80	/// <value>The storage provider used by the harness.</value>
81	public IJobStorage Storage { get; }
82
83	/// <summary>Builds atomic batches against the harness storage and fake clock.</summary>
84	/// <value>The batch scheduler configured for the harness.</value>
85	public IJobBatchScheduler Batches { get; }
86
87	/// <summary>The production scheduler runner hosted by the harness.</summary>
88	/// <value>The scheduler service configured for the harness.</value>
89	public JobSchedulerService Scheduler { get; }
90
91	/// <summary>Runs every invocation currently due and returns when the due queue is empty.</summary>
92	/// <param name="cancellationToken">A token that can cancel draining.</param>
93	/// <returns>A task that completes when no currently due work remains.</returns>
94	public ValueTask DrainAsync(CancellationToken cancellationToken = default) =>
95	Scheduler.DrainAsync(cancellationToken);	4✔
96
97	/// <summary>Advances fake time and then runs every invocation that became due.</summary>
98	/// <param name="amount">The amount by which to advance the clock.</param>
99	/// <param name="cancellationToken">A token that can cancel draining.</param>
100	/// <returns>A task that completes when no newly due work remains.</returns>
101	public async ValueTask AdvanceTimeAndDrainAsync(
102	TimeSpan amount,
103	CancellationToken cancellationToken = default
104	)
105	{
106	if (amount < TimeSpan.Zero)	4✔
107	throw new ArgumentOutOfRangeException(nameof(amount), "Fake time cannot move backwards.");	×
108
109	TimeProvider.Advance(amount);	4✔
110	await DrainAsync(cancellationToken).ConfigureAwait(false);	4✔
111	}	4✔
112
113	/// <summary>Advances fake time to an absolute instant and drains newly due work.</summary>
114	/// <param name="instant">The absolute time to which the clock is advanced.</param>
115	/// <param name="cancellationToken">A token that can cancel draining.</param>
116	/// <returns>A task that completes when no newly due work remains.</returns>
117	public ValueTask AdvanceTimeAndDrainAsync(
118	DateTimeOffset instant,
119	CancellationToken cancellationToken = default
120	)
121	{
122	var amount = instant - TimeProvider.GetUtcNow();	×
123	if (amount < TimeSpan.Zero)	×
124	throw new ArgumentOutOfRangeException(nameof(instant), "Fake time cannot move backwards.");	×
125	return AdvanceTimeAndDrainAsync(amount, cancellationToken);	×
126	}
127
128	/// <summary>Returns persisted jobs matching a monitoring query.</summary>
129	/// <param name="query">The optional monitoring query. When omitted, up to 1,000 jobs are returned.</param>
130	/// <param name="cancellationToken">A token that can cancel the query.</param>
131	/// <returns>The persisted jobs that match the query.</returns>
132	public ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(
133	JobQuery? query = null,
134	CancellationToken cancellationToken = default
135	) => Storage.QueryJobsAsync(query ?? new() { Take = 1000 }, cancellationToken);	4✔
136
137	/// <summary>Finds an invocation by identifier, or throws a test assertion exception.</summary>
138	/// <param name="jobId">The invocation identifier.</param>
139	/// <param name="cancellationToken">A token that can cancel the query.</param>
140	/// <returns>The persisted job record.</returns>
141	public async ValueTask<JobRecord> GetJobAsync(string jobId, CancellationToken cancellationToken = default)
142	{
143	var jobs = await Storage.QueryJobsAsync(new() { Take = 1000 }, cancellationToken).ConfigureAwait(false);	4✔
144	return jobs.FirstOrDefault(job => string.Equals(job.Id, jobId, StringComparison.Ordinal))	4✔
145	?? throw new JobTestAssertionException($"Expected job '{jobId}' to have been enqueued, but it was not found.");	4✔
146	}	3✔
147
148	/// <summary>Finds an invocation returned by a typed scheduler call.</summary>
149	/// <param name="job">The handle returned by the scheduler.</param>
150	/// <param name="cancellationToken">A token that can cancel the query.</param>
151	/// <returns>The persisted job record.</returns>
152	public ValueTask<JobRecord> GetJobAsync(JobHandle job, CancellationToken cancellationToken = default) =>
153	GetJobAsync(job.Id, cancellationToken);	4✔
154
155	/// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
156	/// <typeparam name="TPayload">The expected payload type.</typeparam>
157	/// <param name="jobId">The invocation identifier.</param>
158	/// <param name="expectedState">The expected durable state, or <see langword="null"/> to accept any state.</param>
159	/// <param name="cancellationToken">A token that can cancel the query.</param>
160	/// <returns>The durable record paired with its deserialized payload.</returns>
161	public async ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
162	string jobId,
163	JobState? expectedState = null,
164	CancellationToken cancellationToken = default
165	)
166	{
167	var job = await GetJobAsync(jobId, cancellationToken).ConfigureAwait(false);	4✔
168	if (expectedState is { } state && job.State != state)	4✔
169	{
170	throw new JobTestAssertionException(	×
171	$"Expected job '{jobId}' to be {state}, but it was {job.State}."	×
172	);	×
173	}
174
175	TPayload payload;
176	try
177	{
178	payload = _serviceProvider.GetRequiredService<IJobSerializer>().Deserialize<TPayload>(job.Payload);	4✔
179	}	4✔
180	catch (Exception exception)	×
181	{
182	throw new JobTestAssertionException(	×
183	$"Job '{jobId}' did not contain a valid {typeof(TPayload).FullName} payload: {exception.Message}"	×
184	);	×
185	}
186
187	return new(job, payload);	4✔
188	}	3✔
189
190	/// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
191	/// <typeparam name="TPayload">The expected payload type.</typeparam>
192	/// <param name="job">The handle returned by the scheduler.</param>
193	/// <param name="expectedState">The expected durable state, or <see langword="null"/> to accept any state.</param>
194	/// <param name="cancellationToken">A token that can cancel the query.</param>
195	/// <returns>The durable record paired with its deserialized payload.</returns>
196	public ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
197	JobHandle job,
198	JobState? expectedState = null,
199	CancellationToken cancellationToken = default
200	) => AssertEnqueuedAsync<TPayload>(job.Id, expectedState, cancellationToken);	4✔
201
202	/// <summary>Asserts that a committed batch and exactly the expected number of members are visible together.</summary>
203	/// <param name="batch">The committed batch handle.</param>
204	/// <param name="expectedMembers">The expected number of committed batch members.</param>
205	/// <param name="cancellationToken">A token that can cancel the assertion query.</param>
206	/// <returns>A task that completes when the assertion succeeds.</returns>
207	public async ValueTask AssertBatchCommittedAtomicallyAsync(
208	BatchHandle batch,
209	int expectedMembers,
210	CancellationToken cancellationToken = default
211	)
212	{
213	ArgumentNullException.ThrowIfNull(batch);	×
214	var status = await _graphStorage.GetBatchStatusAsync(batch.Id, cancellationToken).ConfigureAwait(false)	×
215	?? throw new JobTestAssertionException($"Expected batch '{batch.Id}' to be committed, but it was not found.");	×
216	var members = await _graphStorage.QueryBatchMembersAsync(	×
217	batch.Id,	×
218	new() { Take = Math.Max(1, expectedMembers + 1) },	×
219	cancellationToken	×
220	).ConfigureAwait(false);	×
221	if (status.Total != expectedMembers \|\| members.Count != expectedMembers)	×
222	{
223	throw new JobTestAssertionException(	×
224	string.Create(CultureInfo.InvariantCulture, $"Expected batch '{batch.Id}' to contain {expectedMembers} jobs, but its header reports {status.Total} and {members.Count} members were found.")	×
225	);	×
226	}
227	}	×
228
229	/// <summary>Asserts that the child has a persisted dependency on the supplied parent.</summary>
230	/// <param name="parent">The expected parent invocation.</param>
231	/// <param name="child">The expected child invocation.</param>
232	/// <param name="cancellationToken">A token that can cancel the assertion query.</param>
233	/// <returns>A task that completes when the assertion succeeds.</returns>
234	public async ValueTask AssertContinuationReleasedAfterAsync(
235	JobHandle parent,
236	JobHandle child,
237	CancellationToken cancellationToken = default
238	)
239	{
240	var childStatus = await Storage.GetJobStatusAsync(child.Id, cancellationToken).ConfigureAwait(false)	4✔
241	?? throw new JobTestAssertionException($"Expected continuation '{child.Id}', but it was not found.");	4✔
242	if (!childStatus.DependsOn.Any(edge => string.Equals(edge.ParentJobId, parent.Id, StringComparison.Ordinal)))	4✔
243	{
244	throw new JobTestAssertionException(	×
245	$"Expected job '{child.Id}' to depend on '{parent.Id}', but no such edge was persisted."	×
246	);	×
247	}
248
249	if (childStatus.State is JobState.Cancelled or JobState.Skipped)	4✔
250	throw new JobTestAssertionException($"Expected continuation '{child.Id}' to be released, but it was {childStatus.State}.");	4✔
251	if (childStatus.State == JobState.AwaitingContinuation)	×
252	throw new JobTestAssertionException($"Expected continuation '{child.Id}' to be released, but it is still waiting.");	×
253	}	×
254
255	/// <summary>Asserts that every supplied invocation was skipped by a dependency cascade.</summary>
256	/// <param name="subtree">The invocations expected to be cascade-skipped.</param>
257	/// <param name="cancellationToken">A token that can cancel the assertion query.</param>
258	/// <returns>A task that completes when the assertion succeeds.</returns>
259	public async ValueTask AssertCascadeSkippedAsync(
260	IReadOnlyCollection<JobHandle> subtree,
261	CancellationToken cancellationToken = default
262	)
263	{
264	ArgumentNullException.ThrowIfNull(subtree);	×
265	foreach (var handle in subtree)	×
266	{
267	var job = await GetJobAsync(handle, cancellationToken).ConfigureAwait(false);	×
NEW 268	if (job.State != JobState.Skipped)	×
NEW 269	throw new JobTestAssertionException($"Expected job '{handle.Id}' to be cascade-skipped, but it was {job.State}.");	×
270	}
271	}	×
272
273	/// <summary>Compatibility alias for <see cref="AssertCascadeSkippedAsync"/>.</summary>
274	/// <param name="subtree">The invocations expected to be cascade-skipped.</param>
275	/// <param name="cancellationToken">A token that can cancel the assertion query.</param>
276	/// <returns>A task that completes when the assertion succeeds.</returns>
277	public ValueTask AssertCascadeCancelledAsync(
278	IReadOnlyCollection<JobHandle> subtree,
279	CancellationToken cancellationToken = default
NEW 280	) => AssertCascadeSkippedAsync(subtree, cancellationToken);	×
281
282	/// <summary>Runs one generated invoker, including its compile-time behavior pipeline, outside durable state.</summary>
283	/// <typeparam name="TPayload">The payload type accepted by the generated invoker.</typeparam>
284	/// <param name="definition">The generated job definition to invoke.</param>
285	/// <param name="payload">The payload supplied to the invoker.</param>
286	/// <param name="cancellationToken">A token that can cancel invocation.</param>
287	/// <returns>A task that completes when the behavior pipeline finishes.</returns>
288	public async ValueTask RunThroughPipelineAsync<TPayload>(
289	JobDefinition definition,
290	TPayload payload,
291	CancellationToken cancellationToken = default
292	)
293	{
294	ArgumentNullException.ThrowIfNull(definition);	×
295	var now = TimeProvider.GetUtcNow();	×
296	var record = new JobRecord	×
297	{	×
298	Id = _serviceProvider.GetRequiredService<IIdGenerator>().CreateId(IdKind.Job),	×
299	JobName = definition.Name,	×
300	QueueName = definition.Queue.Name,	×
301	Payload = _serviceProvider.GetRequiredService<IJobSerializer>().Serialize(payload),	×
302	State = JobState.Active,	×
303	DueAt = now,	×
304	CreatedAt = now,	×
305	Attempt = 1,	×
306	};	×
307	await using var scope = _serviceProvider.CreateAsyncScope();	×
308	await definition.Invoker.InvokeAsync(	×
309	scope.ServiceProvider,	×
310	new(record, definition, cancellationToken)	×
311	).ConfigureAwait(false);	×
312	}	×
313
314	/// <inheritdoc />
315	public void Dispose()
316	{
317	if (_disposed)	×
318	return;	×
319	_disposed = true;	×
320	_serviceProvider.Dispose();	×
321	}	×
322
323	/// <inheritdoc />
324	public async ValueTask DisposeAsync()
325	{
326	if (_disposed)	4✔
327	return;	×
328	_disposed = true;	4✔
329	await _serviceProvider.DisposeAsync().ConfigureAwait(false);	4✔
330	}	4✔
331	}
332
333	/// <summary>A durable invocation paired with its strongly typed deserialized payload.</summary>
334	/// <typeparam name="TPayload">The deserialized payload type.</typeparam>
335	/// <param name="Record">The persisted job record.</param>
336	/// <param name="Payload">The deserialized payload.</param>
337	public sealed record EnqueuedJob<TPayload>(JobRecord Record, TPayload Payload);	4✔

ImmediatePlatform / Immediate.Jobs / 30629636973

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous