cesnimda
259 lines
13 KiB
C#
259 lines
13 KiB
C#
using System.Data.Common;
|
|
using System.Security.Claims;
|
|
using JobTrackerApi.Controllers;
|
|
using JobTrackerApi.Data;
|
|
using JobTrackerApi.Models;
|
|
using JobTrackerApi.Services;
|
|
using Microsoft.AspNetCore.Http;
|
|
using Microsoft.AspNetCore.Identity;
|
|
using Microsoft.AspNetCore.Mvc;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.EntityFrameworkCore.Diagnostics;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Moq;
|
|
using Xunit;
|
|
|
|
namespace JobTrackerApi.MariaDb.Tests;
|
|
|
|
public class CvImportLifecycleConcurrencyTests
|
|
{
|
|
private const string Owner = "rd21-owner";
|
|
private static readonly ProfileCvController.ApplyReviewedRequest Review = new(0, new[] { new CvImportDecision("skills.0", "use") });
|
|
|
|
[Fact]
|
|
public async Task Document_only_claim_is_owner_bound_and_translates_on_MariaDb_without_restricting_general_claims()
|
|
{
|
|
await using var database = await MariaDbDatabase.CreateAsync();
|
|
Guid documentId, legacyId, strategyId, mismatchedId, wrongTriggerId;
|
|
await using (var seed = database.Context())
|
|
{
|
|
seed.Users.AddRange(new ApplicationUser { Id = Owner, UserName = "claim-owner" },
|
|
new ApplicationUser { Id = "other-owner", UserName = "claim-other" });
|
|
var document = new CvExtractionRun { OwnerUserId = Owner, ParserVersion = ProfileCvController.DocumentParserVersion, Trigger = "document_import", Status = "queued" };
|
|
var foreign = new CvExtractionRun { OwnerUserId = "other-owner", ParserVersion = ProfileCvController.DocumentParserVersion, Trigger = "document_import", Status = "queued" };
|
|
var legacy = new CvExtractionRun { OwnerUserId = Owner, ParserVersion = "legacy-text", Trigger = "document_import", Status = "queued" };
|
|
var wrongTrigger = new CvExtractionRun { OwnerUserId = Owner, ParserVersion = ProfileCvController.DocumentParserVersion, Trigger = "text_extraction", Status = "queued" };
|
|
seed.CvExtractionRuns.AddRange(document, foreign, legacy, wrongTrigger);
|
|
await seed.SaveChangesAsync();
|
|
UserOperation Queued(CvExtractionRun run, int priority)
|
|
{
|
|
var operation = Operation(run.Id, OperationStatuses.Queued);
|
|
operation.IdempotencyKey = operation.Id.ToString("N");
|
|
operation.LeaseToken = null;
|
|
operation.Priority = priority;
|
|
return operation;
|
|
}
|
|
var eligible = Queued(document, 100);
|
|
var legacyOperation = Queued(legacy, 1000);
|
|
var mismatched = Queued(foreign, 900);
|
|
var wrongTriggerOperation = Queued(wrongTrigger, 800);
|
|
var strategy = Queued(document, 1100);
|
|
strategy.TaskType = StrategySnapshotService.TaskType;
|
|
strategy.SubjectType = "job_application";
|
|
seed.UserOperations.AddRange(eligible, legacyOperation, mismatched, wrongTriggerOperation, strategy);
|
|
await seed.SaveChangesAsync();
|
|
(documentId, legacyId, strategyId, mismatchedId, wrongTriggerId) =
|
|
(eligible.Id, legacyOperation.Id, strategy.Id, mismatched.Id, wrongTriggerOperation.Id);
|
|
}
|
|
|
|
await using var neutral = database.Context(null!);
|
|
var store = new UserOperationStore(neutral, TimeProvider.System);
|
|
var claimed = Assert.IsType<UserOperationLease>(await store.ClaimNextAsync(TimeSpan.FromMinutes(1), default, documentImportsOnly: true));
|
|
Assert.Equal(documentId, claimed.OperationId);
|
|
Assert.Equal(Owner, claimed.OwnerUserId);
|
|
Assert.Null(await store.ClaimNextAsync(TimeSpan.FromMinutes(1), default, documentImportsOnly: true));
|
|
var generalCv = Assert.IsType<UserOperationLease>(await store.ClaimNextAsync(TimeSpan.FromMinutes(1), default, new[] { CvProcessingQueue.TaskType }));
|
|
Assert.Equal(legacyId, generalCv.OperationId);
|
|
var generalOther = Assert.IsType<UserOperationLease>(await store.ClaimNextAsync(TimeSpan.FromMinutes(1), default, new[] { StrategySnapshotService.TaskType }));
|
|
Assert.Equal(strategyId, generalOther.OperationId);
|
|
var excluded = await neutral.UserOperations.IgnoreQueryFilters().AsNoTracking()
|
|
.Where(operation => operation.Id == mismatchedId || operation.Id == wrongTriggerId).ToListAsync();
|
|
Assert.Equal(2, excluded.Count);
|
|
Assert.All(excluded, operation =>
|
|
{
|
|
Assert.Equal(OperationStatuses.Queued, operation.Status);
|
|
Assert.Equal(0, operation.AttemptCount);
|
|
Assert.Null(operation.LeaseToken);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Acknowledged_discard_prevents_an_apply_that_already_read_pending()
|
|
{
|
|
await using var database = await MariaDbDatabase.CreateAsync();
|
|
var id = await Seed(database);
|
|
var gate = new RunReadGate();
|
|
await using var applying = database.Context(Owner, gate);
|
|
var apply = Controller(applying).ApplyReviewed(id, Review);
|
|
try
|
|
{
|
|
await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15));
|
|
await using var discarding = database.Context();
|
|
Assert.IsType<NoContentResult>(await Controller(discarding).DiscardRun(id));
|
|
}
|
|
finally { gate.Release.TrySetResult(); }
|
|
Assert.IsType<ConflictObjectResult>(await apply.WaitAsync(TimeSpan.FromSeconds(15)));
|
|
await applying.SaveChangesAsync();
|
|
await using var verify = database.Context();
|
|
Assert.Equal("discarded", (await verify.CvExtractionRuns.SingleAsync()).Status);
|
|
Assert.Empty(await verify.CareerProfileVersions.ToListAsync());
|
|
Assert.Empty(await verify.CareerSkills.ToListAsync());
|
|
Assert.Null((await verify.Users.SingleAsync()).CurrentCvExtractionRunId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Applied_review_prevents_a_discard_that_already_read_pending()
|
|
{
|
|
await using var database = await MariaDbDatabase.CreateAsync();
|
|
var id = await Seed(database);
|
|
var gate = new RunReadGate();
|
|
await using var discarding = database.Context(Owner, gate);
|
|
var discard = Controller(discarding).DiscardRun(id);
|
|
try
|
|
{
|
|
await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15));
|
|
await using var applying = database.Context();
|
|
Assert.IsType<OkObjectResult>(await Controller(applying).ApplyReviewed(id, Review));
|
|
}
|
|
finally { gate.Release.TrySetResult(); }
|
|
Assert.IsType<ConflictObjectResult>(await discard.WaitAsync(TimeSpan.FromSeconds(15)));
|
|
await using var verify = database.Context();
|
|
Assert.Equal("applied", (await verify.CvExtractionRuns.SingleAsync()).Status);
|
|
Assert.Equal("C#", (await verify.CareerSkills.SingleAsync()).Name);
|
|
Assert.Single(await verify.CareerProfileVersions.ToListAsync());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Retention_with_a_stale_pending_snapshot_preserves_applied_result_and_replay()
|
|
{
|
|
await using var database = await MariaDbDatabase.CreateAsync();
|
|
var paths = database.App.Services.GetRequiredService<AppPaths>();
|
|
var source = Path.Combine(paths.CvArtifactsRoot, "lifecycle-synthetic.pdf");
|
|
await File.WriteAllTextAsync(source, "Synthetic source");
|
|
var id = await Seed(database, source);
|
|
var gate = new RunReadGate();
|
|
await using var sweeping = database.Context(Owner, gate);
|
|
var sweep = CvImportRetentionHostedService.SweepAsync(sweeping, paths, DateTimeOffset.UtcNow, default);
|
|
string? candidate = null;
|
|
string? receipt = null;
|
|
try
|
|
{
|
|
await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15));
|
|
await using var applying = database.Context();
|
|
Assert.IsType<OkObjectResult>(await Controller(applying).ApplyReviewed(id, Review));
|
|
var saved = await applying.CvExtractionRuns.AsNoTracking().SingleAsync();
|
|
candidate = saved.StructuredProfileJson;
|
|
receipt = saved.ReviewJson;
|
|
}
|
|
finally { gate.Release.TrySetResult(); }
|
|
await sweep.WaitAsync(TimeSpan.FromSeconds(15));
|
|
await using var verify = database.Context();
|
|
var run = await verify.CvExtractionRuns.SingleAsync();
|
|
Assert.Equal("applied", run.Status);
|
|
Assert.Equal(candidate, run.StructuredProfileJson);
|
|
Assert.Equal(receipt, run.ReviewJson);
|
|
Assert.False(File.Exists(source));
|
|
Assert.Equal("", (await verify.CvUploadArtifacts.SingleAsync()).StoragePath);
|
|
Assert.IsType<OkObjectResult>(await Controller(verify).ApplyReviewed(id, Review));
|
|
Assert.Single(await verify.CareerProfileVersions.ToListAsync());
|
|
Assert.Equal("C#", (await verify.CareerSkills.SingleAsync()).Name);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Cancellation_acknowledgement_preserves_a_discarded_review()
|
|
{
|
|
await using var database = await MariaDbDatabase.CreateAsync();
|
|
var id = await Seed(database);
|
|
await using var db = database.Context();
|
|
var operation = Operation(id, OperationStatuses.Running);
|
|
operation.CancellationRequestedAtUtc = DateTime.UtcNow;
|
|
db.UserOperations.Add(operation);
|
|
await db.SaveChangesAsync();
|
|
Assert.IsType<NoContentResult>(await Controller(db).DiscardRun(id));
|
|
var store = new UserOperationStore(db, TimeProvider.System);
|
|
Assert.Equal(1, await store.AcknowledgeCancellationAsync(operation.Id, "lease", default));
|
|
await using var verify = database.Context();
|
|
Assert.Equal("discarded", (await verify.CvExtractionRuns.SingleAsync()).Status);
|
|
Assert.Equal(OperationStatuses.Cancelled, (await verify.UserOperations.SingleAsync()).Status);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData("discarded")]
|
|
[InlineData("expired")]
|
|
[InlineData("applied")]
|
|
public async Task Retry_cannot_reopen_a_terminal_review(string status)
|
|
{
|
|
await using var database = await MariaDbDatabase.CreateAsync();
|
|
var id = await Seed(database);
|
|
await using var db = database.Context();
|
|
var run = await db.CvExtractionRuns.SingleAsync();
|
|
run.Status = status;
|
|
var operation = Operation(id, OperationStatuses.Cancelled);
|
|
db.UserOperations.Add(operation);
|
|
await db.SaveChangesAsync();
|
|
Assert.False(await new UserOperationStore(db, TimeProvider.System).RetryAsync(operation.Id, default));
|
|
await using var verify = database.Context();
|
|
Assert.Equal(status, (await verify.CvExtractionRuns.SingleAsync()).Status);
|
|
Assert.Equal(OperationStatuses.Cancelled, (await verify.UserOperations.SingleAsync()).Status);
|
|
}
|
|
|
|
private static UserOperation Operation(int runId, string status) => new()
|
|
{
|
|
Id = Guid.NewGuid(), OwnerUserId = Owner, TaskType = CvProcessingQueue.TaskType,
|
|
SubjectType = CvProcessingQueue.SubjectType, SubjectId = runId.ToString(System.Globalization.CultureInfo.InvariantCulture),
|
|
IdempotencyKey = "lifecycle", Status = status, LeaseToken = "lease",
|
|
CreatedAtUtc = DateTime.UtcNow, AvailableAtUtc = DateTime.UtcNow,
|
|
};
|
|
|
|
private static async Task<int> Seed(MariaDbDatabase database, string? source = null)
|
|
{
|
|
await using var db = database.Context();
|
|
db.Users.Add(new() { Id = Owner, UserName = "lifecycle-synthetic" });
|
|
var run = new CvExtractionRun
|
|
{
|
|
OwnerUserId = Owner, Trigger = "document_import", ParserVersion = ProfileCvController.DocumentParserVersion,
|
|
Status = "pending_review", StartedAtUtc = DateTimeOffset.UtcNow.AddDays(-8),
|
|
RawExtractedText = "C#", DocumentJson = "{}",
|
|
StructuredProfileJson = StructuredCvProfileJson.SerializePersisted(new() { Skills = { "C#" } }),
|
|
Artifact = source is null ? null : new() { OwnerUserId = Owner, StoragePath = source, OriginalFileName = "synthetic.pdf" },
|
|
};
|
|
db.CvExtractionRuns.Add(run);
|
|
await db.SaveChangesAsync();
|
|
return run.Id;
|
|
}
|
|
|
|
private static ProfileCvController Controller(JobTrackerContext db)
|
|
{
|
|
var user = db.Users.Single(u => u.Id == Owner);
|
|
var users = TestHostFactory.CreateUserManager();
|
|
users.Setup(u => u.GetUserAsync(It.IsAny<ClaimsPrincipal>())).ReturnsAsync(user);
|
|
users.Setup(u => u.UpdateAsync(user)).Returns(async () =>
|
|
{
|
|
await db.SaveChangesAsync();
|
|
return IdentityResult.Success;
|
|
});
|
|
return new(users.Object, Mock.Of<ISummarizerService>(), db, null!)
|
|
{ ControllerContext = new() { HttpContext = new DefaultHttpContext() } };
|
|
}
|
|
|
|
// Pause after the server has executed the SELECT, so the result contains the old row.
|
|
private sealed class RunReadGate : DbCommandInterceptor
|
|
{
|
|
public TaskCompletionSource Entered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
private int _entered;
|
|
|
|
public override async ValueTask<DbDataReader> ReaderExecutedAsync(DbCommand command, CommandExecutedEventData eventData, DbDataReader result, CancellationToken cancellationToken = default)
|
|
{
|
|
if (command.CommandText.TrimStart().StartsWith("SELECT", StringComparison.OrdinalIgnoreCase)
|
|
&& command.CommandText.Contains("`CvExtractionRuns`", StringComparison.Ordinal)
|
|
&& Interlocked.Exchange(ref _entered, 1) == 0)
|
|
{
|
|
Entered.TrySetResult();
|
|
await Release.Task.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken);
|
|
}
|
|
return result;
|
|
}
|
|
}
|
|
}
|