git.stella-ops.org/tests/reachability/StellaOps.Signals.Reachability.Tests/RuntimeFactsIngestionServiceTests.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using FluentAssertions;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Time.Testing;
using StellaOps.Signals.Models;
using StellaOps.Signals.Persistence;
using StellaOps.Signals.Services;
using Xunit;

namespace StellaOps.Signals.Reachability.Tests;

public sealed class RuntimeFactsIngestionServiceTests
{
    private readonly FakeReachabilityFactRepository repository = new();
    private readonly FakeReachabilityCache cache = new();
    private readonly FakeEventsPublisher eventsPublisher = new();
    private readonly FakeScoringService scoringService = new();
    private readonly FakeProvenanceNormalizer provenanceNormalizer = new();
    private readonly FakeTimeProvider timeProvider = new(DateTimeOffset.Parse("2025-11-09T10:15:00Z", null, System.Globalization.DateTimeStyles.AssumeUniversal));
    private readonly RuntimeFactsIngestionService sut;

    public RuntimeFactsIngestionServiceTests()
    {
        sut = new RuntimeFactsIngestionService(
            repository,
            timeProvider,
            cache,
            eventsPublisher,
            scoringService,
            provenanceNormalizer,
            NullLogger<RuntimeFactsIngestionService>.Instance);
    }

    [Fact]
    public async Task IngestAsync_InsertsAggregatedFacts()
    {
        var request = new RuntimeFactsIngestRequest
        {
            CallgraphId = "cg-123",
            Subject = new ReachabilitySubject { ScanId = "scan-1" },
            Events = new List<RuntimeFactEvent>
            {
                new()
                {
                    SymbolId = "symbol::foo",
                    HitCount = 3,
                    ProcessId = 100,
                    ProcessName = "worker",
                    ContainerId = "ctr-1",
                    SocketAddress = "10.0.0.5:443",
                    Metadata = new Dictionary<string, string?> { ["thread"] = "main" }
                },
                new()
                {
                    SymbolId = "symbol::foo",
                    HitCount = 2
                },
                new()
                {
                    SymbolId = "symbol::bar",
                    CodeId = "elf:abcd",
                    LoaderBase = "0x4000",
                    HitCount = 1
                }
            },
            Metadata = new Dictionary<string, string?> { ["source"] = "zastava" }
        };

        var response = await sut.IngestAsync(request, CancellationToken.None);

        response.SubjectKey.Should().Be("scan-1");
        response.CallgraphId.Should().Be("cg-123");
        response.RuntimeFactCount.Should().Be(2);
        response.TotalHitCount.Should().Be(6);
        response.StoredAt.Should().Be(timeProvider.GetUtcNow());

        repository.LastUpsert.Should().NotBeNull();
        repository.LastUpsert!.RuntimeFacts.Should().NotBeNull();
        repository.LastUpsert!.RuntimeFacts.Should().HaveCount(2);
        repository.LastUpsert!.RuntimeFacts![0].SymbolId.Should().Be("symbol::bar");
        repository.LastUpsert!.RuntimeFacts![0].HitCount.Should().Be(1);
        repository.LastUpsert!.RuntimeFacts![1].SymbolId.Should().Be("symbol::foo");
        repository.LastUpsert!.RuntimeFacts![1].ProcessId.Should().Be(100);
        repository.LastUpsert!.RuntimeFacts![1].ContainerId.Should().Be("ctr-1");
        repository.LastUpsert!.RuntimeFacts![1].HitCount.Should().Be(5);
        repository.LastUpsert!.Metadata.Should().ContainKey("source");
    }

    [Fact]
    public async Task IngestAsync_MergesExistingDocument()
    {
        var existing = new ReachabilityFactDocument
        {
            Id = "507f1f77bcf86cd799439011",
            Subject = new ReachabilitySubject { ImageDigest = "sha256:abc" },
            SubjectKey = "sha256:abc",
            CallgraphId = "cg-old",
            RuntimeFacts = new List<RuntimeFactDocument>
            {
                new() { SymbolId = "old::symbol", HitCount = 1, Metadata = new Dictionary<string, string?> { ["thread"] = "bg" } }
            }
        };

        repository.LastUpsert = existing;

        var request = new RuntimeFactsIngestRequest
        {
            Subject = new ReachabilitySubject { ImageDigest = "sha256:abc" },
            CallgraphId = "cg-new",
            Events = new List<RuntimeFactEvent>
            {
                new() { SymbolId = "new::symbol", HitCount = 2, ProcessName = "svc" },
                new() { SymbolId = "old::symbol", HitCount = 3, ProcessId = 200, Metadata = new Dictionary<string, string?> { ["thread"] = "main" } }
            }
        };

        var response = await sut.IngestAsync(request, CancellationToken.None);

        response.FactId.Should().Be(existing.Id);
        repository.LastUpsert!.RuntimeFacts.Should().HaveCount(2);
        repository.LastUpsert!.RuntimeFacts![0].SymbolId.Should().Be("new::symbol");
        repository.LastUpsert!.RuntimeFacts![1].SymbolId.Should().Be("old::symbol");
        repository.LastUpsert!.RuntimeFacts![1].HitCount.Should().Be(4);
        repository.LastUpsert!.RuntimeFacts![1].ProcessId.Should().Be(200);
        repository.LastUpsert!.RuntimeFacts![1].Metadata.Should().ContainKey("thread").WhoseValue.Should().Be("main");
    }

