Files
itworx-knowledge-hub-Public/tests/backend/ItWorx.KnowledgeHub.IntegrationTests/Jobs/BackgroundJobsApiTests.cs
T

543 lines
29 KiB
C#

using System.Net;
using System.Net.Http.Headers;
using System.Net.Http.Json;
using System.Text.Json;
using ItWorx.KnowledgeHub.Application.Identity;
using ItWorx.KnowledgeHub.Application.Jobs;
using ItWorx.KnowledgeHub.Application.Jobs.UseCases;
using ItWorx.KnowledgeHub.Application.Knowledge;
using ItWorx.KnowledgeHub.Application.Knowledge.UseCases;
using ItWorx.KnowledgeHub.Contracts.Admin;
using ItWorx.KnowledgeHub.Contracts.Jobs;
using ItWorx.KnowledgeHub.Contracts.Knowledge;
using ItWorx.KnowledgeHub.Domain.Jobs;
using ItWorx.KnowledgeHub.Infrastructure.Persistence;
using ItWorx.KnowledgeHub.IntegrationTests.Infrastructure;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
namespace ItWorx.KnowledgeHub.IntegrationTests.Jobs;
[Collection(PostgresTestGroup.Name)]
public sealed class BackgroundJobsApiTests(PostgresContainerFixture postgres)
: OwnerSetupTestBase(postgres, new Dictionary<string, string?> { ["KnowledgeHub:Jobs:Enabled"] = "false" })
{
[Fact]
public async Task Completed_jobs_still_count_toward_the_daily_work_budget()
{
using var owner = await AuthenticatedClientAsync();
Factory.Services.GetRequiredService<BackgroundJobsOptions>().MaxCreatedPerDay = 1;
using var first = await QueueBackupAsync(owner, "daily-first");
Assert.Equal(HttpStatusCode.Accepted, first.StatusCode);
using var cancelled = await CancelAsync(owner, await ReadIdAsync(first));
Assert.Equal(HttpStatusCode.OK, cancelled.StatusCode);
using var excess = await QueueBackupAsync(owner, "daily-excess");
Assert.Equal(HttpStatusCode.TooManyRequests, excess.StatusCode);
using var scope = Factory.Services.CreateScope();
Assert.Equal(1, await scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>().BackgroundJobs.CountAsync());
}
[Fact]
public async Task Concurrent_distinct_jobs_cannot_exceed_the_durable_queue_budget()
{
using var owner = await AuthenticatedClientAsync();
var rootId = await CreatePageAsync(owner, "Bounded job queue");
var responses = await Task.WhenAll(Enumerable.Range(0, 6)
.Select(index => QueueCourseAsync(owner, rootId, "course:bounded", $"bounded-{index}")));
try
{
Assert.Equal(3, responses.Count(response => response.StatusCode == HttpStatusCode.Accepted));
Assert.Equal(3, responses.Count(response => response.StatusCode == HttpStatusCode.TooManyRequests));
var acceptedIndex = Array.FindIndex(responses, response => response.StatusCode == HttpStatusCode.Accepted);
var jobId = await ReadIdAsync(responses[acceptedIndex]);
using var replay = await QueueCourseAsync(owner, rootId, "course:bounded", $"bounded-{acceptedIndex}");
Assert.Equal(HttpStatusCode.Accepted, replay.StatusCode);
Assert.Equal(jobId, await ReadIdAsync(replay));
using var scope = Factory.Services.CreateScope();
var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
Assert.Equal(3, await db.BackgroundJobs.CountAsync());
Assert.Equal(3, await db.BackgroundJobItems.CountAsync());
using var cancelled = await CancelAsync(owner, jobId);
Assert.Equal(HttpStatusCode.OK, cancelled.StatusCode);
Assert.False(await db.BackgroundJobItems.AnyAsync(item => item.JobId == jobId));
Assert.True(await db.BackgroundJobs.AnyAsync(job => job.Id == jobId && job.Status == BackgroundJobStatus.Cancelled));
using var next = await QueueCourseAsync(owner, rootId, "course:bounded", "bounded-after-cancel");
Assert.Equal(HttpStatusCode.Accepted, next.StatusCode);
}
finally
{
foreach (var response in responses) response.Dispose();
}
}
[Fact]
public async Task A_stale_tracked_worker_cannot_overwrite_a_reclaimed_lease()
{
using var owner = await AuthenticatedClientAsync();
using var queued = await QueueBackupAsync(owner, "stale-lease-regression");
Assert.Equal(HttpStatusCode.Accepted, queued.StatusCode);
var jobId = await ReadIdAsync(queued);
using var workspaceScope = Factory.Services.CreateScope();
var workspaceId = await workspaceScope.ServiceProvider
.GetRequiredService<IWorkspaceRepository>().GetDefaultWorkspaceIdAsync(CancellationToken.None);
Assert.NotNull(workspaceId);
var firstStart = DateTimeOffset.UtcNow;
var firstLease = Guid.NewGuid();
using var staleScope = Factory.Services.CreateScope();
var staleDb = staleScope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
var staleRepository = staleScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
Assert.Equal(jobId, await staleRepository.TryClaimNextAsync(
firstLease, firstStart, firstStart.AddSeconds(30), CancellationToken.None));
var staleJob = await staleRepository.FindLeasedAsync(jobId, firstLease, CancellationToken.None);
Assert.NotNull(staleJob);
Assert.Equal(firstLease, staleJob.LeaseToken);
Assert.Equal(1, staleJob.AttemptCount);
var secondStart = firstStart.AddSeconds(31);
var secondLease = Guid.NewGuid();
using var reclaimedScope = Factory.Services.CreateScope();
var reclaimedRepository = reclaimedScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
Assert.Equal(jobId, await reclaimedRepository.TryClaimNextAsync(
secondLease, secondStart, secondStart.AddMinutes(2), CancellationToken.None));
var reclaimedJob = await reclaimedRepository.FindLeasedAsync(jobId, secondLease, CancellationToken.None);
Assert.NotNull(reclaimedJob);
Assert.Equal(2, reclaimedJob.AttemptCount);
staleJob.Complete(secondStart.AddSeconds(1), "{\"stale\":true}");
await Assert.ThrowsAsync<DbUpdateConcurrencyException>(() => staleDb.SaveChangesAsync(CancellationToken.None));
using var verifyScope = Factory.Services.CreateScope();
var current = await verifyScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>()
.FindAsync(workspaceId.Value, jobId, CancellationToken.None);
Assert.NotNull(current);
Assert.Equal(BackgroundJobStatus.Running, current.Status);
Assert.Equal(secondLease, current.LeaseToken);
Assert.Equal(2, current.AttemptCount);
Assert.Null(current.CompletedAt);
Assert.Null(current.ResultJson);
}
[Fact]
public async Task Queue_is_idempotent_actor_scoped_and_rejects_payload_conflicts()
{
using var owner = await AuthenticatedClientAsync();
var rootId = await CreatePageAsync(owner, "Job idempotency root", markdown: "unchanged body");
const string key = "course-tag-repeat";
using var first = await QueueCourseAsync(owner, rootId, " course:ccna ", key);
using var replay = await QueueCourseAsync(owner, rootId, "course:CCNA", key);
using var conflict = await QueueCourseAsync(owner, rootId, "course:other", key);
Assert.Equal(HttpStatusCode.Accepted, first.StatusCode);
Assert.Equal(HttpStatusCode.Accepted, replay.StatusCode);
Assert.Equal(HttpStatusCode.Conflict, conflict.StatusCode);
var firstId = await ReadIdAsync(first);
Assert.Equal(firstId, await ReadIdAsync(replay));
var created = await SetupHelper.PostWithCsrfAsync(
owner,
"/api/v1/admin/users",
new CreateUserRequest("Tweede beheerder", "tweede-beheerder@itworx.test", "correct-horse-battery-staple2", "Administrator"));
Assert.True(created.IsSuccessStatusCode, await created.Content.ReadAsStringAsync());
using var secondAdmin = Factory.CreateClient();
using var login = await SetupHelper.LoginAsync(secondAdmin, "tweede-beheerder@itworx.test", "correct-horse-battery-staple2");
Assert.Equal(HttpStatusCode.OK, login.StatusCode);
using var actorScoped = await QueueCourseAsync(secondAdmin, rootId, "course:ccna", key);
Assert.Equal(HttpStatusCode.Accepted, actorScoped.StatusCode);
Assert.NotEqual(firstId, await ReadIdAsync(actorScoped));
using var cancelFirst = await CancelAsync(owner, firstId);
using var cancelSecond = await CancelAsync(secondAdmin, await ReadIdAsync(actorScoped));
Assert.Equal(HttpStatusCode.OK, cancelFirst.StatusCode);
Assert.Equal(HttpStatusCode.OK, cancelSecond.StatusCode);
}
[Fact]
public async Task Job_mutations_require_csrf_even_when_an_authorization_header_is_present()
{
using var client = await AuthenticatedClientAsync();
var rootId = await CreatePageAsync(client, "Job CSRF root");
using var request = new HttpRequestMessage(HttpMethod.Post, "/api/v1/jobs/course-tagging")
{
Content = JsonContent.Create(new QueueCourseTaggingJobRequest(rootId, "course:csrf"))
};
request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", "not-a-service-token");
request.Headers.Add("Idempotency-Key", "csrf-missing");
using var response = await client.SendAsync(request);
Assert.Equal(HttpStatusCode.BadRequest, response.StatusCode);
Assert.Contains("csrf-validation-failed", await response.Content.ReadAsStringAsync(), StringComparison.Ordinal);
}
[Fact]
public async Task Job_endpoints_reject_anonymous_and_non_administrator_users()
{
using var anonymous = Factory.CreateClient();
using var anonymousResponse = await anonymous.GetAsync("/api/v1/jobs");
Assert.Equal(HttpStatusCode.Unauthorized, anonymousResponse.StatusCode);
using var owner = await AuthenticatedClientAsync();
using var created = await SetupHelper.PostWithCsrfAsync(
owner,
"/api/v1/admin/users",
new CreateUserRequest("Lezer", "job-lezer@itworx.test", "job-reader-password-strong", "Reader"));
Assert.Equal(HttpStatusCode.OK, created.StatusCode);
using var reader = Factory.CreateClient();
using var login = await SetupHelper.LoginAsync(reader, "job-lezer@itworx.test", "job-reader-password-strong");
Assert.Equal(HttpStatusCode.OK, login.StatusCode);
using var response = await reader.GetAsync("/api/v1/jobs");
Assert.Equal(HttpStatusCode.Forbidden, response.StatusCode);
}
[Fact]
public async Task Queued_jobs_can_be_cancelled_and_terminal_events_close_with_the_latest_snapshot()
{
using var client = await AuthenticatedClientAsync();
using var queued = await QueueBackupAsync(client, "cancel-backup");
Assert.Equal(HttpStatusCode.Accepted, queued.StatusCode);
var id = await ReadIdAsync(queued);
using var canceled = await CancelAsync(client, id);
Assert.Equal(HttpStatusCode.OK, canceled.StatusCode);
using var response = JsonDocument.Parse(await canceled.Content.ReadAsStringAsync());
Assert.Equal("cancelled", response.RootElement.GetProperty("status").GetString());
Assert.True(response.RootElement.GetProperty("cancelRequested").GetBoolean());
using var events = await client.GetAsync($"/api/v1/jobs/{id}/events");
Assert.Equal("text/event-stream", events.Content.Headers.ContentType?.MediaType);
var body = await events.Content.ReadAsStringAsync();
Assert.Contains("event: terminal", body, StringComparison.Ordinal);
Assert.Contains("cancelled", body, StringComparison.Ordinal);
}
[Fact]
public async Task Sse_emits_terminal_and_closes_even_when_terminal_version_matches_the_last_progress_event()
{
using var client = await AuthenticatedClientAsync();
using var queued = await QueueBackupAsync(client, "same-version-terminal-event");
Assert.Equal(HttpStatusCode.Accepted, queued.StatusCode);
var jobId = await ReadIdAsync(queued);
var workspaceId = await Factory.Services.GetRequiredService<IWorkspaceRepository>()
.GetDefaultWorkspaceIdAsync(CancellationToken.None);
Assert.NotNull(workspaceId);
var initial = await Factory.Services.GetRequiredService<IBackgroundJobRepository>()
.FindAsync(workspaceId.Value, jobId, CancellationToken.None);
Assert.NotNull(initial);
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
using var response = await client.GetAsync(
$"/api/v1/jobs/{jobId:D}/events", HttpCompletionOption.ResponseHeadersRead, timeout.Token);
Assert.Equal(HttpStatusCode.OK, response.StatusCode);
await using var stream = await response.Content.ReadAsStreamAsync(timeout.Token);
using var reader = new StreamReader(stream);
var sawInitialJob = false;
while (await reader.ReadLineAsync(timeout.Token) is { } line)
{
if (line == "event: job")
sawInitialJob = true;
if (line.Length == 0 && sawInitialJob)
break;
}
Assert.True(sawInitialJob);
using (var scope = Factory.Services.CreateScope())
{
var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
Assert.Equal(1, await db.Database.ExecuteSqlInterpolatedAsync($"""
UPDATE background_jobs
SET status = 'Cancelled', completed_at = {DateTimeOffset.UtcNow}
WHERE id = {jobId}
"""));
var terminal = await db.BackgroundJobs.AsNoTracking().SingleAsync(job => job.Id == jobId);
Assert.Equal(initial.Version, terminal.Version);
}
var remaining = await reader.ReadToEndAsync(timeout.Token);
Assert.Contains("event: terminal", remaining, StringComparison.Ordinal);
Assert.Contains("cancelled", remaining, StringComparison.Ordinal);
}
[Fact]
public async Task Running_job_cancellation_is_persisted_until_the_current_lease_holder_stops()
{
using var client = await AuthenticatedClientAsync();
using var queued = await QueueBackupAsync(client, "cancel-running-backup");
var jobId = await ReadIdAsync(queued);
using var scope = Factory.Services.CreateScope();
var repository = scope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
var workspaceId = await scope.ServiceProvider.GetRequiredService<IWorkspaceRepository>()
.GetDefaultWorkspaceIdAsync(CancellationToken.None);
Assert.NotNull(workspaceId);
var now = DateTimeOffset.UtcNow;
var lease = Guid.NewGuid();
Assert.Equal(jobId, await repository.TryClaimNextAsync(lease, now, now.AddMinutes(2), CancellationToken.None));
using var cancellation = await CancelAsync(client, jobId);
Assert.Equal(HttpStatusCode.OK, cancellation.StatusCode);
using var cancellationJson = JsonDocument.Parse(await cancellation.Content.ReadAsStringAsync());
Assert.Equal("running", cancellationJson.RootElement.GetProperty("status").GetString());
Assert.True(cancellationJson.RootElement.GetProperty("cancelRequested").GetBoolean());
Assert.True(await repository.IsCancellationRequestedAsync(jobId, lease, CancellationToken.None));
await repository.CancelLeasedJobAsync(jobId, lease, now.AddSeconds(1), CancellationToken.None);
var cancelled = await repository.FindAsync(workspaceId.Value, jobId, CancellationToken.None);
Assert.NotNull(cancelled);
Assert.Equal(BackgroundJobStatus.Cancelled, cancelled.Status);
Assert.Null(cancelled.LeaseToken);
Assert.Null(cancelled.LeaseUntil);
}
[Fact]
public async Task Job_listing_is_bounded_and_scoped_to_the_workspace()
{
using var client = await AuthenticatedClientAsync();
using var queued = await QueueBackupAsync(client, "list-backup");
var id = await ReadIdAsync(queued);
using var response = await client.GetAsync("/api/v1/jobs?limit=1");
using var json = JsonDocument.Parse(await response.Content.ReadAsStringAsync());
Assert.Equal(HttpStatusCode.OK, response.StatusCode);
var items = json.RootElement.GetProperty("items");
Assert.Equal(1, items.GetArrayLength());
Assert.Equal(id, items[0].GetProperty("id").GetGuid());
Assert.Equal("backup", items[0].GetProperty("type").GetString());
Assert.Equal("queued", items[0].GetProperty("status").GetString());
Assert.Equal(0, items[0].GetProperty("progress").GetInt32());
Assert.Equal(1, items[0].GetProperty("total").GetInt32());
Assert.Equal(0, items[0].GetProperty("processed").GetInt32());
Assert.Empty(items[0].GetProperty("errors").EnumerateArray());
using var cancel = await CancelAsync(client, id);
Assert.Equal(HttpStatusCode.OK, cancel.StatusCode);
}
[Fact]
public async Task Concurrent_workers_claim_distinct_queued_jobs()
{
using var client = await AuthenticatedClientAsync();
var first = await QueueBackupAsync(client, "claim-one");
var second = await QueueBackupAsync(client, "claim-two");
var firstId = await ReadIdAsync(first);
var secondId = await ReadIdAsync(second);
first.Dispose();
second.Dispose();
using var firstScope = Factory.Services.CreateScope();
using var secondScope = Factory.Services.CreateScope();
var firstRepository = firstScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
var secondRepository = secondScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
var now = DateTimeOffset.UtcNow;
var claims = await Task.WhenAll(
firstRepository.TryClaimNextAsync(Guid.NewGuid(), now, now.AddMinutes(2), CancellationToken.None),
secondRepository.TryClaimNextAsync(Guid.NewGuid(), now, now.AddMinutes(2), CancellationToken.None));
var claimedIds = claims.Select(claim => claim ?? throw new Xunit.Sdk.XunitException("A queued job was not claimed.")).ToArray();
Assert.Equal(2, claimedIds.Distinct().Count());
Assert.Contains(firstId, claimedIds);
Assert.Contains(secondId, claimedIds);
}
[Fact]
public async Task Postgres_recovers_an_expired_lease_rejects_the_stale_worker_and_bounds_retry_failures()
{
using var client = await AuthenticatedClientAsync();
using var queued = await QueueBackupAsync(client, "retry-recovery-backup");
Assert.Equal(HttpStatusCode.Accepted, queued.StatusCode);
var jobId = await ReadIdAsync(queued);
using var scope = Factory.Services.CreateScope();
var repository = scope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
var workspaceId = await scope.ServiceProvider.GetRequiredService<IWorkspaceRepository>()
.GetDefaultWorkspaceIdAsync(CancellationToken.None);
Assert.NotNull(workspaceId);
const string safeMessage = "De tijdelijke verwerking is mislukt; de taak wordt opnieuw geprobeerd.";
var firstStart = DateTimeOffset.UtcNow;
var firstLease = Guid.NewGuid();
Assert.Equal(jobId, await repository.TryClaimNextAsync(
firstLease, firstStart, firstStart.AddSeconds(30), CancellationToken.None));
await repository.RetryLeasedJobAsync(
jobId, firstLease, firstStart, firstStart.AddSeconds(1),
"processing_retry", safeMessage, CancellationToken.None);
Assert.False(await repository.RenewLeaseAsync(
jobId, firstLease, firstStart.AddMinutes(1), CancellationToken.None));
var secondStart = firstStart.AddSeconds(1);
var secondLease = Guid.NewGuid();
Assert.Equal(jobId, await repository.TryClaimNextAsync(
secondLease, secondStart, secondStart.AddSeconds(30), CancellationToken.None));
// Simulate a process stop with the second lease still running. PostgreSQL must
// make the same durable job claimable again after its lease expires.
var thirdStart = secondStart.AddSeconds(31);
var thirdLease = Guid.NewGuid();
Assert.Equal(jobId, await repository.TryClaimNextAsync(
thirdLease, thirdStart, thirdStart.AddSeconds(30), CancellationToken.None));
var resumed = await repository.FindLeasedAsync(jobId, thirdLease, CancellationToken.None);
Assert.NotNull(resumed);
Assert.Equal(3, resumed.AttemptCount);
await Assert.ThrowsAsync<InvalidOperationException>(() => repository.CompleteLeasedJobAsync(
jobId, secondLease, "{}", thirdStart, CancellationToken.None));
await repository.RetryLeasedJobAsync(
jobId, thirdLease, thirdStart, thirdStart.AddSeconds(1),
"processing_retry", safeMessage, CancellationToken.None);
var fourthStart = thirdStart.AddSeconds(1);
var fourthLease = Guid.NewGuid();
Assert.Equal(jobId, await repository.TryClaimNextAsync(
fourthLease, fourthStart, fourthStart.AddSeconds(30), CancellationToken.None));
var finalAttempt = await repository.FindLeasedAsync(jobId, fourthLease, CancellationToken.None);
Assert.NotNull(finalAttempt);
Assert.Equal(4, finalAttempt.AttemptCount);
await repository.RetryLeasedJobAsync(
jobId, fourthLease, fourthStart, fourthStart.AddSeconds(1),
"processing_retry", safeMessage, CancellationToken.None);
var completed = await repository.FindAsync(workspaceId.Value, jobId, CancellationToken.None);
Assert.NotNull(completed);
Assert.Equal(BackgroundJobStatus.Failed, completed.Status);
Assert.Equal(1, completed.Processed);
Assert.Equal(1, completed.Failed);
var errors = JsonSerializer.Deserialize<BackgroundJobError[]>(completed.ErrorsJson);
Assert.NotNull(errors);
Assert.Equal(3, errors.Length);
Assert.All(errors, error =>
{
Assert.Equal("processing_retry", error.Code);
Assert.Equal(safeMessage, error.Message);
Assert.True(error.Message.Length <= BackgroundJob.MaxErrorMessageLength);
});
Assert.True(completed.ErrorsJson.Length < 4096);
}
[Fact]
public async Task A_crash_after_a_page_update_resumes_the_processing_item_without_a_duplicate_revision()
{
using var client = await AuthenticatedClientAsync();
var pageId = await CreatePageAsync(client, "Job resume page", markdown: "do not change", tags: ["existing"]);
using var response = await QueueCourseAsync(client, pageId, "course:resume", "resume-course-tag");
Assert.Equal(HttpStatusCode.Accepted, response.StatusCode);
var jobId = await ReadIdAsync(response);
using var firstScope = Factory.Services.CreateScope();
var repository = firstScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
var pageRepository = firstScope.ServiceProvider.GetRequiredService<IPageRepository>();
var updatePage = firstScope.ServiceProvider.GetRequiredService<UpdatePageUseCase>();
var actorId = await OwnerIdAsync(firstScope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>());
var start = DateTimeOffset.UtcNow;
var crashedLease = Guid.NewGuid();
Assert.Equal(jobId, await repository.TryClaimNextAsync(crashedLease, start, start.AddSeconds(30), CancellationToken.None));
var workItem = await repository.FindNextPageItemAsync(jobId, crashedLease, CancellationToken.None);
Assert.NotNull(workItem);
await repository.MarkPageItemProcessingAsync(jobId, crashedLease, workItem.Ordinal, start, CancellationToken.None);
var page = await pageRepository.FindAsync(pageId, CancellationToken.None);
var etag = await pageRepository.GetETagAsync(pageId, CancellationToken.None);
Assert.NotNull(page);
Assert.NotNull(etag);
var update = await updatePage.ExecuteAsync(new UpdatePageCommand(
pageId, page.TitleValue, page.Markdown, page.DocumentType.ToString(), page.DocumentStatus.ToString(),
"Cursustag course:resume toegevoegd", page.Tags.Append("course:resume").ToArray(), etag,
actorId, "crash-resume-test"), CancellationToken.None);
Assert.Equal(PageOperationStatus.Success, update.Status);
// Simulate a process crash here: the published page tag is durable while the job
// item is still Processing and its lease is allowed to expire.
using var resumedScope = Factory.Services.CreateScope();
var resumedRepository = resumedScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
var resumedLease = Guid.NewGuid();
var resumedAt = start.AddSeconds(31);
Assert.Equal(jobId, await resumedRepository.TryClaimNextAsync(resumedLease, resumedAt, resumedAt.AddMinutes(2), CancellationToken.None));
var resumedItem = await resumedRepository.FindNextPageItemAsync(jobId, resumedLease, CancellationToken.None);
Assert.NotNull(resumedItem);
Assert.Equal(BackgroundJobItemState.Processing, resumedItem.State);
Assert.False(resumedItem.InitiallyTagged);
var currentPage = await resumedScope.ServiceProvider.GetRequiredService<IPageRepository>()
.FindAsync(pageId, CancellationToken.None);
Assert.NotNull(currentPage);
Assert.Contains(currentPage.Tags, tag => string.Equals(tag, "course:resume", StringComparison.OrdinalIgnoreCase));
await resumedRepository.RecordPageItemOutcomeAsync(
jobId, resumedLease, resumedItem.Ordinal, BackgroundJobItemState.Succeeded,
null, null, null, resumedAt, CancellationToken.None);
await resumedRepository.CompleteLeasedJobAsync(jobId, resumedLease, "{}", resumedAt, CancellationToken.None);
var completed = await resumedRepository.FindAsync(
(await resumedRepository.FindByIdempotencyAsync(
(await resumedScope.ServiceProvider.GetRequiredService<IWorkspaceRepository>().GetDefaultWorkspaceIdAsync(CancellationToken.None))!.Value,
actorId,
await KeyHashAsync(), CancellationToken.None))!.WorkspaceId,
jobId,
CancellationToken.None);
Assert.Equal(BackgroundJobStatus.Succeeded, completed!.Status);
var revisions = await resumedScope.ServiceProvider.GetRequiredService<IPageRepository>()
.GetRevisionsAsync(pageId, CancellationToken.None);
Assert.Equal(2, revisions.Count);
Assert.Equal("do not change", currentPage.Markdown);
static Task<string> KeyHashAsync() => Task.FromResult(Convert.ToHexStringLower(
System.Security.Cryptography.SHA256.HashData(System.Text.Encoding.UTF8.GetBytes("resume-course-tag"))));
}
private async Task<HttpClient> AuthenticatedClientAsync()
{
var client = Factory.CreateClient();
var login = await SetupHelper.LoginAsync(client, OwnerEmail, OwnerPassword);
Assert.Equal(HttpStatusCode.OK, login.StatusCode);
return client;
}
private static async Task<Guid> CreatePageAsync(HttpClient client, string title, string markdown = "page body", IReadOnlyList<string>? tags = null)
{
using var response = await SetupHelper.PostWithCsrfAsync(
client, "/api/v1/pages", new CreatePageRequest(null, title, "WikiPage", markdown, null, tags));
Assert.True(response.IsSuccessStatusCode, await response.Content.ReadAsStringAsync());
using var json = JsonDocument.Parse(await response.Content.ReadAsStringAsync());
return json.RootElement.GetProperty("id").GetGuid();
}
private static async Task<HttpResponseMessage> QueueCourseAsync(HttpClient client, Guid rootPageId, string tag, string key)
{
var csrf = await SetupHelper.GetCsrfTokenAsync(client);
using var request = new HttpRequestMessage(HttpMethod.Post, "/api/v1/jobs/course-tagging")
{
Content = JsonContent.Create(new QueueCourseTaggingJobRequest(rootPageId, tag))
};
request.Headers.Add(csrf.HeaderName, csrf.Value);
request.Headers.Add("Idempotency-Key", key);
return await client.SendAsync(request);
}
private static async Task<HttpResponseMessage> QueueBackupAsync(HttpClient client, string key)
{
var csrf = await SetupHelper.GetCsrfTokenAsync(client);
using var request = new HttpRequestMessage(HttpMethod.Post, "/api/v1/jobs/backups")
{
Content = JsonContent.Create(new QueueBackupJobRequest())
};
request.Headers.Add(csrf.HeaderName, csrf.Value);
request.Headers.Add("Idempotency-Key", key);
return await client.SendAsync(request);
}
private static async Task<HttpResponseMessage> CancelAsync(HttpClient client, Guid jobId) =>
await SetupHelper.PostWithCsrfAsync(client, $"/api/v1/jobs/{jobId:D}/cancel", new { });
private static async Task<Guid> ReadIdAsync(HttpResponseMessage response)
{
using var json = JsonDocument.Parse(await response.Content.ReadAsStringAsync());
return json.RootElement.GetProperty("id").GetGuid();
}
private static async Task<Guid> OwnerIdAsync(KnowledgeHubDbContext db) =>
await db.Users.AsNoTracking().Where(user => user.Email == "eigenaar@itworx.test")
.Select(user => user.Id).SingleAsync();
}