50 lines
3.0 KiB
C#
50 lines
3.0 KiB
C#
using System.Text.Json;
|
|
using WxAgent.Service;
|
|
using Xunit;
|
|
|
|
namespace WxAgent.Service.Tests;
|
|
|
|
public sealed class OperationQueueTests
|
|
{
|
|
[Fact]
|
|
public async Task QueueCancelsBeforeExecutionRevokesAtExecutionAndNeverOverlapsTimedOutAction()
|
|
{
|
|
var directory = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString("N"));
|
|
Directory.CreateDirectory(directory);
|
|
var options = new ServiceOptions { DataDirectory = directory, CredentialFile = Path.Combine(directory, "auth.json") };
|
|
var hash = ServiceOptions.HashToken(new string('A', 43));
|
|
var identity = new ServiceIdentity("alice", hash, ["read", "write"], ["account"]);
|
|
void Credentials(string value) => File.WriteAllText(options.CredentialFile,
|
|
JsonSerializer.Serialize(new[] { new ServiceCredential("alice", value, identity.Permissions, identity.AccountIds) }));
|
|
Credentials(hash);
|
|
var capability = new AgentCapability("send", true, true, true, true, true, "write", false, 30, null, []);
|
|
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
try
|
|
{
|
|
using var store = new OperationStore(options);
|
|
using var queue = new OperationQueue(store, new ServiceSecurity(options));
|
|
await queue.StartAsync(default);
|
|
var first = queue.Submit(identity, "account", capability, "first", "one", async _ => { entered.SetResult(); await release.Task; });
|
|
await entered.Task.WaitAsync(TimeSpan.FromSeconds(10));
|
|
var calls = 0;
|
|
var second = queue.Submit(identity, "account", capability, "second", "two", _ => { Interlocked.Increment(ref calls); return Task.CompletedTask; });
|
|
Assert.Equal("Cancelled", queue.Cancel(identity, second.Id).State);
|
|
var third = queue.Submit(identity, "account", capability, "third", "three", _ => { Interlocked.Increment(ref calls); return Task.CompletedTask; });
|
|
queue.Cancel(identity, first.Id);
|
|
// Cancel is cooperative: the first action still owns the lane until it really exits.
|
|
Assert.Equal("Running", store.Get(first.Id, "alice").State);
|
|
Assert.Equal("Queued", store.Get(third.Id, "alice").State);
|
|
Credentials(ServiceOptions.HashToken(new string('B', 43)));
|
|
release.SetResult();
|
|
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
while (store.Get(third.Id, "alice").State == "Queued") await Task.Delay(10, deadline.Token);
|
|
Assert.Equal(0, calls);
|
|
Assert.Equal("Unconfirmed", store.Get(first.Id, "alice").State);
|
|
Assert.Equal("AuthorizationRevoked", store.Get(third.Id, "alice").ErrorCode);
|
|
await queue.StopAsync(default);
|
|
}
|
|
finally { release.TrySetResult(); Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); Directory.Delete(directory, true); }
|
|
}
|
|
}
|