716 lines
31 KiB
C#
716 lines
31 KiB
C#
using Explorer.Application;
|
|
using Explorer.Domain;
|
|
using Explorer.Domain.Abstractions;
|
|
using Explorer.FileOperations;
|
|
using Explorer.Storage.Sqlite;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
|
|
namespace Explorer.FileOperations.Tests;
|
|
|
|
public class TransferQueueTests
|
|
{
|
|
[Fact]
|
|
public async Task Copies_run_one_after_another()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var firstStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var releaseFirst = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var secondStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
ctx.Shell.OnCopy = src =>
|
|
{
|
|
if (src.EndsWith("a.txt", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
firstStarted.TrySetResult();
|
|
releaseFirst.Task.GetAwaiter().GetResult();
|
|
}
|
|
else
|
|
{
|
|
secondStarted.TrySetResult();
|
|
}
|
|
};
|
|
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt"), ctx.File("b.txt")], ctx.Dest);
|
|
await firstStarted.Task.WaitAsync(TimeSpan.FromSeconds(3));
|
|
Assert.False(secondStarted.Task.IsCompleted);
|
|
Assert.Equal(1, ctx.Shell.CopyCount);
|
|
|
|
releaseFirst.TrySetResult();
|
|
await secondStarted.Task.WaitAsync(TimeSpan.FromSeconds(3));
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Count(j => j.Status == TransferStatus.Done) == 2);
|
|
Assert.Equal(new[] { ctx.File("a.txt"), ctx.File("b.txt") }, ctx.Shell.Copied);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Pause_keeps_later_jobs_from_starting()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var firstStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var awaitingPause = true;
|
|
ctx.Shell.OnCopy = src =>
|
|
{
|
|
if (src.EndsWith("a.txt", StringComparison.OrdinalIgnoreCase) && awaitingPause)
|
|
{
|
|
firstStarted.TrySetResult();
|
|
WaitUntil(() => ctx.Shell.PauseRequested?.Invoke() == true, TimeSpan.FromSeconds(3))
|
|
.GetAwaiter().GetResult();
|
|
awaitingPause = false;
|
|
}
|
|
};
|
|
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt"), ctx.File("b.txt")], ctx.Dest);
|
|
await firstStarted.Task.WaitAsync(TimeSpan.FromSeconds(3));
|
|
ctx.Queue.PauseAll();
|
|
await WaitUntil(() => ctx.Queue.Snapshot()[0].Status == TransferStatus.Paused);
|
|
Assert.DoesNotContain(ctx.Shell.Copied, p => p.EndsWith("b.txt", StringComparison.OrdinalIgnoreCase));
|
|
|
|
ctx.Queue.ResumeAll();
|
|
await WaitUntil(() => ctx.Queue.Snapshot().All(j => j.Status == TransferStatus.Done));
|
|
Assert.Contains(ctx.Shell.Copied, p => p.EndsWith("b.txt", StringComparison.OrdinalIgnoreCase));
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Remove_drops_a_queued_step()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
ctx.Queue.PauseAll();
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt"), ctx.File("b.txt")], ctx.Dest);
|
|
var jobs = ctx.Queue.Snapshot();
|
|
Assert.Equal(2, jobs.Count);
|
|
ctx.Queue.Cancel(jobs[1].Id);
|
|
ctx.Queue.ResumeAll();
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j => j.Status == TransferStatus.Done));
|
|
Assert.Equal(new[] { ctx.File("a.txt") }, ctx.Shell.Copied);
|
|
Assert.DoesNotContain(ctx.Queue.Snapshot(), j => j.SourcePath.EndsWith("b.txt", StringComparison.OrdinalIgnoreCase)
|
|
&& j.Status is TransferStatus.Queued or TransferStatus.Running or TransferStatus.Done);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Cancel_stops_the_running_copy()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var firstStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var releaseFirst = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
ctx.Shell.OnCopy = src =>
|
|
{
|
|
if (src.EndsWith("a.txt", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
firstStarted.TrySetResult();
|
|
releaseFirst.Task.GetAwaiter().GetResult();
|
|
}
|
|
};
|
|
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt"), ctx.File("b.txt")], ctx.Dest);
|
|
await firstStarted.Task.WaitAsync(TimeSpan.FromSeconds(3));
|
|
var running = ctx.Queue.Snapshot().Single(j => j.SourcePath.EndsWith("a.txt", StringComparison.OrdinalIgnoreCase));
|
|
ctx.Queue.Cancel(running.Id);
|
|
releaseFirst.TrySetResult();
|
|
await WaitUntil(() => ctx.Shell.Copied.Contains(ctx.File("b.txt")));
|
|
Assert.DoesNotContain(ctx.Shell.Copied, p => p.EndsWith("a.txt", StringComparison.OrdinalIgnoreCase));
|
|
Assert.DoesNotContain(ctx.Queue.Snapshot(), j => j.Id == running.Id && j.Status == TransferStatus.Done);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Pause_queued_job_skips_it()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
ctx.Queue.PauseAll();
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt"), ctx.File("b.txt")], ctx.Dest);
|
|
var jobs = ctx.Queue.Snapshot();
|
|
ctx.Queue.Pause(jobs[0].Id);
|
|
ctx.Queue.ResumeAll();
|
|
await WaitUntil(() => ctx.Shell.Copied.Contains(ctx.File("b.txt")));
|
|
Assert.Equal(new[] { ctx.File("b.txt") }, ctx.Shell.Copied);
|
|
Assert.Equal(TransferStatus.Paused, ctx.Queue.Snapshot().First(j => j.Id == jobs[0].Id).Status);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reorder_changes_which_job_runs_first()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
ctx.Queue.PauseAll();
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt"), ctx.File("b.txt")], ctx.Dest);
|
|
var jobs = ctx.Queue.Snapshot();
|
|
Assert.True(ctx.Queue.MoveUp(jobs[1].Id));
|
|
ctx.Queue.ResumeAll();
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Count(j => j.Status == TransferStatus.Done) == 2);
|
|
Assert.Equal(new[] { ctx.File("b.txt"), ctx.File("a.txt") }, ctx.Shell.Copied);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Restores_paused_and_queued_jobs_after_restart()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
ctx.Queue.PauseAll();
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt"), ctx.File("b.txt")], ctx.Dest);
|
|
var jobs = ctx.Queue.Snapshot();
|
|
ctx.Queue.Pause(jobs[0].Id);
|
|
await WaitUntil(async () =>
|
|
{
|
|
var stored = await ctx.Store.Transfers.GetIncompleteAsync();
|
|
return stored.Any(j => j.Id == jobs[0].Id && j.Status == TransferStatus.Paused);
|
|
});
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
|
|
var restored = ctx.CreateQueue();
|
|
restored.PauseAll();
|
|
await restored.StartAsync(CancellationToken.None);
|
|
await WaitUntil(() => restored.Snapshot().Count == 2);
|
|
Assert.Equal(TransferStatus.Paused, restored.Snapshot().First(j => j.Id == jobs[0].Id).Status);
|
|
Assert.Equal(TransferStatus.Queued, restored.Snapshot().First(j => j.Id == jobs[1].Id).Status);
|
|
restored.Resume(jobs[0].Id);
|
|
await WaitUntil(() => restored.Snapshot().Count(j => j.Status == TransferStatus.Done) == 2);
|
|
await restored.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Restores_running_jobs_as_paused()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var id = await ctx.Store.Transfers.InsertAsync(new TransferJob
|
|
{
|
|
Op = TransferOp.Copy,
|
|
SourcePath = ctx.File("a.txt"),
|
|
DestinationPath = Path.Combine(ctx.Dest, "a.txt"),
|
|
Status = TransferStatus.Running,
|
|
CreatedUtc = DateTimeOffset.UtcNow,
|
|
StartedUtc = DateTimeOffset.UtcNow
|
|
});
|
|
var restored = ctx.CreateQueue();
|
|
await restored.StartAsync(CancellationToken.None);
|
|
await WaitUntil(() => restored.Snapshot().Any(j => j.Id == id && j.Status == TransferStatus.Paused));
|
|
Assert.DoesNotContain(restored.Snapshot(), j => j.Status == TransferStatus.Running);
|
|
await restored.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Waiting_jobs_do_not_run_until_destination_is_reachable()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
ctx.Volumes.Reachable = false;
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt")], ctx.Dest);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j => j.Status == TransferStatus.Waiting));
|
|
await Task.Delay(80);
|
|
Assert.Equal(0, ctx.Shell.CopyCount);
|
|
Assert.Equal(FileOperationErrors.DestinationUnavailable, ctx.Queue.Snapshot()[0].WaitReason);
|
|
|
|
ctx.Volumes.Reachable = true;
|
|
ctx.Queue.NotifyAvailability();
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Done);
|
|
Assert.Equal(new[] { ctx.File("a.txt") }, ctx.Shell.Copied);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Retry_wakes_a_waiting_job_when_the_destination_is_back()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
ctx.Volumes.Reachable = false;
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt")], ctx.Dest);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j => j.Status == TransferStatus.Waiting));
|
|
|
|
ctx.Volumes.Reachable = true;
|
|
ctx.Queue.Retry(ctx.Queue.Snapshot()[0].Id);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Done);
|
|
Assert.Equal(1, ctx.Queue.Snapshot()[0].RetryCount);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Retry_reruns_a_failed_job()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
ctx.Shell.FailRemaining = 1;
|
|
ctx.Shell.FailError = "disk full";
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt")], ctx.Dest);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Failed);
|
|
Assert.Equal("disk full", ctx.Queue.Snapshot()[0].Error);
|
|
|
|
ctx.Queue.Retry(ctx.Queue.Snapshot()[0].Id);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Done);
|
|
Assert.Equal(1, ctx.Queue.Snapshot()[0].RetryCount);
|
|
Assert.Contains(ctx.File("a.txt"), ctx.Shell.Copied);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task File_lock_fails_and_can_be_retried()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
ctx.Shell.FailRemaining = 1;
|
|
ctx.Shell.FailError = "The process cannot access the file because it is being used by another process.";
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt")], ctx.Dest);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Failed);
|
|
Assert.Equal(FileOperationErrors.FileInUse, ctx.Queue.Snapshot()[0].Error);
|
|
|
|
ctx.Queue.Retry(ctx.Queue.Snapshot()[0].Id);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Done);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Queued_rename_collision_stays_failed()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueRenameAsync(ctx.File("a.txt"), "b.txt");
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Failed);
|
|
Assert.Equal(FileOperationErrors.NameExists, ctx.Queue.Snapshot()[0].Error);
|
|
Assert.True(System.IO.File.Exists(ctx.File("a.txt")));
|
|
Assert.True(System.IO.File.Exists(ctx.File("b.txt")));
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Queued_rename_moves_the_file()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueRenameAsync(ctx.File("a.txt"), "renamed.txt");
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Done);
|
|
Assert.False(System.IO.File.Exists(ctx.File("a.txt")));
|
|
Assert.True(System.IO.File.Exists(ctx.File("renamed.txt")));
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ClearFinished_keeps_history_and_does_not_restore_dismissed_jobs()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ctx.Queue.EnqueueCopyAsync([ctx.File("a.txt")], ctx.Dest);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Single().Status == TransferStatus.Done);
|
|
ctx.Queue.ClearFinished();
|
|
Assert.Empty(ctx.Queue.Snapshot());
|
|
await WaitUntil(async () =>
|
|
{
|
|
var history = await ctx.Store.Transfers.GetHistoryAsync(10);
|
|
return history.Any(j => j.Status == TransferStatus.Done && j.Dismissed);
|
|
});
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
|
|
var restored = ctx.CreateQueue();
|
|
await restored.StartAsync(CancellationToken.None);
|
|
await Task.Delay(50);
|
|
Assert.Empty(restored.Snapshot());
|
|
await restored.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Batch_rename_enqueues_each_item()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
ctx.Queue.PauseAll();
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ops.EnqueueRenameAsync([(ctx.File("a.txt"), "a2.txt"), (ctx.File("b.txt"), "b2.txt")]);
|
|
Assert.Equal(2, ctx.Queue.Snapshot().Count);
|
|
Assert.All(ctx.Queue.Snapshot(), j => Assert.Equal(TransferOp.Rename, j.Op));
|
|
ctx.Queue.ResumeAll();
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Count(j => j.Status == TransferStatus.Done) == 2);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Batch_rename_then_undo_restores_names()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
var batches = new RenameBatchService(new RenamePlanner(), ctx.Store, new LocalIndexMutations(ctx.Store), ops);
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
var subjects = new[]
|
|
{
|
|
new RenameSubject(ctx.File("a.txt"), "a.txt", false),
|
|
new RenameSubject(ctx.File("b.txt"), "b.txt", false)
|
|
};
|
|
var plan = batches.Preview(subjects, new RenameRuleSet { Prefix = "x_" });
|
|
Assert.True(plan.CanEnqueue);
|
|
await batches.EnqueueAsync(plan);
|
|
await WaitUntil(() => File.Exists(ctx.File("x_a.txt")) && File.Exists(ctx.File("x_b.txt")));
|
|
Assert.False(File.Exists(ctx.File("a.txt")));
|
|
var undo = await batches.UndoLastAsync();
|
|
Assert.True(undo.CanEnqueue);
|
|
await WaitUntil(() => File.Exists(ctx.File("a.txt")) && File.Exists(ctx.File("b.txt")));
|
|
Assert.False(File.Exists(ctx.File("x_a.txt")));
|
|
Assert.Null(await batches.GetUndoableAsync());
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Extract_fails_when_7zip_is_missing()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ops.ExtractAsync(ctx.File("a.txt"), Path.Combine(ctx.Dest, "out"));
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j =>
|
|
j.Op == TransferOp.Extract && j.Status == TransferStatus.Failed));
|
|
var job = ctx.Queue.Snapshot().Single(j => j.Op == TransferOp.Extract);
|
|
Assert.Contains("7-Zip", job.Error, StringComparison.OrdinalIgnoreCase);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Fake_extract_writes_files_and_can_be_cancelled()
|
|
{
|
|
var fake = new FakeArchiveExecutor { ExtractDelay = TimeSpan.FromSeconds(20) };
|
|
await using var ctx = await Harness.CreateAsync(fake);
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
var dest = Path.Combine(ctx.Dest, "pack");
|
|
await ops.ExtractAsync(ctx.File("a.txt"), dest);
|
|
await fake.Started.Task.WaitAsync(TimeSpan.FromSeconds(4));
|
|
var job = Assert.Single(ctx.Queue.Snapshot(), j => j.Op == TransferOp.Extract);
|
|
ctx.Queue.Cancel(job.Id);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j =>
|
|
j.Op == TransferOp.Extract && j.Status == TransferStatus.Cancelled));
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Fake_extract_completes_and_writes_output()
|
|
{
|
|
var fake = new FakeArchiveExecutor();
|
|
await using var ctx = await Harness.CreateAsync(fake);
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
var dest = Path.Combine(ctx.Dest, "pack");
|
|
await ops.ExtractAsync(ctx.File("a.txt"), dest);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j =>
|
|
j.Op == TransferOp.Extract && j.Status == TransferStatus.Done));
|
|
Assert.True(File.Exists(Path.Combine(dest, "out.txt")));
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Extract_refuses_online_only_cloud_archives()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync(new FakeArchiveExecutor(), new AlwaysHydrate());
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ops.ExtractAsync(ctx.File("a.txt"), Path.Combine(ctx.Dest, "out"));
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j =>
|
|
j.Op == TransferOp.Extract && j.Status == TransferStatus.Failed));
|
|
Assert.Equal(FileOperationErrors.CloudHydration, ctx.Queue.Snapshot().Single().Error);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Convert_fails_when_ffmpeg_is_missing()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ops.ConvertAsync(ctx.File("a.txt"), Path.Combine(ctx.Dest, "a.mp4"), ConversionKind.VideoToMp4);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j =>
|
|
j.Op == TransferOp.Convert && j.Status == TransferStatus.Failed));
|
|
var job = ctx.Queue.Snapshot().Single(j => j.Op == TransferOp.Convert);
|
|
Assert.Contains("FFmpeg", job.Error, StringComparison.OrdinalIgnoreCase);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Fake_convert_writes_output()
|
|
{
|
|
var fake = new FakeConversionExecutor();
|
|
await using var ctx = await Harness.CreateAsync(conversion: fake);
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
var dest = Path.Combine(ctx.Dest, "a.mp4");
|
|
await ops.ConvertAsync(ctx.File("a.txt"), dest, ConversionKind.VideoToMp4);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j =>
|
|
j.Op == TransferOp.Convert && j.Status == TransferStatus.Done));
|
|
Assert.True(File.Exists(dest));
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Convert_refuses_online_only_cloud_files()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync(hydration: new AlwaysHydrate(), conversion: new FakeConversionExecutor());
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ops.ConvertAsync(ctx.File("a.txt"), Path.Combine(ctx.Dest, "a.mp4"), ConversionKind.VideoToMp4);
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j =>
|
|
j.Op == TransferOp.Convert && j.Status == TransferStatus.Failed));
|
|
Assert.Equal(FileOperationErrors.CloudHydration, ctx.Queue.Snapshot().Single().Error);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Empty_recycle_bin_goes_through_the_queue()
|
|
{
|
|
await using var ctx = await Harness.CreateAsync();
|
|
var ops = new FileOperationService(ctx.Queue, ctx.Shell, new DiskEnum());
|
|
await ctx.Queue.StartAsync(CancellationToken.None);
|
|
await ops.EmptyRecycleBinAsync();
|
|
await WaitUntil(() => ctx.Queue.Snapshot().Any(j =>
|
|
j.Op == TransferOp.EmptyRecycleBin && j.Status == TransferStatus.Done));
|
|
Assert.Equal(1, ctx.Shell.EmptyCount);
|
|
await ctx.Queue.StopAsync(CancellationToken.None);
|
|
}
|
|
|
|
private static async Task WaitUntil(Func<bool> condition, TimeSpan? timeout = null)
|
|
=> await WaitUntil(() => Task.FromResult(condition()), timeout);
|
|
|
|
private static async Task WaitUntil(Func<Task<bool>> condition, TimeSpan? timeout = null)
|
|
{
|
|
var limit = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(4));
|
|
while (!await condition().ConfigureAwait(false))
|
|
{
|
|
if (DateTime.UtcNow > limit)
|
|
{
|
|
throw new TimeoutException("Condition was not met.");
|
|
}
|
|
|
|
await Task.Delay(20);
|
|
}
|
|
}
|
|
|
|
private sealed class Harness : IAsyncDisposable
|
|
{
|
|
public required TransferQueue Queue { get; init; }
|
|
public required GateShell Shell { get; init; }
|
|
public required SqliteIndexStore Store { get; init; }
|
|
public required ControlledVolumes Volumes { get; init; }
|
|
public required string Dest { get; init; }
|
|
public required string Root { get; init; }
|
|
|
|
public string File(string name) => Path.Combine(Root, name);
|
|
|
|
public static async Task<Harness> CreateAsync(
|
|
IArchiveExecutor? archives = null,
|
|
IHydrationGuard? hydration = null,
|
|
IMediaConversionProvider? conversion = null)
|
|
{
|
|
var root = Path.Combine(Path.GetTempPath(), "ew-xfer", Guid.NewGuid().ToString("N"));
|
|
Directory.CreateDirectory(root);
|
|
var dest = Path.Combine(root, "dest");
|
|
Directory.CreateDirectory(dest);
|
|
System.IO.File.WriteAllText(Path.Combine(root, "a.txt"), "a");
|
|
System.IO.File.WriteAllText(Path.Combine(root, "b.txt"), "b");
|
|
var db = Path.Combine(root, "index.db");
|
|
var store = new SqliteIndexStore(db, NullLogger<SqliteIndexStore>.Instance);
|
|
await store.OpenAsync();
|
|
var shell = new GateShell();
|
|
var volumes = new ControlledVolumes();
|
|
var queue = new TransferQueue(
|
|
new NativeFileOperationExecutor(shell, new DiskEnum(), archives, hydration, conversion),
|
|
store,
|
|
volumes,
|
|
NullLogger<TransferQueue>.Instance);
|
|
return new Harness
|
|
{
|
|
Queue = queue,
|
|
Shell = shell,
|
|
Store = store,
|
|
Volumes = volumes,
|
|
Dest = dest,
|
|
Root = root
|
|
};
|
|
}
|
|
|
|
public TransferQueue CreateQueue()
|
|
=> new(
|
|
new NativeFileOperationExecutor(Shell, new DiskEnum()),
|
|
Store,
|
|
Volumes,
|
|
NullLogger<TransferQueue>.Instance);
|
|
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
await Store.DisposeAsync();
|
|
try { Directory.Delete(Root, true); } catch { /* ignore */ }
|
|
}
|
|
}
|
|
}
|
|
|
|
internal sealed class DiskEnum : IFileSystemEnumerator
|
|
{
|
|
public IEnumerable<FileSystemItem> EnumerateChildren(string directoryPath) => EnumerateChildrenSafe(directoryPath, out _);
|
|
|
|
public IReadOnlyList<FileSystemItem> EnumerateChildrenSafe(string directoryPath, out string? error)
|
|
{
|
|
error = null;
|
|
return Directory.Exists(directoryPath)
|
|
? Directory.GetFileSystemEntries(directoryPath).Select(GetRequired).ToList()
|
|
: [];
|
|
}
|
|
|
|
public FileSystemItem? GetItem(string path)
|
|
=> System.IO.File.Exists(path) || Directory.Exists(path) ? GetRequired(path) : null;
|
|
|
|
private static FileSystemItem GetRequired(string path)
|
|
{
|
|
var isDir = Directory.Exists(path);
|
|
return new FileSystemItem
|
|
{
|
|
FullPath = path,
|
|
Name = Path.GetFileName(path),
|
|
IsDirectory = isDir,
|
|
SizeBytes = isDir ? 0 : new FileInfo(path).Length
|
|
};
|
|
}
|
|
}
|
|
|
|
internal sealed class GateShell : IShellFileOperations
|
|
{
|
|
public int CopyCount { get; private set; }
|
|
public List<string> Copied { get; } = [];
|
|
public Action<string>? OnCopy { get; set; }
|
|
public Func<bool>? PauseRequested { get; private set; }
|
|
public int FailRemaining { get; set; }
|
|
public string? FailError { get; set; }
|
|
public int EmptyCount { get; private set; }
|
|
|
|
public void Open(string path) { }
|
|
public bool DeleteToRecycleBin(IReadOnlyList<string> paths, out string? error) => Delete(paths, true, out error);
|
|
public bool Delete(IReadOnlyList<string> paths, bool recycle, out string? error) { error = null; return true; }
|
|
public bool CreateShortcut(string targetPath, string shortcutPath, out string? error)
|
|
{ error = null; return true; }
|
|
public bool EmptyRecycleBin(out string? error)
|
|
{
|
|
EmptyCount++;
|
|
error = null;
|
|
return true;
|
|
}
|
|
|
|
public bool CopyFileWithProgress(string source, string destination, bool overwrite, IProgress<long>? progress, CancellationToken cancellationToken, out string? error, Func<bool>? pauseRequested = null)
|
|
{
|
|
CopyCount++;
|
|
PauseRequested = pauseRequested;
|
|
progress?.Report(1);
|
|
OnCopy?.Invoke(source);
|
|
if (cancellationToken.IsCancellationRequested)
|
|
{
|
|
error = "Cancelled";
|
|
return false;
|
|
}
|
|
|
|
if (pauseRequested?.Invoke() == true)
|
|
{
|
|
error = "Paused";
|
|
return false;
|
|
}
|
|
|
|
if (FailRemaining > 0)
|
|
{
|
|
FailRemaining--;
|
|
error = FailError ?? "failed";
|
|
return false;
|
|
}
|
|
|
|
Copied.Add(source);
|
|
Directory.CreateDirectory(Path.GetDirectoryName(destination)!);
|
|
System.IO.File.Copy(source, destination, overwrite);
|
|
progress?.Report(new FileInfo(source).Length);
|
|
error = null;
|
|
return true;
|
|
}
|
|
|
|
public bool MoveFileWithProgress(string source, string destination, bool overwrite, IProgress<long>? progress, CancellationToken cancellationToken, out string? error, Func<bool>? pauseRequested = null)
|
|
=> CopyFileWithProgress(source, destination, overwrite, progress, cancellationToken, out error, pauseRequested);
|
|
}
|
|
|
|
internal sealed class ControlledVolumes : IVolumeService
|
|
{
|
|
public bool Reachable { get; set; } = true;
|
|
|
|
public IReadOnlyList<VolumeFingerprint> EnumerateOnlineVolumes() => [];
|
|
|
|
public VolumeFingerprint? Probe(string path) => null;
|
|
|
|
public VolumeSpace GetSpace(string path) => default;
|
|
|
|
public bool IsPathReachable(string path) => Reachable;
|
|
}
|
|
|
|
internal sealed class FakeArchiveExecutor : IArchiveExecutor
|
|
{
|
|
public bool IsAvailable { get; set; } = true;
|
|
public string MissingHint => SevenZipLocator.MissingHint;
|
|
public TimeSpan ExtractDelay { get; set; }
|
|
public TaskCompletionSource Started { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
public async Task ExtractAsync(
|
|
string archivePath,
|
|
string destinationDirectory,
|
|
IProgress<ArchiveProgress>? progress,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
Started.TrySetResult();
|
|
if (ExtractDelay > TimeSpan.Zero)
|
|
{
|
|
await Task.Delay(ExtractDelay, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
|
|
Directory.CreateDirectory(destinationDirectory);
|
|
await File.WriteAllTextAsync(Path.Combine(destinationDirectory, "out.txt"), "ok", cancellationToken)
|
|
.ConfigureAwait(false);
|
|
progress?.Report(new ArchiveProgress(100, 1, "out.txt"));
|
|
}
|
|
|
|
public Task CompressAsync(
|
|
IReadOnlyList<string> sources,
|
|
string archivePath,
|
|
ArchiveFormat format,
|
|
IProgress<ArchiveProgress>? progress,
|
|
CancellationToken cancellationToken)
|
|
=> Task.CompletedTask;
|
|
|
|
public Task AddAsync(
|
|
string archivePath,
|
|
IReadOnlyList<string> sources,
|
|
IProgress<ArchiveProgress>? progress,
|
|
CancellationToken cancellationToken)
|
|
=> Task.CompletedTask;
|
|
|
|
public Task VerifyAsync(
|
|
string archivePath,
|
|
IProgress<ArchiveProgress>? progress,
|
|
CancellationToken cancellationToken)
|
|
=> Task.CompletedTask;
|
|
}
|
|
|
|
internal sealed class FakeConversionExecutor : IMediaConversionProvider
|
|
{
|
|
public bool IsAvailable { get; set; } = true;
|
|
public string MissingHint => FfmpegLocator.MissingHint;
|
|
|
|
public async Task ConvertAsync(
|
|
string sourcePath,
|
|
string destinationPath,
|
|
ConversionKind kind,
|
|
IProgress<ConversionProgress>? progress,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
Directory.CreateDirectory(Path.GetDirectoryName(destinationPath)!);
|
|
await File.WriteAllTextAsync(destinationPath, "converted:" + kind, cancellationToken).ConfigureAwait(false);
|
|
progress?.Report(new ConversionProgress(100, destinationPath));
|
|
}
|
|
}
|
|
|
|
internal sealed class AlwaysHydrate : IHydrationGuard
|
|
{
|
|
public bool WouldHydrateOnRead(FileSystemItem item) => true;
|
|
public bool WouldHydrateOnRead(int attributes, CloudAvailability? availability) => true;
|
|
public Task<bool> WouldHydrateOnReadAsync(string path, CancellationToken cancellationToken = default)
|
|
=> Task.FromResult(true);
|
|
}
|