using System.Runtime.CompilerServices;
using System.Security.Cryptography;
using System.Text;
using AvParser.Core.Collecting;
using AvParser.Core.Parsing;
using AvParser.Core.Settings;
using AvParser.Infrastructure.Collecting;
using AvParser.Infrastructure.Media;
using AvParser.Infrastructure.Storage;
using Microsoft.Extensions.Logging.Abstractions;
using ReactiveUI.Primitives.Signals;
namespace AvParser.Infrastructure.Tests.Collecting;
/// A source that hands back exactly the addresses the test names.
internal sealed class ListSource(params string[] urls) : IMediaSource
{
public string Id => "test-source";
public string DisplayName => "Test source";
public string Description => string.Empty;
public int Listings { get; private set; }
public bool CanParse(MediaQuery input) => true;
public async IAsyncEnumerable> ParseAsync(
MediaQuery input,
IProgress? progress,
[EnumeratorCancellation] CancellationToken cancellationToken
)
{
Listings++;
for (var index = 0; index < urls.Length; index++)
{
cancellationToken.ThrowIfCancellationRequested();
await Task.Yield();
yield return ParseOutcome.Success(
new MediaCandidate(new Uri(urls[index])) { SourceId = Id, Ordinal = index + 1 }
);
}
}
}
/// A fetcher that stages bytes from a table instead of using a network.
internal sealed class ScriptedFetcher(BlobStore blobs) : IMediaFetcher
{
public Dictionary Content { get; } = new(StringComparer.Ordinal);
public HashSet Failing { get; } = new(StringComparer.Ordinal);
public List Fetched { get; } = [];
public TimeSpan Delay { get; set; }
public async Task FetchAsync(
MediaCandidate candidate,
FetchOptions options,
CancellationToken cancellationToken = default
)
{
var url = candidate.Url.AbsoluteUri;
lock (Fetched)
{
Fetched.Add(url);
}
if (Delay > TimeSpan.Zero)
{
await Task.Delay(Delay, cancellationToken);
}
cancellationToken.ThrowIfCancellationRequested();
if (Failing.Contains(url) || !Content.TryGetValue(url, out var bytes))
{
return new FetchResult(SeenOutcome.Failed, candidate.Url) { ErrorCode = "RequestFailed", HttpStatus = 500 };
}
var temp = blobs.CreateTempPath();
await File.WriteAllBytesAsync(temp, bytes, cancellationToken);
var hash = Convert.ToHexStringLower(SHA256.HashData(bytes));
return new FetchResult(SeenOutcome.Stored, candidate.Url)
{
Blob = MediaBlob.Create(hash, MediaKind.Png, bytes.Length),
TempPath = temp,
HttpStatus = 200,
};
}
}
/// In-memory settings, so the runner never reads the developer's real profile.
internal sealed class FixedSettings(AppSettings? initial = null) : ISettingsService, IDisposable
{
private readonly BehaviorSignal _current = new(initial ?? new AppSettings());
public AppSettings Current => _current.Value;
public IObservable Changes => _current;
public void Update(Func mutate) => _current.OnNext(mutate(_current.Value));
public Task FlushAsync(CancellationToken cancellationToken = default) => Task.CompletedTask;
public void Dispose() => _current.Dispose();
}
public sealed class CollectRunnerTests : IAsyncLifetime
{
private readonly string _root = Path.Combine(Path.GetTempPath(), "AvParserTests", Guid.NewGuid().ToString("N"));
private AppPaths _paths = null!;
private SqliteMediaIndex _index = null!;
private BlobStore _blobs = null!;
private MediaStore _store = null!;
private ScriptedFetcher _fetcher = null!;
private FixedSettings _settings = null!;
private HostThrottle _throttle = null!;
private CollectRunner _runner = null!;
public async ValueTask InitializeAsync()
{
_paths = new AppPaths(_root);
_paths.EnsureCreated();
_index = new SqliteMediaIndex(_paths, NullLogger.Instance);
_blobs = new BlobStore(_paths, NullLogger.Instance);
_store = new MediaStore(
_index,
_blobs,
new ShowcaseLinker(_paths, _blobs, NullLogger.Instance),
NullLogger.Instance
);
_settings = new FixedSettings(new AppSettings());
_fetcher = new ScriptedFetcher(_blobs);
_throttle = new HostThrottle(4, TimeSpan.Zero, NullLogger.Instance);
_runner = new CollectRunner(_fetcher, _store, _throttle, NullLogger.Instance);
await _store.InitialiseAsync(TestContext.Current.CancellationToken);
}
public ValueTask DisposeAsync()
{
_settings.Dispose();
_throttle.Dispose();
_index.Dispose();
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
if (Directory.Exists(_root))
{
try
{
Directory.Delete(_root, recursive: true);
}
catch (IOException)
{
// Not worth failing a green test over.
}
}
return ValueTask.CompletedTask;
}
private static byte[] Image(string seed) => [.. Samples.Png(), .. Encoding.UTF8.GetBytes(seed)];
private async Task>> RunAsync(IMediaSource source, CollectOptions? options = null)
{
var results = new List>();
await foreach (
var outcome in _runner.RunAsync(
source,
new MediaQuery(),
options ?? new CollectOptions(),
null,
TestContext.Current.CancellationToken
)
)
{
results.Add(outcome);
}
return results;
}
[Fact]
public async Task Everything_the_source_finds_is_downloaded_and_stored()
{
var source = new ListSource("https://a.test/1.png", "https://a.test/2.png");
_fetcher.Content["https://a.test/1.png"] = Image("one");
_fetcher.Content["https://a.test/2.png"] = Image("two");
var results = await RunAsync(source);
results.Count.ShouldBe(2);
results.ShouldAllBe(r => r.IsSuccess);
results.ShouldAllBe(r => r.Value!.Status == CollectStatus.Stored);
var stats = await _store.GetStatsAsync(TestContext.Current.CancellationToken);
stats.BlobCount.ShouldBe(2);
}
[Fact]
public async Task A_second_run_over_the_same_list_fetches_nothing()
{
// The journal exists precisely so that re-running a list costs no requests and no proxy.
var source = new ListSource("https://a.test/1.png", "https://a.test/2.png");
_fetcher.Content["https://a.test/1.png"] = Image("one");
_fetcher.Content["https://a.test/2.png"] = Image("two");
await RunAsync(source);
_fetcher.Fetched.Clear();
var second = await RunAsync(source);
_fetcher.Fetched.ShouldBeEmpty();
second.Count.ShouldBe(2);
second.ShouldAllBe(r => r.Value!.Status == CollectStatus.Skipped);
}
[Fact]
public async Task Forcing_a_refetch_ignores_the_journal()
{
var source = new ListSource("https://a.test/1.png");
_fetcher.Content["https://a.test/1.png"] = Image("one");
await RunAsync(source);
_fetcher.Fetched.Clear();
await RunAsync(source, new CollectOptions { ForceRefetch = true });
_fetcher.Fetched.ShouldHaveSingleItem();
}
[Fact]
public async Task A_failure_is_reported_and_retried_next_time()
{
// Failures describe the moment, not the resource; a flaky network must not lose content.
var source = new ListSource("https://a.test/1.png");
_fetcher.Failing.Add("https://a.test/1.png");
var first = await RunAsync(source);
first.ShouldHaveSingleItem().IsSuccess.ShouldBeFalse();
_fetcher.Failing.Clear();
_fetcher.Content["https://a.test/1.png"] = Image("recovered");
_fetcher.Fetched.Clear();
var second = await RunAsync(source);
_fetcher.Fetched.ShouldHaveSingleItem();
second.ShouldHaveSingleItem().Value!.Status.ShouldBe(CollectStatus.Stored);
}
[Fact]
public async Task The_same_bytes_at_two_addresses_are_stored_once()
{
var source = new ListSource("https://a.test/1.png", "https://a.test/2.png");
var shared = Image("identical");
_fetcher.Content["https://a.test/1.png"] = shared;
_fetcher.Content["https://a.test/2.png"] = shared;
var results = await RunAsync(source, new CollectOptions { MaxConcurrentDownloads = 1 });
results.Count(r => r.Value!.Status == CollectStatus.Stored).ShouldBe(1);
results.Count(r => r.Value!.Status == CollectStatus.Duplicate).ShouldBe(1);
(await _store.GetStatsAsync(TestContext.Current.CancellationToken)).BlobCount.ShouldBe(1);
}
[Fact]
public async Task A_source_that_throws_does_not_take_the_run_down()
{
var results = await RunAsync(new ThrowingSource());
results.ShouldHaveSingleItem().Error!.Code.ShouldBe("SourceFailed");
}
[Fact]
public async Task Cancelling_stops_the_run_and_nothing_writes_afterwards()
{
// The runner owns its workers and waits for them; otherwise a stopped run keeps writing to
// the store after the page has already said it stopped.
var urls = Enumerable.Range(0, 40).Select(i => $"https://a.test/{i}.png").ToArray();
var source = new ListSource(urls);
foreach (var url in urls)
{
_fetcher.Content[url] = Image(url);
}
_fetcher.Delay = TimeSpan.FromMilliseconds(50);
using var cancellation = new CancellationTokenSource();
var results = new List>();
await Should.ThrowAsync(async () =>
{
await foreach (
var outcome in _runner.RunAsync(
source,
new MediaQuery(),
new CollectOptions(),
null,
cancellation.Token
)
)
{
results.Add(outcome);
if (results.Count == 3)
{
await cancellation.CancelAsync();
}
}
});
var afterReturn = await _store.GetStatsAsync(TestContext.Current.CancellationToken);
await Task.Delay(200, TestContext.Current.CancellationToken);
var later = await _store.GetStatsAsync(TestContext.Current.CancellationToken);
later.ItemCount.ShouldBe(afterReturn.ItemCount);
Directory.EnumerateFiles(_paths.MediaTempDirectory).ShouldBeEmpty();
}
private sealed class ThrowingSource : IMediaSource
{
public string Id => "throwing";
public string DisplayName => "Throwing source";
public string Description => string.Empty;
public bool CanParse(MediaQuery input) => true;
public async IAsyncEnumerable> ParseAsync(
MediaQuery input,
IProgress? progress,
[EnumeratorCancellation] CancellationToken cancellationToken
)
{
await Task.Yield();
throw new InvalidOperationException("the listing endpoint moved");
#pragma warning disable CS0162
yield break;
#pragma warning restore CS0162
}
}
}