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.Instance); } [Fact] public async Task IngestAsync_InsertsAggregatedFacts() { var request = new RuntimeFactsIngestRequest { CallgraphId = "cg-123", Subject = new ReachabilitySubject { ScanId = "scan-1" }, Events = new List { new() { SymbolId = "symbol::foo", HitCount = 3, ProcessId = 100, ProcessName = "worker", ContainerId = "ctr-1", SocketAddress = "10.0.0.5:443", Metadata = new Dictionary { ["thread"] = "main" } }, new() { SymbolId = "symbol::foo", HitCount = 2 }, new() { SymbolId = "symbol::bar", CodeId = "elf:abcd", LoaderBase = "0x4000", HitCount = 1 } }, Metadata = new Dictionary { ["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 { new() { SymbolId = "old::symbol", HitCount = 1, Metadata = new Dictionary { ["thread"] = "bg" } } } }; repository.LastUpsert = existing; var request = new RuntimeFactsIngestRequest { Subject = new ReachabilitySubject { ImageDigest = "sha256:abc" }, CallgraphId = "cg-new", Events = new List { new() { SymbolId = "new::symbol", HitCount = 2, ProcessName = "svc" }, new() { SymbolId = "old::symbol", HitCount = 3, ProcessId = 200, Metadata = new Dictionary { ["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 { new() { SymbolId = "foo" } } }; await Assert.ThrowsAsync(() => sut.IngestAsync(request, CancellationToken.None)); } private sealed class FakeReachabilityFactRepository : IReachabilityFactRepository { public ReachabilityFactDocument? LastUpsert { get; set; } public Task UpsertAsync(ReachabilityFactDocument document, CancellationToken cancellationToken) { LastUpsert = document; return Task.FromResult(document); } public Task GetBySubjectAsync(string subjectKey, CancellationToken cancellationToken) => Task.FromResult(LastUpsert is { SubjectKey: not null } doc && doc.SubjectKey == subjectKey ? doc : null); public Task> GetExpiredAsync(DateTimeOffset olderThan, int limit, CancellationToken cancellationToken) => Task.FromResult>(Array.Empty()); public Task DeleteAsync(string subjectKey, CancellationToken cancellationToken) => Task.FromResult(true); public Task GetRuntimeFactsCountAsync(string subjectKey, CancellationToken cancellationToken) => Task.FromResult(0); public Task TrimRuntimeFactsAsync(string subjectKey, int maxCount, CancellationToken cancellationToken) => Task.CompletedTask; } private sealed class FakeReachabilityCache : IReachabilityCache { private readonly Dictionary storage = new(StringComparer.Ordinal); public Task 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 Published { get; } = new(); public Task PublishFactUpdatedAsync(ReachabilityFactDocument fact, CancellationToken cancellationToken) { Published.Add(fact); return Task.CompletedTask; } } private sealed class FakeScoringService : IReachabilityScoringService { public List Requests { get; } = new(); public Task 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 events, ReachabilitySubject subject, string callgraphId, Dictionary? metadata, DateTimeOffset generatedAt) => new() { FeedId = "fixture", GeneratedAt = generatedAt, CorrelationId = callgraphId, Records = new List() }; public ContextFacts CreateContextFacts( IEnumerable events, ReachabilitySubject subject, string callgraphId, Dictionary? metadata, DateTimeOffset timestamp) => new() { Provenance = NormalizeToFeed(events, subject, callgraphId, metadata, timestamp), LastUpdatedAt = timestamp, RecordCount = events is ICollection collection ? collection.Count : 0 }; } }