30300691995

Committed 27 Jul 2026 08:00PM UTC coverage: 64.845%. First build

Build # 30300691995

Build Type

Pull #20

github

Committed by

web-flow

Commit Message

Merge a73d5f4ed into e8a73c280

Pull Request Pull Request #20: Initial draft

Coverage Stats

1300 of 1892 new or added lines in 29 files covered. (68.71%)

1649 of 2543 relevant lines covered (64.84%)

2.58 hits per line

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

55.95

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

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 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>
        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>();
                Scheduler = _serviceProvider.GetRequiredService<JobSchedulerService>();
        }

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

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

        /// <summary>The in-memory durable-state abstraction.</summary>
        public IJobStorage Storage { get; }

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

        /// <summary>Runs every invocation currently due and returns when the due queue is empty.</summary>
        public ValueTask DrainAsync(CancellationToken cancellationToken = default) =>
                Scheduler.DrainAsync(cancellationToken);

        /// <summary>Advances fake time and then runs every invocation that became due.</summary>
        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>
        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>
        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>
        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 => job.Id == jobId)
                        ?? throw new JobTestAssertionException($"Expected job '{jobId}' to have been enqueued, but it was not found.");
        }

        /// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
        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>Runs one generated invoker, including its compile-time behavior pipeline, outside durable state.</summary>
        public async ValueTask RunThroughPipelineAsync<TPayload>(
                JobDefinition definition,
                TPayload payload,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(definition);
                var now = TimeProvider.GetUtcNow();
                var record = new JobRecord
                {
                        Id = Guid.NewGuid().ToString("N"),
                        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>
public sealed record EnqueuedJob<TPayload>(JobRecord Record, TPayload Payload);

1	using Microsoft.Extensions.DependencyInjection;
2	using Microsoft.Extensions.Logging;
3	using Microsoft.Extensions.Logging.Abstractions;
4	using Microsoft.Extensions.Time.Testing;
5
6	namespace Immediate.Jobs.Testing;
7
8	/// <summary>
9	/// Hosts Immediate.Jobs with in-memory storage and a controllable clock without starting background threads.
10	/// </summary>
11	public sealed class JobTestHarness : IAsyncDisposable, IDisposable
12	{
13	private readonly ServiceProvider _serviceProvider;
14	private bool _disposed;
15
16	/// <summary>Creates a harness at the Unix epoch.</summary>
17	/// <param name="configureServices">
18	/// Registers generated job definitions and the services used by their handlers. Calling the generated
19	/// <c>AddImmediateJobs</c> method here is supported; the harness clock and in-memory provider remain authoritative.
20	/// </param>
21	public JobTestHarness(Action<IServiceCollection>? configureServices = null)
22	: this(DateTimeOffset.UnixEpoch, configureServices)	4✔
23	{
24	}	4✔
25
26	/// <summary>Creates a harness at a specified UTC instant.</summary>
27	public JobTestHarness(	4✔
28	DateTimeOffset start,	4✔
29	Action<IServiceCollection>? configureServices = null	4✔
30	)	4✔
31	{
32	TimeProvider = new(start);	4✔
33	var services = new ServiceCollection();	4✔
34	_ = services.AddSingleton<TimeProvider>(TimeProvider);	4✔
35	_ = services.AddSingleton(TimeProvider);	4✔
36	_ = services.AddSingleton(typeof(ILogger<>), typeof(NullLogger<>));	4✔
37	_ = services.AddImmediateJobsCore(options =>	4✔
38	{	4✔
39	_ = options.UseInMemory();	4✔
40	options.MaxParallelJobs = 1;	4✔
41	});	4✔
42	configureServices?.Invoke(services);	4✔
43	_serviceProvider = services.BuildServiceProvider(new ServiceProviderOptions	4✔
44	{	4✔
45	ValidateScopes = true,	4✔
46	ValidateOnBuild = true,	4✔
47	});	4✔
48
49	Services = _serviceProvider;	4✔
50	Storage = _serviceProvider.GetRequiredService<IJobStorage>();	4✔
51	Scheduler = _serviceProvider.GetRequiredService<JobSchedulerService>();	4✔
52	}	4✔
53
54	/// <summary>The controllable clock used by schedulers, storage, retries, and cron materialization.</summary>
55	public FakeTimeProvider TimeProvider { get; }
56
57	/// <summary>The harness service provider.</summary>
58	public IServiceProvider Services { get; }
59
60	/// <summary>The in-memory durable-state abstraction.</summary>
61	public IJobStorage Storage { get; }
62
63	/// <summary>The production scheduler runner hosted by the harness.</summary>
64	public JobSchedulerService Scheduler { get; }
65
66	/// <summary>Runs every invocation currently due and returns when the due queue is empty.</summary>
67	public ValueTask DrainAsync(CancellationToken cancellationToken = default) =>
68	Scheduler.DrainAsync(cancellationToken);	4✔
69
70	/// <summary>Advances fake time and then runs every invocation that became due.</summary>
71	public async ValueTask AdvanceTimeAndDrainAsync(
72	TimeSpan amount,
73	CancellationToken cancellationToken = default
74	)
75	{
76	if (amount < TimeSpan.Zero)	4✔
NEW 77	throw new ArgumentOutOfRangeException(nameof(amount), "Fake time cannot move backwards.");	×
78
79	TimeProvider.Advance(amount);	4✔
80	await DrainAsync(cancellationToken).ConfigureAwait(false);	4✔
81	}	4✔
82
83	/// <summary>Advances fake time to an absolute instant and drains newly due work.</summary>
84	public ValueTask AdvanceTimeAndDrainAsync(
85	DateTimeOffset instant,
86	CancellationToken cancellationToken = default
87	)
88	{
NEW 89	var amount = instant - TimeProvider.GetUtcNow();	×
NEW 90	if (amount < TimeSpan.Zero)	×
NEW 91	throw new ArgumentOutOfRangeException(nameof(instant), "Fake time cannot move backwards.");	×
NEW 92	return AdvanceTimeAndDrainAsync(amount, cancellationToken);	×
93	}
94
95	/// <summary>Returns persisted jobs matching a monitoring query.</summary>
96	public ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(
97	JobQuery? query = null,
98	CancellationToken cancellationToken = default
99	) => Storage.QueryJobsAsync(query ?? new() { Take = 1000 }, cancellationToken);	4✔
100
101	/// <summary>Finds an invocation by identifier, or throws a test assertion exception.</summary>
102	public async ValueTask<JobRecord> GetJobAsync(string jobId, CancellationToken cancellationToken = default)
103	{
104	var jobs = await Storage.QueryJobsAsync(new() { Take = 1000 }, cancellationToken).ConfigureAwait(false);	4✔
105	return jobs.FirstOrDefault(job => job.Id == jobId)	4✔
106	?? throw new JobTestAssertionException($"Expected job '{jobId}' to have been enqueued, but it was not found.");	4✔
107	}	4✔
108
109	/// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
110	public async ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
111	string jobId,
112	JobState? expectedState = null,
113	CancellationToken cancellationToken = default
114	)
115	{
116	var job = await GetJobAsync(jobId, cancellationToken).ConfigureAwait(false);	4✔
117	if (expectedState is { } state && job.State != state)	4✔
118	{
NEW 119	throw new JobTestAssertionException(	×
NEW 120	$"Expected job '{jobId}' to be {state}, but it was {job.State}."	×
NEW 121	);	×
122	}
123
124	TPayload payload;
125	try
126	{
127	payload = _serviceProvider.GetRequiredService<IJobSerializer>().Deserialize<TPayload>(job.Payload);	4✔
128	}	4✔
NEW 129	catch (Exception exception)	×
130	{
NEW 131	throw new JobTestAssertionException(	×
NEW 132	$"Job '{jobId}' did not contain a valid {typeof(TPayload).FullName} payload: {exception.Message}"	×
NEW 133	);	×
134	}
135
136	return new(job, payload);	4✔
137	}	4✔
138
139	/// <summary>Runs one generated invoker, including its compile-time behavior pipeline, outside durable state.</summary>
140	public async ValueTask RunThroughPipelineAsync<TPayload>(
141	JobDefinition definition,
142	TPayload payload,
143	CancellationToken cancellationToken = default
144	)
145	{
NEW 146	ArgumentNullException.ThrowIfNull(definition);	×
NEW 147	var now = TimeProvider.GetUtcNow();	×
NEW 148	var record = new JobRecord	×
NEW 149	{	×
NEW 150	Id = Guid.NewGuid().ToString("N"),	×
NEW 151	JobName = definition.Name,	×
NEW 152	QueueName = definition.Queue.Name,	×
NEW 153	Payload = _serviceProvider.GetRequiredService<IJobSerializer>().Serialize(payload),	×
NEW 154	State = JobState.Active,	×
NEW 155	DueAt = now,	×
NEW 156	CreatedAt = now,	×
NEW 157	Attempt = 1,	×
NEW 158	};	×
NEW 159	await using var scope = _serviceProvider.CreateAsyncScope();	×
NEW 160	await definition.Invoker.InvokeAsync(	×
NEW 161	scope.ServiceProvider,	×
NEW 162	new(record, definition, cancellationToken)	×
NEW 163	).ConfigureAwait(false);	×
NEW 164	}	×
165
166	/// <inheritdoc />
167	public void Dispose()
168	{
NEW 169	if (_disposed)	×
NEW 170	return;	×
NEW 171	_disposed = true;	×
NEW 172	_serviceProvider.Dispose();	×
NEW 173	}	×
174
175	/// <inheritdoc />
176	public async ValueTask DisposeAsync()
177	{
178	if (_disposed)	4✔
NEW 179	return;	×
180	_disposed = true;	4✔
181	await _serviceProvider.DisposeAsync().ConfigureAwait(false);	4✔
182	}	4✔
183	}
184
185	/// <summary>A durable invocation paired with its strongly typed deserialized payload.</summary>
186	public sealed record EnqueuedJob<TPayload>(JobRecord Record, TPayload Payload);	4✔

ImmediatePlatform / Immediate.Jobs / 30300691995

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