av-parser/tests/AvParser.Infrastructure.Tests/Collecting/CollectRunnerTests.cs

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;

/// <summary>A source that hands back exactly the addresses the test names.</summary>
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<ParseOutcome<MediaCandidate>> ParseAsync(
        MediaQuery input,
        IProgress<ParseProgress>? progress,
        [EnumeratorCancellation] CancellationToken cancellationToken
    )
    {
        Listings++;

        for (var index = 0; index < urls.Length; index++)
        {
            cancellationToken.ThrowIfCancellationRequested();
            await Task.Yield();

            yield return ParseOutcome<MediaCandidate>.Success(
                new MediaCandidate(new Uri(urls[index])) { SourceId = Id, Ordinal = index + 1 }
            );
        }
    }
}

/// <summary>A fetcher that stages bytes from a table instead of using a network.</summary>
internal sealed class ScriptedFetcher(BlobStore blobs) : IMediaFetcher
{
    public Dictionary<string, byte[]> Content { get; } = new(StringComparer.Ordinal);

    public HashSet<string> Failing { get; } = new(StringComparer.Ordinal);

    public List<string> Fetched { get; } = [];

    public TimeSpan Delay { get; set; }

    public async Task<FetchResult> 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,
        };
    }
}

/// <summary>In-memory settings, so the runner never reads the developer's real profile.</summary>
internal sealed class FixedSettings(AppSettings? initial = null) : ISettingsService, IDisposable
{
    private readonly BehaviorSignal<AppSettings> _current = new(initial ?? new AppSettings());

    public AppSettings Current => _current.Value;

    public IObservable<AppSettings> Changes => _current;

    public void Update(Func<AppSettings, AppSettings> 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<SqliteMediaIndex>.Instance);
        _blobs = new BlobStore(_paths, NullLogger<BlobStore>.Instance);
        _store = new MediaStore(
            _index,
            _blobs,
            new ShowcaseLinker(_paths, _blobs, NullLogger<ShowcaseLinker>.Instance),
            NullLogger<MediaStore>.Instance
        );

        // Direct is allowed here: these tests are about the runner, not the proxy gate.
        _settings = new FixedSettings(new AppSettings { AllowDirectConnection = true });
        _fetcher = new ScriptedFetcher(_blobs);
        _throttle = new HostThrottle(4, TimeSpan.Zero, NullLogger<CollectRunnerTests>.Instance);
        _runner = new CollectRunner(_fetcher, _store, _throttle, NullLogger<CollectRunner>.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<List<ParseOutcome<CollectedItem>>> RunAsync(IMediaSource source, CollectOptions? options = null)
    {
        var results = new List<ParseOutcome<CollectedItem>>();

        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<ParseOutcome<CollectedItem>>();

        await Should.ThrowAsync<OperationCanceledException>(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<ParseOutcome<MediaCandidate>> ParseAsync(
            MediaQuery input,
            IProgress<ParseProgress>? progress,
            [EnumeratorCancellation] CancellationToken cancellationToken
        )
        {
            await Task.Yield();
            throw new InvalidOperationException("the listing endpoint moved");
#pragma warning disable CS0162
            yield break;
#pragma warning restore CS0162
        }
    }
}

public class OwnServiceListingTests
{
    private static readonly Uri Endpoint = new("https://own.test/api/list");

    [Fact]
    public void An_object_with_items_is_read()
    {
        var page = OwnServiceSource.ReadPage(
            """
            { "items": [ { "url": "https://own.test/a.png", "id": "42", "name": "kitten",
                           "published": "2026-08-13T10:00:00Z", "size": 4096, "tags": ["cats"] } ],
              "next": "page2" }
            """,
            Endpoint
        );

        var item = page.Items.ShouldHaveSingleItem();
        item.Url.AbsoluteUri.ShouldBe("https://own.test/a.png");
        item.ExternalId.ShouldBe("42");
        item.SuggestedName.ShouldBe("kitten");
        item.ExpectedLength.ShouldBe(4096);
        item.Tags.ShouldBe(["cats"]);
        page.Next.ShouldBe("page2");
    }

    [Fact]
    public void A_bare_array_of_addresses_is_read()
    {
        // The service on the other end is the user's own; it should not have to be rewritten to
        // match a schema we invented.
        var page = OwnServiceSource.ReadPage("""["https://own.test/a.png", "https://own.test/b.gif"]""", Endpoint);

        page.Items.Count.ShouldBe(2);
        page.Next.ShouldBeNull();
    }

    [Fact]
    public void Relative_addresses_resolve_against_the_endpoint()
    {
        var page = OwnServiceSource.ReadPage("""{"items":[{"url":"/files/a.png"}]}""", Endpoint);

        page.Items.ShouldHaveSingleItem().Url.AbsoluteUri.ShouldBe("https://own.test/files/a.png");
    }

    [Fact]
    public void Entries_without_a_usable_address_are_dropped()
    {
        var page = OwnServiceSource.ReadPage(
            """{"items":[{"name":"no url"}, {"url":"data:image/png;base64,AA"}, {"url":"https://own.test/ok.png"}]}""",
            Endpoint
        );

        page.Items.ShouldHaveSingleItem().Url.AbsoluteUri.ShouldBe("https://own.test/ok.png");
    }

    [Fact]
    public void An_empty_listing_is_not_an_error()
    {
        OwnServiceSource.ReadPage("""{"items":[]}""", Endpoint).Items.ShouldBeEmpty();
        OwnServiceSource.ReadPage("[]", Endpoint).Items.ShouldBeEmpty();
    }
}