    [Theory]
    [InlineData(null)]
    [InlineData("")]
    public async Task IngestAsync_ValidatesCallgraphId(string? callgraphId)
    {
        var request = new RuntimeFactsIngestRequest
        {
            Subject = new ReachabilitySubject { ScanId = "scan" },
            CallgraphId = callgraphId ?? string.Empty,
            Events = new List<RuntimeFactEvent> { new() { SymbolId = "foo" } }
        };

        await Assert.ThrowsAsync<RuntimeFactsValidationException>(() => sut.IngestAsync(request, CancellationToken.None));
    }

    private sealed class FakeReachabilityFactRepository : IReachabilityFactRepository
    {
        public ReachabilityFactDocument? LastUpsert { get; set; }

        public Task<ReachabilityFactDocument> UpsertAsync(ReachabilityFactDocument document, CancellationToken cancellationToken)
        {
            LastUpsert = document;
            return Task.FromResult(document);
        }

        public Task<ReachabilityFactDocument?> GetBySubjectAsync(string subjectKey, CancellationToken cancellationToken) =>
            Task.FromResult(LastUpsert is { SubjectKey: not null } doc && doc.SubjectKey == subjectKey ? doc : null);
    }

    private sealed class FakeReachabilityCache : IReachabilityCache
    {
        private readonly Dictionary<string, ReachabilityFactDocument> storage = new(StringComparer.Ordinal);

        public Task<ReachabilityFactDocument?> GetAsync(string subjectKey, CancellationToken cancellationToken)
        {
            storage.TryGetValue(subjectKey, out var document);
            return Task.FromResult(document);
        }

        public Task SetAsync(ReachabilityFactDocument document, CancellationToken cancellationToken)
        {
            storage[document.SubjectKey] = document;
            return Task.CompletedTask;
        }

        public Task InvalidateAsync(string subjectKey, CancellationToken cancellationToken)
        {
            storage.Remove(subjectKey);
            return Task.CompletedTask;
        }
    }

    private sealed class FakeEventsPublisher : IEventsPublisher
    {
        public List<ReachabilityFactDocument> Published { get; } = new();

        public Task PublishFactUpdatedAsync(ReachabilityFactDocument fact, CancellationToken cancellationToken)
        {
            Published.Add(fact);
            return Task.CompletedTask;
        }
    }

    private sealed class FakeScoringService : IReachabilityScoringService
    {
        public List<ReachabilityRecomputeRequest> Requests { get; } = new();

        public Task<ReachabilityFactDocument> RecomputeAsync(ReachabilityRecomputeRequest request, CancellationToken cancellationToken)
        {
            Requests.Add(request);
            return Task.FromResult(new ReachabilityFactDocument
            {
                Subject = request.Subject,
                SubjectKey = request.Subject.ToSubjectKey(),
                CallgraphId = request.CallgraphId,
                ComputedAt = TimeProvider.System.GetUtcNow()
            });
        }
    }

    private sealed class FakeProvenanceNormalizer : IRuntimeFactsProvenanceNormalizer
    {
        public ProvenanceFeed NormalizeToFeed(
            IEnumerable<RuntimeFactEvent> events,
            ReachabilitySubject subject,
            string callgraphId,
            Dictionary<string, string?>? metadata,
            DateTimeOffset generatedAt) => new()
        {
            FeedId = "fixture",
            GeneratedAt = generatedAt,
            CorrelationId = callgraphId,
            Records = new List<ProvenanceRecord>()
        };

        public ContextFacts CreateContextFacts(
            IEnumerable<RuntimeFactEvent> events,
            ReachabilitySubject subject,
            string callgraphId,
            Dictionary<string, string?>? metadata,
            DateTimeOffset timestamp) => new()
        {
            Provenance = NormalizeToFeed(events, subject, callgraphId, metadata, timestamp),
            LastUpdatedAt = timestamp,
            RecordCount = events is ICollection<RuntimeFactEvent> collection ? collection.Count : 0
        };
    }
}